如何在Google OR-Tools Python封装中使用LocalSearchOperator子类？求示例

嘿，这就是你要找的对应C++版jobshop_heuristic.cc的Python实现啦！我专门针对Job Shop问题做了大邻域搜索（LNS）的示例，核心就是自定义LocalSearchOperator子类来实现邻域操作逻辑，下面是完整的代码和说明：

1. 导入依赖与定义问题实例 #

首先我们导入OR-Tools的约束求解模块，然后定义经典的FT06 Job Shop问题实例（6个作业、6台机器）：

from ortools.constraint_solver import pywrapcp
import random

# 定义Job Shop问题实例（FT06问题：6个作业，6台机器）
# 格式：每个作业是一个列表，元素为(机器ID, 加工时间)
jobs_data = [
    [(0, 1), (1, 3), (2, 6), (3, 7), (4, 3), (5, 6)],
    [(0, 8), (1, 5), (2, 10), (3, 10), (4, 10), (5, 4)],
    [(0, 5), (1, 4), (2, 8), (3, 9), (4, 1), (5, 7)],
    [(0, 5), (1, 5), (2, 5), (3, 3), (4, 8), (5, 9)],
    [(0, 9), (1, 3), (2, 5), (3, 4), (4, 3), (5, 1)],
    [(0, 3), (1, 3), (2, 9), (3, 10), (4, 4), (5, 1)]
]

machines_count = 1 + max(task[0] for job in jobs_data for task in job)
jobs_count = len(jobs_data)
all_machines = range(machines_count)
all_jobs = range(jobs_count)

2. 自定义LocalSearchOperator子类 #

这里我们实现一个同一机器上任务交换的邻域操作，继承pywrapcp.LocalSearchOperator，核心是Apply方法实现交换逻辑，并确保操作后不违反作业内的任务顺序约束：

class MachineSwapOperator(pywrapcp.LocalSearchOperator):
    def __init__(self, solver, machines, tasks_per_machine, start_vars):
        super().__init__(solver)
        self.machines = machines
        self.tasks_per_machine = tasks_per_machine
        self.start_vars = start_vars
        self.solver = solver

    def OnStart(self):
        # 初始化操作，可选实现
        pass

    def Apply(self, delta):
        # 随机选择一台有至少2个任务的机器
        eligible_machines = [m for m in self.machines if len(self.tasks_per_machine[m]) >= 2]
        if not eligible_machines:
            return False
        machine = random.choice(eligible_machines)
        tasks = self.tasks_per_machine[machine]

        # 随机选两个不同的任务
        idx1, idx2 = random.sample(range(len(tasks)), 2)
        task_a, task_b = tasks[idx1], tasks[idx2]

        # 获取任务的开始时间变量和加工时间
        start_a = self.start_vars[task_a[0]][task_a[1]]
        start_b = self.start_vars[task_b[0]][task_b[1]]
        duration_a = task_a[2]
        duration_b = task_b[2]

        # 交换两个任务的开始时间
        delta.MakeAssignDelta(start_a, start_b.Value())
        delta.MakeAssignDelta(start_b, start_a.Value())

        # 检查交换后是否违反同一作业的任务顺序约束
        job_a, task_idx_a = task_a[0], task_a[1]
        if task_idx_a > 0:
            prev_end_a = self.start_vars[job_a][task_idx_a - 1].Value() + jobs_data[job_a][task_idx_a - 1][1]
            if start_a.Value() < prev_end_a:
                return False

        job_b, task_idx_b = task_b[0], task_b[1]
        if task_idx_b > 0:
            prev_end_b = self.start_vars[job_b][task_idx_b - 1].Value() + jobs_data[job_b][task_idx_b - 1][1]
            if start_b.Value() < prev_end_b:
                return False

        return True

    def CanBeApplied(self):
        # 判断是否存在可执行交换操作的机器
        return any(len(self.tasks_per_machine[m]) >= 2 for m in self.machines)

3. 构建求解器与约束，运行LNS求解 #

接下来我们构建Job Shop问题的核心约束，定义目标函数（最小化最大完工时间），然后将自定义的邻域操作传入LNS求解器并运行：

def main():
    # 创建求解器实例
    solver = pywrapcp.Solver("job_shop_lns")

    # 创建任务的开始/结束时间变量
    start_vars = []
    end_vars = []
    for job_idx, job in enumerate(jobs_data):
        job_start = []
        job_end = []
        for task_idx, (machine, duration) in enumerate(job):
            start = solver.IntVar(0, 1000, f"start_job_{job_idx}_task_{task_idx}")
            end = solver.IntVar(0, 1000, f"end_job_{job_idx}_task_{task_idx}")
            job_start.append(start)
            job_end.append(end)
            solver.Add(end == start + duration)
        start_vars.append(job_start)
        end_vars.append(job_end)

    # 约束1：同一作业的任务必须按顺序执行（前一个任务结束后才能开始下一个）
    for job_idx in all_jobs:
        for task_idx in range(1, len(jobs_data[job_idx])):
            solver.Add(start_vars[job_idx][task_idx] >= end_vars[job_idx][task_idx - 1])

    # 约束2：同一机器上的任务不能重叠
    tasks_per_machine = {m: [] for m in all_machines}
    for job_idx in all_jobs:
        for task_idx, (machine, duration) in enumerate(jobs_data[job_idx]):
            tasks_per_machine[machine].append((job_idx, task_idx, duration))

    for machine in all_machines:
        tasks = tasks_per_machine[machine]
        for i in range(len(tasks)):
            for j in range(i + 1, len(tasks)):
                task_i = tasks[i]
                task_j = tasks[j]
                start_i = start_vars[task_i[0]][task_i[1]]
                end_i = end_vars[task_i[0]][task_i[1]]
                start_j = start_vars[task_j[0]][task_j[1]]
                end_j = end_vars[task_j[0]][task_j[1]]
                # 两个任务不能重叠：要么i在j前，要么j在i前
                solver.Add(end_i <= start_j).OnlyEnforceIf(solver.IsLessOrEqual(start_i, start_j))
                solver.Add(end_j <= start_i).OnlyEnforceIf(solver.IsGreater(start_i, start_j))

    # 目标函数：最小化最大完工时间（Makespan）
    makespan = solver.IntVar(0, 1000, "makespan")
    solver.Add(makespan >= max(end_vars[job][-1] for job in all_jobs))
    objective = solver.Minimize(makespan, 1)

    # 设置局部搜索参数
    parameters = pywrapcp.DefaultPhaseParameters()
    parameters.heuristic_period = 1000  # 每1000步触发一次启发式
    parameters.num_failures_limit = 10000  # 失败次数上限

    # 注册自定义邻域操作
    swap_operator = MachineSwapOperator(solver, all_machines, tasks_per_machine, start_vars)
    local_search_ops = [swap_operator]

    # 创建LNS求解器并运行
    lns_solver = solver.LnsSolver(objective, local_search_ops)
    solution_collector = solver.FirstSolutionCollector()
    solution_collector.Add(makespan)
    for job_start in start_vars:
        for var in job_start:
            solution_collector.Add(var)

    status = lns_solver.Solve(parameters, solution_collector)

    # 输出求解结果
    if status in (solver.OPTIMAL, solver.FEASIBLE):
        print(f"找到的最小完工时间: {solution_collector.Value(0, makespan)}")
        for job_idx in all_jobs:
            print(f"\n作业 {job_idx} 的调度详情:")
            for task_idx, (machine, duration) in enumerate(jobs_data[job_idx]):
                start = solution_collector.Value(0, start_vars[job_idx][task_idx])
                end = start + duration
                print(f"  任务 {task_idx} (机器 {machine}): 开始时间 {start}, 结束时间 {end}")
    else:
        print("未找到可行解")

if __name__ == "__main__":
    main()

关键说明 #

• 这个示例完全对应C++版jobshop_heuristic.cc的核心逻辑：通过自定义邻域操作来探索解空间，用LNS求解Job Shop问题。
• 你可以扩展LocalSearchOperator的子类，实现其他邻域操作（比如作业内任务调整、机器间任务迁移等）来提升求解效果。
• 运行代码后会输出每个作业的任务调度时间，以及最终的最小完工时间。

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