使用Google OR Tools求解带时间窗VRP，结果不满足时间窗约束

问题概述 #

使用Google OR-Tools的Python库求解带时间窗约束的车辆路径问题（VRPTW）时，得到的解决方案未遵守设定的时间窗约束。以下是原始代码：

def create_data_model():
    """Stores the data for the problem."""
    data_dict = {}
    data_dict["time_matrix"] = time_matrix.values.tolist()
    data_dict["time_windows"] = time_window["time_windows"].values.tolist()
    data_dict["num_vehicles"] = num_vehicles
    data_dict["depot"] = 0
    data_dict["demands"] = [0] + [1] * (len(time_matrix) - 1)
    data_dict["vehicle_capacities"] = [12]*data_dict["num_vehicles"]
    data_dict["vehicle_load_time"] = 5
    data_dict["vehicle_unload_time"] = 5

    return data_dict


def print_solution(data, manager, routing, solution):

    print('Objective: {} miles'.format(solution.ObjectiveValue()))
    route_distance = 0
    time_dimension = routing.GetDimensionOrDie("Time")
    for vehicle_id in range(data["num_vehicles"]):
        print(f"Vehicle = {vehicle_id}")
        index = routing.Start(vehicle_id)
        plan_output = 'Route:\n'
        while not routing.IsEnd(index):
            node_index = manager.IndexToNode(index)
            arrival_time = solution.Min(time_dimension.CumulVar(index))
            departure_time = solution.Max(time_dimension.CumulVar(index))
            time_window = data['time_windows'][node_index]
            print(f"Client id {node_index}: Arrival Time {arrival_time}, Departure Time {departure_time} | Time Window {time_window}")

            plan_output += f' {node_index} ({arrival_time}) ->'
            previous_index = index
            index = solution.Value(routing.NextVar(index))
            route_distance += routing.GetArcCostForVehicle(previous_index, index, 0)
        plan_output += f' {manager.IndexToNode(index)}\n'
        print(plan_output)
        print('Route time: {} minutes'.format(route_distance))
    

def get_routes(solution, routing, manager):
  # Get vehicle routes from a solution and store them in an array.
  # Get vehicle routes and store them in a two dimensional array whose
  # i,j entry is the jth location visited by vehicle i along its route.
  routes = []
  for route_nbr in range(routing.vehicles()):
    index = routing.Start(route_nbr)
    route = [manager.IndexToNode(index)]
    while not routing.IsEnd(index):
      index = solution.Value(routing.NextVar(index))
      route.append(manager.IndexToNode(index))
    routes.append(route)
  return routes


def optimize():
    
    # Instantiate the data problem
    data_dict = create_data_model()
    
    # Create the routing index manager.
    manager = pywrapcp.RoutingIndexManager(
        len(data_dict["time_matrix"]), data_dict["num_vehicles"], data_dict["depot"]
    )
    
    # Create and register a transit callback.
    def time_callback(from_index, to_index):
        """Returns the travel time between the two nodes."""
        # Convert from routing variable Index to time matrix NodeIndex.
        from_node = manager.IndexToNode(from_index)
        to_node = manager.IndexToNode(to_index)
        return data_dict["time_matrix"][from_node][to_node]
    
    # Create Routing Model.
    routing = pywrapcp.RoutingModel(manager)
    transit_callback_index = routing.RegisterTransitCallback(time_callback)
    
    # Define cost of each arc.
    routing.SetArcCostEvaluatorOfAllVehicles(transit_callback_index)
    
    # Add Capacity constraint.
    def demand_callback(from_index):
        """Returns the demand of the node."""
        # Convert from routing variable Index to demands NodeIndex.
        from_node = manager.IndexToNode(from_index)
        return data_dict["demands"][from_node]

    demand_callback_index = routing.RegisterUnaryTransitCallback(demand_callback)
    routing.AddDimensionWithVehicleCapacity(
        demand_callback_index,
        0,  # null capacity slack
        data_dict["vehicle_capacities"],  # vehicle maximum capacities
        True,  # start cumul to zero
        "Capacity",
    )
    
    max_time_truck_can_travel = 540 ## 9hr

    # Add Time Windows constraint.
    time = "Time"
    routing.AddDimension(
        transit_callback_index,
        30,  # allow waiting time
        max_time_truck_can_travel,  # maximum time per vehicle
        False,  # Don't force start cumul to zero.
        time,
    )
    time_dimension = routing.GetDimensionOrDie(time)
    
    for location_idx, time_window in enumerate(data_dict["time_windows"]):
        if location_idx == data_dict["depot"]:
            continue
        index = manager.NodeToIndex(location_idx)
        time_dimension.CumulVar(index).SetMax(max_time_truck_can_travel)

    # Add time window constraints for each vehicle start node.
    depot_idx = data_dict["depot"]
    for vehicle_id in range(data_dict["num_vehicles"]):
        index = routing.Start(vehicle_id)
        time_dimension.CumulVar(index).SetRange(
            round(data_dict["time_windows"][depot_idx][0]), round(data_dict["time_windows"][depot_idx][1])
        )
        
    # Instantiate route start and end times to produce feasible times.
    for i in range(data_dict["num_vehicles"]):
        routing.AddVariableMinimizedByFinalizer(
            time_dimension.CumulVar(routing.Start(i))
        )
        routing.AddVariableMinimizedByFinalizer(time_dimension.CumulVar(routing.End(i)))
        
    # Setting first solution heuristic.
    search_parameters = pywrapcp.DefaultRoutingSearchParameters()
    search_parameters.first_solution_strategy = (
        routing_enums_pb2.FirstSolutionStrategy.PATH_CHEAPEST_ARC
    )
    search_parameters.local_search_operators.use_shortest_path_swap_active = "BOOL_FALSE"
    
    # Solve the problem.
    solution = routing.SolveWithParameters(search_parameters)
    
    if solution:
        print_solution(data_dict, manager, routing, solution)
        result = get_routes(solution, routing, manager)
    else:
        print("No Solution exist")
        result = []
        
    return result

核心问题分析 #

代码存在三个关键错误导致时间窗约束失效：

1. 未正确应用节点时间窗：仅为非depot节点设置了累计时间上限max_time_truck_can_travel，但没有将data_dict["time_windows"]中的具体时间范围（[start, end]）绑定到节点的累计时间变量。
2. 未包含服务时间：代码定义了装卸货时间，但time_callback仅返回节点间旅行时间，未将depot的装货时间、客户节点的卸货时间加入时间维度计算。
3. 搜索参数配置限制：禁用了最短路径交换的局部搜索算子，可能导致算法无法找到满足约束的最优解。

routing.AddDimension 参数详解 #

针对你询问的时间维度配置函数：

routing.AddDimension(
    transit_callback_index,
    30,  # allow waiting time
    max_time_truck_can_travel,  # maximum time per vehicle
    False,  # Don't force start cumul to zero.
    time,
)

各参数作用：

• transit_callback_index：注册的时间回调函数索引，用于计算节点间的旅行（+服务）时间。
• slack_max（允许等待时间）：节点处允许等待的最长时间单位。例如设为30，意味着车辆到达节点时间早于时间窗开始时，最多可等待30个时间单位（如分钟）。若需严格遵守时间窗，可结合节点的CumulVar.SetRange来限制，此参数仅控制等待时间的上限。
• capacity（车辆最大行驶时间）：单辆车从出发到返回depot的累计时间上限，需与时间矩阵、时间窗的单位保持一致（此处为540分钟=9小时）。
• fix_start_cumul：设为False时，车辆的出发时间可在depot的时间窗范围内灵活调整；设为True则强制车辆从时间0出发。

修正后的代码 #

def create_data_model():
    """Stores the data for the problem."""
    data_dict = {}
    data_dict["time_matrix"] = time_matrix.values.tolist()
    data_dict["time_windows"] = time_window["time_windows"].values.tolist()
    data_dict["num_vehicles"] = num_vehicles
    data_dict["depot"] = 0
    data_dict["demands"] = [0] + [1] * (len(time_matrix) - 1)
    data_dict["vehicle_capacities"] = [12]*data_dict["num_vehicles"]
    data_dict["vehicle_load_time"] = 5
    data_dict["vehicle_unload_time"] = 5

    return data_dict


def print_solution(data, manager, routing, solution):
    print('Objective: {} minutes'.format(solution.ObjectiveValue()))
    time_dimension = routing.GetDimensionOrDie("Time")
    for vehicle_id in range(data["num_vehicles"]):
        print(f"\nVehicle = {vehicle_id}")
        index = routing.Start(vehicle_id)
        plan_output = 'Route:\n'
        total_time = 0
        while not routing.IsEnd(index):
            node_index = manager.IndexToNode(index)
            arrival_time = solution.Min(time_dimension.CumulVar(index))
            departure_time = solution.Max(time_dimension.CumulVar(index))
            time_window = data['time_windows'][node_index]
            print(f"Client {node_index}: Arrival={arrival_time}, Departure={departure_time} | Time Window={time_window}")

            plan_output += f' {node_index} ({arrival_time}) ->'
            previous_index = index
            index = solution.Value(routing.NextVar(index))
            total_time += routing.GetArcCostForVehicle(previous_index, index, vehicle_id)
        plan_output += f' {manager.IndexToNode(index)}'
        print(plan_output)
        print(f'Total route time: {total_time} minutes')
    

def get_routes(solution, routing, manager):
    routes = []
    for route_nbr in range(routing.vehicles()):
        index = routing.Start(route_nbr)
        route = [manager.IndexToNode(index)]
        while not routing.IsEnd(index):
            index = solution.Value(routing.NextVar(index))
            route.append(manager.IndexToNode(index))
        routes.append(route)
    return routes


def optimize():
    data_dict = create_data_model()
    
    manager = pywrapcp.RoutingIndexManager(
        len(data_dict["time_matrix"]), data_dict["num_vehicles"], data_dict["depot"]
    )
    
    # 修正时间回调：加入服务时间
    def time_callback(from_index, to_index):
        from_node = manager.IndexToNode(from_index)
        to_node = manager.IndexToNode(to_index)
        travel_time = data_dict["time_matrix"][from_node][to_node]
        # 添加服务时间：depot是装货时间，其他节点是卸货时间
        if to_node == data_dict["depot"]:
            service_time = 0  # 返回depot无需服务时间
        elif from_node == data_dict["depot"]:
            service_time = data_dict["vehicle_load_time"]
        else:
            service_time = data_dict["vehicle_unload_time"]
        return travel_time + service_time
    
    routing = pywrapcp.RoutingModel(manager)
    transit_callback_index = routing.RegisterTransitCallback(time_callback)
    
    routing.SetArcCostEvaluatorOfAllVehicles(transit_callback_index)
    
    # 容量约束保持不变
    def demand_callback(from_index):
        from_node = manager.IndexToNode(from_index)
        return data_dict["demands"][from_node]

    demand_callback_index = routing.RegisterUnaryTransitCallback(demand_callback)
    routing.AddDimensionWithVehicleCapacity(
        demand_callback_index,
        0,
        data_dict["vehicle_capacities"],
        True,
        "Capacity",
    )
    
    max_time_truck_can_travel = 540  # 9小时（分钟）

    # 添加时间维度
    time_dimension_name = "Time"
    routing.AddDimension(
        transit_callback_index,
        30,  # 最大等待时间
        max_time_truck_can_travel,
        False,
        time_dimension_name,
    )
    time_dimension = routing.GetDimensionOrDie(time_dimension_name)
    
    # 为所有节点设置时间窗约束（含depot）
    for location_idx, (tw_start, tw_end) in enumerate(data_dict["time_windows"]):
        index = manager.NodeToIndex(location_idx)
        time_dimension.CumulVar(index).SetRange(int(tw_start), int(tw_end))
    
    # 最小化出发和返回时间
    for i in range(data_dict["num_vehicles"]):
        routing.AddVariableMinimizedByFinalizer(time_dimension.CumulVar(routing.Start(i)))
        routing.AddVariableMinimizedByFinalizer(time_dimension.CumulVar(routing.End(i)))
        
    # 优化搜索参数：启用局部搜索
    search_parameters = pywrapcp.DefaultRoutingSearchParameters()
    search_parameters.first_solution_strategy = routing_enums_pb2.FirstSolutionStrategy.PATH_CHEAPEST_ARC
    # 启用引导式局部搜索，提升解的质量和约束满足度
    search_parameters.local_search_metaheuristic = routing_enums_pb2.LocalSearchMetaheuristic.GUIDED_LOCAL_SEARCH
    search_parameters.time_limit.seconds = 30  # 设置求解时间限制
    
    solution = routing.SolveWithParameters(search_parameters)
    
    if solution:
        print_solution(data_dict, manager, routing, solution)
        result = get_routes(solution, routing, manager)
    else:
        print("No Solution exists")
        result = []
        
    return result

关键修正点说明 #

1. 时间回调加入服务时间：计算节点间时间时，加入depot的装货时间和客户节点的卸货时间，确保时间维度的累计值准确反映实际耗时。
2. 绑定节点时间窗：遍历所有节点，将time_windows的[start, end]范围直接设置到对应节点的累计时间变量，强制算法遵守时间窗约束。
3. 优化搜索参数：启用引导式局部搜索，并设置时间限制，帮助算法在合理时间内找到满足所有约束的可行解。

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