数据¶
我们提供来自 Chassaing 等人数据集 的电话叫车问题 (DARP) 实例。每个数据文件都是一个 JSON 文件,包含:
- 车辆数量
- 节点数量
- 最大行驶时间(对所有客户相同)
- 车辆载客能力
- 用于计算两个不同节点之间行驶时间的缩放因子
- 车辆的速度
- 客户数量
- 每对节点之间的距离矩阵
- 关于配送中心的信息:标识符、装车时间、客户数量、路线开始时间和最迟结束时间
- 关于每位客户的信息:乘客数量,以及接送点和送达点的索引、时间窗口、装卸时间与最大行驶时间
建模思路¶
电话叫车问题 (DARP) 的 OptAgent 模型使用列表决策变量来表示每辆车访问的节点序列,并使用浮点决策变量表示车辆在配送中心的出发时间以及各节点的额外等待时间。对所有路线添加 partition 约束,确保每个接送或送达节点恰好被访问一次。
车辆载客量和节点离开时间都通过递归数组计算。对于每个客户,使用 find 找到接送点和送达点所属的车辆,使用 index 获取它们在路线中的位置,从而约束同车服务且接送先于送达。
模型按声明顺序进行字典序多目标优化:
- 最小化所有路线的节点迟到与返回配送中心迟到
- 最小化超过客户最大乘车时间的总时长
- 最小化缩放后的总行驶距离
Python 实现¶
import json
from pathlib import Path
from optagent import OptModel, solve
def read_input_darp(instance_file):
instance = json.loads(Path(instance_file).read_text(encoding="utf-8"))
nb_clients = instance["nbClients"]
nb_nodes = instance["nbNodes"]
quantities = [0] * nb_nodes
starts = [0.0] * nb_nodes
ends = [0.0] * nb_nodes
loading_times = [0.0] * nb_nodes
max_travel_times = [0.0] * nb_nodes
for client_index, client in enumerate(instance["clients"]):
delivery_index = client_index + nb_clients
quantities[client_index] = client["nbClients"]
quantities[delivery_index] = -client["nbClients"]
for node_index, visit in (
(client_index, client["pickup"]),
(delivery_index, client["delivery"]),
):
starts[node_index] = visit["start"]
ends[node_index] = visit["end"]
loading_times[node_index] = visit["loadingTime"]
max_travel_times[node_index] = visit["maxTravelTime"]
distances = instance["distanceMatrix"]
factor = 1.0 / (instance["scale"] * instance["speed"])
distance_depot = [distances[0][node + 1] for node in range(nb_nodes)]
time_depot = [distance * factor for distance in distance_depot]
distance_matrix = [
[distances[i + 1][j + 1] for j in range(nb_nodes)]
for i in range(nb_nodes)
]
time_matrix = [
[distance * factor for distance in row] for row in distance_matrix
]
return {
"nb_clients": nb_clients,
"nb_nodes": nb_nodes,
"nb_vehicles": instance["nbVehicles"],
"depot_tw_end": instance["depot"]["twEnd"],
"capacity": instance["capacity"],
"scale": instance["scale"],
"quantities": quantities,
"starts": starts,
"ends": ends,
"loading_times": loading_times,
"max_travel_times": max_travel_times,
"distance_depot": distance_depot,
"time_depot": time_depot,
"distance_matrix": distance_matrix,
"time_matrix": time_matrix,
}
def build_darp_model(data):
nb_clients = data["nb_clients"]
nb_nodes = data["nb_nodes"]
nb_vehicles = data["nb_vehicles"]
depot_tw_end = data["depot_tw_end"]
model = OptModel()
routes = [
model.list(
nb_nodes,
default=[
node
for client in range(nb_clients)
if client % nb_vehicles == vehicle
for node in (client, client + nb_clients)
],
)
for vehicle in range(nb_vehicles)
]
depot_starts = [
model.float(0, depot_tw_end)
for vehicle in range(nb_vehicles)
]
waiting = [
model.float(0, depot_tw_end)
for node in range(nb_nodes)
]
model.constraint(model.partition(routes))
# 注意：quantities 已经提前记录好了每个节点会让车辆载客量变化多少
quantities = model.array(data["quantities"])
starts = model.array(data["starts"])
ends = model.array(data["ends"])
loading_times = model.array(data["loading_times"])
max_travel_times = model.array(data["max_travel_times"])
distance_depot = model.array(data["distance_depot"])
time_depot = model.array(data["time_depot"])
distance_matrix = model.array(data["distance_matrix"])
time_matrix = model.array(data["time_matrix"])
waiting_array = model.array(waiting)
route_times = []
route_lateness = []
home_lateness = []
route_distances = []
for vehicle, route in enumerate(routes):
count = route.count()
positions = model.range(0, count)
# 容量限制
route_quantities = model.array(
positions,
model.lambda_function(
lambda i, previous: previous + quantities[route[i]]
),
0,
)
model.constraint(
model.and_(
positions,
model.lambda_function(
lambda i: route_quantities[i] <= data["capacity"]
),
),
)
# 递推进行节点时间计算：max(Ei,Ai)+Wi+Li, 其中 Ei 是时间窗开始，Ai 是实际到达时间，Wi 是额外等待，Li 是装卸/服务时间。
route_time = model.array(
positions,
model.lambda_function(
lambda i, previous: model.max(
starts[route[i]],
model.iif(
i == 0,
depot_starts[vehicle] + time_depot[route[i]],
previous + time_matrix[route[i - 1], route[i]],
),
)
+ waiting_array[route[i]]
+ loading_times[route[i]]
),
0,
)
route_times.append(route_time)
route_lateness.append(
model.sum(
positions,
model.lambda_function(
lambda i: model.max(
0,
route_time[i]
- loading_times[route[i]]
- ends[route[i]],
)
),
)
)
home_lateness.append(
model.iif(
count > 0,
model.max(
0,
route_time[count - 1]
+ time_depot[route[count - 1]]
- depot_tw_end,
),
0,
)
)
route_distances.append(
model.sum(
model.range(1, count),
model.lambda_function(
lambda i: distance_matrix[route[i - 1], route[i]]
),
)
+ model.iif(
count > 0,
distance_depot[route[0]] + distance_depot[route[count - 1]],
0,
)
)
routes_array = model.array(routes)
route_times_array = model.array(route_times)
client_lateness = []
for client in range(nb_clients):
# 保证“同车接送，而且先接后送”
pickup_vehicle = model.find(routes_array, client)
delivery_node = client + nb_clients
delivery_vehicle = model.find(routes_array, delivery_node)
# 1）同一个客户必须由同一辆车接送
model.constraint(
pickup_vehicle == delivery_vehicle,
)
pickup_route = routes_array[pickup_vehicle]
delivery_route = routes_array[delivery_vehicle]
pickup_index = model.index(pickup_route, client)
delivery_index = model.index(delivery_route, delivery_node)
# 2）必须先 pickup，再 delivery
model.constraint(
pickup_index < delivery_index,
)
pickup_times = route_times_array[pickup_vehicle]
delivery_times = route_times_array[delivery_vehicle]
pickup_time = pickup_times[pickup_index]
delivery_time = (
delivery_times[delivery_index]
- loading_times[delivery_node]
)
client_lateness.append(
model.max(
delivery_time - pickup_time - max_travel_times[client], 0
)
)
total_lateness = model.sum(route_lateness + home_lateness)
total_client_lateness = model.sum(client_lateness)
total_distance = model.sum(route_distances) / data["scale"]
model.minimize(total_lateness)
model.minimize(total_client_lateness)
model.minimize(total_distance)
# 注意：在当前建模下，延后 depot 出发不会改善当前三个目标
expressions = {
"total_lateness": total_lateness,
"total_client_lateness": total_client_lateness,
"total_distance": total_distance,
**{f"route_{i}": route for i, route in enumerate(routes)},
**{f"depot_start_{i}": start for i, start in enumerate(depot_starts)},
**{f"waiting_{i}": value for i, value in enumerate(waiting)},
}
return model, expressions
def main(instance_file, output_file=None, time_limit=20):
data = read_input_darp(instance_file)
model, expressions = build_darp_model(data)
solution = solve(model, time_limit_s=float(time_limit))
values = {name: expression.value for name, expression in expressions.items()}
lines = [
f"Total route lateness = {values['total_lateness']}; "
f"Total client lateness = {values['total_client_lateness']}; "
f"Total distance = {values['total_distance']:.2f}; "
f"Status = {solution.feasible}"
]
for vehicle in range(data["nb_vehicles"]):
visits = ", ".join(
f"{node} ({values[f'waiting_{node}']:.2f})"
for node in values[f"route_{vehicle}"]
)
lines.append(
f"Vehicle {vehicle + 1} ({values[f'depot_start_{vehicle}']:.2f}): "
f"{visits}"
)
result_text = "\n".join(lines)
print(result_text)
if output_file is not None:
Path(output_file).write_text(result_text + "\n", encoding="utf-8")
return solution
本地运行¶
INSTANCE_DIR = Path.cwd() / "instances"
solution = main(INSTANCE_DIR / "a5-40.json", time_limit=10)