数据¶
我们提供的带时间窗与固定休息的车辆路径问题算例来自 Solomon 算例。数据文件的格式如下:
第一行给出算例的名称
第五行包含车辆数量及其公共容量
从第 10 行起,每个客户(从配送中心开始):
客户的索引
x 坐标
y 坐标
需求
最早到达时间
最晚到达时间
服务时间
建模思路¶
带时间窗与固定休息的容量约束车辆路径问题的 OptAgent 模型在 CVRPTW 模型的基础上扩展得到。关于该问题的路径与时间窗部分,我们请读者参阅该模型。
为了对此建模,我们引入了整型决策变量,表示每辆卡车相邻休息之间的时间间隔。通过以休息频率作为这些决策的上界,我们确保休息在整个规划时段内均匀分布。实际的休息时间则通过对这些间隔进行累积求和得到。
将休息纳入路径时间安排遵循以下原则:无论休息发生在行驶段、等待期还是服务期间,其固定时长都会被加到当前时间,从而使路径上的所有后续事件相应延后。
最后,目标与 CVRPTW 相同:我们按字典序依次最小化总延误、所用车辆数量以及总行驶距离。
Python 实现¶
import math
from pathlib import Path
from optagent import OptModel, solve
# Breaks parameters (15 minutes every 4 hours)
BREAKFREQUENCY = 60 * 4
BREAKDURATION = 15
def read_elem(filename):
return Path(filename).read_text(encoding="utf-8").split()
def read_input_cvrptwrb(filename):
file_it = iter(read_elem(filename))
for _ in range(4):
next(file_it)
nb_trucks = int(next(file_it))
truck_capacity = int(next(file_it))
for _ in range(13):
next(file_it)
depot_x = int(next(file_it))
depot_y = int(next(file_it))
for _ in range(2):
next(file_it)
max_horizon = int(next(file_it))
next(file_it)
customers_x = []
customers_y = []
demands = []
earliest_start = []
latest_end = []
service_time = []
while True:
value = next(file_it, None)
if value is None:
break
customer_index = int(value) - 1
customers_x.append(int(next(file_it)))
customers_y.append(int(next(file_it)))
demands.append(int(next(file_it)))
ready = int(next(file_it))
due = int(next(file_it))
duration = int(next(file_it))
earliest_start.append(ready)
latest_end.append(due + duration)
service_time.append(duration)
nb_customers = customer_index + 1
distance_matrix = compute_distance_matrix(customers_x, customers_y)
distance_depots = compute_distance_depots(
depot_x, depot_y, customers_x, customers_y
)
return (
nb_customers,
nb_trucks,
truck_capacity,
distance_matrix,
distance_depots,
demands,
service_time,
earliest_start,
latest_end,
max_horizon,
)
def compute_distance_matrix(customers_x, customers_y):
nb_customers = len(customers_x)
distance_matrix = [
[None for _ in range(nb_customers)] for _ in range(nb_customers)
]
for i in range(nb_customers):
distance_matrix[i][i] = 0
for j in range(nb_customers):
dist = compute_dist(
customers_x[i], customers_x[j], customers_y[i], customers_y[j]
)
distance_matrix[i][j] = dist
distance_matrix[j][i] = dist
return distance_matrix
def compute_distance_depots(depot_x, depot_y, customers_x, customers_y):
nb_customers = len(customers_x)
distance_depots = [None] * nb_customers
for i in range(nb_customers):
distance_depots[i] = compute_dist(
depot_x, customers_x[i], depot_y, customers_y[i]
)
return distance_depots
def compute_dist(xi, xj, yi, yj):
return math.sqrt(math.pow(xi - xj, 2) + math.pow(yi - yj, 2))
def next_available_time(customer, time, model, earliest):
return model.max(time, earliest[customer])
def needs_break(break_start, start, end, model):
return model.and_(start <= break_start, end > break_start)
def travel_end(
vehicle,
index,
time,
customers_sequences,
model,
dist_depot,
dist_matrix,
nb_breaks,
breaks_start_times,
):
sequence = customers_sequences[vehicle]
customer = sequence.at(index, default=0)
previous_customer = sequence.at(index - 1, default=0)
travel_duration = model.iif(
index == 0,
dist_depot[customer],
dist_matrix[previous_customer, customer],
)
end_with_breaks = time + travel_duration
for break_index in range(nb_breaks):
break_start = model.at(breaks_start_times[vehicle], break_index)
end_with_breaks = model.iif(
needs_break(break_start, time, end_with_breaks, model),
end_with_breaks + BREAKDURATION,
end_with_breaks,
)
return end_with_breaks
def waiting_and_service_end(
vehicle,
customer,
time,
model,
earliest,
service_time,
nb_breaks,
breaks_start_times,
):
next_start_without_break = next_available_time(customer, time, model, earliest)
end_with_breaks = next_start_without_break + service_time[customer]
for break_index in range(nb_breaks):
break_start = model.at(breaks_start_times[vehicle], break_index)
end_with_breaks = model.iif(
needs_break(break_start, time, end_with_breaks, model),
next_available_time(
customer, break_start + BREAKDURATION, model, earliest
)
+ service_time[customer],
end_with_breaks,
)
return end_with_breaks
def returning_home_time(
vehicle,
customer,
time,
model,
dist_depot,
nb_breaks,
breaks_start_times,
):
end_with_breaks = time + dist_depot[customer]
for break_index in range(nb_breaks):
break_start = model.at(breaks_start_times[vehicle], break_index)
end_with_breaks = model.iif(
needs_break(break_start, time, end_with_breaks, model),
end_with_breaks + BREAKDURATION,
end_with_breaks,
)
return end_with_breaks
def build_cvrptwrb_model(data):
nb_customers = data["nb_customers"]
nb_trucks = data["nb_trucks"]
truck_capacity = data["truck_capacity"]
max_horizon = data["max_horizon"]
nb_breaks = int(math.ceil(max_horizon / BREAKFREQUENCY) + 1)
model = OptModel()
customers_sequences = [model.list(nb_customers) for k in range(nb_trucks)]
model.constraint(model.partition(customers_sequences))
demands_array = model.array(data["demands"])
earliest_array = model.array(data["earliest_start"])
latest_array = model.array(data["latest_end"])
service_time_array = model.array(data["service_time"])
dist_matrix_array = model.array(data["distance_matrix"])
dist_depot_array = model.array(data["distance_depots"])
# Break gap decision variables and cumulative start times per truck.
breaks_start_times = []
for k in range(nb_trucks):
gaps_k = [
model.int(1, BREAKFREQUENCY)
for b in range(nb_breaks)
]
# Cumulative sum: each break_start = sum of prior gaps + offsets.
truck_breaks = []
cumulative = 0
for b in range(nb_breaks):
cumulative = cumulative + gaps_k[b]
truck_breaks.append(cumulative + BREAKDURATION * b)
breaks_start_times.append(model.array(truck_breaks))
# Last break must extend past the planning horizon.
model.constraint(
model.at(breaks_start_times[k], nb_breaks - 1) >= max_horizon + 1,
)
trucks_used = [(model.count(customers_sequences[k]) > 0) for k in range(nb_trucks)]
dist_routes = []
end_times = []
for k in range(nb_trucks):
sequence = customers_sequences[k]
c = model.count(sequence)
demand_lambda = model.lambda_function(lambda j: demands_array[j])
route_quantity = model.sum(sequence, demand_lambda)
model.constraint(route_quantity <= truck_capacity)
dist_lambda = model.lambda_function(
lambda i: dist_matrix_array[sequence[i - 1], sequence[i]]
)
dist_routes.append(
model.sum(model.range(1, c), dist_lambda)
+ model.iif(
c > 0,
dist_depot_array[sequence[0]] + dist_depot_array[sequence[c - 1]],
0,
)
)
end_time_lambda = model.lambda_function(
lambda i, prev: waiting_and_service_end(
k,
sequence[i],
travel_end(
k,
i,
prev,
customers_sequences,
model,
dist_depot_array,
dist_matrix_array,
nb_breaks,
breaks_start_times,
),
model,
earliest_array,
service_time_array,
nb_breaks,
breaks_start_times,
)
)
end_time_k = model.array(model.range(0, c), end_time_lambda, 0)
end_times.append(end_time_k)
# Lateness terms.
total_lateness_terms = []
for k in range(nb_trucks):
sequence = customers_sequences[k]
c = model.count(sequence)
end_time_k = end_times[k]
def home_lateness():
last_end = end_time_k.at(c - 1, default=0)
last_customer = sequence.at(c - 1, default=0)
return model.max(
0,
returning_home_time(
k,
last_customer,
last_end,
model,
dist_depot_array,
nb_breaks,
breaks_start_times,
)
- max_horizon,
)
home_term = model.iif(trucks_used[k], home_lateness(), 0)
def visit_lateness(i):
customer = sequence.at(i, default=0)
return model.max(0, end_time_k[i] - latest_array[customer])
visit_term = model.sum(model.range(0, c), model.lambda_function(visit_lateness))
total_lateness_terms.append(home_term + visit_term)
total_lateness = model.sum(*total_lateness_terms)
nb_trucks_used = model.sum(*trucks_used)
total_distance = model.round(100 * model.sum(*dist_routes)) / 100
model.minimize(total_lateness)
model.minimize(nb_trucks_used)
model.minimize(total_distance)
return model, customers_sequences, total_lateness, nb_trucks_used, total_distance
def main(instance_file, output_file=None, time_limit=20):
(
nb_customers,
nb_trucks,
truck_capacity,
distance_matrix,
distance_depots,
demands,
service_time,
earliest_start,
latest_end,
max_horizon,
) = read_input_cvrptwrb(instance_file)
data = {
"nb_customers": nb_customers,
"nb_trucks": nb_trucks,
"truck_capacity": truck_capacity,
"distance_matrix": distance_matrix,
"distance_depots": distance_depots,
"demands": demands,
"service_time": service_time,
"earliest_start": earliest_start,
"latest_end": latest_end,
"max_horizon": max_horizon,
}
print(
f"customers={data['nb_customers']} trucks={data['nb_trucks']} "
f"capacity={data['truck_capacity']} horizon={data['max_horizon']}"
)
model, customers_sequences, total_lateness, nb_trucks_used, total_distance = build_cvrptwrb_model(data)
solution = solve(model, time_limit_s=float(time_limit))
result_values = {'total_lateness': total_lateness.value, 'trucks_used': nb_trucks_used.value, 'total_distance': total_distance.value, **{f'truck_{k}': sequence.value for k, sequence in enumerate(customers_sequences)}}
lines = [
f"Total lateness = {result_values['total_lateness']}; "
f"Trucks used = {result_values['trucks_used']}; "
f"Total distance = {result_values['total_distance']}; "
f"Status = {solution.feasible}"
]
for truck in range(data["nb_trucks"]):
sequence = result_values[f"truck_{truck}"]
if sequence:
customers = " ".join(str(customer + 1) for customer in sequence)
lines.append(f"Truck {truck + 1}: {customers}")
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
本地运行¶
Notebook 直接调用 main 并显式传入实例路径。以下三个代码格相互独立,可以按需要单独运行;调整 time_limit 可以控制每个实例的求解时间。
INSTANCE_DIR = Path.cwd() / "instances"
print("Instances:", INSTANCE_DIR)
solution_c101_25 = main(
INSTANCE_DIR / "C101.25.txt",
time_limit=10,
)
solution_c101_50 = main(
INSTANCE_DIR / "C101.50.txt",
time_limit=10,
)
solution_c101_100 = main(
INSTANCE_DIR / "C101.100.txt",
time_limit=10,
)