变邻域搜索(VNS)从入门到进阶:如何用Python复现经典算法框架
变邻域搜索(VNS)实战:用Python构建一个可交互的TSP求解器
如果你曾经尝试过用启发式算法解决旅行商问题(TSP),大概率会碰到一个令人沮丧的瓶颈:算法在某个解附近徘徊不前,无论怎么迭代都无法突破。这就像爬山时被困在一个小山丘上,虽然能看到远处有更高的山峰,但眼前的路却都是下坡。变邻域搜索(Variable Neighborhood Search, VNS)正是为解决这种困境而生的——它教会算法如何“战略性撤退”,通过变换视角来发现通往更高峰的路径。
今天,我们不只讲理论,而是要动手实现一个完整的VNS框架。我会带你用Python从零开始构建一个TSP求解器,这个求解器不仅能找到高质量的解,还能用matplotlib动态展示搜索过程,让你亲眼看到算法是如何跳出局部最优的。与常见的C++实现不同,我们将采用更易读、更易扩展的Python代码,特别适合算法学习者和需要快速原型验证的研究人员。
1. 环境准备与问题建模
在开始编码之前,我们需要搭建一个合适的工作环境。我推荐使用Python 3.8或更高版本,因为我们将用到一些较新的语言特性。如果你习惯使用Jupyter Notebook进行探索性编程,那也没问题,但为了代码的模块化和可重用性,我建议创建一个标准的Python项目结构。
1.1 安装必要的依赖库
打开终端,执行以下命令安装所需的Python包:
pip install numpy matplotlib tqdm
这里简单说明一下每个库的作用:
- numpy:处理数值计算和数组操作,比Python原生列表快得多
- matplotlib:用于可视化,我们将用它绘制TSP路径和搜索过程动画
- tqdm:添加进度条,在长时间运行时能直观看到算法进展
提示:如果你打算进行更复杂的实验,可以考虑安装
scipy用于距离计算,或者pandas用于结果分析。但为了保持简洁,我们的核心实现只依赖上述三个库。
1.2 TSP问题表示与数据加载
TSP问题通常以坐标点列表的形式给出。我们首先定义一个TSPInstance类来封装问题实例:
import numpy as np
from typing import List, Tuple
import math
class TSPInstance:
"""TSP问题实例的封装类"""
def __init__(self, name: str, coordinates: List[Tuple[float, float]]):
self.name = name
self.coordinates = np.array(coordinates)
self.n = len(coordinates)
self._distance_matrix = None
@property
def distance_matrix(self) -> np.ndarray:
"""惰性计算距离矩阵"""
if self._distance_matrix is None:
n = self.n
dist_mat = np.zeros((n, n))
coords = self.coordinates
for i in range(n):
for j in range(i+1, n):
# 欧几里得距离
dx = coords[i][0] - coords[j][0]
dy = coords[i][1] - coords[j][1]
dist = math.sqrt(dx*dx + dy*dy)
dist_mat[i][j] = dist_mat[j][i] = dist
self._distance_matrix = dist_mat
return self._distance_matrix
def tour_length(self, tour: List[int]) -> float:
"""计算给定路径的总长度"""
if len(tour) != self.n:
raise ValueError(f"路径长度应为{self.n},实际为{len(tour)}")
total = 0.0
dist_mat = self.distance_matrix
for i in range(self.n):
j = (i + 1) % self.n # 循环回到起点
total += dist_mat[tour[i]][tour[j]]
return total
为了测试方便,我们可以创建一些标准测试实例。这里我提供一个生成随机TSP实例的实用函数:
def generate_random_tsp(n_cities: int = 20, seed: int = 42) -> TSPInstance:
"""生成随机TSP实例"""
np.random.seed(seed)
coordinates = [(np.random.uniform(0, 100), np.random.uniform(0, 100))
for _ in range(n_cities)]
return TSPInstance(f"random_{n_cities}", coordinates)
现在我们已经有了问题表示的基础设施。接下来,让我们深入VNS的核心机制。
2. VNS算法框架的Python实现
变邻域搜索的精髓在于“变”字——它通过系统性地改变邻域结构来避免陷入局部最优。一个完整的VNS框架包含三个关键组件:Shaking(扰动)、Local Search(局部搜索)和Neighborhood Change(邻域切换)。让我们逐一实现它们。
2.1 邻域结构的设计与实现
在TSP问题中,邻域结构定义了如何从一个解变换到另一个“相邻”的解。不同的邻域结构具有不同的搜索特性:
| 邻域类型 | 操作描述 | 搜索强度 | 计算复杂度 |
|---|---|---|---|
| 2-opt | 反转路径中的一段 | 中等 | O(n²) |
| Swap | 交换两个城市的位置 | 较弱 | O(n²) |
| Relocate | 将一个城市移动到新位置 | 中等 | O(n²) |
| 3-opt | 移除三条边并重新连接 | 较强 | O(n³) |
我们先实现最常用的2-opt邻域。2-opt操作通过反转路径中的一段来生成新解:
def two_opt_swap(tour: List[int], i: int, j: int) -> List[int]:
"""执行2-opt交换:反转i到j之间的路径段"""
if i > j:
i, j = j, i
# 创建新路径:保留0到i-1,反转i到j,保留j+1到末尾
new_tour = tour[:i] + tour[i:j+1][::-1] + tour[j+1:]
return new_tour
def two_opt_neighborhood(tour: List[int], instance: TSPInstance,
first_improvement: bool = True) -> Tuple[List[int], float]:
"""
在2-opt邻域中搜索改进解
first_improvement: True表示找到第一个改进就返回,False表示搜索整个邻域找最优
"""
best_tour = tour.copy()
best_length = instance.tour_length(tour)
improved = False
n = instance.n
# 遍历所有可能的(i, j)对
for i in range(n):
for j in range(i+2, n + (i if i>0 else 0)):
j_mod = j % n
if j_mod == (i + 1) % n:
continue
# 计算2-opt操作带来的长度变化(增量计算,避免重复计算)
a, b = tour[i], tour[(i+1)%n]
c, d = tour[j_mod], tour[(j_mod+1)%n]
dist_ab = instance.distance_matrix[a][b]
dist_cd = instance.distance_matrix[c][d]
dist_ac = instance.distance_matrix[a][c]
dist_bd = instance.distance_matrix[b][d]
delta = (dist_ac + dist_bd) - (dist_ab + dist_cd)
if delta < -1e-10: # 有改进
new_tour = two_opt_swap(tour, i, j_mod)
new_length = best_length + delta
if first_improvement:
return new_tour, new_length
else:
if new_length < best_length:
best_tour = new_tour
best_length = new_length
improved = True
return best_tour if improved else tour, best_length
接下来实现Swap邻域,它通过交换两个城市的位置来生成新解:
def swap_neighborhood(tour: List[int], instance: TSPInstance,
first_improvement: bool = True) -> Tuple[List[int], float]:
"""Swap邻域:交换两个城市的位置"""
best_tour = tour.copy()
best_length = instance.tour_length(tour)
n = instance.n
improved = False
for i in range(n):
for j in range(i+1, n):
# 计算交换操作带来的长度变化
# 需要考虑四个相关边:i-1→i, i→i+1, j-1→j, j→j+1
i_prev = tour[(i-1)%n]
i_curr = tour[i]
i_next = tour[(i+1)%n]
j_prev = tour[(j-1)%n]
j_curr = tour[j]
j_next = tour[(j+1)%n]
# 原始长度贡献
old_part = (instance.distance_matrix[i_prev][i_curr] +
instance.distance_matrix[i_curr][i_next] +
instance.distance_matrix[j_prev][j_curr] +
instance.distance_matrix[j_curr][j_next])
# 特殊情况处理:如果i和j相邻
if (j == (i+1)%n):
# i和j交换后,原来的i→j和j→i边会交换
new_part = (instance.distance_matrix[i_prev][j_curr] +
instance.distance_matrix[j_curr][i_curr] +
instance.distance_matrix[i_curr][j_next])
elif (i == (j+1)%n):
new_part = (instance.distance_matrix[j_prev][i_curr] +
instance.distance_matrix[i_curr][j_curr] +
instance.distance_matrix[j_curr][i_next])
else:
# 一般情况
new_part = (instance.distance_matrix[i_prev][j_curr] +
instance.distance_matrix[j_curr][i_next] +
instance.distance_matrix[j_prev][i_curr] +
instance.distance_matrix[i_curr][j_next])
delta = new_part - old_part
if delta < -1e-10: # 有改进
new_tour = tour.copy()
new_tour[i], new_tour[j] = new_tour[j], new_tour[i]
new_length = best_length + delta
if first_improvement:
return new_tour, new_length
else:
if new_length < best_length:
best_tour = new_tour
best_length = new_length
improved = True
return best_tour if improved else tour, best_length
2.2 Shaking机制:跳出局部最优的关键
Shaking是VNS区别于传统局部搜索的核心。当算法在当前邻域中找不到改进时,Shaking会施加一个较强的扰动,让搜索跳出当前的局部最优区域:
class ShakingOperator:
"""Shaking操作器,提供不同强度的扰动"""
def __init__(self, instance: TSPInstance):
self.instance = instance
def double_bridge(self, tour: List[int], k: int = 4) -> List[int]:
"""
双桥扰动:将路径分成4段,然后重新连接
k控制扰动强度(分段数)
"""
n = len(tour)
if k < 2 or k > n//2:
k = 4 # 默认使用经典的双桥扰动
# 随机选择k个切割点
cut_points = sorted(np.random.choice(n, k, replace=False))
# 构建新路径:按特定顺序重新连接段
new_tour = []
# 经典双桥顺序:段1-段4-段3-段2
segments = []
start = 0
for end in cut_points:
segments.append(tour[start:end])
start = end
segments.append(tour[start:])
# 重新排列段
if k == 4:
# 经典双桥:A-B-C-D → A-D-C-B
new_tour = segments[0] + segments[3] + segments[2] + segments[1]
else:
# 随机重排段
order = list(range(len(segments)))
np.random.shuffle(order)
for idx in order:
new_tour.extend(segments[idx])
return new_tour
def random_relocate(self, tour: List[int], n_moves: int = 5) -> List[int]:
"""随机重定位:随机选择n_moves个城市移动到随机位置"""
new_tour = tour.copy()
n = len(tour)
for _ in range(n_moves):
if n <= 1:
break
# 随机选择要移动的城市
city_idx = np.random.randint(0, n)
city = new_tour.pop(city_idx)
# 随机选择新位置(不能是原来的位置)
new_pos = np.random.randint(0, n-1)
if new_pos >= city_idx:
new_pos += 1
new_tour.insert(new_pos, city)
n = len(new_tour) # 更新n值
return new_tour
def shaking(self, tour: List[int], shaking_type: str = "double_bridge",
strength: int = 1) -> List[int]:
"""
执行shaking操作
strength: 扰动强度,1-5,数值越大扰动越强
"""
if shaking_type == "double_bridge":
# strength控制分段数
k = min(2 + strength * 2, len(tour)//2)
return self.double_bridge(tour, k)
elif shaking_type == "random_relocate":
n_moves = min(strength * 3, len(tour)//2)
return self.random_relocate(tour, n_moves)
else:
raise ValueError(f"未知的shaking类型: {shaking_type}")
2.3 完整的VNS算法实现
现在我们可以将各个组件组合成完整的VNS算法。我将实现两种变体:基本VNS和广义VNS(GVNS):
class VariableNeighborhoodSearch:
"""变邻域搜索算法主类"""
def __init__(self, instance: TSPInstance,
neighborhoods: List[str] = None,
shaking_types: List[str] = None):
self.instance = instance
self.shaking_op = ShakingOperator(instance)
# 默认邻域结构序列
self.neighborhoods = neighborhoods or ["2-opt", "swap", "relocate"]
self.shaking_types = shaking_types or ["double_bridge", "random_relocate"]
# 算法参数
self.max_iterations = 1000
self.max_no_improve = 50
self.k_max = 5 # 最大shaking强度
# 记录搜索历史
self.history = {
'best_lengths': [],
'current_lengths': [],
'neighborhood_used': [],
'shaking_strength': []
}
def _local_search(self, tour: List[int], neighborhood: str,
first_improvement: bool = True) -> Tuple[List[int], float]:
"""在指定邻域执行局部搜索"""
if neighborhood == "2-opt":
return two_opt_neighborhood(tour, self.instance, first_improvement)
elif neighborhood == "swap":
return swap_neighborhood(tour, self.instance, first_improvement)
elif neighborhood == "relocate":
# 这里可以调用relocate_neighborhood的实现
# 为简洁起见,我们先使用swap作为替代
return swap_neighborhood(tour, self.instance, first_improvement)
else:
raise ValueError(f"未知邻域: {neighborhood}")
def _vnd(self, tour: List[int]) -> Tuple[List[int], float]:
"""变邻域下降:顺序尝试不同邻域结构"""
current_tour = tour.copy()
current_length = self.instance.tour_length(tour)
k = 0
improved = True
while k < len(self.neighborhoods) and improved:
improved = False
neighborhood = self.neighborhoods[k]
# 在当前邻域搜索
new_tour, new_length = self._local_search(
current_tour, neighborhood, first_improvement=True
)
if new_length < current_length - 1e-10:
current_tour, current_length = new_tour, new_length
improved = True
k = 0 # 回到第一个邻域
else:
k += 1 # 尝试下一个邻域
return current_tour, current_length
def basic_vns(self, initial_tour: List[int] = None,
verbose: bool = True) -> Tuple[List[int], float]:
"""基本VNS算法"""
import time
from tqdm import tqdm
# 初始化
if initial_tour is None:
current_tour = list(range(self.instance.n))
np.random.shuffle(current_tour)
else:
current_tour = initial_tour.copy()
current_length = self.instance.tour_length(current_tour)
best_tour = current_tour.copy()
best_length = current_length
# 重置历史记录
self.history = {k: [] for k in self.history.keys()}
start_time = time.time()
no_improve_count = 0
# 主循环
pbar = tqdm(total=self.max_iterations, disable=not verbose)
for iteration in range(self.max_iterations):
k = 1 # 从最小强度开始
while k <= self.k_max:
# 1. Shaking
shaking_type = self.shaking_types[(iteration + k) % len(self.shaking_types)]
shaken_tour = self.shaking_op.shaking(
current_tour, shaking_type, strength=k
)
# 2. 局部搜索(这里使用简单的2-opt)
candidate_tour, candidate_length = self._local_search(
shaken_tour, "2-opt", first_improvement=True
)
# 3. 邻域切换决策
if candidate_length < current_length - 1e-10:
# 找到改进解,接受并回到最小邻域
current_tour, current_length = candidate_tour, candidate_length
k = 1
# 更新全局最优
if candidate_length < best_length - 1e-10:
best_tour, best_length = candidate_tour, candidate_length
no_improve_count = 0
else:
no_improve_count += 1
else:
# 没有改进,增加扰动强度
k += 1
# 记录历史
self.history['best_lengths'].append(best_length)
self.history['current_lengths'].append(current_length)
self.history['neighborhood_used'].append(k-1) # 记录使用的邻域索引
self.history['shaking_strength'].append(k)
pbar.update(1)
pbar.set_description(f"Best: {best_length:.2f}, Current: {current_length:.2f}")
# 提前终止条件
no_improve_count += 1
if no_improve_count >= self.max_no_improve:
if verbose:
print(f"提前终止:连续{self.max_no_improve}次迭代无改进")
break
pbar.close()
elapsed = time.time() - start_time
if verbose:
print(f"VNS完成,迭代{iteration+1}次,耗时{elapsed:.2f}秒")
print(f"初始解长度: {self.instance.tour_length(initial_tour) if initial_tour else 'N/A'}")
print(f"最终解长度: {best_length}")
return best_tour, best_length
def gvns(self, initial_tour: List[int] = None,
verbose: bool = True) -> Tuple[List[int], float]:
"""广义VNS:在局部搜索阶段使用VND"""
import time
from tqdm import tqdm
# 初始化
if initial_tour is None:
current_tour = list(range(self.instance.n))
np.random.shuffle(current_tour)
else:
current_tour = initial_tour.copy()
current_length = self.instance.tour_length(current_tour)
best_tour = current_tour.copy()
best_length = current_length
# 重置历史记录
self.history = {k: [] for k in self.history.keys()}
start_time = time.time()
no_improve_count = 0
# 主循环
pbar = tqdm(total=self.max_iterations, disable=not verbose)
for iteration in range(self.max_iterations):
k = 1
while k <= self.k_max:
# 1. Shaking
shaking_type = self.shaking_types[(iteration + k) % len(self.shaking_types)]
shaken_tour = self.shaking_op.shaking(
current_tour, shaking_type, strength=k
)
# 2. 使用VND进行局部搜索(比基本VNS更强)
candidate_tour, candidate_length = self._vnd(shaken_tour)
# 3. 邻域切换决策
if candidate_length < current_length - 1e-10:
current_tour, current_length = candidate_tour, candidate_length
k = 1
if candidate_length < best_length - 1e-10:
best_tour, best_length = candidate_tour, candidate_length
no_improve_count = 0
else:
no_improve_count += 1
else:
k += 1
# 记录历史
self.history['best_lengths'].append(best_length)
self.history['current_lengths'].append(current_length)
self.history['neighborhood_used'].append(k-1)
self.history['shaking_strength'].append(k)
pbar.update(1)
pbar.set_description(f"Best: {best_length:.2f}, Current: {current_length:.2f}")
no_improve_count += 1
if no_improve_count >= self.max_no_improve:
if verbose:
print(f"提前终止:连续{self.max_no_improve}次迭代无改进")
break
pbar.close()
elapsed = time.time() - start_time
if verbose:
print(f"GVNS完成,迭代{iteration+1}次,耗时{elapsed:.2f}秒")
print(f"最终解长度: {best_length}")
return best_tour, best_length
3. 可视化:让搜索过程一目了然
对于学习算法来说,可视化是理解其工作原理的最佳方式。我们将创建几个可视化函数来展示VNS的搜索过程:
import matplotlib.pyplot as plt
from matplotlib.animation import FuncAnimation
import matplotlib.patches as mpatches
class VNSVisualizer:
"""VNS算法可视化工具"""
def __init__(self, instance: TSPInstance):
self.instance = instance
self.fig = None
self.axs = None
def plot_tour(self, tour: List[int], ax=None, title: str = "TSP路径",
show_labels: bool = True, color: str = 'blue'):
"""绘制TSP路径"""
if ax is None:
fig, ax = plt.subplots(figsize=(10, 8))
coords = self.instance.coordinates
# 绘制城市点
ax.scatter(coords[:, 0], coords[:, 1], c='red', s=100, zorder=5)
if show_labels:
for i, (x, y) in enumerate(coords):
ax.text(x, y, f'{i}', fontsize=12, ha='center', va='center')
# 绘制路径
tour_coords = coords[tour]
tour_coords = np.vstack([tour_coords, tour_coords[0]]) # 回到起点
ax.plot(tour_coords[:, 0], tour_coords[:, 1],
color=color, linewidth=2, alpha=0.7, zorder=4)
ax.set_xlabel('X坐标')
ax.set_ylabel('Y坐标')
ax.set_title(title)
ax.grid(True, alpha=0.3)
# 添加总长度信息
length = self.instance.tour_length(tour)
ax.text(0.02, 0.98, f'路径长度: {length:.2f}',
transform=ax.transAxes, fontsize=12,
verticalalignment='top',
bbox=dict(boxstyle='round', facecolor='wheat', alpha=0.8))
return ax
def plot_search_history(self, history: dict, figsize: tuple = (15, 10)):
"""绘制搜索过程历史"""
fig, axs = plt.subplots(2, 2, figsize=figsize)
# 1. 最优解长度变化
axs[0, 0].plot(history['best_lengths'], linewidth=2)
axs[0, 0].set_xlabel('迭代次数')
axs[0, 0].set_ylabel('最优解长度')
axs[0, 0].set_title('最优解收敛曲线')
axs[0, 0].grid(True, alpha=0.3)
# 2. 当前解长度变化
axs[0, 1].plot(history['current_lengths'], linewidth=1, alpha=0.7)
axs[0, 1].set_xlabel('迭代次数')
axs[0, 1].set_ylabel('当前解长度')
axs[0, 1].set_title('当前解变化轨迹')
axs[0, 1].grid(True, alpha=0.3)
# 3. 邻域使用情况
unique_neighborhoods = sorted(set(history['neighborhood_used']))
neighborhood_counts = [history['neighborhood_used'].count(n)
for n in unique_neighborhoods]
colors = plt.cm.Set3(np.linspace(0, 1, len(unique_neighborhoods)))
axs[1, 0].bar(unique_neighborhoods, neighborhood_counts, color=colors)
axs[1, 0].set_xlabel('邻域索引')
axs[1, 0].set_ylabel('使用次数')
axs[1, 0].set_title('邻域结构使用统计')
axs[1, 0].grid(True, alpha=0.3, axis='y')
# 4. Shaking强度分布
shaking_counts = [history['shaking_strength'].count(s)
for s in range(1, max(history['shaking_strength'])+1)]
axs[1, 1].bar(range(1, len(shaking_counts)+1), shaking_counts,
color='skyblue', alpha=0.7)
axs[1, 1].set_xlabel('Shaking强度')
axs[1, 1].set_ylabel('使用次数')
axs[1, 1].set_title('Shaking强度分布')
axs[1, 1].grid(True, alpha=0.3, axis='y')
plt.tight_layout()
return fig
def create_search_animation(self, instance: TSPInstance,
tours_history: List[List[int]],
lengths_history: List[float],
interval: int = 200):
"""创建搜索过程动画"""
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(16, 8))
# 初始化
coords = instance.coordinates
current_tour = tours_history[0]
# 左侧:TSP路径图
scat = ax1.scatter(coords[:, 0], coords[:, 1], c='red', s=100, zorder=5)
line, = ax1.plot([], [], 'b-', linewidth=2, alpha=0.7, zorder=4)
# 添加城市标签
for i, (x, y) in enumerate(coords):
ax1.text(x, y, f'{i}', fontsize=10, ha='center', va='center')
ax1.set_xlim(coords[:, 0].min()-10, coords[:, 0].max()+10)
ax1.set_ylim(coords[:, 1].min()-10, coords[:, 1].max()+10)
ax1.set_xlabel('X坐标')
ax1.set_ylabel('Y坐标')
ax1.grid(True, alpha=0.3)
# 右侧:收敛曲线
ax2.set_xlim(0, len(lengths_history))
ax2.set_ylim(min(lengths_history)*0.95, max(lengths_history)*1.05)
ax2.set_xlabel('迭代次数')
ax2.set_ylabel('路径长度')
ax2.set_title('收敛过程')
ax2.grid(True, alpha=0.3)
conv_line, = ax2.plot([], [], 'r-', linewidth=2)
current_point, = ax2.plot([], [], 'bo', markersize=8)
# 添加长度文本
length_text = ax1.text(0.02, 0.98, '', transform=ax1.transAxes,
fontsize=12, verticalalignment='top',
bbox=dict(boxstyle='round', facecolor='wheat', alpha=0.8))
iteration_text = ax1.text(0.02, 0.92, '', transform=ax1.transAxes,
fontsize=12, verticalalignment='top',
bbox=dict(boxstyle='round', facecolor='lightblue', alpha=0.8))
def update(frame):
# 更新路径图
current_tour = tours_history[frame]
tour_coords = coords[current_tour]
tour_coords = np.vstack([tour_coords, tour_coords[0]])
line.set_data(tour_coords[:, 0], tour_coords[:, 1])
# 更新收敛曲线
conv_line.set_data(range(frame+1), lengths_history[:frame+1])
current_point.set_data([frame], [lengths_history[frame]])
# 更新文本
length_text.set_text(f'长度: {lengths_history[frame]:.2f}')
iteration_text.set_text(f'迭代: {frame+1}/{len(tours_history)}')
# 更新右侧图的x轴范围
if frame > 10:
ax2.set_xlim(0, frame+10)
return line, conv_line, current_point, length_text, iteration_text
anim = FuncAnimation(fig, update, frames=len(tours_history),
interval=interval, blit=True, repeat=False)
plt.tight_layout()
return anim
4. 实验设计与性能对比
现在让我们进行一系列实验来验证VNS算法的有效性,并比较不同配置下的性能差异。我们将设计三个实验:1) 不同邻域结构的对比;2) 基本VNS vs 广义VNS;3) 与经典启发式算法的对比。
4.1 实验设置与评估指标
首先,我们定义评估指标和实验框架:
class ExperimentRunner:
"""算法实验运行器"""
def __init__(self, instance_sizes: List[int] = [20, 50, 100],
n_trials: int = 10, random_seed: int = 42):
self.instance_sizes = instance_sizes
self.n_trials = n_trials
self.random_seed = random_seed
self.results = {}
np.random.seed(random_seed)
def run_comparison(self, algorithms: dict):
"""运行算法对比实验"""
for size in self.instance_sizes:
print(f"\n{'='*60}")
print(f"测试问题规模: {size}个城市")
print(f"{'='*60}")
size_results = {}
for algo_name, algo_config in algorithms.items():
print(f"\n测试算法: {algo_name}")
lengths = []
times = []
best_tours = []
for trial in range(self.n_trials):
# 生成测试实例
instance = generate_random_tsp(size, seed=self.random_seed + trial)
# 创建算法实例
if algo_name == "Basic VNS":
vns = VariableNeighborhoodSearch(
instance,
neighborhoods=algo_config.get('neighborhoods', ['2-opt', 'swap']),
shaking_types=algo_config.get('shaking_types', ['double_bridge'])
)
vns.max_iterations = algo_config.get('max_iterations', 100)
import time
start_time = time.time()
best_tour, best_length = vns.basic_vns(verbose=False)
elapsed = time.time() - start_time
elif algo_name == "GVNS":
vns = VariableNeighborhoodSearch(
instance,
neighborhoods=algo_config.get('neighborhoods', ['2-opt', 'swap', 'relocate']),
shaking_types=algo_config.get('shaking_types', ['double_bridge', 'random_relocate'])
)
vns.max_iterations = algo_config.get('max_iterations', 100)
import time
start_time = time.time()
best_tour, best_length = vns.gvns(verbose=False)
elapsed = time.time() - start_time
elif algo_name == "2-opt Only":
# 作为基准的纯2-opt局部搜索
instance = generate_random_tsp(size, seed=self.random_seed + trial)
import time
start_time = time.time()
# 随机初始解
initial_tour = list(range(size))
np.random.shuffle(initial_tour)
current_tour = initial_tour.copy()
current_length = instance.tour_length(current_tour)
# 简单迭代改进
for _ in range(100): # 固定迭代次数
new_tour, new_length = two_opt_neighborhood(
current_tour, instance, first_improvement=True
)
if new_length < current_length - 1e-10:
current_tour, current_length = new_tour, new_length
else:
break
elapsed = time.time() - start_time
best_tour, best_length = current_tour, current_length
lengths.append(best_length)
times.append(elapsed)
best_tours.append(best_tour)
# 统计结果
size_results[algo_name] = {
'mean_length': np.mean(lengths),
'std_length': np.std(lengths),
'mean_time': np.mean(times),
'std_time': np.std(times),
'best_length': np.min(lengths),
'worst_length': np.max(lengths),
'all_lengths': lengths,
'all_times': times
}
print(f" 平均长度: {np.mean(lengths):.2f} ± {np.std(lengths):.2f}")
print(f" 平均时间: {np.mean(times):.2f}秒")
self.results[size] = size_results
def plot_comparison(self):
"""绘制对比结果图"""
n_sizes = len(self.instance_sizes)
n_algorithms = len(next(iter(self.results.values())))
fig, axs = plt.subplots(2, 2, figsize=(15, 12))
# 准备数据
algo_names = list(next(iter(self.results.values())).keys())
colors = plt.cm.tab10(np.linspace(0, 1, len(algo_names)))
# 1. 平均解质量对比
for i, algo_name in enumerate(algo_names):
mean_lengths = [self.results[size][algo_name]['mean_length']
for size in self.instance_sizes]
axs[0, 0].plot(self.instance_sizes, mean_lengths,
marker='o', linewidth=2, label=algo_name, color=colors[i])
axs[0, 0].set_xlabel('问题规模(城市数量)')
axs[0, 0].set_ylabel('平均路径长度')
axs[0, 0].set_title('不同算法解质量对比')
axs[0, 0].legend()
axs[0, 0].grid(True, alpha=0.3)
# 2. 平均运行时间对比
for i, algo_name in enumerate(algo_names):
mean_times = [self.results[size][algo_name]['mean_time']
for size in self.instance_sizes]
axs[0, 1].plot(self.instance_sizes, mean_times,
marker='s', linewidth=2, label=algo_name, color=colors[i])
axs[0, 1].set_xlabel('问题规模(城市数量)')
axs[0, 1].set_ylabel('平均运行时间(秒)')
axs[0, 1].set_title('不同算法运行时间对比')
axs[0, 1].legend()
axs[0, 1].grid(True, alpha=0.3)
axs[0, 1].set_yscale('log') # 对数尺度,因为时间增长可能很快
# 3. 解质量分布(箱线图)
box_data = []
positions = []
pos = 1
for algo_name in algo_names:
# 取中等规模问题的结果
size = self.instance_sizes[len(self.instance_sizes)//2]
box_data.append(self.results[size][algo_name]['all_lengths'])
positions.append(pos)
pos += 1
bp = axs[1, 0].boxplot(box_data, positions=positions, patch_artist=True)
# 设置箱线图颜色
for patch, color in zip(bp['boxes'], colors):
patch.set_facecolor(color)
patch.set_alpha(0.7)
axs[1, 0].set_xticks(positions)
axs[1, 0].set_xticklabels(algo_names, rotation=45)
axs[1, 0].set_ylabel('路径长度')
axs[1, 0].set_title('解质量分布(箱线图)')
axs[1, 0].grid(True, alpha=0.3, axis='y')
# 4. 时间-质量散点图
for i, algo_name in enumerate(algo_names):
# 取所有规模的所有结果
all_lengths = []
all_times = []
for size in self.instance_sizes:
all_lengths.extend(self.results[size][algo_name]['all_lengths'])
all_times.extend(self.results[size][algo_name]['all_times'])
axs[1, 1].scatter(all_times, all_lengths, alpha=0.6,
label=algo_name, color=colors[i], s=50)
axs[1, 1].set_xlabel('运行时间(秒)')
axs[1, 1].set_ylabel('路径长度')
axs[1, 1].set_title('时间-质量权衡分析')
axs[1, 1].legend()
axs[1, 1].grid(True, alpha=0.3)
plt.tight_layout()
return fig
4.2 运行对比实验
现在让我们运行一个实际的对比实验:
# 配置实验
experiment = ExperimentRunner(
instance_sizes=[20, 30, 40, 50], # 测试不同规模
n_trials=5, # 每个配置运行5次
random_seed=2024
)
# 定义要对比的算法配置
algorithms = {
"2-opt Only": {
"description": "纯2-opt局部搜索,作为基准"
},
"Basic VNS": {
"neighborhoods": ["2-opt", "swap"],
"shaking_types": ["double_bridge"],
"max_iterations": 50
},
"GVNS (3 neighborhoods)": {
"neighborhoods": ["2-opt", "swap", "relocate"],
"shaking_types": ["double_bridge", "random_relocate"],
"max_iterations": 50
}
}
# 运行实验
experiment.run_comparison(algorithms)
# 绘制结果
fig = experiment.plot_comparison()
plt.show()
# 输出详细结果表格
print("\n详细结果汇总:")
print("="*80)
for size in experiment.instance_sizes:
print(f"\n问题规模: {size}个城市")
print("-"*40)
print(f"{'算法':<25} {'平均长度':<12} {'标准差':<10} {'平均时间(秒)':<15}")
print("-"*40)
for algo_name in algorithms.keys():
result = experiment.results[size][algo_name]
print(f"{algo_name:<25} {result['mean_length']:<12.2f} "
f"{result['std_length']:<10.2f} {result['mean_time']:<15.2f}")
4.3 邻域结构性能分析
除了整体算法对比,我们还需要深入分析不同邻域结构的具体表现。让我们设计一个专门测试邻域结构的实验:
def analyze_neighborhood_performance(instance: TSPInstance,
initial_tour: List[int],
neighborhoods: List[str]):
"""分析不同邻域结构的性能"""
results = {}
for neighborhood in neighborhoods:
print(f"\n测试邻域: {neighborhood}")
# 复制初始解
current_tour = initial_tour.copy()
current_length = instance.tour_length(current_tour)
improvements = []
lengths_history = [current_length]
# 运行固定次数的局部搜索
for iteration in range(100):
new_tour, new_length = two_opt_neighborhood(
current_tour, instance, first_improvement=True
) if neighborhood == "2-opt" else swap_neighborhood(
current_tour, instance, first_improvement=True
)
improvement = current_length - new_length
if improvement > 1e-10:
current_tour, current_length = new_tour, new_length
improvements.append(improvement)
lengths_history.append(current_length)
# 计算统计指标
results[neighborhood] = {
'final_length': current_length,
'total_improvement': instance.tour_length(initial_tour) - current_length,
'num_improvements': len(improvements),
'avg_improvement': np.mean(improvements) if improvements else 0,
'improvement_rates': improvements,
'lengths_history': lengths_history
}
print(f" 最终长度: {current_length:.2f}")
print(f" 总改进量: {results[neighborhood]['total_improvement']:.2f}")
print(f" 改进次数: {len(improvements)}")
# 可视化对比
fig, axs = plt.subplots(2, 2, figsize=(14, 10))
# 1. 收敛曲线对比
for neighborhood in neighborhoods:
axs[0, 0].plot(results[neighborhood]['lengths_history'],
label=neighborhood, linewidth=2)
axs[0, 0].set_xlabel('迭代次数')
axs[0, 0].set_ylabel('路径长度')
axs[0, 0].set_title('不同邻域收敛速度对比')
axs[0, 0].legend()
axs[0, 0].grid(True, alpha=0.3)
# 2. 总改进量对比
total_improvements = [results[n]['total_improvement'] for n in neighborhoods]
bars = axs[0, 1].bar(neighborhoods, total_improvements,
color=['skyblue', 'lightgreen', 'salmon'])
axs[0, 1].set_ylabel('总改进量')
axs[0, 1].set_title('不同邻域总改进量对比')
axs[0, 1].grid(True, alpha=0.3, axis='y')
# 在柱子上添加数值标签
for bar, val in zip(bars, total_improvements):
height = bar.get_height()
axs[0, 1].text(bar.get_x() + bar.get_width()/2., height + 0.1,
f'{val:.1f}', ha='center', va='bottom')
# 3. 改进次数对比
num_improvements = [results[n]['num_improvements'] for n in neighborhoods]
bars = axs[1, 0].bar(neighborhoods, num_improvements,
color=['lightblue', 'lightgreen', 'pink'])
axs[1, 0].set_ylabel('改进次数')
axs[1, 0].set_title('不同邻域改进频率对比')
axs[1, 0].grid(True, alpha=0.3, axis='y')
for bar, val in zip(bars, num_improvements):
height = bar.get_height()
axs[1, 0].text(bar.get_x() + bar.get_width()/2., height + 0.5,
f'{val}', ha='center', va='bottom')
# 4. 平均改进量对比
avg_improvements = [results[n]['avg_improvement'] for n in neighborhoods]
bars = axs[1, 1].bar(neighborhoods, avg_improvements,
color=['cornflowerblue', 'limegreen', 'tomato'])
axs[1, 1].set_ylabel('平均改进量')
axs[1, 1].set_title('不同邻域平均改进幅度对比')
axs[1, 1].grid(True, alpha=0.3, axis='y')
for bar, val in zip(bars, avg_improvements):
height = bar.get_height()
axs[1, 1].text(bar.get_x() + bar.get_width()/2., height + 0.1,
f'{val:.2f}', ha='center', va='bottom')
plt.tight_layout()
return fig, results
5. 高级技巧与实战建议
在实现和使用了基本的VNS框架后,我想分享一些在实际项目中积累的经验和技巧。这些建议能帮助你更好地应用VNS解决实际问题。
5.1 邻域结构的设计原则
设计好的邻域结构是VNS成功的关键。根据我的经验,有效的邻域结构应该满足以下几个原则:
-
多样性原则:不同的邻域结构应该具有不同的搜索特性。例如:
- 有的邻域擅长局部微调(如swap)
- 有的邻域适合中等范围的改进(如2-opt)
- 有的邻域能够实现大幅度的结构调整(如双桥扰动)
-
渐进性原则:邻域结构应该按照搜索强度从小到大排列。VNS的核心思想就是从弱扰动开始,逐步增加扰动强度。一个典型的序列可能是:
弱扰动 → 中等扰动 → 强扰动 swap → 2-opt → relocate → 双桥扰动 -
互补性原则:不同的邻域结构应该能够互补彼此的不足。比如,swap邻域可能在某些情况下陷入停滞,而2-opt邻域可能正好能打破这种停滞。
5.2 参数调优策略
VNS的参数相对较少,但合理的参数设置仍然很重要:
class VNSParameterTuner:
"""VNS参数调优工具"""
@staticmethod
def suggest_parameters(problem_size: int, time_budget: float):
"""根据问题规模和时间预算推荐参数"""
base_params = {
'max_iterations': 100,
'k_max': 5,
'max_no_improve': 20
}
# 根据问题规模调整
if problem_size < 30:
suggestions = {
**base_params,
'max_iterations': 50,
'neighborhoods': ['swap', '2-opt'],
'shaking_types': ['double_bridge']
}
elif problem_size < 100:
suggestions = {
**base_params,
'max_iterations': 100,
'neighborhoods': ['2-opt', 'swap', 'relocate'],
'shaking_types': ['double_bridge', 'random_relocate']
}
else: # 大规模问题
suggestions = {
**base_params,
'max_iterations': 200,
'k_max': 3, # 减少扰动强度以加快速度
'neighborhoods': ['2-opt'], # 只使用最有效的邻域
'shaking_types': ['double_bridge'],
'max_no_improve': 10 # 更早终止
}
# 根据时间预算调整迭代次数
if time_budget < 10: # 10秒以内
suggestions['max_iterations'] = min(suggestions['max_iterations'], 30)
elif time_budget > 300: # 5分钟以上
suggestions['max_iterations'] = suggestions['max_iterations'] * 3
return suggestions
@staticmethod
def adaptive_parameter_adjustment(current_performance: dict,
iteration: int):
"""根据当前性能自适应调整参数"""
adjustments = {}
# 如果最近改进很少,增加扰动强度
if current_performance.get('improvements_last_10', 0) < 2:
adjustments['k_max'] = min(
current_performance.get('k_max', 3) + 1, 5
)
# 如果改进频率很高,减少扰动强度以加快收敛
elif current_performance.get('improvements_last_10', 0) > 8:
adjustments['k_max'] = max(
current_performance.get('k_max', 3) - 1, 1
)
# 随着迭代进行,逐渐减少最大迭代次数
if iteration > 50:
remaining_iterations = current_performance.get('max_iterations', 100) - iteration
if remaining_iterations > 50:
adjustments['max_no_improve'] = min(
current_performance.get('max_no_improve', 20),
10 # 后期更早终止
)
return adjustments
5.3 处理大规模问题的技巧
当问题规模很大时(比如超过500个城市),基本的VNS实现可能会遇到性能瓶颈。以下是一些优化技巧:
- 增量计算:避免重复计算整个路径的长度。只计算受操作影响的部分:
def incremental_two_opt_delta(tour: List[int], i: int, j: int,
instance: TSPInstance) -> float:
"""增量计算2-opt操作的长度变化"""
n = len(tour)
# 获取相关节点
a = tour[i]
b = tour[(i+1)%n]
c = tour[j]
d = tour[(j+1)%n]
# 计算变化量
old_edges = instance.distance_matrix[a][b] + instance.distance_matrix[c][d]
new_edges = instance.distance_matrix[a][c] + instance.distance_matrix[b][d]
return new_edges - old_edges
- 候选列表策略:不是检查所有可能的邻域移动,而是只检查最有希望的移动:
def generate_candidate_moves(tour: List[int], instance: TSPInstance,
n_candidates: int = 50):
"""生成候选移动列表"""
n = len(tour)
candidates = []
# 基于距离启发式选择候选
for i in range(n):
current_city = tour[i]
# 找到距离当前城市最近的一些城市
distances = instance.distance_matrix[current_city]
nearest_indices = np.argsort(distances)[:n_candidates//n + 1]
for idx in nearest_indices:
if idx != current_city:
# 找到idx在tour中的位置
j = tour.index(idx)
if abs(i - j) > 1: # 避免相邻城市的无效交换
delta = incremental_two_opt_delta(tour, i, j, instance)
candidates.append((i, j, delta))
# 按改进潜力排序
candidates.sort(key=lambda x: x[2])
return candidates[:n_candidates]
- 并行化处理:利用多核CPU并行评估多个邻域移动:
from concurrent.futures import ProcessPoolExecutor
import multiprocessing
def parallel_neighborhood_evaluation(tour: List[int], instance: TSPInstance,
n_workers: int = None):
"""并行评估邻域移动"""
if n_workers is None:
n_workers = multiprocessing.cpu_count()
n = len(tour)
moves = []
# 生成要评估的移动
for i in range(n):
for j in range(i+2, n + (i if i>0 else 0)):
j_mod = j % n
if j_mod != (i+1) % n:
moves.append((i, j_mod))
# 并行计算
def evaluate_move(args):
i, j = args
delta = incremental_two_opt_delta(tour, i, j, instance)
return i, j, delta
with ProcessPoolExecutor(max_workers=n_workers) as executor:
results = list(executor.map(evaluate_move, moves))
# 找到最佳移动
best_move = min(results, key=lambda x: x[2])
return best_move
5.4 与其他启发式算法的结合
VNS的一个强大之处在于它能很容易地与其他启发式算法结合。这里给出一个与模拟退火结合的示例:
class HybridVNS_SA:
"""VNS与模拟退火混合算法"""
def __init__(self, instance: TSPInstance,
initial_temperature: float = 1000,
cooling_rate: float = 0.995):
self.instance = instance
self.vns = VariableNeighborhoodSearch(instance)
self.T = initial_temperature
self.cooling_rate = cooling_rate
self.min_temperature = 1e-3
def simulated_annealing_acceptance(self, current_length: float,
new_length: float) -> bool:
"""模拟退火接受准则"""
if new_length < current_length:
return True
# 以一定概率接受劣解
delta = new_length - current_length
probability = np.exp(-delta / self.T)
return np.random.random() < probability
def hybrid_search(self, initial_tour: List[int],
max_iterations: int = 500):
"""混合搜索主循环"""
current_tour = initial_tour.copy()
current_length = self.instance.tour_length(current_tour)
best_tour = current_tour.copy()
best_length = current_length
history = []
for iteration in range(max_iterations):
# VNS阶段:使用当前温度控制扰动强度
k_max = max(1, int(3 * (self.T / 1000))) # 温度高时扰动强
# 执行一次VNS迭代
k = 1
while k <= k_max:
# Shaking
shaken_tour = self.vns.shaking_op.shaking(
current_tour, "double_bridge", strength=k
)
# 局部搜索
candidate_tour, candidate_length = self.vns._local_search(
shaken_tour, "2-opt", first_improvement=True
)
# 模拟退火接受准则
if self.simulated_annealing_acceptance(current_length, candidate_length):
current_tour, current_length = candidate_tour, candidate_length
k = 1 # 回到最小邻域
# 更新全局最优
if candidate_length < best_length:
best_tour, best_length = candidate_tour, candidate_length
else:
k += 1
# 降温
self.T *= self.cooling_rate
# 记录历史
history.append({
'iteration': iteration,
'temperature': self.T,
'current_length': current_length,
'best_length': best_length,
'k_max_used': k_max
})
# 终止条件
if self.T < self.min_temperature:
break
return best_tour, best_length, history
5.5 实际项目中的调试技巧
在将VNS应用到实际项目时,我总结了一些调试技巧:
-
可视化调试:不要只依赖数值结果,一定要可视化查看搜索过程。有时候算法看似收敛了,但可视化后可能发现路径有明显的交叉。
-
多样性检查:定期检查解的多样性。如果所有运行都收敛到相同的解,可能说明算法缺乏探索能力:
def check_solution_diversity(solutions: List[List[int]],
instance: TSPInstance) -> dict:
"""检查解集的多样性"""
n_solutions = len(solutions)
# 计算两两之间的差异
differences = []
for i in range(n_solutions):
for j in range(i+1, n_solutions):
# 计算解之间的汉明距离(归一化)
diff = sum(1 for a, b in zip(solutions[i], solutions[j]) if a != b)
differences.append(diff / len(solutions[i]))
return {
'mean_difference': np.mean(differences),
'std_difference': np.std(differences),
'min_difference': np.min(differences),
'max_difference': np.max(differences)
}
- 性能剖面分析:使用性能剖面来评估算法在不同问题实例上的表现:
def performance_profiles(algorithms_results: dict,
baseline_algo: str = "2-opt Only"):
"""生成性能剖面图"""
fig, ax = plt.subplots(figsize=(10, 6))
for algo_name, results in algorithms_results.items():
# 计算相对于基准算法的性能比率
ratios = []
for instance_key in results.keys():
if instance_key in algorithms_results[baseline_algo]:
ratio = (results[instance_key]['best_length'] /
algorithms_results[baseline_algo][instance_key]['best_length'])
ratios.append(ratio)
# 排序并计算累积分布
ratios.sort()
n = len(ratios)
y = [i/n for i in range(n)]
ax.plot(ratios, y, label=algo_name, linewidth=2)
ax.set_xlabel('性能比率(相对于基准)')
ax.set_ylabel('累积分布')
ax.set_title('算法性能剖面')
ax.legend()
ax.grid(True, alpha=0.3)
ax.set_xlim(0.9, 1.2) # 通常关注1附近的区域
return fig
通过这些高级技巧和实战建议,你应该能够更有效地应用VNS解决实际问题。记住,启发式算法的魅力在于它的灵活性和适应性——不要害怕根据具体问题调整算法的各个组件。有时候,一个针对特定问题设计的简单邻域结构,可能比通用的复杂结构更有效。
更多推荐


所有评论(0)