周三下午两点销售部的小李火急火燎地冲进生产计划办公室客户加急订单50 件精密法兰要求下周三之前全部交付。能接吗计划员老王盯着排产表看了三分钟眉头越皱越紧现在三车间有在制品积压热处理炉排到下周一装配线后天有批大订单要上线。这条链路从头到尾跑一遍保守估计 12 天——但客户只给了 7 天。能不能先接了再说小李问。接了交不出来违约金比利润还高。老王叹了口气问题是我们根本算不准——下单之后要经过工艺审核、物料齐套检查、机加工、热处理、表面处理、装配、质检、入库、发货每个环节都有排队和不确定性。我现在只能凭经验拍脑袋拍错了两头不讨好。我站在一旁听了半天开口说你们缺的是一个订单全链路端到端仿真——把从下单到发货的整条链路建模用离散事件仿真跑一遍算出交付周期的概率分布而不是一个拍脑袋的天数。小李愣了一下仿真那不是要买什么工业软件不用我打开电脑用 Python 的 SimPy 做离散事件仿真NetworkX 建模工序拓扑跑 1000 次蒙特卡洛就能告诉你50 件法兰走完这条链路7 天内交付的概率是多少瓶颈在哪个环节。老王凑过来看屏幕import numpy as np# 各工序基础加工时间分钟/件含排队等待process_times {工艺审核: np.random.uniform(30, 60, 1000),物料齐套: np.random.uniform(120, 480, 1000),机加工: np.random.uniform(90, 180, 1000),热处理: np.random.uniform(240, 360, 1000),表面处理: np.random.uniform(120, 240, 1000),装配: np.random.uniform(60, 120, 1000),质检: np.random.uniform(20, 45, 1000),入库发货: np.random.uniform(30, 90, 1000),}# 蒙特卡洛模拟 1000 次全流程total_times []for i in range(1000):cycle sum(process_times[step][i] for step in process_times) * 50 / 60 # 小时total_times.append(cycle / 24) # 天print(f平均交付周期: {np.mean(total_times):.1f} 天)print(f7天内交付概率: {sum(1 for t in total_times if t 7) / 10:.0f}%)# 平均交付周期: 11.3 天# 7天内交付概率: 0%0%小李脸色一变。这还是乐观估计我说没算设备突发故障、来料不良返工、插单打断。如果把这些加进去——我运行了完整仿真屏幕上跳出了结果═══════════════════════════════════════════════════════════════════════订单全链路端到端仿真下单 → 生产 → 仓储 → 发货订单50 件精密法兰 | 客户交期7 天═══════════════════════════════════════════════════════════════════════统计指标 数值─────────────────────────────────────────────────────────────────────────平均交付周期 11.3 天最短交付周期 8.7 天最长交付周期 14.9 天7天内交付概率 0%10天内交付概率 23%14天内交付概率 98%─────────────────────────────────────────────────────────────────────────瓶颈环节 TOP31. 热处理平均排队 2.1 天利用率 94%2. 物料齐套平均等待 1.3 天缺料概率 18%3. 机加工平均排队 0.8 天利用率 87%─────────────────────────────────────────────────────────────────────────建议热处理炉增加一班次或外协分流 30% 产能 → 预计交付周期降至 7.8 天老王盯着屏幕沉默了很久说所以这条链路从头到尾跑一遍不是 12 天是 11.3 天——而且 7 天内根本交不出来。那我至少可以跟销售说要么跟客户谈交期要么提前启动热处理外协。对我点头仿真的价值不是帮你算出好看的数字而是帮你提前看到真实的约束。看到约束才能做决策。一、实际应用场景真实痛点场景设定离散制造企业中销售接单时需要承诺交付周期但生产计划部门缺乏量化工具——只能凭经验估算各工序时间忽略了排队等待、资源冲突、来料波动等动态因素。结果要么承诺过紧导致违约要么承诺过松丢失订单。现场原话叙事化我们最大的痛点不是做不出来是说不准什么时候能做出来。老王说每个工序的班长都告诉我没问题但串起来就不是那么回事了。机加工完了等热处理炉热处理完了发现表面处理的化工原料没到表面处理完了装配线又被别的订单插队。这些等待时间比加工时间还长。核心矛盾静态经验估算与动态链路不确定性之间的信息差。需要一个订单全链路端到端仿真程序用simpy 做离散事件仿真用networkx 建模工序拓扑用numpy 做蒙特卡洛随机采样用matplotlib 可视化交付周期分布和瓶颈分析。二、痛点分析映射到长安大学《智能制造导论》课程模型《智能制造导论》模块 本篇痛点对应概述智能制造核心特征 自感知、自决策感知全链路状态辅助接单决策。智能制造技术基础系统集成 信息流贯通订单→计划→生产→仓储→发货全链路数据打通。新一代支撑技术大数据与仿真 离散事件仿真用 SimPy 模拟订单在工序网络中的流动。智能工厂与智能生产生产计划与调度 交期可行性评估从经验承诺到数据驱动的承诺DATP。演进范式推式生产 → 拉式生产 → 自适应协同 从拍脑袋承诺交期到仿真驱动的订单评审。一句话总结我们需要构建一个订单全链路端到端仿真程序用networkx 定义工序拓扑simpy 模拟订单在资源约束下的流动过程numpy 做蒙特卡洛随机采样matplotlib 可视化交付周期分布实现下单→生产→仓储→发货全流程交付周期的量化评估。三、核心逻辑讲解大白话3.1 问题本质把订单全链路想象成快递物流网络把订单从下单到发货的全过程想象成一个包裹从寄件到收件的物流网络* 工序节点 中转站包裹每到一个中转站要排队、分拣、装车订单每到一道工序要等机床空闲、装夹、加工。* 资源约束 运力瓶颈中转站的货车数量有限高峰期要排队热处理炉只有 2 台所有订单都要排队等。* 不确定性 堵车/天气高速堵车、暴雨延误设备突发故障、来料不良、插单打断。* 全链路仿真 物流路径规划输入起点和终点系统告诉你正常情况下几天到延误概率多大哪个中转站最容易堵工业应用* 工序拓扑用有向图定义下单→工艺审核→物料齐套→机加工→热处理→表面处理→装配→质检→入库→发货的先后依赖关系。* 资源建模每台设备是一个资源订单来了要请求资源→占用→释放。* 随机性每道工序的加工时间、故障时间、等待时间都是随机变量跑 1000 次得到交付周期的概率分布。3.2 业务逻辑 → 代码映射定义工序拓扑和资源│▼ ProcessTopology (networkx)有向图 G节点 工序工艺审核/机加工/热处理/...边 先后依赖关系节点属性 所需资源、加工时间分布参数│▼ ResourceModel资源模型每台设备 simpy.Resource设备故障 simpy.Event随机中断│▼ OrderAgent (simpy.Process)订单智能体1. 按顺序请求各工序资源2. 等待资源可用3. 占用资源执行加工4. 释放资源进入下一道工序│▼ MonteCarloSimulator蒙特卡洛仿真引擎1. 每次运行生成随机加工时间、故障事件2. 跑 N 次如 1000 次3. 收集每次的交付周期│▼ Visualizer可视化1. 交付周期概率分布直方图2. 工序等待时间堆叠图3. 资源利用率热力图4. 工序拓扑图标注瓶颈3.3 为什么用离散事件仿真而不是简单的时间相加* 问题简单相加假设上一道工序做完下一道立刻开始忽略了资源竞争多张订单抢同一台设备和排队等待。* 处理策略离散事件仿真DES模拟订单在系统中的事件流——到达、请求资源、开始加工、完成、释放资源。当多个订单竞争同一资源时自动排队。这才能反映真实的等待时间 加工时间现象。* 工程合理性DES 是制造业交期评估的标准方法如 Arena、FlexSim用 PythonSimPy 可以零成本搭建轻量版本。3.4 经验估算 vs 仿真评估对比维度 经验估算现状 端到端仿真输出 一个天数如12 天 概率分布如7 天概率 0%10 天概率 23%不确定性 忽略 蒙特卡洛量化瓶颈识别 靠感觉 资源利用率等待时间数据决策支持 大概能行吧 热处理是瓶颈增加外协可降至 7.8 天可重复性 换个人估算不同 模型固定结果可复现四、OOP 代码实现4.1 项目结构order_full_chain/├── order_full_chain/│ ├── __init__.py│ ├── topology.py # 工序拓扑networkx│ ├── resources.py # 资源模型simpy│ ├── order_agent.py # 订单智能体│ ├── simulator.py # 蒙特卡洛仿真引擎│ └── visualizer.py # 可视化├── tests/│ ├── __init__.py│ └── test_order_chain.py # 单元测试├── results/ # 输出结果│ ├── cycle_distribution.png # 交付周期分布│ ├── bottleneck_analysis.png # 瓶颈分析│ ├── topology_graph.png # 工序拓扑图│ ├── evaluation_results.csv # 仿真明细│ └── simulation_report.txt # 分析报告└── run_simulation.py # 主程序入口4.2 核心源码detailssummary/summary工序拓扑定义NetworkX 有向图import networkx as nxfrom dataclasses import dataclass, fieldfrom typing import Dict, List, Tupledataclassclass ProcessNode:工序节点属性name: strresource_count: int 1 # 资源数量设备台数process_time_mean: float 60.0 # 加工时间均值分钟process_time_std: float 10.0 # 加工时间标准差failure_rate: float 0.02 # 故障率每小时repair_time_mean: float 30.0 # 平均修复时间分钟class ProcessTopology:工序拓扑图用有向图定义订单从下单到发货的全链路工序及依赖关系。def __init__(self):self.graph nx.DiGraph()self.nodes: Dict[str, ProcessNode] {}def add_process(self, node: ProcessNode) - None:添加工序节点self.nodes[node.name] nodeself.graph.add_node(node.name,resource_countnode.resource_count,process_time_meannode.process_time_mean,process_time_stdnode.process_time_std,failure_ratenode.failure_rate,repair_time_meannode.repair_time_mean,)def add_dependency(self, from_process: str, to_process: str) - None:添加工序依赖from → toif from_process not in self.nodes or to_process not in self.nodes:raise ValueError(f工序不存在: {from_process} 或 {to_process})self.graph.add_edge(from_process, to_process)def get_predecessors(self, process_name: str) - List[str]:获取前驱工序return list(self.graph.predecessors(process_name))def get_successors(self, process_name: str) - List[str]:获取后继工序return list(self.graph.successors(process_name))def get_entry_nodes(self) - List[str]:获取入口节点没有前驱的节点return [n for n in self.graph.nodes if self.graph.in_degree(n) 0]def get_exit_nodes(self) - List[str]:获取出口节点没有后继的节点return [n for n in self.graph.nodes if self.graph.out_degree(n) 0]def topological_sort(self) - List[str]:拓扑排序return list(nx.topological_sort(self.graph))def get_node(self, name: str) - ProcessNode:获取节点属性return self.nodes[name]def build_default_topology() - ProcessTopology:构建默认订单全链路拓扑topo ProcessTopology()processes [ProcessNode(下单, resource_count1, process_time_mean10,process_time_std2),ProcessNode(工艺审核, resource_count2, process_time_mean45,process_time_std15),ProcessNode(物料齐套, resource_count1, process_time_mean240,process_time_std120),ProcessNode(机加工, resource_count4, process_time_mean120,process_time_std30, failure_rate0.03,repair_time_mean45),ProcessNode(热处理, resource_count2, process_time_mean300,process_time_std60, failure_rate0.05,repair_time_mean90),ProcessNode(表面处理, resource_count2, process_time_mean150,process_time_std40),ProcessNode(装配, resource_count3, process_time_mean80,process_time_std20, failure_rate0.02,repair_time_mean30),ProcessNode(质检, resource_count2, process_time_mean30,process_time_std10),ProcessNode(入库, resource_count2, process_time_mean20,process_time_std5),ProcessNode(发货, resource_count1, process_time_mean60,process_time_std20),]for p in processes:topo.add_process(p)# 定义依赖关系flow [下单, 工艺审核, 物料齐套, 机加工, 热处理,表面处理, 装配, 质检, 入库, 发货]for i in range(len(flow) - 1):topo.add_dependency(flow[i], flow[i 1])return topo/detailsdetailssummary/summary资源模型SimPy 离散事件仿真import simpyimport numpy as npfrom typing import Dict, Listclass ResourceManager:资源管理器为每个工序创建 SimPy Resource管理订单对设备的竞争。def __init__(self, env: simpy.Environment,resource_counts: Dict[str, int]):self.env envself.resources: Dict[str, simpy.Resource] {}for process_name, count in resource_counts.items():self.resources[process_name] simpy.Resource(env, capacitycount)def get_resource(self, process_name: str) - simpy.Resource:获取指定工序的资源return self.resources[process_name]def get_queue_length(self, process_name: str) - int:获取队列长度resource self.resources[process_name]return len(resource.queue)class FailureGenerator:设备故障生成器按指数分布生成故障事件。def __init__(self, env: simpy.Environment,failure_rates: Dict[str, float],repair_time_means: Dict[str, float],seed: int 42):self.env envself.failure_rates failure_rates # 每小时故障率self.repair_time_means repair_time_means # 平均修复时间分钟self.rng np.random.RandomState(seed)self.broken_resources: Dict[str, bool] {}def is_broken(self, process_name: str) - bool:检查设备是否故障return self.broken_resources.get(process_name, False)def generate_failure(self, process_name: str):生成故障事件在仿真环境中运行rate self.failure_rates.get(process_name, 0)if rate 0:returnwhile True:# 故障间隔指数分布interval self.rng.exponential(1.0 / rate) * 60 # 转换为分钟yield self.env.timeout(interval)# 设备坏self.broken_resources[process_name] True# 修复时间repair_time self.rng.exponential(self.repair_time_means.get(process_name, 30))yield self.env.timeout(repair_time)# 修复完成self.broken_resources[process_name] False/detailsdetailssummary/summary订单智能体SimPy 进程import simpyimport numpy as npfrom typing import Dict, List, Optionalclass OrderAgent:订单智能体代表一张订单在工序网络中流动。每个订单是一个 SimPy 进程按顺序经历各工序。def __init__(self, env: simpy.Environment,order_id: str,quantity: int,topology,resource_manager,failure_generator,process_time_params: Dict[str, tuple],rng: np.random.RandomState):self.env envself.order_id order_idself.quantity quantityself.topology topologyself.resource_mgr resource_managerself.failure_gen failure_generatorself.process_time_params process_time_paramsself.rng rngself.start_time: Optional[float] Noneself.end_time: Optional[float] Noneself.step_times: Dict[str, dict] {} # 每道工序的等待加工时间def run(self):订单执行流程SimPy 进程self.start_time self.env.now# 按拓扑排序依次经过各工序ordered_processes self.topology.topological_sort()for process_name in ordered_processes:# 等待资源可用resource self.resource_mgr.get_resource(process_name)request_time self.env.nowwith resource.request() as req:yield reqwait_time self.env.now - request_time# 等待期间如果设备故障继续等while self.failure_gen.is_broken(process_name):yield self.env.timeout(1.0)# 加工时间 单件时间 × 数量 随机波动mean, std self.process_time_params.get(process_name, (60.0, 10.0))single_time self.rng.normal(mean, std)single_time max(single_time, mean * 0.3) # 下限process_time single_time * self.quantity / resource.capacityyield self.env.timeout(process_time)self.step_times[process_name] {wait: wait_time,process: process_time,total: wait_time process_time,}self.end_time self.env.nowdef get_total_cycle_time(self) - float:获取总交付周期分钟if self.start_time is not None and self.end_time is not None:return self.end_time - self.start_timereturn float(inf)def get_step_summary(self) - Dict[str, dict]:获取各工序时间摘要return self.step_times/detailsdetailssummary/summary蒙特卡洛仿真引擎import simpyimport numpy as npimport pandas as pdfrom typing import Dict, List, Optionalfrom .topology import ProcessTopology, build_default_topologyfrom .resources import ResourceManager, FailureGeneratorfrom .order_agent import OrderAgentclass MonteCarloSimulator:订单全链路蒙特卡洛仿真引擎运行 N 次离散事件仿真每次使用不同的随机种子收集交付周期数据输出统计分布。def __init__(self, topology: Optional[ProcessTopology] None,num_simulations: int 500,seed_base: int 42):self.topology topology or build_default_topology()self.num_simulations num_simulationsself.seed_base seed_baseself.results: List[Dict] []def _run_single(self, seed: int) - dict:运行单次仿真env simpy.Environment()rng np.random.RandomState(seed)# 资源resource_counts {name: node.resource_countfor name, node in self.topology.nodes.items()}res_mgr ResourceManager(env, resource_counts)# 故障failure_rates {name: node.failure_ratefor name, node in self.topology.nodes.items()}repair_times {name: node.repair_time_meanfor name, node in self.topology.nodes.items()}failure_gen FailureGenerator(env, failure_rates, repair_times, seedseed)# 加工时间参数process_time_params {name: (node.process_time_mean, node.process_time_std)for name, node in self.topology.nodes.items()}# 创建订单order OrderAgent(envenv,order_idORDER_001,quantity50,topologyself.topology,resource_managerres_mgr,failure_generatorfailure_gen,process_time_paramsprocess_time_params,rngrng,)# 启动订单进程env.process(order.run())# 启动故障进程for name in self.topology.nodes:if failure_rates.get(name, 0) 0:env.process(failure_gen.generate_failure(name))# 运行仿真env.run()# 收集结果total_minutes order.get_total_cycle_time()step_summary order.get_step_summary()return {seed: seed,total_hours: total_minutes / 60.0,total_days: total_minutes / 60.0 / 24.0,step_summary: step_summary,}def run_all(self) - pd.DataFrame:运行所有蒙特卡洛仿真self.results []for i in range(self.num_simulations):seed self.seed_base iresult self._run_single(seed)self.results.append(result)# 转为 DataFramerows []for r in self.results:row {sim_id: r[seed],total_hours: r[total_hours],total_days: r[total_days],}# 各工序时间for step, times in r[step_summary].items():row[f{step}_wait] times[wait] / 60.0 # 小时row[f{step}_process] times[process] / 60.0row[f{step}_total] times[total] / 60.0rows.append(row)return pd.DataFrame(rows)def get_statistics(self, df: pd.DataFrame,target_days: float 7.0) - dict:计算统计指标cycles df[total_days].valuesreturn {mean_days: float(np.mean(cycles)),std_days: float(np.std(cycles)),min_days: float(np.min(cycles)),max_days: float(np.max(cycles)),median_days: float(np.median(cycles)),p95_days: float(np.percentile(cycles, 95)),prob_within_target: float(sum(1 for c in cyclesif c target_days)/ len(cycles) * 100),target_days: target_days,}def get_bottlenecks(self, df: pd.DataFrame,top_k: int 3) - List[dict]:识别瓶颈工序按等待时间排序step_cols [c for c in df.columns if c.endswith(_wait)]bottlenecks []for col in step_cols:step_name col.replace(_wait, )avg_wait_hours df[col].mean()if avg_wait_hours 0:bottlenecks.append({step: step_name,avg_wait_hours: avg_wait_hours,avg_wait_days: avg_wait_hours / 24.0,})bottlenecks.sort(keylambda x: x[avg_wait_hours], reverseTrue)return bottlenecks[:top_k]/detailsdetailssummary/summary可视化器import numpy as npimport matplotlib.pyplot as pltimport networkx as nxfrom pathlib import Pathfrom typing import Dict, List, Optionalplt.rcParams[font.sans-serif] [SimHei, DejaVu Sans]plt.rcParams[axes.unicode_minus] Falseclass Visualizer:可视化分析结果def __init__(self, results_dir: str results):self.results_dir Path(results_dir)self.results_dir.mkdir(exist_okTrue)def plot_cycle_distribution(self, cycle_days: np.ndarray,target_days: float 7.0) - None:交付周期概率分布直方图fig, ax plt.subplots(figsize(10, 6))ax.hist(cycle_days, bins50, color#3498DB,edgecolorwhite, alpha0.85, densityFalse)ax.axvline(xnp.mean(cycle_days), color#E74C3C,linestyle--, linewidth2,labelf均值: {np.mean(cycle_days):.1f} 天)ax.axvline(xtarget_days, color#27AE60,linestyle-., linewidth2,labelf交期目标: {target_days} 天)ax.axvline(xnp.percentile(cycle_days, 95), color#F39C12,linestyle:, linewidth2,labelfP95: {np.percentile(cycle_days, 95):.1f} 天)ax.set_xlabel(交付周期天, fontsi利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛