如何在OR-Tools CP-Sat中实现跨停机时段的任务续加工？
2026-7-14
用OR-Tools CP-Sat实现支持机器停机中断续加工的柔性作业车间调度
问题概述
需要实现基于OR-Tools CP-Sat的柔性作业车间调度，核心需求是：任务在机器上加工时可被机器停机中断，停机结束后继续完成剩余加工；最终目标是最小化完工时间。现有方案要么无法避免任务在停机时段加工，要么不支持任务跨停机时段续加工，因此需要重新建模。
解决方案思路
- 预处理机器停机区间，生成连续的可用时段
- 对每个任务操作，选定机器后将其加工拆分到机器的多个可用时段中，总加工时长严格等于任务要求的时长
- 约束每个可用时段内的任务加工区间不重叠，保证机器独占性
- 结合工件操作的先后工艺约束，最小化整体完工时间
完整代码实现
from ortools.sat.python import cp_model from typing import List, Dict, Tuple def main(): # ---------------------- 问题参数定义 ---------------------- # 柔性作业车间：每个工件的操作序列，每个操作可选机器及对应加工时长 # 格式：jobs[工件id][操作id] = [(机器id, 加工时长), ...] jobs = [ # 工件0：2个操作 [(0, 3), (1, 4)], # 工件1：3个操作 [(1, 2), (2, 3), (0, 2)], # 工件2：2个操作 [(2, 5), (0, 3)] ] # 机器停机区间：key=机器id，value=停机时段列表(start, end) machine_downtime_intervals = { 0: [(2, 5)], 1: [(2, 3), (6, 7), (9, 14)], 2: [(3, 5)] } num_machines = max(machine_downtime_intervals.keys()) + 1 num_jobs = len(jobs) # ---------------------- 预处理机器可用区间 ---------------------- # 预估最大完工时间（可根据任务总时长+停机总时长估算） total_processing_time = sum( min(dur for (_, dur) in op) for job in jobs for op in job ) total_downtime = sum( end - start for dt_list in machine_downtime_intervals.values() for (start, end) in dt_list ) horizon = total_processing_time + total_downtime + 10 # 加冗余量 machine_available_intervals: Dict[int, List[Tuple[int, int]]] = {} for machine in range(num_machines): dt_intervals = sorted(machine_downtime_intervals.get(machine, [])) available = [] current_start = 0 for (dt_start, dt_end) in dt_intervals: if current_start < dt_start: available.append((current_start, dt_start)) current_start = dt_end # 添加最后一个可用区间到预估的horizon available.append((current_start, horizon)) machine_available_intervals[machine] = available # ---------------------- 创建CP-Sat模型 ---------------------- model = cp_model.CpModel() # 变量定义 # 1. 操作选中机器的布尔变量：selected[job_id][op_id][machine_id] selected = {} # 2. 操作在机器可用区间内的加工变量： # op_start[job_id][op_id][machine_id][interval_idx] # op_end[job_id][op_id][machine_id][interval_idx] # op_duration[job_id][op_id][machine_id][interval_idx] op_start = {} op_end = {} op_duration = {} # 3. 每个操作的实际开始/结束时间（跨可用区间的整体时间） job_op_start = {} job_op_end = {} # 4. 完工时间 makespan = model.NewIntVar(0, horizon, "makespan") # 存储各机器的加工区间，用于无重叠约束 machine_intervals = {} for job_id in range(num_jobs): for op_id in range(len(jobs[job_id])): # 操作的可选机器 allowed_machines = [m for (m, _) in jobs[job_id][op_id]] # 操作的要求加工时长（按机器区分） required_duration = {m: dur for (m, dur) in jobs[job_id][op_id]} # 创建选中机器的布尔变量 selected[(job_id, op_id)] = {} for machine in allowed_machines: selected[(job_id, op_id)][machine] = model.NewBoolVar(f"selected_job{job_id}_op{op_id}_machine{machine}") # 约束：必须选中一台机器 model.Add(sum(selected[(job_id, op_id)][m] for m in allowed_machines) == 1) # 初始化操作的整体开始/结束时间变量 job_op_start[(job_id, op_id)] = model.NewIntVar(0, horizon, f"job{job_id}_op{op_id}_start") job_op_end[(job_id, op_id)] = model.NewIntVar(0, horizon, f"job{job_id}_op{op_id}_end") for machine in allowed_machines: avail_intervals = machine_available_intervals[machine] op_start[(job_id, op_id, machine)] = [] op_end[(job_id, op_id, machine)] = [] op_duration[(job_id, op_id, machine)] = [] total_dur = model.NewIntVar(0, required_duration[machine], f"total_dur_job{job_id}_op{op_id}_machine{machine}") # 选中机器时，总加工时长等于要求值；未选中时为0 model.Add(total_dur == required_duration[machine]).OnlyEnforceIf(selected[(job_id, op_id)][machine]) model.Add(total_dur == 0).OnlyEnforceIf(selected[(job_id, op_id)][machine].Not()) # 为每个可用区间创建加工变量 interval_vars = [] for idx, (avail_start, avail_end) in enumerate(avail_intervals): s = model.NewIntVar(avail_start, avail_end, f"start_job{job_id}_op{op_id}_machine{machine}_int{idx}") e = model.NewIntVar(avail_start, avail_end, f"end_job{job_id}_op{op_id}_machine{machine}_int{idx}") d = model.NewIntVar(0, avail_end - avail_start, f"dur_job{job_id}_op{op_id}_machine{machine}_int{idx}") op_start[(job_id, op_id, machine)].append(s) op_end[(job_id, op_id, machine)].append(e) op_duration[(job_id, op_id, machine)].append(d) # 约束：子加工时段的时长=结束-开始 model.Add(e - s == d) # 未选中机器时，子加工时长为0 model.Add(d == 0).OnlyEnforceIf(selected[(job_id, op_id)][machine].Not()) # 创建区间变量，用于无重叠约束 interval = model.NewIntervalVar(s, d, e, f"interval_job{job_id}_op{op_id}_machine{machine}_int{idx}") interval_vars.append(interval) # 约束：该机器上所有可用区间的加工时长之和等于任务要求的总时长 model.Add(sum(op_duration[(job_id, op_id, machine)]) == total_dur) # 约束：操作的整体开始时间是所有子加工时段的最早开始时间 model.AddMinEquality(job_op_start[(job_id, op_id)], op_start[(job_id, op_id, machine)]).OnlyEnforceIf(selected[(job_id, op_id)][machine]) # 约束：操作的整体结束时间是所有子加工时段的最晚结束时间 model.AddMaxEquality(job_op_end[(job_id, op_id)], op_end[(job_id, op_id, machine)]).OnlyEnforceIf(selected[(job_id, op_id)][machine]) # 将子加工区间加入机器的区间列表 if machine not in machine_intervals: machine_intervals[machine] = [] machine_intervals[machine].extend(interval_vars) # ---------------------- 添加工艺约束（工件操作先后顺序） ---------------------- for job_id in range(num_jobs): for op_id in range(len(jobs[job_id]) - 1): # 前一个操作结束后，后一个操作才能开始 model.Add(job_op_start[(job_id, op_id + 1)] >= job_op_end[(job_id, op_id)]) # ---------------------- 添加机器无重叠约束 ---------------------- for machine in range(num_machines): if machine in machine_intervals: model.AddNoOverlap(machine_intervals[machine]) # ---------------------- 设置目标函数 ---------------------- # 完工时间是所有操作结束时间的最大值 all_op_ends = [job_op_end[(job_id, op_id)] for job_id in range(num_jobs) for op_id in range(len(jobs[job_id]))] model.AddMaxEquality(makespan, all_op_ends) model.Minimize(makespan) # ---------------------- 求解模型 ---------------------- solver = cp_model.CpSolver() solver.parameters.max_time_in_seconds = 60.0 # 设置超时时间 status = solver.Solve(model) # ---------------------- 输出结果 ---------------------- if status == cp_model.OPTIMAL or status == cp_model.FEASIBLE: print(f"最小完工时间: {solver.Value(makespan)}") print("\n任务调度详情：") for job_id in range(num_jobs): print(f"\n工件 {job_id}:") for op_id in range(len(jobs[job_id])): # 找到选中的机器 selected_machine = None for m in selected[(job_id, op_id)]: if solver.Value(selected[(job_id, op_id)][m]): selected_machine = m break assert selected_machine is not None start = solver.Value(job_op_start[(job_id, op_id)]) end = solver.Value(job_op_end[(job_id, op_id)]) print(f" 操作 {op_id}: 机器 {selected_machine}, 总时段 [{start}, {end}]") # 输出各可用区间内的加工子时段 avail_intervals = machine_available_intervals[selected_machine] for idx in range(len(avail_intervals)): d = solver.Value(op_duration[(job_id, op_id, selected_machine)][idx]) if d > 0: s = solver.Value(op_start[(job_id, op_id, selected_machine)][idx]) e = solver.Value(op_end[(job_id, op_id, selected_machine)][idx]) print(f" 子加工时段: [{s}, {e}], 时长 {d}") else: print("未找到可行解") if __name__ == "__main__": main()
代码关键逻辑说明
- 可用区间预处理：将机器的停机区间转换为连续的可用时段，确保任务加工只能落在这些时段内。
- 变量分层设计：为操作的选中机器、子加工时段、整体时段分别创建变量，既满足柔性选机需求，又能跟踪任务跨停机的完整调度周期。
- 核心约束：
- 强制任务总加工时长等于要求值，子加工时长之和匹配总时长
- 子加工时段严格落在机器可用区间内，同一机器同一可用区间内的加工时段无重叠
- 保证工件操作的先后执行顺序
- 目标优化：以所有操作的最晚结束时间为完工时间，通过CP-Sat求解器找到最小值。
内容的提问来源于stack exchange，提问作者rellorolo
相关产品推荐
相关产品推荐
