用Python手把手实现ALNS算法：从TSP路径规划到代码实战（附完整源码）

用Python手把手实现ALNS算法从TSP路径规划到代码实战旅行商问题TSP是组合优化中最经典的NP难问题之一如何在合理时间内找到近似最优解一直是算法研究的重点。自适应大邻域搜索ALNS作为LNS算法的增强版本通过动态调整算子权重显著提升了搜索效率。本文将用Python从零实现ALNS算法并针对TSP问题给出完整解决方案。1. 算法核心组件拆解1.1 破坏与修复算子设计破坏算子和修复算子是ALNS的核心操作单元它们的组合决定了邻域搜索的多样性。我们实现两种典型算子随机移除算子def random_destroy(solution, num_remove3): 随机移除num_remove个城市节点 destroyed solution.copy() removed [] for _ in range(num_remove): idx random.randint(0, len(destroyed)-1) removed.append(destroyed.pop(idx)) return destroyed, removed最差移除算子移除当前路径中最长的边def worst_destroy(solution, dist_matrix, num_remove3): 移除当前路径中贡献最大的num_remove个节点 destroyed solution.copy() removed [] edge_costs [] # 计算各边的成本贡献 for i in range(len(solution)): prev solution[i-1] if i0 else solution[-1] next solution[i1] if ilen(solution)-1 else solution[0] cost dist_matrix[prev][solution[i]] dist_matrix[solution[i]][next] edge_costs.append((i, cost)) # 按成本降序排序并移除 edge_costs.sort(keylambda x: -x[1]) for i in range(num_remove): idx edge_costs[i][0] removed.append(destroyed.pop(idx)) return destroyed, removed对应的修复算子实现贪婪插入算子def greedy_insert(destroyed, removed, dist_matrix): 将移除的节点插入到使路径增长最小的位置 solution destroyed.copy() for city in removed: min_cost float(inf) best_pos 0 for pos in range(len(solution)1): temp solution.copy() temp.insert(pos, city) cost calculate_total_distance(temp, dist_matrix) if cost min_cost: min_cost cost best_pos pos solution.insert(best_pos, city) return solution1.2 权重自适应机制ALNS通过记录算子的历史表现动态调整选择概率。我们实现权重更新公式$$ \omega_i (1-\rho)\omega_i \rho\frac{\sigma_i}{u_i} $$其中$\rho$为衰减系数$\sigma_i$为算子得分$u_i$为使用次数。Python实现def update_weights(weights, scores, usage, rho0.1): 更新算子权重 new_weights [] for w, s, u in zip(weights, scores, usage): if u 0: new_w (1-rho)*w rho*(s/u) else: new_w w new_weights.append(new_w) return new_weights1.3 轮盘赌选择实现根据算子权重进行概率选择def select_operator(weights): 轮盘赌选择算子 total sum(weights) pick random.uniform(0, total) current 0 for i, w in enumerate(weights): current w if current pick: return i return len(weights)-12. 完整算法框架搭建2.1 模拟退火接受准则ALNS通常与模拟退火结合以一定概率接受劣解def accept_solution(current_cost, new_cost, temperature): 模拟退火接受准则 if new_cost current_cost: return True else: prob math.exp((current_cost - new_cost)/temperature) return random.random() prob2.2 主算法流程def alns_tsp(dist_matrix, max_iter1000, initial_temp100, cooling_rate0.99): # 初始化 n_cities len(dist_matrix) current_solution random.sample(range(n_cities), n_cities) best_solution current_solution.copy() temperature initial_temp # 算子设置 destroy_weights [1, 1] # 随机移除、最差移除 repair_weights [1] # 贪婪插入 destroy_scores [0, 0] destroy_usage [0, 0] for iteration in range(max_iter): # 选择算子 d_idx select_operator(destroy_weights) if d_idx 0: destroyed, removed random_destroy(current_solution) else: destroyed, removed worst_destroy(current_solution, dist_matrix) # 修复解 new_solution greedy_insert(destroyed, removed, dist_matrix) # 评估解 current_cost calculate_total_distance(current_solution, dist_matrix) new_cost calculate_total_distance(new_solution, dist_matrix) # 更新权重分数 destroy_usage[d_idx] 1 if new_cost calculate_total_distance(best_solution, dist_matrix): destroy_scores[d_idx] 1.5 best_solution new_solution.copy() elif new_cost current_cost: destroy_scores[d_idx] 1.2 elif accept_solution(current_cost, new_cost, temperature): destroy_scores[d_idx] 0.8 # 更新当前解 if accept_solution(current_cost, new_cost, temperature): current_solution new_solution.copy() # 每100次迭代更新权重 if iteration % 100 0: destroy_weights update_weights( destroy_weights, destroy_scores, destroy_usage) destroy_scores [0, 0] destroy_usage [0, 0] # 降温 temperature * cooling_rate return best_solution3. 算法优化技巧3.1 加速距离计算频繁的路径距离计算是性能瓶颈我们实现增量计算优化def calculate_delta_cost(solution, new_solution, dist_matrix, current_cost): 计算解变化前后的成本差异 delta 0 n len(solution) for i in range(n): prev solution[i-1] if i0 else solution[-1] curr solution[i] next solution[i1] if in-1 else solution[0] delta - dist_matrix[prev][curr] dist_matrix[curr][next] new_prev new_solution[i-1] if i0 else new_solution[-1] new_curr new_solution[i] new_next new_solution[i1] if in-1 else new_solution[0] delta dist_matrix[new_prev][new_curr] dist_matrix[new_curr][new_next] return current_cost delta3.2 多线程算子评估对于大规模问题可以并行评估不同算子组合from concurrent.futures import ThreadPoolExecutor def parallel_evaluate_operators(current_solution, dist_matrix, destroy_methods): 并行评估多个破坏算子 with ThreadPoolExecutor() as executor: futures [] for method in destroy_methods: if method random: future executor.submit(random_destroy, current_solution.copy()) else: future executor.submit(worst_destroy, current_solution.copy(), dist_matrix) futures.append(future) results [f.result() for f in futures] return results4. 实战案例与参数调优4.1 TSPLIB标准测试我们使用att48数据集48个城市进行测试# 加载TSPLIB数据 def load_tsplib(filename): with open(filename) as f: lines f.readlines() # 解析坐标数据 coords [] reading_nodes False for line in lines: if line.startswith(NODE_COORD_SECTION): reading_nodes True continue if line.startswith(EOF): break if reading_nodes: parts line.strip().split() coords.append((float(parts[1]), float(parts[2]))) # 构建距离矩阵 n len(coords) dist_matrix [[0]*n for _ in range(n)] for i in range(n): for j in range(n): xd coords[i][0] - coords[j][0] yd coords[i][1] - coords[j][1] dist int(round(math.sqrt(xd*xd yd*yd))) dist_matrix[i][j] dist return dist_matrix4.2 参数调优指南通过网格搜索寻找最优参数组合参数推荐范围影响说明初始温度50-200越高接受劣解概率越大冷却率0.95-0.999越接近1降温越慢破坏节点数3-10%城市数决定邻域大小权重更新频率50-200迭代影响自适应速度ρ值0.05-0.2权重更新幅度实际测试中发现对于48城市问题初始温度100、冷却率0.995、每次移除5个城市、每100次迭代更新权重、ρ0.1的组合效果最佳。4.3 可视化分析实现搜索过程可视化有助于理解算法行为import matplotlib.pyplot as plt def plot_search_process(costs, temperatures): 绘制成本与温度变化曲线 fig, ax1 plt.subplots() color tab:red ax1.set_xlabel(Iteration) ax1.set_ylabel(Cost, colorcolor) ax1.plot(costs, colorcolor) ax1.tick_params(axisy, labelcolorcolor) ax2 ax1.twinx() color tab:blue ax2.set_ylabel(Temperature, colorcolor) ax2.plot(temperatures, colorcolor) ax2.tick_params(axisy, labelcolorcolor) plt.title(ALNS Search Process) plt.show()5. 进阶优化方向5.1 混合局部搜索在ALNS中嵌入2-opt等局部搜索提升解质量def two_opt_improve(solution, dist_matrix): 2-opt局部优化 improved True while improved: improved False for i in range(len(solution)-1): for j in range(i2, len(solution)): # 计算交换后的成本变化 a, b solution[i], solution[i1] c, d solution[j], solution[j1] if j1 len(solution) else solution[0] current dist_matrix[a][b] dist_matrix[c][d] new dist_matrix[a][c] dist_matrix[b][d] if new current: solution[i1:j1] solution[j:i:-1] improved True return solution5.2 自适应破坏强度根据搜索阶段动态调整破坏的节点数量def adaptive_destroy_size(iteration, max_iter, min_remove3, max_remove10): 随迭代次数调整破坏强度 progress iteration / max_iter if progress 0.3: # 初期大范围探索 return max_remove elif progress 0.7: # 中期平衡 return (min_remove max_remove) // 2 else: # 后期精细搜索 return min_remove5.3 记忆机制引入禁忌表避免重复搜索from collections import deque class SolutionCache: def __init__(self, max_size100): self.cache deque(maxlenmax_size) def add(self, solution): key tuple(solution) self.cache.append(key) def contains(self, solution): return tuple(solution) in self.cache在ALNS主循环中加入记忆检查if not cache.contains(new_solution): # 评估和解接受逻辑 cache.add(new_solution)通过以上实现我们构建了一个完整的ALNS算法框架。在实际TSP问题测试中该实现能够在合理时间内找到接近最优的解且明显优于基本LNS算法。算法的核心优势在于其自适应性——随着搜索进行表现良好的算子会被更频繁地使用从而自动聚焦于有潜力的搜索方向。