用Python+粒子群算法搞定物流配送路径规划(CVRP实战,附完整代码)

发布时间:2026/5/28 2:05:30

用Python+粒子群算法搞定物流配送路径规划(CVRP实战,附完整代码) Python粒子群算法实战智能物流配送路径优化全解析物流配送路径规划CVRP是供应链管理中的经典难题如何在满足客户需求的同时最小化运输成本直接影响企业的运营效率。本文将手把手带你用Python实现基于粒子群算法PSO的CVRP解决方案从数学建模到代码实现完整呈现智能路径规划的实战过程。1. 问题定义与数学建模1.1 CVRP问题核心要素容量约束车辆路径问题Capacitated Vehicle Routing Problem, CVRP包含以下关键要素配送中心所有车辆的起始和返回点客户点需要服务的节点每个点有特定需求量车辆具有固定容量和最大行驶距离限制目标在满足所有约束条件下最小化总运输成本典型约束条件每辆车的装载量不超过最大容量每辆车的行驶距离不超过上限每个客户点只能被一辆车服务所有车辆必须从配送中心出发并返回1.2 数学模型构建我们用数学语言精确描述CVRP问题决策变量x_ijk {1, 如果车辆k从i行驶到j {0, 否则目标函数最小化 Σ(C0*y_k) ΣΣ(C1*d_ij*x_ijk)其中C0车辆启动成本y_k是否使用车辆kC1单位距离成本d_ij点i到j的距离约束条件每个客户点只被服务一次ΣΣx_ijk 1, ∀j∈客户点车辆容量限制Σq_i ≤ Q, ∀k∈车辆路径连续性Σx_ihk - Σx_hjk 0, ∀h∈节点消除子环路ΣΣx_ijk ≤ |S|-1, ∀S⊆客户点2. 粒子群算法设计原理2.1 算法核心思想粒子群算法模拟鸟群觅食行为通过群体智能寻找最优解。在CVRP问题中粒子位置表示一个可能的路径序列速度更新根据个体经验和群体经验调整路径适应度函数评估路径优劣总成本关键参数参数说明典型值w惯性权重0.2-0.9c1个体学习因子1.5-2.0c2社会学习因子1.5-2.0粒子数种群规模20-1002.2 CVRP特殊处理编码方案# 示例客户点编号为1-31配送中心为0 particle [3,15,22,7,...,28] # 随机排列的客户点序列解码策略按顺序遍历粒子位置为每个客户点分配当前车辆直到超过容量或距离限制开启新车继续分配提示贪婪解码时建议保留10%-20%的容量余量应对突发需求3. Python完整实现3.1 基础数据结构import numpy as np import matplotlib.pyplot as plt class CVRPInstance: def __init__(self): self.coordinates [] # 节点坐标 self.demands [] # 需求量 self.capacity 120 # 车辆容量 self.max_distance 250 # 最大行驶距离 self.fixed_cost 30 # 车辆固定成本 self.per_km_cost 1 # 单位距离成本3.2 距离矩阵计算def calculate_distance_matrix(coords): 计算欧式距离矩阵 n len(coords) dist_mat np.zeros((n,n)) for i in range(n): for j in range(n): dx coords[i][0] - coords[j][0] dy coords[i][1] - coords[j][1] dist_mat[i,j] np.sqrt(dx**2 dy**2) return dist_mat3.3 粒子群算法核心class PSOSolver: def __init__(self, instance, num_particles50, max_iter1000): self.instance instance self.num_particles num_particles self.max_iter max_iter # 初始化粒子群 self.particles self._initialize_particles() self.velocities np.random.rand(num_particles, len(instance.demands)-1) # 记录最优解 self.global_best None self.global_best_fitness float(inf) def _initialize_particles(self): 使用贪婪算法初始化粒子 particles [] for _ in range(self.num_particles): # 随机打乱客户点顺序 customers list(range(1, len(self.instance.demands))) np.random.shuffle(customers) particles.append(customers) return particles def _evaluate(self, solution): 评估解的质量 # 解码并计算总成本 routes self._decode(solution) total_cost self._calculate_cost(routes) return total_cost, routes def _decode(self, solution): 将编码转换为实际路径 routes [] current_route [0] # 从配送中心出发 current_load 0 current_distance 0 for node in solution: demand self.instance.demands[node] # 检查约束条件 if (current_load demand self.instance.capacity or current_distance self._get_distance(current_route[-1], node) self._get_distance(node, 0) self.instance.max_distance): # 结束当前路径 current_route.append(0) routes.append(current_route) # 开始新路径 current_route [0, node] current_load demand current_distance self._get_distance(0, node) else: # 继续当前路径 current_distance self._get_distance(current_route[-1], node) current_route.append(node) current_load demand # 添加最后一条路径 current_route.append(0) routes.append(current_route) return routes def _get_distance(self, i, j): 获取两点间距离 coord_i self.instance.coordinates[i] coord_j self.instance.coordinates[j] return np.sqrt((coord_i[0]-coord_j[0])**2 (coord_i[1]-coord_j[1])**2) def _calculate_cost(self, routes): 计算路径总成本 total_distance 0 for route in routes: route_distance 0 for i in range(len(route)-1): route_distance self._get_distance(route[i], route[i1]) total_distance route_distance return len(routes)*self.instance.fixed_cost total_distance*self.instance.per_km_cost def optimize(self): 执行优化过程 for iter in range(self.max_iter): for i in range(self.num_particles): # 评估当前粒子 fitness, routes self._evaluate(self.particles[i]) # 更新个体最优 if fitness self.personal_best_fitness[i]: self.personal_best[i] self.particles[i].copy() self.personal_best_fitness[i] fitness # 更新全局最优 if fitness self.global_best_fitness: self.global_best self.particles[i].copy() self.global_best_fitness fitness self.best_routes routes # 更新粒子位置和速度 self._update_particles() # 打印当前最优解 if iter % 100 0: print(fIteration {iter}: Best Cost {self.global_best_fitness:.2f}) return self.best_routes, self.global_best_fitness4. 结果可视化与调优技巧4.1 路径可视化def plot_routes(instance, routes): 绘制配送路径图 plt.figure(figsize(10,8)) # 绘制客户点 x [c[0] for c in instance.coordinates[1:]] y [c[1] for c in instance.coordinates[1:]] plt.scatter(x, y, cblue, labelCustomers) # 绘制配送中心 plt.scatter(instance.coordinates[0][0], instance.coordinates[0][1], cred, markers, s100, labelDepot) # 绘制路径 colors plt.cm.rainbow(np.linspace(0,1,len(routes))) for i, route in enumerate(routes): x [instance.coordinates[node][0] for node in route] y [instance.coordinates[node][1] for node in route] plt.plot(x, y, o-, colorcolors[i], alpha0.6, labelfVehicle {i1}) plt.xlabel(X Coordinate) plt.ylabel(Y Coordinate) plt.title(Optimized Delivery Routes) plt.legend() plt.grid() plt.show()4.2 参数调优指南惯性权重w的影响较大值0.7增强全局搜索能力较小值0.3)增强局部搜索能力推荐调整策略初始阶段使用较大w0.9随着迭代线性减小到0.4w w_max - (w_max-w_min)*(iter/max_iter)学习因子设置c1 c2 1.5 通常效果良好若算法过早收敛尝试增大c1个体认知若算法收敛慢尝试增大c2社会认知4.3 性能优化技巧加速距离计算# 使用numpy向量化计算 def calculate_distance_matrix(coords): coords np.array(coords) diff coords[:,None,:] - coords[None,:,:] return np.sqrt(np.sum(diff**2, axis2))并行评估from multiprocessing import Pool def evaluate_particles(particles): with Pool() as p: results p.map(evaluate, particles) return results5. 进阶优化方向5.1 混合算法设计PSO与局部搜索结合使用PSO进行全局搜索对优质解应用2-opt局部优化def two_opt_swap(route, i, k): 反转i到k之间的节点 new_route route[:i] route[i:k1][::-1] route[k1:] return new_route5.2 动态需求处理实时调整策略预留部分车辆容量10-20%设计插入新需求的评估函数def evaluate_insertion(route, new_node, position): 评估在指定位置插入新节点的成本增量 prev_node route[position-1] next_node route[position] added_distance (distance[prev_node][new_node] distance[new_node][next_node] - distance[prev_node][next_node]) return added_distance5.3 多目标优化考虑多个优化目标总成本主要目标车辆使用数量最长单一路径长度客户等待时间处理方法def multi_objective_fitness(routes): cost calculate_cost(routes) num_vehicles len(routes) max_route_length max(calculate_route_length(r) for r in routes) return [cost, num_vehicles, max_route_length]

相关新闻