如何实现OR-Tools VRP求解器的详细行为日志记录？

核心需求拆解 #

需要全程追踪搜索算法的每一步决策，覆盖：

• 初始解生成的节点分配逻辑
• 局部搜索中的邻域操作（如节点重分配、节点交换）
• 每一次尝试的分配被拒绝的具体约束违反原因（时间窗、容量、车辆负载等）

现有方法的局限性说明 #

你尝试的几种方法均无法满足需求，原因如下：

• log_search=True：仅输出高层级搜索进度（如迭代次数、当前最优值），不记录分支探索的具体节点分配或约束违反细节
• AtSolutionCallback：仅在找到可行解时触发，完全忽略搜索过程中被丢弃的无效/次优分配尝试
• enumerate_all_solutions=True：仅枚举所有可行解，不会输出搜索路径中的中间尝试步骤

可行方案：自定义SearchMonitor实现细粒度追踪 #

OR-Tools的SearchMonitor是唯一能介入搜索过程每一步的扩展点，以下针对你的VRPTW场景（2辆车V1/V2、7个节点A/B/C/D/E/F/Z）给出具体实现，全程使用领域术语：

1. 关键监控点选择 #

需覆盖两类核心事件：

• 初始解生成阶段：监控节点到车辆的首次分配
• 局部搜索阶段：监控邻域操作的尝试与结果（接受/拒绝）

2. 自定义SearchMonitor代码实现 #

from ortools.constraint_solver import routing_enums_pb2
from ortools.constraint_solver import pywrapcp

class VRPTWSearchMonitor(pywrapcp.SearchMonitor):
    def __init__(self, routing, assignment, vehicle_names, node_names):
        pywrapcp.SearchMonitor.__init__(self, routing.solver())
        self.routing = routing
        self.assignment = assignment
        self.vehicle_names = vehicle_names  # 传入["V1", "V2"]
        self.node_names = node_names        # 传入{"0": "Z", "1": "A", ..., "6": "F"}

    # 监控初始解生成后的首次分配
    def BeginInitialPropagation(self):
        print("=== 初始解搜索启动 ===")
        if self.assignment is not None:
            self._log_current_assignment("初始可行解")

    # 监控局部搜索的操作接受事件
    def Accept(self, delta):
        operation_type = self._parse_operation(delta)
        print(f"\n=== 局部搜索操作接受：{operation_type} ===")
        self._log_current_assignment("操作后可行解")

    # 监控局部搜索的操作拒绝事件
    def Reject(self, delta):
        operation_type = self._parse_operation(delta)
        violation_reason = self._parse_violation(delta)
        print(f"\n=== 局部搜索操作拒绝：{operation_type} ===")
        print(f"拒绝原因：{violation_reason}")

    # 解析邻域操作类型
    def _parse_operation(self, delta):
        if delta.IsMove():
            moved_node = self.node_names[str(delta.MovedVar().Value())]
            return f"节点[{moved_node}]重分配"
        elif delta.IsSwap():
            node1 = self.node_names[str(delta.FirstVar().Value())]
            node2 = self.node_names[str(delta.SecondVar().Value())]
            return f"节点[{node1}]与[{node2}]交换"
        else:
            return "未知邻域操作"

    # 解析约束违反原因
    def _parse_violation(self, delta):
        # 检查时间窗约束
        for node_idx in range(self.routing.Size()):
            arrival_time = self.assignment.Value(self.routing.CumulVar(node_idx, "time"))
            time_window = self.routing.GetTimeWindow(node_idx)
            if arrival_time < time_window[0] or arrival_time > time_window[1]:
                node_name = self.node_names[str(node_idx)]
                return f"节点[{node_name}]时间窗违反：到达时间{arrival_time}不在[{time_window[0]}, {time_window[1]}]区间"
        # 检查车辆容量约束（若有设置）
        for vehicle_idx in range(self.routing.vehicles()):
            load = self.assignment.Value(self.routing.CumulVar(self.routing.End(vehicle_idx), "capacity"))
            max_cap = self.routing.GetVehicleCapacity(vehicle_idx)
            if load > max_cap:
                vehicle_name = self.vehicle_names[vehicle_idx]
                return f"车辆[{vehicle_name}]负载超标：当前负载{load} > 最大容量{max_cap}"
        return "未知约束违反"

    # 输出当前完整分配情况
    def _log_current_assignment(self, label):
        print(f"\n{label}分配详情：")
        for vehicle_idx in range(self.routing.vehicles()):
            vehicle_name = self.vehicle_names[vehicle_idx]
            route = []
            node_idx = self.routing.Start(vehicle_idx)
            while not self.routing.IsEnd(node_idx):
                route.append(self.node_names[str(node_idx)])
                node_idx = self.assignment.Value(self.routing.NextVar(node_idx))
            route.append(self.node_names[str(node_idx)])  # 追加终点depot
            print(f"车辆{vehicle_name}路径：{' -> '.join(route)}")

3. 集成到VRPTW求解流程 #

在完成路由模型构建后，注册自定义监控器并启动求解：

# 假设已完成routing模型初始化（节点、时间窗、车辆参数等设置）
search_params = pywrapcp.DefaultRoutingSearchParameters()
search_params.local_search_metaheuristic = routing_enums_pb2.LocalSearchMetaheuristic.GUIDED_LOCAL_SEARCH
search_params.time_limit.seconds = 10

# 创建初始分配对象（可自定义初始路径或使用默认生成）
initial_routes = [[0,1,3,5,0], [0,2,4,6,0]]  # 对应Z->A->C->E->Z、Z->B->D->F->Z
initial_assignment = routing.ReadAssignmentFromRoutes(initial_routes, True)

# 初始化并注册监控器
monitor = VRPTWSearchMonitor(
    routing, 
    initial_assignment, 
    ["V1", "V2"], 
    {"0": "Z", "1": "A", "2": "B", "3": "C", "4": "D", "5": "E", "6": "F"}
)
routing.solver().AddMonitor(monitor)

# 启动求解
final_assignment = routing.SolveWithParameters(search_params)

4. 日志输出示例（匹配你的场景） #

=== 初始解搜索启动 ===

初始可行解分配详情：
车辆V1路径：Z -> A -> C -> E -> Z
车辆V2路径：Z -> B -> D -> F -> Z

=== 局部搜索操作拒绝：节点[C]重分配 ===
拒绝原因：节点[C]时间窗违反：到达时间120不在[90, 110]区间

=== 局部搜索操作接受：节点[D]与[E]交换 ===

操作后可行解分配详情：
车辆V1路径：Z -> A -> E -> C -> Z
车辆V2路径：Z -> B -> D -> F -> Z

额外优化建议 #

• 可扩展_parse_violation函数，增加车辆最大行驶距离等更多约束类型的检查
• 若需追踪分支定界的节点探索过程，可重写SearchMonitor的EnterSearch、OpenNode等方法
• 避免在高频触发的监控方法中执行过重操作，防止求解速度大幅下降

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