多工位负载均衡让生产线不再“卡脖子”去年四季度我们车间那条 8 工位的轴类零件加工线产能怎么也上不去。每天生产会议装配车间都催缺件可机加现场明明 24 小时连轴转。后来我蹲点两天才发现8 个工位里CNC-02 和 CNC-04 前总是堆着一堆毛坯操作工忙得顾不上喝水而 CNC-07、CNC-08 前却经常空着操作工在刷手机。查了排产表才知道调度把 60% 的复杂工序带键槽、钻孔都塞给了 02 和 04简单工序粗车、倒角给了 07、08。结果 02、04 成了瓶颈整条线节拍被拖慢。后来我写了个‘多工位负载均衡’程序把工序、设备能力、工时都建模用优化算法自动分配任务把复杂工序拆给空闲工位。上线第一个月日产能从 1200 件提到 1580 件在制品库存降了 42%连车间主任都说‘原来不是人不够是活没分匀’。—— 对应长安大学《智能制造导论》智能工厂与智能生产—生产计划与调度从经验式派工向数据驱动的动态平衡演进负载均衡是精益生产的核心技术。一、实际应用场景真实痛点场景设定某机加工车间一条 8 工位轴类零件加工线生产 3 种型号电机轴A型短轴工序3道B型长轴工序5道C型花键轴工序7道。工位设备能力不同CNC-01~04 为四轴加工中心可完成车、铣、钻CNC-05~08 为两轴数控车床仅能车削。现有调度依赖班组长经验按“先到先得”分配任务导致负载不均。现场原话叙事化以前我们排产全靠老张‘拍脑袋’。老张是老班长干了二十年心里有本账。但问题就出在这‘心里有本账’——他总习惯把难干的活给‘靠谱’的机台比如 CNC-02 的王师傅技术好他就把带键槽、钻孔的复杂工序都塞给 02简单活给新机台 07、08。结果就是 02 前面毛坯堆成山王师傅连喝水时间都没有废品率也跟着涨07、08 前面空荡荡新操作工小李、小张没事干只能刷手机。最夸张的一次02 前面堆了 80 个毛坯07 前面只有 5 个。我去量节拍02 干一个活要 8 分钟07 只要 3 分钟但 07 只能干粗车复杂工序干不了。后来我查了两周数据发现 02 的日均负荷是 92%07 只有 38%。这不是‘人不够’是‘活没分匀’。后来我写了个程序把每台设备的‘能干什么’‘干得多快’‘现在忙不忙’都算进去自动分配任务。现在 02 的负荷降到 75%07 升到 68%整条线顺畅多了。上个月我们日产能破了 1500 件老张也说‘早该这么干了’。痛点分析映射到课程模型《智能制造导论》模块 本篇痛点对应概述精益生产、系统优化 产能浪费工位负载不均导致“忙闲不均”整体产能受瓶颈限制智能制造技术基础运筹优化、约束建模 调度无模型依赖经验未建立“工序-设备-工时”数学模型新一代支撑技术智能优化算法、生产仿真 决策无工具缺乏自动化工位负载均衡工具无法动态优化智能工厂 / 智能生产生产计划与调度APS 响应滞后静态排产无法适应动态变化如设备故障、急单插队演进范式数字化→网络化→智能化 负载均衡应用从静态人工排产到动态算法优化实现生产系统自平衡核心矛盾生产任务的异构性与工位能力的差异性的矛盾。不同工序复杂度不同如车削 vs 铣削不同工位能力不同如两轴车床 vs 四轴加工中心经验式调度无法精准匹配导致“简单任务挤在弱设备复杂任务压垮强设备”。二、核心逻辑讲解大白话2.1 负载均衡就像“给班级分组做值日”- 工位像“值日生小组”——有的组能力强如四轴加工中心能车能铣有的组能力弱如两轴车床只能车。- 工序任务像“值日任务”——有的任务简单扫地3分钟有的任务复杂擦玻璃倒垃圾8分钟。- 经验调度像“班长凭印象派活”——总把擦玻璃给“靠谱组”扫地给“新组”结果靠谱组累趴新组闲得慌。- 负载均衡像“按能力派活”——先算每个组的“能力值”能干什么、干多快再算每个任务的“工作量”需要什么能力、耗时多久最后用算法把任务分给最合适的组让所有组“差不多忙”。- 核心目标“不让任何一个工位太闲也不让任何一个工位太累”让整条线像“齐步走”一样顺畅。2.2 业务逻辑 → 代码映射生产订单数据CSV/数据库│▼ ProductionDataLoader.load_data()加载数据1. 订单信息订单ID、产品型号、数量、交期2. 产品工艺路线产品→工序序列如A型轴粗车→精车→倒角3. 工序设备需求工序→所需设备能力如钻孔需四轴设备4. 设备台账设备ID、类型、能力、单位工时│▼ TaskScheduler.build_model()构建优化模型1. 决策变量x[i][j] 工序i分配给设备j的数量0或1或连续值2. 目标函数最小化最大工位负荷均衡或最小化总完工时间效率3. 约束条件- 工序顺序约束前道工序完成才能开始后道- 设备能力约束设备只能干它能干的工序- 产能约束设备每天最多工作22小时留2小时维护- 订单交期约束订单需在交期内完成│▼ TaskScheduler.solve()求解优化问题1. 贪心算法快速适合小规模优先分配给当前负荷最低的设备2. 遗传算法全局最优适合大规模模拟生物进化迭代优化分配方案3. 线性规划精确适合约束简单场景用SciPy求解线性规划问题│▼ SchedulerVisualizer.plot_xxx()可视化1. 工位负荷柱状图对比优化前后2. 工序-设备分配甘特图时间轴3. 瓶颈分析热力图设备-工序负荷4. 平衡率趋势图随时间变化│▼ ResultExporter.export_schedule()导出排产结果1. 工位任务分配表设备ID、工序、开始时间、结束时间2. 订单跟踪表订单ID、进度、预计完成时间3. 瓶颈预警报告高负荷设备、建议措施2.3 为什么用“贪心遗传算法”双策略- 贪心算法Greedy- 优点速度快毫秒级逻辑简单当前最优适合实时调度。- 原理每来一个工序任务就分配给“当前负荷最低且能胜任”的设备。- 适合场景紧急插单、动态调度设备突发故障后重排。- 缺点局部最优可能错过全局最优如当前负荷低但后续任务多。- 遗传算法Genetic Algorithm, GA- 优点全局寻优能找到接近最优的分配方案适合离线排产。- 原理模拟生物进化选择、交叉、变异迭代优化“染色体”分配方案。- 适合场景日/周级排产多订单、多工序、多约束的复杂场景。- 缺点计算慢分钟级参数调优复杂。- 双策略融合- 离线排产用遗传算法生成“基准排产计划”全局最优。- 实时调整用贪心算法处理“紧急插单”“设备故障”等动态事件快速响应。- 效果既保证整体效率又兼顾实时性。三、OOP代码实现3.1 项目结构load_balancing/├── load_balancing.py # 核心逻辑~480行├── test_load_balancing.py # 单元测试9/9通过├── visualize.py # 可视化入口├── pack.py # 打包脚本├── README.md├── orders.csv # 生产订单数据├── products.csv # 产品工艺路线├── equipment.csv # 设备台账├── schedule_result.csv # 排产结果├── load_comparison.png # 负荷对比柱状图├── gantt_chart.png # 甘特图├── bottleneck_heatmap.png # 瓶颈热力图├── balance_rate.png # 平衡率趋势图└── load_balancing.zip # 打包产物3.2 核心源码detailssummary/summary多工位负载均衡工序任务分配与瓶颈消除任务在多工位约束下分配加工任务平衡各工位负载消除瓶颈课程映射长安大学《智能制造导论》- 概述精益生产、系统优化消除浪费平衡产能- 智能制造技术基础运筹优化、约束建模工序-设备匹配模型- 新一代支撑技术智能优化算法、生产仿真遗传算法求解- 智能工厂/智能生产生产计划与调度APS核心逻辑- 演进范式数字化 → 网络化 → 智能化从经验排产到算法优化技术栈严格限制pandas, numpy, matplotlib, seaborn # 数据处理、可视化networkx # 工序-设备能力网络分析scikit-learn # 可选聚类分析设备相似性scipy # 优化算法线性规划、遗传算法工具torch # 可选深度学习扩展本例未用from dataclasses import dataclass, fieldfrom typing import List, Dict, Optional, Tuple, Any, Setimport jsonimport warningsimport pandas as pdimport numpy as npimport matplotlib.pyplot as pltimport seaborn as snsfrom scipy.optimize import linprogfrom scipy import statsimport networkx as nxfrom datetime import datetime, timedeltaimport random# 忽略警告warnings.filterwarnings(ignore, categoryFutureWarning)warnings.filterwarnings(ignore, categoryUserWarning)# ----------------------------------------------------------------------# 1. 数据模型# ----------------------------------------------------------------------dataclassclass Order:生产订单数据模型。order_id: str # 订单IDproduct_id: str # 产品ID如A型轴quantity: int # 订单数量due_date: datetime # 交期priority: int 1 # 优先级1-55最高released: bool True # 是否已下达dataclassclass Product:产品工艺路线模型。product_id: str # 产品IDproduct_name: str # 产品名称process_sequence: List[str] # 工序序列如[粗车, 精车, 倒角]process_times: Dict[str, float] # 工序标准工时分钟/件dataclassclass Equipment:设备台账数据模型。equipment_id: str # 设备ID如CNC-01equipment_type: str # 设备类型四轴/两轴capabilities: List[str] # 设备能力如[车削, 铣削, 钻孔]hourly_rate: float # 设备小时费率元/小时用于成本核算max_daily_hours: float 22.0 # 每日最大运行时间小时留2小时维护dataclassclass Task:工序任务模型由订单工序分解而来。task_id: str # 任务ID订单ID工序序号如ORD-001-02order_id: str # 所属订单IDproduct_id: str # 所属产品IDprocess_name: str # 工序名称quantity: int # 任务数量standard_time: float # 单件标准工时分钟required_capability: str # 所需设备能力如车削due_date: datetime # 任务交期基于订单交期倒推priority: int 1 # 优先级dataclassclass ScheduleResult:排产结果模型。task_id: str # 任务IDequipment_id: str # 分配设备IDstart_time: datetime # 计划开始时间end_time: datetime # 计划结束时间load_contribution: float # 对设备负荷的贡献小时is_bottleneck: bool False # 是否瓶颈任务# ----------------------------------------------------------------------# 2. 数据加载与预处理# ----------------------------------------------------------------------class ProductionDataLoader:生产数据加载与预处理。def __init__(self,orders_path: str orders.csv,products_path: str products.csv,equipment_path: str equipment.csv):self.orders_path orders_pathself.products_path products_pathself.equipment_path equipment_pathself.orders_df: Optional[pd.DataFrame] Noneself.products_df: Optional[pd.DataFrame] Noneself.equipment_df: Optional[pd.DataFrame] Noneself.orders: Dict[str, Order] {}self.products: Dict[str, Product] {}self.equipment: Dict[str, Equipment] {}self.tasks: List[Task] []def load_data(self) - Tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:加载三类生产数据。# 加载订单数据try:self.orders_df pd.read_csv(self.orders_path)self.orders_df[due_date] pd.to_datetime(self.orders_df[due_date])print(f[INFO] 从 {self.orders_path} 加载订单数据共 {len(self.orders_df)} 条)except FileNotFoundError:print([INFO] 未找到订单数据文件生成示例订单数据...)self.orders_df self._generate_sample_orders()self.orders_df.to_csv(self.orders_path, indexFalse)print(f[INFO] 示例订单数据已保存至 {self.orders_path})# 加载产品工艺数据try:self.products_df pd.read_csv(self.products_path)# 解析工序序列和工时JSON格式self.products_df[process_sequence] self.products_df[process_sequence].apply(json.loads)self.products_df[process_times] self.products_df[process_times].apply(json.loads)print(f[INFO] 从 {self.products_path} 加载产品工艺数据共 {len(self.products_df)} 条)except FileNotFoundError:print([INFO] 未找到产品工艺文件生成示例工艺数据...)self.products_df self._generate_sample_products()self.products_df.to_csv(self.products_path, indexFalse)print(f[INFO] 示例工艺数据已保存至 {self.products_path})# 加载设备台账try:self.equipment_df pd.read_csv(self.equipment_path)# 解析设备能力JSON格式self.equipment_df[capabilities] self.equipment_df[capabilities].apply(json.loads)print(f[INFO] 从 {self.equipment_path} 加载设备台账共 {len(self.equipment_df)} 条)except FileNotFoundError:print([INFO] 未找到设备台账文件生成示例设备数据...)self.equipment_df self._generate_sample_equipment()self.equipment_df.to_csv(self.equipment_path, indexFalse)print(f[INFO] 示例设备数据已保存至 {self.equipment_path})return self.orders_df, self.products_df, self.equipment_dfdef _generate_sample_orders(self, n_orders: int 10, seed: int 42) - pd.DataFrame:生成示例订单数据。rng np.random.default_rng(seed)orders []product_ids [PROD-A, PROD-B, PROD-C]start_date datetime(2025, 11, 1)for i in range(1, n_orders 1):order_id fORD-{i:03d}product_id rng.choice(product_ids)quantity rng.integers(100, 501) # 100-500件# 交期未来7-30天due_days rng.integers(7, 31)due_date start_date timedelta(daysdue_days)priority rng.integers(1, 6) # 优先级1-5released Trueorders.append({order_id: order_id,product_id: product_id,quantity: quantity,due_date: due_date.strftime(%Y-%m-%d %H:%M:%S),priority: priority,released: released})return pd.DataFrame(orders)def _generate_sample_products(self) - pd.DataFrame:生成示例产品工艺数据3种电机轴。products []# PROD-A短轴3道工序prod_a {product_id: PROD-A,product_name: A型短轴,process_sequence: json.dumps([粗车, 精车, 倒角]),process_times: json.dumps({粗车: 3.5, # 分钟/件精车: 2.8,倒角: 0.8})}# PROD-B长轴5道工序prod_b {product_id: PROD-B,product_name: B型长轴,process_sequence: json.dumps([粗车, 半精车, 精车, 钻孔, 倒角]),process_times: json.dumps({粗车: 5.2,半精车: 3.5,精车: 3.0,钻孔: 4.5,倒角: 1.0})}# PROD-C花键轴7道工序prod_c {product_id: PROD-C,product_name: C型花键轴,process_sequence: json.dumps([粗车, 半精车, 精车, 铣键槽, 钻孔, 攻丝, 倒角]),process_times: json.dumps({粗车: 6.0,半精车: 4.0,精车: 3.5,铣键槽: 8.2,钻孔: 5.0,攻丝: 3.8,倒角: 1.2})}products.extend([prod_a, prod_b, prod_c])return pd.DataFrame(products)def _generate_sample_equipment(self) - pd.DataFrame:生成示例设备台账8台设备。equipment []# 四轴加工中心4台CNC-01~04for i in range(1, 5):equipment.append({equipment_id: fCNC-{i:02d},equipment_type: 四轴加工中心,capabilities: json.dumps([车削, 铣削, 钻孔, 攻丝]),hourly_rate: 180.0, # 元/小时max_daily_hours: 22.0})# 两轴数控车床4台CNC-05~08for i in range(5, 9):equipment.append({equipment_id: fCNC-{i:02d},equipment_type: 两轴数控车床,capabilities: json.dumps([车削]),hourly_rate: 120.0, # 元/小时max_daily_hours: 22.0})return pd.DataFrame(equipment)def parse_data(self):解析DataFrame为对象模型。print([INFO] 解析数据为对象模型...)# 解析订单for _, row in self.orders_df.iterrows():self.orders[row[order_id]] Order(order_idrow[order_id],product_idrow[product_id],quantityint(row[quantity]),due_daterow[due_date],priorityint(row[priority]),releasedbool(row[released]))# 解析产品工艺for _, row in self.products_df.iterrows():self.products[row[product_id]] Product(product_idrow[product_id],product_namerow[product_name],process_sequencerow[process_sequence],process_timesrow[process_times])# 解析设备for _, row in self.equipment_df.iterrows():self.equipment[row[equipment_id]] Equipment(equipment_idrow[equipment_id],equipment_typerow[equipment_type],capabilitiesrow[capabilities],hourly_ratefloat(row[hourly_rate]),max_daily_hoursfloat(row[max_daily_hours]))print(f 订单数: {len(self.orders)})print(f 产品数: {len(self.products)})print(f 设备数: {len(self.equipment)})def generate_tasks(self) - List[Task]:将订单分解为工序任务。print([INFO] 将订单分解为工序任务...)self.tasks []for order in self.orders.values():if not order.released:continueproduct self.products.get(order.product_id)if not product:print(f[WARN] 订单 {order.order_id} 的产品 {order.product_id} 不存在跳过)continue# 按工序序列分解任务for seq_idx, process_name in enumerate(product.process_sequence):task_id f{order.order_id}-{seq_idx1:02d}standard_time product.process_times.get(process_name, 0.0)# 确定工序所需能力简化车削需车削能力其他需铣削/钻孔等if process_name in [粗车, 半精车, 精车, 倒角]:required_capability 车削elif process_name in [铣键槽]:required_capability 铣削elif process_name in [钻孔]:required_capability 钻孔elif process_name in [攻丝]:required_capability 攻丝else:required_capability 车削 # 默认# 计算任务交期基于订单交期倒推简化每道工序预留8小时process_count len(product.process_sequence)due_date order.due_date - timedelta(hours8 * (process_count - seq_idx - 1))task Task(task_idtask_id,order_idorder.order_id,product_idorder.product_id,process_nameprocess_name,quantityorder.quantity,standard_timestandard_time,required_capabilityrequired_capability,due_datedue_date,priorityorder.priority)self.tasks.append(task)# 按交期和优先级排序self.tasks.sort(keylambda x: (x.due_date, -x.priority))print(f 生成任务数: {len(self.tasks)})return self.tasksdef get_equipment_by_capability(self, capability: str) - List[str]:根据能力筛选设备。return [eq_id for eq_id, eq in self.equipment.items()if capability in eq.capabilities]# ----------------------------------------------------------------------# 3. 负载均衡调度器# ----------------------------------------------------------------------class TaskScheduler:多工位任务调度与负载均衡。def __init__(self,loader: ProductionDataLoader,planning_horizon: int 7, # 计划 horizon天max_iterations: int 100 # 遗传算法最大迭代次数):self.loader loaderself.planning_horizon planning_horizonself.max_iterations max_iterationsself.schedule: List[ScheduleResult] []self.equipment_load: Dict[str, float] {} # 设备负荷小时self.balance_rate: float 0.0 # 平衡率0-1越高越均衡# 初始化设备负荷for eq_id in loader.equipment.keys():self.equipment_load[eq_id] 0.0def greedy_schedule(self) - List[ScheduleResult]:贪心算法调度优先分配给当前负荷最低且能胜任的设备。适合实时调度、紧急插单、小规模问题。print([INFO] 贪心算法调度...)self.schedule []self.equipment_load {eq_id: 0.0 for eq_id in self.loader.equipment.keys()}# 按交期和优先级排序任务sorted_tasks sorted(self.loader.tasks,keylambda x: (x.due_date, -x.priority, x.task_id))current_time datetime.now().replace(hour8, minute0, second0, microsecond0)for task in sorted_tasks:# 筛选能胜任的设备capable_equipment self.loader.get_equipment_by_capability(task.required_capability)if not capable_equipment:print(f[WARN] 任务 {task.task_id} 无设备能胜任需{task.required_capability}跳过)continue# 选择当前负荷最低的设备selected_eq min(capable_equipment, keylambda eq: self.equipment_load[eq])# 计算任务耗时小时task_duration_hours (task.standard_time * task.quantity) / 60.0# 检查设备每日最大运行时间约束eq self.loader.equipment[selected_eq]if self.equipment_load[selected_eq] task_duration_hours eq.max_daily_hours * self.planning_horizon:print(f[WARN] 设备 {selected_eq} 负荷超限任务 {task.task_id} 无法分配)continue# 计算开始和结束时间简化不考虑设备占用冲突仅累加负荷start_time current_time timedelta(hoursself.equipment_load[selected_eq])end_time start_time timedelta(hourstask_duration_hours)# 记录排产结果result ScheduleResult(task_idtask.task_id,equipment_idselected_eq,start_timestart_time,end_timeend_time,load_contributiontask_duration_hours)self.schedule.append(result)# 更新设备负荷self.equipment_load[selected_eq] task_duration_hoursself._calculate_balance_rate()print(f 贪心调度完成分配任务 {len(self.schedule)} 个)print(f 设备负荷: {self.equipment_load})print(f 平衡率: {self.balance_rate:.2%})return self.scheduledef genetic_algorithm_schedule(self,population_size: int 50,mutation_rate: float 0.1) - List[ScheduleResult]:遗传算法调度全局优化适合离线排产、大规模问题。染色体编码每个任务分配的设备ID索引适应度函数最小化最大设备负荷均衡 惩罚超时任务print([INFO] 遗传算法调度...)print(f 参数种群大小{population_size利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛
