Python Google OR Tools CP：修正约束避免多任务同时运行

问题定位 #

原代码未对单台机器的作业执行排他性约束，导致同一机器可同时处理多个任务的作业；错误使用AddCumulative/AddNoOverlap时，未按机器维度分组设置约束，导致容量限制与任务分配逻辑冲突，引发模型不可行。

核心修正思路 #

1. 为每台机器收集所有分配到它的作业任务
2. 对每台机器的作业集合添加AddNoOverlap约束（单台机器容量固定为1，同一时间仅能处理一个作业）
3. 以所有作业的结束时间最大值作为总完工时间（Makespan），将其设为目标函数进行最小化

完整修正代码 #

from ortools.sat.python import cp_model

def main():
    # 示例任务数据：每个任务包含多道工序，每道工序可选（机器ID, 加工时间）
    jobs_data = [
        [[(0, 3), (1, 2)], [(0, 2), (1, 3)]],  # 任务0：2道工序
        [[(0, 2), (1, 1)], [(0, 1), (1, 4)]]   # 任务1：2道工序
    ]
    machines_count = 2
    all_machines = range(machines_count)

    # 创建模型
    model = cp_model.CpModel()

    # 定义变量：每个工序的开始时间、结束时间，以及选择的机器
    all_tasks = []
    machine_to_tasks = {machine: [] for machine in all_machines}

    for job_id, job in enumerate(jobs_data):
        for task_id, task in enumerate(job):
            start_var = model.NewIntVar(0, 100, f"start_job{job_id}_task{task_id}")
            end_var = model.NewIntVar(0, 100, f"end_job{job_id}_task{task_id}")
            machine_var = model.NewIntVarFromDomain(
                cp_model.Domain.FromValues([m for m, _ in task]),
                f"machine_job{job_id}_task{task_id}"
            )
            duration_var = model.NewIntVar(0, 100, f"dur_job{job_id}_task{task_id}")
            
            # 绑定加工时间与所选机器的关系
            for m, t in task:
                model.Add(machine_var == m).OnlyEnforceIf(duration_var == t)
            
            all_tasks.append((start_var, end_var, duration_var, machine_var))
            # 将任务按机器分组（后续用于添加约束）
            assign_bools = []
            for m, _ in task:
                bool_var = model.NewBoolVar(f"assign_job{job_id}_task{task_id}_to{m}")
                model.Add(machine_var == m).OnlyEnforceIf(bool_var)
                # 仅当任务分配到该机器时，才将其加入机器的任务列表
                model.Add(start_var >= 0).OnlyEnforceIf(bool_var)
                model.Add(end_var >= start_var + duration_var).OnlyEnforceIf(bool_var)
                machine_to_tasks[m].append((start_var, end_var, bool_var))

    # 1. 添加任务内工序先后约束：同一任务的下一道工序必须在上一道结束后开始
    for job_id, job in enumerate(jobs_data):
        for task_id in range(len(job)-1):
            prev_end = all_tasks[job_id*len(job) + task_id][1]
            curr_start = all_tasks[job_id*len(job) + task_id+1][0]
            model.Add(curr_start >= prev_end)

    # 2. 添加机器排他约束：每台机器上的作业不能重叠
    for machine in all_machines:
        tasks = machine_to_tasks[machine]
        if not tasks:
            continue
        # 提取当前机器上所有可能分配的作业的时间变量与分配标记
        starts = [t[0] for t in tasks]
        ends = [t[1] for t in tasks]
        is_assigned = [t[2] for t in tasks]
        # 使用AddNoOverlap约束，仅对实际分配到该机器的作业生效
        model.AddNoOverlap(starts, ends, is_assigned)

    # 定义总完工时间（Makespan）：所有作业结束时间的最大值
    makespan_var = model.NewIntVar(0, 100, "makespan")
    all_ends = [t[1] for t in all_tasks]
    model.AddMaxEquality(makespan_var, all_ends)

    # 设置目标：最小化总完工时间
    model.Minimize(makespan_var)

    # 求解模型
    solver = cp_model.CpSolver()
    status = solver.Solve(model)

    # 输出结果
    if status == cp_model.OPTIMAL or status == cp_model.FEASIBLE:
        print(f"总完工时间: {solver.Value(makespan_var)}")
        for job_id, job in enumerate(jobs_data):
            for task_id, task in enumerate(job):
                start = solver.Value(all_tasks[job_id*len(job)+task_id][0])
                end = solver.Value(all_tasks[job_id*len(job)+task_id][1])
                machine = solver.Value(all_tasks[job_id*len(job)+task_id][3])
                print(f"任务{job_id} 工序{task_id}: 机器{machine}, 开始时间{start}, 结束时间{end}")
    else:
        print("未找到可行解")

if __name__ == "__main__":
    main()

关键说明 #

• 机器排他约束：通过machine_to_tasks按机器分组作业，结合分配标记变量使用AddNoOverlap，确保仅对实际分配到该机器的作业生效，避免全局约束导致的冲突
• 工序时间绑定：用OnlyEnforceIf关联机器选择与加工时间，保证选中某台机器时，使用对应的加工时长
• Makespan计算：通过AddMaxEquality将所有作业的结束时间最大值设为总完工时间，确保目标函数准确反映调度的整体效率

内容的提问来源于stack exchange，提问作者AJT