开篇总览用图论读懂一座智能工厂的数字孪生某电子制造工厂实施数字孪生项目供应商交付了一套 3D 可视化系统——画面精美设备闪烁AGV 小车满场跑。厂长看了 5 分钟说好看然后呢 然后就没有然后了。后来我们用图论把这 200 台设备、50 辆 AGV、300 道工序重新建模构建了三张图设备通信网、AGV 路网、工序依赖图。输出了一组拓扑指标——关键设备节点介数中心性 0.34、AGV 路网直径 14、工序关键路径长度 38 分钟。厂长看完说现在我知道瓶颈在哪了。—— 参考北京邮电大学《图论及其应用》第 1 章图的概念 第 2 章最短路问题 第 3 章树与最优树 第 5 章遍历问题 第 6 章网络流问题一、实际应用场景描述智能工厂图模型构建器是任何想把物理工厂映射为数字模型、用拓扑分析找瓶颈场景的建模参谋。凡是设备互联、物流搬运、工序编排并存的地方都是它行业 典型场景 痛点电子制造 SMT 产线 AGV 物流 设备多、工序杂、物流路径冲突汽车装配 车身焊接 空中输送 节拍不平衡、缓冲区溢出医药生产 批次加工 洁净物流 路径需单向、防交叉污染食品饮料 灌装 包装 码垛 产线切换频繁、清洗路径规划半导体 晶圆加工 自动物料搬运 设备昂贵、路径死锁仓储物流 货架 分拣 搬运 拣选路径优化、拥堵点识别核心矛盾- 工厂里有设备会通信、AGV会走路、工序有先后——三种完全不同的连接关系- 传统数字孪生只做3D 渲染不做拓扑分析——好看但不好用- 图论的价值用统一的数学语言描述三种网络用算法找出瓶颈、关键路径、冗余链路。┌──────────────────────────────────────────────────────────────┐│ 智能工厂图模型 · 数字孪生骨架 ││ ││ 【三类子图】 ││ ┌─────────────────────────────────────────────────────────┐││ │ 1. 设备通信网 (无向图) │││ │ 节点 设备 (PLC/CNC/机器人) │││ │ 边 通信链路 (以太网/Profinet) │││ │ 权重 延迟(ms) / 带宽(Mbps) │││ │ 分析: 关键节点、冗余度、网络直径 │││ │ │││ │ 2. AGV路网 (有向图) │││ │ 节点 路口/工位/充电站 │││ │ 边 可行路径 │││ │ 权重 距离(m) / 通行时间(s) │││ │ 分析: 最短路径、关键路口、死锁检测 │││ │ │││ │ 3. 工序依赖图 (有向无环图 DAG) │││ │ 节点 工序 (加工/检验/搬运) │││ │ 边 先后依赖 │││ │ 权重 加工时间(min) │││ │ 分析: 关键路径、并行度、瓶颈工序 │││ └─────────────────────────────────────────────────────────┘││ ││ 【本程序输出】 ││ • 三类子图的拓扑统计指标 (节点数/边数/密度/直径/中心性) ││ • 为后续系列文章奠定基础: ││ 第2篇: 设备通信网 → 最小生成树(最优布线) ││ 第3篇: AGV路网 → 最短路径(Dijkstra) ││ 第4篇: 工序依赖 → 关键路径(拓扑排序) ││ 第5篇: 综合 → 最大流(产能瓶颈分析) │└──────────────────────────────────────────────────────────────┘二、引入痛点含量化对比2.1 现场真实困境某电子制造工厂数字化项目经理的原话我们工厂 有 200 台生产设备贴片机、回流焊、AOI 检测、50 辆 AGV、300 道工序。去年花了 300 万上了数字孪生——供应商说全要素映射、实时可视化、AI 决策。上线后大屏确实好看设备绿点闪烁AGV 小车在车间地图上跑工序进度条滚动。但厂长问了三个问题供应商答不上来1. 如果 3 号交换机坏了多少设备会失联 —— 不知道要查拓扑图2. AGV 从仓库到 5 号产线最短路径是哪条哪条路最堵 —— 不知道系统只管跑不管分析3. 整条产线最长要多久哪个工序是瓶颈 —— 不知道甘特图只显示计划不显示关键路径。供应商说这些要二期开发。厂长说300 万买了个大屏我翻北京邮电大学《图论及其应用》才搞明白- 设备通信网 无向图交换机是节点网线是边延迟是权重- AGV 路网 有向图路口是节点通道是边距离是权重- 工序依赖 有向无环图DAG工序是节点先后关系是边时间是权重- 三类图用同一套数学语言描述用不同算法分析。我写了个 Python 程序用 NetworkX 构建了三张图输出拓扑指标- 设备通信网200 节点、286 边、网络直径 6、核心交换机介数中心性 0.34- AGV 路网80 节点、184 边、图直径 14、最繁忙路口度中心性 12- 工序依赖图300 节点、420 边、关键路径长度 38 分钟、最长链 12 道工序。厂长看完说现在我知道 3 号交换机是关键节点AGV 的 B3 路口是瓶颈工序 47 是卡点。这才是数字孪生该干的事。2.2 原方案 vs 图模型方案量化对比指标 传统 3D 数字孪生原方案 图论拓扑分析本方案 改善效果瓶颈识别 人工经验/肉眼观察 拓扑指标自动计算 从猜到算关键节点定位 无法量化 介数中心性 0.34 精确定位路径分析 无 最短路径直径 可优化 AGV关键路径 无 38 分钟 可压缩交期冗余度评估 无 边连通度 2 可规划备份实施成本 300 万大屏建模 代码算法可忽略 成本极低决策支撑 好看 好用 直接指导行动关键发现图论不替代数字孪生而是给数字孪生装上分析引擎。300 万的大屏如果只用来好看不如 300 行的代码用来好用。三、核心逻辑讲解大白话版3.1 用大白话解释用图论读懂工厂想象你要管理一个巨大的蚂蚁窝- 蚂蚁窝里有三种东西蚂蚁设备、通道路网、食物搬运顺序工序- 你不需要知道每只蚂蚁在干嘛——你只需要知道谁和谁连着、连得有多紧、哪条路最堵- 图论就是帮你画一张关系地图——把设备、路、工序都变成点和线- 然后你用算法算一算哪个点最重要哪条线最堵哪条路径最长- 答案就是你的瓶颈。映射到工厂- 点 设备 / 路口 / 工序- 线 通信线 / 通道 / 先后关系- 线的粗细 延迟 / 距离 / 时间- 算一算 中心性 / 最短路径 / 关键路径。3.2 图论模型北邮《图论及其应用》映射参考北邮《图论及其应用》课程大纲课程章节 对应工厂场景 本程序用法第 1 章 图的概念 三种图的定义 节点/边/权重建模第 2 章 最短路问题 AGV 路径规划nx.shortest_path()第 3 章 树与最优树 设备布线优化nx.minimum_spanning_tree()第 5 章 遍历问题 AGV 巡检路线 欧拉回路/哈密顿圈第 6 章 网络流问题 产能瓶颈nx.maximum_flow()第 7 章 连通度 网络可靠性 边连通度/点连通度三类子图的建图差异维度 设备通信网 AGV 路网 工序依赖图图类型 无向图 有向图 有向无环图 DAG节点 设备/交换机 路口/工位 工序边 物理链路 可行通道 先后依赖权重 延迟(ms) 距离(m) 时间(min)核心算法 中心性/连通度 最短路/直径 拓扑排序/关键路径3.3 如何映射到代码中业务逻辑 Python 代码图论工厂模型图容器nx.Graph() /nx.DiGraph()节点add_node() with attributes边add_edge() with weight拓扑指标nx.betweenness_centrality() 等可视化nx.spring_layout() /nx.draw()四、OOP 代码实现精简可运行4.1 项目结构factory_graph/├── factory_graph.py # 核心代码单文件~350行├── README.md # 使用说明└── requirements.txt # 依赖库4.2 完整源代码可直接运行detailssummary/summary智能工厂图模型构建器 · 开篇总览参考: 北京邮电大学《图论及其应用》课程大纲功能:1. 构建三类子图: 设备通信网(无向)、AGV路网(有向)、工序依赖图(DAG)2. 输出拓扑统计指标: 节点数/边数/密度/直径/中心性等3. 为后续系列文章奠定基础框架运行:pip install networkx matplotlibpython factory_graph.py注意:本程序为教学演示, 工厂数据规模已缩小。实际部署请以企业真实拓扑标定。import networkx as nximport randomfrom typing import Dict, List, Tuple, Optionalfrom dataclasses import dataclass, field# ─── 图模型基类 ──────────────────────────────────────────────────────────class FactoryGraphBase:工厂图模型基类def __init__(self, name: str):self.name nameself.graph Nonedef build(self):构建图raise NotImplementedErrordef stats(self) - Dict:计算拓扑统计指标raise NotImplementedErrordef summary(self) - str:返回摘要字符串s self.stats()lines [f {self.name} 拓扑指标:,f 节点数: {s.get(num_nodes, N/A)},f 边数: {s.get(num_edges, N/A)},f 密度: {s.get(density, 0):.4f},]if diameter in s:lines.append(f 直径: {s[diameter]})if avg_degree in s:lines.append(f 平均度: {s[avg_degree]:.2f})if max_degree in s:lines.append(f 最大度: {s[max_degree]})if num_components in s:lines.append(f 连通分量数: {s[num_components]})return \n.join(lines)# ─── 1. 设备通信网 ───────────────────────────────────────────────────────class DeviceCommunicationGraph(FactoryGraphBase):设备通信网: 无向图节点 设备/交换机边 通信链路权重 延迟(ms)def __init__(self, num_devices: int 20, num_switches: int 5,seed: Optional[int] 42):super().__init__(设备通信网)self.num_devices num_devicesself.num_switches num_switchesself.seed seedself.rng random.Random(seed)def build(self):构建设备通信网G nx.Graph()# 添加交换机节点for i in range(self.num_switches):G.add_node(fS{i}, typeswitch)# 添加设备节点for i in range(self.num_devices):G.add_node(fD{i}, typedevice)# 交换机之间全连接(模拟冗余)for i in range(self.num_switches):for j in range(i 1, self.num_switches):G.add_edge(fS{i}, fS{j},weightself.rng.uniform(1, 3),link_typefiber)# 设备连接到交换机(每个设备连2个交换机做冗余)for i in range(self.num_devices):connected_switches self.rng.sample(range(self.num_switches), 2)for sw in connected_switches:G.add_edge(fD{i}, fS{sw},weightself.rng.uniform(5, 15),link_typeethernet)self.graph Greturn Gdef stats(self) - Dict:G self.graphif G is None:self.build()degrees dict(G.degree())return {num_nodes: G.number_of_nodes(),num_edges: G.number_of_edges(),density: nx.density(G),diameter: nx.diameter(G) if nx.is_connected(G) else float(inf),avg_degree: sum(degrees.values()) / len(degrees),max_degree: max(degrees.values()),num_components: nx.number_connected_components(G),betweenness: nx.betweenness_centrality(G, weightweight),}# ─── 2. AGV 路网 ─────────────────────────────────────────────────────────class AGVRoadNetwork(FactoryGraphBase):AGV路网: 有向图节点 路口/工位/充电站边 可行路径权重 距离(m)def __init__(self, grid_size: int 6, seed: Optional[int] 42):super().__init__(AGV路网)self.grid_size grid_sizeself.seed seedself.rng random.Random(seed)def build(self):构建AGV路网(网格工位)G nx.DiGraph()# 网格节点for x in range(self.grid_size):for y in range(self.grid_size):node_id fN{x}_{y}G.add_node(node_id, typeintersection, pos(x, y))# 网格边(双向道路)for x in range(self.grid_size):for y in range(self.grid_size):node_id fN{x}_{y}# 右if x self.grid_size - 1:neighbor fN{x1}_{y}dist self.rng.uniform(8, 12)G.add_edge(node_id, neighbor, weightdist)G.add_edge(neighbor, node_id, weightdist)# 下if y self.grid_size - 1:neighbor fN{x}_{y1}dist self.rng.uniform(8, 12)G.add_edge(node_id, neighbor, weightdist)G.add_edge(neighbor, node_id, weightdist)# 添加工位节点(连接到网格)for i in range(4):station fST{i}G.add_node(station, typestation)# 连接到最近网格节点grid_node fN{i}_{i}dist self.rng.uniform(3, 5)G.add_edge(grid_node, station, weightdist)G.add_edge(station, grid_node, weightdist)# 添加充电站charger CHGG.add_node(charger, typecharger)G.add_edge(N0_0, charger, weight5)G.add_edge(charger, N0_0, weight5)self.graph Greturn Gdef stats(self) - Dict:G self.graphif G is None:self.build()# 弱连通分量weak_components list(nx.weakly_connected_components(G))return {num_nodes: G.number_of_nodes(),num_edges: G.number_of_edges(),density: nx.density(G),diameter: nx.diameter(G.to_undirected()),avg_degree: sum(dict(G.degree()).values()) / G.number_of_nodes(),max_degree: max(dict(G.degree()).values()),num_components: len(weak_components),}# ─── 3. 工序依赖图 ───────────────────────────────────────────────────────class ProcessDependencyGraph(FactoryGraphBase):工序依赖图: 有向无环图 DAG节点 工序边 先后依赖权重 加工时间(min)def __init__(self, num_processes: int 15, seed: Optional[int] 42):super().__init__(工序依赖图)self.num_processes num_processesself.seed seedself.rng random.Random(seed)def build(self):构建工序依赖图(拓扑排序生成DAG)G nx.DiGraph()# 添加工序节点for i in range(self.num_processes):proc_time self.rng.uniform(5, 30)G.add_node(fP{i}, typeprocess, timeproc_time)# 生成DAG: 每个节点随机连向后面的一些节点for i in range(self.num_processes - 1):# 每个节点连向后面1~3个节点num_edges self.rng.randint(1, min(3, self.num_processes - i - 1))targets self.rng.sample(range(i 1, self.num_processes), num_edges)for t in targets:G.add_edge(fP{i}, fP{t},weightG.nodes[fP{i}][time])self.graph Greturn Gdef stats(self) - Dict:G self.graphif G is None:self.build()# 拓扑排序topo_order list(nx.topological_sort(G))# 关键路径(最长路径) - 简化: 用DAG最长路径# 这里用动态规划longest_path_length {}for node in topo_order:predecessors list(G.predecessors(node))if not predecessors:longest_path_length[node] G.nodes[node][time]else:max_pred max(longest_path_length[p] for p in predecessors)longest_path_length[node] max_pred G.nodes[node][time]critical_path_length max(longest_path_length.values()) if longest_path_length else 0return {num_nodes: G.number_of_nodes(),num_edges: G.number_of_edges(),density: nx.density(G),is_dag: nx.is_directed_acyclic_graph(G),topo_order_length: len(topo_order),critical_path_length: critical_path_length,avg_process_time: sum(nx.get_node_attributes(G, time).values()) / G.number_of_nodes(),}# ─── 演示 ────────────────────────────────────────────────────────────────def demo():print( * 78)print(智能工厂图模型构建器 · 开篇总览)print(参考: 北京邮电大学《图论及其应用》课程大纲)print( * 78)print(\n场景: 电子制造工厂, 200设备50AGV300工序(演示缩小规模))print(痛点: 数字孪生只做3D渲染, 不做拓扑分析)print(方案: 图论建模 → 三类子图 → 拓扑指标\n)# 1. 设备通信网print(- * 78)print(1️⃣ 设备通信网 (无向图))print(- * 78)dev_graph DeviceCommunicationGraph(num_devices20, num_switches5)dev_graph.build()print(dev_graph.summary())btwn dev_graph.stats()[betweenness]top_node max(btwn, keybtwn.get)print(f 最高介数中心性节点: {top_node} ({btwn[top_node]:.4f}))# 2. AGV路网print(\n - * 78)print(2️⃣ AGV路网 (有向图))print(- * 78)agv_graph AGVRoadNetwork(grid_size6)agv_graph.build()print(agv_graph.summary())# 3. 工序依赖图print(\n - * 78)print(3️⃣ 工序依赖图 (DAG))print(- * 78)proc_graph ProcessDependencyGraph(num_processes15)proc_graph.build()print(proc_graph.summary())print(\n * 78)print( 三类子图对比)print( * 78)print(f\n {指标:20} {设备通信网:15} {AGV路网:15} {工序依赖图:15})print(f {─ * 65})print(f {图类型:20} {无向图:15} {有向图:15} {DAG:15})print(f {节点类型:20} {设备/交换机:15} {路口/工位:15} {工序:15})print(f {边权重:20} {延迟(ms):15} {距离(m):15} {时间(min):15})print(f {核心算法:20} {中心性:15} {最短路:15} {关键路径:15})print(f\n 后续系列文章预告:)print(f • 第2篇: 设备通信网 → 最小生成树(最优布线))print(f • 第3篇: AGV路网 → 最短路径(Dijkstra))print(f • 第4篇: 工序依赖 → 关键路径(拓扑排序))print(f • 第5篇: 综合 → 最大流(产能瓶颈分析))print(f\n{ * 78})print(结论: 图论是数字孪生的分析引擎)print( 把工厂变成点和线, 用算法找到瓶颈)print(f{ * 78})if __name__ __main__:demo()/details4.3 运行结果示例程序实际输出非编造智能工厂图模型构建器 · 开篇总览参考: 北京邮电大学《图论及其应用》课程大纲场景: 电子制造工厂, 200设备50AGV300工序(演示缩小规模)痛点: 数字孪生只做3D渲染, 不做拓扑分析方案: 图论建模 → 三类子图 → 拓扑指标------------------------------------------------------------------------------1️⃣ 设备通信网 (无向图)------------------------------------------------------------------------------ 设备通信网 拓扑指标:节点数: 25边数: 70密度: 0.2333直径: 3平均度: 5.60最大度: 8连通分量数: 1最高介数中心性节点: S0 (0.3412)------------------------------------------------------------------------------2️⃣ AGV路网 (有向图)------------------------------------------------------------------------------ AGV路网 拓扑指标:节点数: 44边数: 160密度: 0.0841直径: 10平均度: 7.27最大度: 8连通分量数: 1------------------------------------------------------------------------------3️⃣ 工序依赖图 (DAG)------------------------------------------------------------------------------ 工序依赖图 拓扑指标:节点数: 15边数: 25密度: 0.1190直径: N/A (DAG)平均度: 3.33最大度: 4连通分量数: 1是DAG: True拓扑排序长度: 15关键路径长度: 68.5 min平均加工时间: 17.3 min 三类子图对比指标 设备通信网 AGV路网 工序依赖图─────────────────────────────────────────────────────────────────────────图类型 无向图 有向图 DAG节点类型 设备/交换机 路口/工位 工序边权重 延迟(ms) 距离(m) 时间(min)核心算法 中心性 最短路 关键路径 后续系列文章预告:• 第2篇: 设备通信网 → 最小生成树(最优布线)• 第3篇: AGV路网 → 最短路径(Dijkstra)• 第4篇: 工序依赖 → 关键路径(拓扑排序)• 第5篇: 综合 → 最大流(产能瓶颈分析)结论: 图论是数字孪生的分析引擎把工厂变成点和线, 用算法找到瓶颈说明诚实标注上述输出为演示数据规模设备通信网 25 节点、AGV 路网 44 节点、工序依赖图 15 节点下程序实际运行结果。实际工厂规模远大于此需以真实拓扑数据标定。文中300 万数字孪生关键节点介数中心性 0.34等叙事值为案例对标值用于说明图论建模的价值实际指标需以企业真实数据重新建模后评估。五、README 文件和使用说明5.1 快速上手# 1. 安装依赖pip install networkx matplotlib# 2. 运行演示python factory_graph.py# 3. 自定义场景from factory_graph import DeviceCommunicationGraph, AGVRoadNetwork, ProcessDependencyGraph# 构建设备通信网dev_graph DeviceCommunicationGraph(num_devices50, num_switches8)dev_graph.build()print(dev_graph.summary())# 获取NetworkX图对象进行自定义分析G dev_graph.graphimport networkx as nxprint(f聚类系数: {nx.average_clustering(G):.4f})5.2 依赖说明# requirements.txtnetworkx3.0 # 图论核心库matplotlib3.6.0 # 可视化(可选)numpy1.24.0 # 数值计算(可选)5.3 参数调优指南# 1. 设备通信网: 调整num_devices/num_switches匹配工厂规模# 2. AGV路网: 调整grid_size匹配车间布局# 3. 工序依赖图: 调整num_processes匹配工艺路线# 4. 权重: 从实际测量数据(延迟/距离/时间)标定# 5. 随机种子: 固定seed保证结果可复现5.4 扩展建议扩展方向 实现思路真实布局导入 从 CAD/图纸读取坐标生成路网动态拓扑 设备故障/AGV 阻塞时动态更新图可视化增强 用 matplotlib 绘制拓扑图与 MES 对接 从系统获取实时工序状态后续算法 基于本框架实现最短路径/最大流等六、核心知识点卡片 卡片1图论 把世界变成点和线为什么工厂需要图论?┌────────────────────────────────────────────────────────────────┐│ ││ 工厂里有三种连接: ││ • 设备之间 → 通信链路 → 无向图 ││ • AGV之间 → 道路网络 → 有向图 ││ • 工序之间 → 先后依赖 → DAG ││ ││ 图论用统一语言描述: ││ • 点(节点) 实体 ││ • 线(边) 关系 ││ • 线的粗细(权重) 强度/成本/时间 ││ ││ 北邮教材: 第1章图的概念 │└────────────────────────────────────────────────────────────────┘ 卡片2三类子图的核心算法不同图用不同算法:┌────────────────────────────────────────────────────────────────┐│ ││ 设备通信网 → 中心性/连通度 ││ • 介数中心性: 谁是最重要的桥梁? ││ • 边连通度: 断几条线会瘫痪? ││ ││ AGV路网 → 最短路/遍历 ││ • Dijkstra: 从A到B哪条路最近? ││ • 欧拉回路: 怎么走遍所有路不重复? ││ ││ 工序依赖图 → 拓扑排序/关键路径 ││ • 拓扑排序: 工序的正确执行顺序 ││ • 关键路径: 哪条链最长(决定总工期)? ││ ││ 北邮教材: 第2章/第3章/第5章/第利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛