如何修改OR-Tools作业车间调度问题代码以支持单作业内的并行工序

当然可以修改！你的需求是让同一作业内同一“Order within job”的工序可以并行执行，只有更高顺序号的工序必须等前面所有低顺序号的工序全部完成后才能启动——这个调整在OR-Tools的CP-SAT模型里完全可以实现，核心是修改作业内部的 precedence（优先级）约束逻辑。

核心修改思路 #

• 重构输入数据：把原来按顺序排列的任务列表，改成按作业的「Order within job」分组的结构，这样能清晰区分哪些任务属于同一并行组
• 调整作业内部约束：从原来的「作业内任务必须严格顺序执行」，改成「同一作业中，所有k级工序必须等所有k-1级工序全部完成后才能启动」——同一k级的工序之间没有顺序约束，自然可以并行执行

修改后的完整代码 #

"""Jobshop example with parallel tasks within a job."""
import collections

from ortools.sat.python import cp_model

def main():
    """Jobshop problem with parallel tasks per job stage."""
    # 新数据结构：每个作业是一个列表，每个元素对应一个order级别的并行任务列表
    # 格式：[(machine, duration), ...] 属于同一order的任务可以并行
    jobs_data = [
        # Job A: order1有2个并行任务，order2有2个并行任务，order3有1个任务
        [(1, 5), (2, 4)],    # order 1
        [(3, 6), (1, 8)],    # order 2
        [(2, 3)],            # order 3
    ], [
        # Job B: order1有1个任务，order2有2个并行任务
        [(3, 7)],            # order 1
        [(1, 4), (3, 3)],    # order 2
    ], [
        # Job C: order1有1个任务，order2有1个任务
        [(2, 8)],            # order 1
        [(1, 1)],            # order 2
    ]

    # 计算机器总数
    machines_count = 1 + max(task[0] for job in jobs_data for stage in job for task in stage)
    all_machines = range(machines_count)

    # 计算时间上限（所有任务时长总和）
    horizon = sum(task[1] for job in jobs_data for stage in job for task in stage)

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

    # 定义存储任务变量的命名元组
    task_type = collections.namedtuple("task_type", "start end interval")
    # 定义存储结果的命名元组
    assigned_task_type = collections.namedtuple(
        "assigned_task_type", "start job stage task duration"
    )

    # 创建任务变量，并按机器分组存储区间变量
    all_tasks = {}
    machine_to_intervals = collections.defaultdict(list)

    for job_id, job in enumerate(jobs_data):
        for stage_id, stage_tasks in enumerate(job):
            for task_id, task in enumerate(stage_tasks):
                machine, duration = task
                suffix = f"_job{job_id}_stage{stage_id}_task{task_id}"
                start_var = model.NewIntVar(0, horizon, "start" + suffix)
                end_var = model.NewIntVar(0, horizon, "end" + suffix)
                interval_var = model.NewIntervalVar(
                    start_var, duration, end_var, "interval" + suffix
                )
                all_tasks[(job_id, stage_id, task_id)] = task_type(
                    start=start_var, end=end_var, interval=interval_var
                )
                machine_to_intervals[machine].append(interval_var)

    # 添加机器的无重叠约束（同一机器上的任务不能并行）
    for machine in all_machines:
        model.AddNoOverlap(machine_to_intervals[machine])

    # 添加作业内部的阶段优先级约束：所有k阶段的任务必须等k-1阶段的所有任务完成后才能开始
    for job_id, job in enumerate(jobs_data):
        for stage_id in range(1, len(job)):
            # 获取当前作业的前一个阶段（stage_id-1）的所有任务的结束时间
            prev_stage_end_times = [
                all_tasks[(job_id, stage_id-1, task_id)].end
                for task_id in range(len(job[stage_id-1]))
            ]
            # 计算前一个阶段的最晚结束时间
            prev_stage_max_end = model.NewIntVar(0, horizon, f"prev_max_end_job{job_id}_stage{stage_id}")
            model.AddMaxEquality(prev_stage_max_end, prev_stage_end_times)

            # 当前阶段的所有任务的开始时间必须 >= 前一个阶段的最晚结束时间
            for task_id in range(len(job[stage_id])):
                model.Add(
                    all_tasks[(job_id, stage_id, task_id)].start >= prev_stage_max_end
                )

    # 定义目标函数：最小化最大完工时间（makespan）
    obj_var = model.NewIntVar(0, horizon, "makespan")
    # 收集所有作业最后一个阶段的所有任务的结束时间
    all_final_stage_ends = [
        all_tasks[(job_id, len(job)-1, task_id)].end
        for job_id, job in enumerate(jobs_data)
        for task_id in range(len(job[-1]))
    ]
    model.AddMaxEquality(obj_var, all_final_stage_ends)
    model.Minimize(obj_var)

    # 创建求解器并求解
    solver = cp_model.CpSolver()
    status = solver.Solve(model)

    if status == cp_model.OPTIMAL or status == cp_model.FEASIBLE:
        print("Solution:")
        # 按机器整理分配的任务
        assigned_jobs = collections.defaultdict(list)
        for job_id, job in enumerate(jobs_data):
            for stage_id, stage_tasks in enumerate(job):
                for task_id, task in enumerate(stage_tasks):
                    machine = task[0]
                    assigned_jobs[machine].append(
                        assigned_task_type(
                            start=solver.Value(all_tasks[(job_id, stage_id, task_id)].start),
                            job=job_id,
                            stage=stage_id,
                            task=task_id,
                            duration=task[1],
                        )
                    )

        # 生成输出内容
        output = ""
        for machine in all_machines:
            # 按开始时间排序
            assigned_jobs[machine].sort()
            sol_line_tasks = f"Machine {machine}: "
            sol_line = "           "
            for assigned_task in assigned_jobs[machine]:
                name = f"job_{assigned_task.job}_stage_{assigned_task.stage}_task_{assigned_task.task}"
                sol_line_tasks += f"{name:25}"
                start = assigned_task.start
                duration = assigned_task.duration
                sol_tmp = f"[{start},{start + duration}]"
                sol_line += f"{sol_tmp:25}"
            sol_line_tasks += "\n"
            sol_line += "\n"
            output += sol_line_tasks
            output += sol_line

        print(f"Optimal Schedule Length: {solver.ObjectiveValue()}")
        print(output)
    else:
        print("No solution found.")

    # 输出统计信息
    print("\nStatistics")
    print(f"  - conflicts: {solver.NumConflicts()}")
    print(f"  - branches : {solver.NumBranches()}")
    print(f"  - wall time: {solver.WallTime()}s")

if __name__ == "__main__":
    main()

关键修改点说明 #

1. 数据结构：
   把原来的jobs_data从「作业→顺序任务」改成了「作业→阶段（对应你的Order within job）→阶段内并行任务」，这样能直观区分哪些任务属于同一并行组。

2. 作业内部约束：
   不再强制作业内任务按顺序执行，而是针对每个阶段（order），确保该阶段的所有任务都要等前一个阶段的所有任务完成后才启动。这里用AddMaxEquality计算前一阶段的最晚结束时间，再让当前阶段所有任务的开始时间都大于等于这个值。

3. 目标函数调整：
   原来只取每个作业最后一个任务的结束时间，现在要取每个作业最后一个阶段的所有任务的结束时间的最大值，这样才是真正的作业完工时间。

4. 输出逻辑：
   调整了任务命名规则，加入了阶段（stage）信息，方便你查看哪些任务属于同一并行组。

备注：内容来源于stack exchange，提问作者Adler Müller