流程监控与关键路径分析

1. 简介

在快节奏的制造业中，监控生产流程对于确保效率、质量控制和按时交付至关重要。本用例展示了如何使用 Neo4j 来监控制造工作流、可视化任务依赖关系，并执行关键路径分析 (CPA) 以优化运营并减少延误。通过利用图数据库，制造商可以全面了解生产线、预测任务持续时间并识别瓶颈，从而提高产量和资源利用率。

2. 应用场景

为了理解制造流程监控和关键路径分析的价值，请考虑生产环境中存在的现实挑战，即效率低下可能导致代价高昂的停机和错过截止日期。以下三个关键领域突显了这些问题

工作流依赖管理
- 复杂的装配线涉及相互依存的任务，如果管理不当，可能会导致连锁延误。
- 如果没有对依赖关系的清晰可见性，意外的瓶颈会破坏准时制生产。
- 忽视任务之间的相互联系会导致跨机器的资源分配效率低下。
资源与队列优化
- 机器通常处理排队任务，但过载或调度不当会导致闲置时间或积压。
- 传统系统无法动态评估工作负载，使得预测完成时间变得复杂。
- 监控不足会增加设备故障或质量问题未被发现的风险。
风险缓解与合规性
- 法规要求生产流程必须可追溯，以符合质量和安全标准。
- 手动跟踪容易出错，难以证明合规性或进行效率优化。
- 如果没有工具来主动识别关键路径和延误，制造商将面临罚款和声誉受损的风险。这些场景强调了使用 Neo4j 制造流程监控和 Cypher 这种先进解决方案的必要性，它利用图技术对工作流进行建模、分析和可视化，为生产规划和优化中的业务及技术用户提供关键洞察。

3. 解决方案

像 Neo4j 这样的先进图数据库对于处理制造业中互联生产数据的复杂性至关重要。它们擅长管理动态关系，使得对任务依赖关系、队列和机器工作负载进行建模变得直观。通过将数据表示为图，组织可以发现关键路径、模拟场景并获得可操作的洞察，从而增强决策能力、运营效率和生产韧性。

3.1. 图数据库如何提供帮助？

图数据库为制造流程监控和关键路径分析的挑战提供了强大的解决方案。以下是图数据库不可或缺的五个关键原因

依赖建模： 图数据库可以自然地处理复杂的任务互联和机器分配，捕获关系型数据库无法有效表示的关系。
实时队列与工作负载分析： 它们能够实现机器队列和待办工作的动态视图，从而能够即时识别瓶颈。
全面的流程可视化： 图数据库提供生产工作流的完整概览，揭示隐藏的效率低下和风险。
关键路径计算： 通过路径聚合等功能，图数据库支持计算预计到达时间 (ETA) 和关键路径，以便进行主动调整。
可扩展的优化： 与图数据科学 (GDS) 的集成允许在大规模下进行高级分析，如最长路径算法。这些能力使图数据库成为获得洞察和解决制造流程监控中多方面问题的核心。

4. 建模

本节演示示例图上的 Cypher 查询。目标是展示查询结构并指导生产中的数据建模。我们将使用一个小型图，包含几个节点，基于以下数据模型

4.1. 数据模型

图 1. 流程监控数据模型

4.1.1 所需数据字段

以下是开始所需的字段

Machine (机器) 节点
- processor_id: 唯一标识符（例如 "M1"）
- name: 机器名称（例如 "AssemblyMachine1"）
- load: 当前负载水平
Process (流程) 节点
- process_id: 唯一标识符（例如 "Prod1"）
- name: 生产流程名称（例如 "WidgetProduction_Q1"）
Job (工作/任务) 节点
- job_id: 唯一标识符（例如 "T0"）
- name: 任务名称（例如 "Shared_MaterialPrep"）
- status: 当前状态（例如 "Completed", "Running", "Pending"）
- duration: 预计或实际持续时间
- quality_score: 质量或风险评分（改编自 risk_score）
- completion_progress: 完成进度百分比（0.0 到 1.0）
DEPENDS_ON 关系：任务依赖关系
WAITS 关系：机器上的队列顺序
RUNS_ON 关系：任务分配给机器
IS_INSTANCE_OF 关系：任务与流程的关联
QUEUE_HEAD 和 QUEUE_TAIL 关系：机器队列边界

对于用于可扩展 CPA 的重构模型：* 附加节点：- :Start（每个任务，带有 job_id）- :End（每个任务，带有 job_id）- :KickOff（代表整个流程开始的单一节点）* 关系：- [:STARTS]（从 :Job 到 :Start）- [:ENDS]（从 :Job 到 :End）- [:TIME]（带有 duration 属性的加权边）：- 从 :Start 到 :End（对于每个任务，持续时间 = 任务执行时间）- 从依赖任务的 :End 到从属任务的 :Start（持续时间 = 3 秒等待时间）- 从 :KickOff 到初始任务的 :Start（持续时间 = 图中最小任务持续时间）

4.1.2 所需参数

本用例不需要显示的查询特定参数，因为它们是在 process_id 或 processor_id 等节点属性上操作的。

4.2. 演示数据

以下 Cypher 语句将在 Neo4j 数据库中创建示例图（适配制造场景）

// Machines
CREATE (m1:Machine {processor_id: 'M1', name: 'AssemblyMachine1', load: 3})
CREATE (m2:Machine {processor_id: 'M2', name: 'AssemblyMachine2', load: 2})
CREATE (m3:Machine {processor_id: 'M3', name: 'AssemblyMachine3', load: 1})
// Production Processes
CREATE (prod1:Process {process_id: 'Prod1', name: 'WidgetProduction_Q1'})
CREATE (prod2:Process {process_id: 'Prod2', name: 'GadgetProduction_Q1'})
CREATE (prod3:Process {process_id: 'Prod3', name: 'ComponentProduction_Q1'})
// Shared Task (part of WidgetProduction_Q1 and GadgetProduction_Q1)
CREATE (t0:Job {job_id: 'T0', name: 'Shared_MaterialPrep', status: 'Completed', duration: 5, quality_score: 0.2, completion_progress: 1.0})
CREATE (t0)-[:RUNS_ON]->(m1)
// Tasks for WidgetProduction_Q1 (Diamond-shaped DAG)
CREATE (t1:Job {job_id: 'T1', name: 'Widget_Assembly1', status: 'Completed', duration: 15, quality_score: 0.7, completion_progress: 1.0})
CREATE (t2:Job {job_id: 'T2', name: 'Widget_Assembly2', status: 'Completed', duration: 12, quality_score: 0.6, completion_progress: 1.0})
CREATE (t3:Job {job_id: 'T3', name: 'Widget_QualityCheck', status: 'Completed', duration: 6, quality_score: 0.4, completion_progress: 1.0})
CREATE (t4:Job {job_id: 'T4', name: 'Widget_Packaging', status: 'Running', duration: 3, quality_score: 0.2, completion_progress: 0.5})
CREATE (t1)-[:RUNS_ON]->(m1), (t2)-[:RUNS_ON]->(m1), (t3)-[:RUNS_ON]->(m2), (t4)-[:RUNS_ON]->(m3)
CREATE (t1)-[:DEPENDS_ON]->(t0), (t2)-[:DEPENDS_ON]->(t0), (t3)-[:DEPENDS_ON]->(t1), (t3)-[:DEPENDS_ON]->(t2), (t4)-[:DEPENDS_ON]->(t3)
CREATE (t4)-[:IS_INSTANCE_OF]->(prod1)
// Tasks for GadgetProduction_Q1 (Parallel Paths DAG)
CREATE (t5:Job {job_id: 'T5', name: 'Gadget_Assembly1', status: 'Completed', duration: 14, quality_score: 0.6, completion_progress: 1.0})
CREATE (t6:Job {job_id: 'T6', name: 'Gadget_Assembly2', status: 'Completed', duration: 11, quality_score: 0.5, completion_progress: 1.0})
CREATE (t7:Job {job_id: 'T7', name: 'Gadget_QualityCheck1', status: 'Completed', duration: 5, quality_score: 0.3, completion_progress: 1.0})
CREATE (t8:Job {job_id: 'T8', name: 'Gadget_QualityCheck2', status: 'Completed', duration: 4, quality_score: 0.2, completion_progress: 1.0})
CREATE (t9:Job {job_id: 'T9', name: 'Gadget_Packaging', status: 'Pending', duration: 2, quality_score: 0.1, completion_progress: 0.0})
CREATE (t5)-[:RUNS_ON]->(m1), (t6)-[:RUNS_ON]->(m1), (t7)-[:RUNS_ON]->(m2), (t8)-[:RUNS_ON]->(m2), (t9)-[:RUNS_ON]->(m3)
CREATE (t5)-[:DEPENDS_ON]->(t0), (t6)-[:DEPENDS_ON]->(t0), (t7)-[:DEPENDS_ON]->(t5), (t8)-[:DEPENDS_ON]->(t6), (t9)-[:DEPENDS_ON]->(t7), (t9)-[:DEPENDS_ON]->(t8)
CREATE (t9)-[:IS_INSTANCE_OF]->(prod2)
// Shared Task (part of GadgetProduction_Q1 and ComponentProduction_Q1)
CREATE (t10:Job {job_id: 'T10', name: 'Shared_ComponentAssembly', status: 'Running', duration: 10, quality_score: 0.5, completion_progress: 0.5})
CREATE (t10)-[:RUNS_ON]->(m2)
// Tasks for ComponentProduction_Q1 (Single Chain DAG)
CREATE (t11:Job {job_id: 'T11', name: 'Component_MaterialPrep', status: 'Completed', duration: 12, quality_score: 0.5, completion_progress: 1.0})
CREATE (t12:Job {job_id: 'T12', name: 'Component_QualityCheck', status: 'Pending', duration: 5, quality_score: 0.3, completion_progress: 0.0})
CREATE (t13:Job {job_id: 'T13', name: 'Component_Inspection', status: 'Pending', duration: 6, quality_score: 0.4, completion_progress: 0.0})
CREATE (t14:Job {job_id: 'T14', name: 'Component_Packaging', status: 'Pending', duration: 4, quality_score: 0.2, completion_progress: 0.0})
CREATE (t11)-[:RUNS_ON]->(m1), (t12)-[:RUNS_ON]->(m2), (t13)-[:RUNS_ON]->(m3), (t14)-[:RUNS_ON]->(m3)
CREATE (t12)-[:DEPENDS_ON]->(t10), (t10)-[:DEPENDS_ON]->(t11), (t13)-[:DEPENDS_ON]->(t12), (t14)-[:DEPENDS_ON]->(t13)
CREATE (t14)-[:IS_INSTANCE_OF]->(prod3)
// Queue for AssemblyMachine1 (t0 -> t1 -> t5 -> t2 -> t6 -> t11)
CREATE (m1)-[:QUEUE_HEAD]->(t0)
CREATE (m1)-[:QUEUE_TAIL]->(t11)
CREATE (t1)-[:WAITS]->(t0), (t5)-[:WAITS]->(t1), (t2)-[:WAITS]->(t5), (t6)-[:WAITS]->(t2), (t11)-[:WAITS]->(t6)
// Queue for AssemblyMachine2 (t3 -> t7 -> t8 -> t10 -> t12)
CREATE (m2)-[:QUEUE_HEAD]->(t3)
CREATE (m2)-[:QUEUE_TAIL]->(t12)
CREATE (t7)-[:WAITS]->(t3), (t8)-[:WAITS]->(t7), (t10)-[:WAITS]->(t8), (t12)-[:WAITS]->(t10)
// Queue for AssemblyMachine3 (t4 -> t9 -> t13 -> t14)
CREATE (m3)-[:QUEUE_HEAD]->(t4)
CREATE (m3)-[:QUEUE_TAIL]->(t14)
CREATE (t9)-[:WAITS]->(t4), (t13)-[:WAITS]->(t9), (t14)-[:WAITS]->(t13);

5. Cypher 查询

这些 Cypher 查询兼容 Neo4j 5.9+ 版本以及 Cypher 5 或 25。

5.1. 显示图模型

此查询可视化整体架构

CALL db.schema.visualization()

图 2. 数据模型

5.2. 显示制造流程

此查询显示特定的生产流程及其依赖关系

MATCH (n:Process {process_id:"Prod1"})<-[i:IS_INSTANCE_OF]-(j:Job)
OPTIONAL MATCH path = (j)-[:DEPENDS_ON]->*()
RETURN path, n, i

5.3. 显示机器队列

此查询显示在特定机器上排队的任务队列

MATCH path = (n:Machine {processor_id: "M3"} )-[:QUEUE_HEAD]->()
(()<-[:WAITS]-())*
()<-[:QUEUE_TAIL]-(n)
RETURN path

5.4. 显示流程中仍需完成的工作

此查询识别生产流程中的待办任务

MATCH (n:Process {process_id:"Prod3"})<-[i:IS_INSTANCE_OF]-(j:Job)
OPTIONAL MATCH path = (j)-[:DEPENDS_ON|WAITS]->*(x WHERE x.status <> "Completed")
RETURN path, n, i

5.5. 流程的关键路径分析

此查询计算生产流程的关键路径和预计时间

MATCH (n:Process {process_id:"Prod3"})<-[i:IS_INSTANCE_OF]-(j:Job WHERE j.status <> "Completed")
OPTIONAL MATCH path = (j)(()-[:DEPENDS_ON|WAITS]->(jobs))*(x WHERE x.status <> "Completed")
// the *duration* property in this context means *expected_duration* because tasks are not completed yet
WITH n, i, path, reduce(duration=0, job IN [j]+jobs |
    duration + job.duration * (1.0-job.completion_progress)) AS total_duration
ORDER BY total_duration DESC LIMIT 1
RETURN n, i, path, total_duration

图 3. CPA 结果

5.6. 基于 GDS 的可扩展关键路径分析

对于更大的图，重构模型以将时间视为关系，并使用 Neo4j 的图数据科学 (GDS) 库进行最长路径计算。这种方法可扩展至数千个任务，识别关键序列以防止中断。

图 4. 可扩展 CPA 的重构数据模型

此重构灵感来自 Neo4j 博客文章在 Neo4j 中解锁 DAG：从基础到关键路径分析。

5.6.1. 为 Merge 创建索引

此查询为高效合并创建索引

CREATE INDEX start_job_id IF NOT EXISTS FOR (s:Start) ON (s.job_id);
CREATE INDEX end_job_id IF NOT EXISTS FOR (e:End) ON (e.job_id);

5.6.2. 将时间作为关系

此查询创建带有 TIME 关系的 Start 和 End 节点，用于表示任务持续时间

MATCH (j:Job)
CALL (j) {
    MERGE (s:Start {job_id: j.job_id})
    MERGE (e:End {job_id: j.job_id})
    MERGE (j)-[:STARTS]->(s)
    MERGE (j)-[:ENDS]->(e)
    MERGE (s)-[:TIME {duration: j.duration}]->(e)
} IN CONCURRENT TRANSACTIONS OF 1000 ROWS;

5.6.3. 依赖关系 3 秒等待时间

此查询为依赖项和等待项添加 3 秒持续时间的 TIME 关系

MATCH (j1)-[:DEPENDS|WAITS]->(j0)
    CALL (j0, j1) {
    MERGE (s:Start {job_id: j1.job_id})
    MERGE (e:End {job_id: j0.job_id})
    MERGE (e)-[:TIME {duration: 3}]->(s)
} IN CONCURRENT TRANSACTIONS OF 1000 ROWS;

5.6.4. Kickoff 节点

此查询创建 KickOff 节点

MERGE (:KickOff);

5.6.5. KickOff 到初始任务

此查询将 KickOff 连接到初始任务

MATCH (j:Job)
WITH j.duration AS duration
ORDER BY duration ASC LIMIT 1
MATCH (ko:KickOff)
WITH ko, duration
MATCH (j:Job)-[:STARTS]->(s)
WHERE NOT EXISTS {(j)-[:DEPENDS|WAITS]->()}
CALL (ko, s, duration) {
    MERGE (ko)-[:TIME {duration: duration}]->(s)
} IN TRANSACTIONS OF 1000 ROWS;

5.6.6. 投影内存图

此查询为 GDS 投影图

MATCH (source:Start|KickOff|End)
OPTIONAL MATCH (source)-[r:TIME]->(target)
RETURN gds.graph.project("g", source, target, {relationshipProperties: r {.duration}})

5.6.7. 流式传输关键路径

此查询流式传输最长路径

CALL gds.dag.longestPath.stream("g", {relationshipWeightProperty: "duration"})
YIELD targetNode as target, totalCost, path, costs
WITH target AS last_activity, totalCost, path, costs
ORDER BY totalCost DESC
WITH last_activity, collect ({totalCost:totalCost, path:path, costs:costs})[0] AS longest
RETURN last_activity, longest.totalCost AS critical_time, longest.path AS path, longest.costs AS costs

图 5. 到达每个任务时间边界的关键路径

5.6.8. 流式传输特定任务的关键时间

此查询流式传输特定任务的关键时间（需要 $job_id_list 参数）

:params {
  job_id_list: ["T11", "T12", "T14"]
}

CALL gds.dag.longestPath.stream("g", {relationshipWeightProperty: "duration"})
YIELD targetNode as target, totalCost, path, costs
WITH gds.util.asNode(target).job_id AS last_activity, totalCost, path, costs
ORDER BY totalCost DESC
WITH last_activity, collect ({totalCost:totalCost, path:path, costs:costs})[0] AS longest
WHERE last_activity IN $job_id_list
WITH last_activity, longest.totalCost AS critical_time, longest.path AS path, longest.costs AS costs
ORDER BY size(last_activity)
RETURN last_activity, critical_time

这些参数返回的结果为

[
    {
        "last_activity": "T11",
        "critical_time": 86.0
    },
    {
        "last_activity": "T12",
        "critical_time": 44.0
    },
    {
        "last_activity": "T14",
        "critical_time": 26.0
    }
]