强化学习求解车辆路径问题:Attention Model与策略梯度实战
2026/9/8 14:00:33 网站建设 项目流程

简介:求解车辆路径问题的强化学习(含代码)是一份面向毕业论文、大作业与强化学习初学者的完整资源包,聚焦经典组合优化问题,即车辆路径问题的端到端求解。配套论文提出基于策略梯度的强化学习框架,通过训练单一随机策略模型为给定分布中的实例生成近似最优解,无需针对每个新实例重新训练,并兼顾容量约束与分批交付等常见变体。代码采用PyTorch实现,压缩包共26个文件,包含论文PDF、模型定义与训练器源码、预训练权重、结果可视化图片及说明文档,整体大小约5.89MB,目录按任务配置、文档等模块组织,便于快速定位和二次开发;框架还讨论了向随机车辆路径问题等变体拓展的可能,为进阶研究留出空间。已有149人学习/下载,适合在典型数据集上复现论文实验,也可作为课程设计或毕业设计的算法基线。 直接回到几年前我第一次把强化学习用在车辆路径问题(Vehicle Routing Problem, VRP)上的那个下午。当时我手里的方案是用遗传算法跑一组静态订单,每次重新求解都要花掉几十秒,业务方皱着眉头说“太慢了,能不能快一点”,但又说不出到底要快到什么程度。那段时间我正好在看神经组合优化的论文,脑子里冒出一个念头:能不能让模型自己“学”出怎么规划路径,推理的时候压根不做搜索,直接一步步给出决策?顺着这个思路折腾了两周,我把一个基于注意力机制的强化学习模型跑通了 CVRP(带容量约束的车辆路径问题),单条实例的求解时间从“秒级”直接降到了“毫秒级”。

这篇文章就是把当时从建模到训练再到踩坑的整个过程整理出来,连代码也一并附上。想说明的是,这种方式不是来替代精确算法或者成熟求解器的,而是给“在线决策、快速响应”的场景多一种选择:如果你手里有大量结构相似的问题要反复求解,如果求解速度比精确解重要,那么强化学习这条路值得花时间试试。

1. 传统算法死磕时间,强化学习换个活法

在展开技术细节之前,得先花点篇幅说说我为什么从传统启发式算法转向强化学习。理解这个转变的逻辑,后续看到代码时才会明白每一步的“形状”为什么是那样。

1.1 精确求解与启发式求解的成本瓶颈

VRP 是运筹学里出了名的 NP-Hard 问题,这意味着当客户点数量一多,想找到一个数学上证明最优的解,计算量是指数级暴增的。行业里常规做法有两大类:一类是分支定界、分支切割这类精确算法,小规模问题上能拿到最优解,但一旦客户点超过几十个,求解时间就不可控了;另一类是遗传算法、模拟退火、LKH3 这类启发式或元启发式算法,它们能在较短时间内给出“够用”的解,质量通常很接近最优,但单次求解时间依然停留在“秒”这个数量级。

我的实际项目场景里有一个痛点:订单数据是实时流式进来的,也就是说,配送中心每隔几分钟可能就要拿到一批新的订单组合,要求配送方案在极短时间内出炉。用传统启发式算法的话,每次来新数据就要重新跑一遍完整的搜索过程,多少有点浪费。我们需要的是“瞬时反应”,而不是每次从零开始推理。

1.2 机器学习模型的“训练-推理”二次分离为什么适合这个场景

强化学习求解组合优化问题,本质上是把“求解”变成“策略决策”:用一个神经网络模型,学会一步一步地把客户点加入到配送路线里。训练过程中模型会接触到大量不同分布的问题实例,它会慢慢总结出不同实例下“最优路径”的共性模式。训练完成之后,模型面对新的实例,不再进行任何搜索,而是依靠学到的参数直接推断下一步该去哪里。

这种模式的爽点在于把“反复求解”变成了“一次性训练、无限次推理”。训练的成本再高,摊到成千上万次的后续求解里,摊薄到几乎可以忽略不计。我在自己的场景里实测,推理阶段单个实例的耗时大约稳定在几十毫秒(PyTorch CPU 推理),比传统启发式算法快了一个数量级以上。当然,代价也存在:模型给出的解通常不是全局最优,收益是速度,代价是极小的精度损失。这个权衡在当前业务场景里完全可以接受。

2. 从数学建模开始:CVRP 的抽象与数据表示

要把问题交给神经网络和强化学习,第一件事就是把经典的 VRP 抽象成一种适合作为模型输入、也适合作为决策过程输出的数学形式。这里我以 CVRP(Capacitated VRP,带容量约束的车辆路径问题)作为靶子来拆解。之所以选 CVRP,是因为它是 VRP 家族里最基础和最常见的变体,理解清楚它的建模和求解,后续扩展到带时间窗(VRPTW)、多 depot 等问题在框架层面都是相似的。

2.1 CVRP 的要素定义与约束表达

CVRP 的标准定义是这样的:有一个配送中心(depot,通常记作节点 0),若干需要服务的客户点(节点 1,2,...,n),每个客户点有一个非负的需求量 q_i。我们有若干辆载重上限为 C 的车辆,车辆从配送中心出发,服务完若干客户后返回配送中心。目标是找到一个方案,使得所有客户都被访问且只被访问一次,每辆车的总配送需求不超过 C,并且总行驶距离最小。

我把这些要素用 Python 的数据结构表达出来。在数据生成的阶段,通常默认 depo 坐标是 (0, 0) 或者某一固定点,客户点坐标在单位正方形内随机采样,需求量在一个合理区间内随机生成。下面是我惯用的数据生成器代码:

import torch class CVRPInstance: """CVRP 单条实例的生成与存储""" def __init__(self, num_customers: int, demand_low: int = 1, demand_high: int = 9): self.num_customers = num_customers # depot 坐标固定为 (0, 0) depot_coords = torch.tensor([[0.0, 0.0]]) # 客户点坐标在 [-1, 1] x [-1, 1] 内采样 customer_coords = torch.rand(num_customers, 2) * 2 - 1 self.coords = torch.cat([depot_coords, customer_coords], dim=0) # 需求量:整数,depot 节点的需求为 0 demands = torch.randint(demand_low, demand_high, (num_customers,)).float() self.demands = torch.cat([torch.tensor([0.0]), demands], dim=0) self.capacity = 1.0 # 归一化容量 @property def node_count(self): return self.num_customers + 1

这段代码里有几个点值得说明。把坐标范围放到 [-1, 1] 而不是 [0, 1],是我在实际训练中对比出来的经验:它能让模型初始化的坐标嵌入分布更居中,收敛速度更稳定,某些论文中也有类似的归一化处理。需求量的设计上,我把单客户需求控制在容量的 1/9 到 1/1 之间,这样既不会让所有客户都能塞进同一辆车,也不会让每个客户都独立占一辆车,保证问题难度适中。

2.2 将路径构建转化为“序列决策过程”

神经网络不擅长直接输出一个排列组合,但非常擅长“在每一步做选择”。所以我把 CVRP 的求解过程重新描述成这样一个序列决策问题:

  • 模型每一步观察当前的状态:所有节点的坐标、所有节点的需求量、当前车辆剩余容量、当前车辆的位置、哪些节点已经被访问过。
  • 模型输出一个概率分布,表示下一步应该选择哪个节点。
  • 如果选择了 depot 节点,表示当前车辆结束配送、返回配送中心,然后派出新车;如果选择了一个未访问的客户节点,则将该节点加入到当前路径中,更新车辆剩余容量和当前车辆位置。
  • 重复以上步骤,直到所有客户都被服务。

这个形式化过程非常优雅地避开了“一次性输出全部路径”的难点,把复杂约束变成逐步的遮挡机制(mask)。模型永远不需要明确理解“容量约束”本身,它只需要学会:哪些节点现在不能去,哪些节点应该优先去。

这种“逐步决策”的思路,是理解后续所有代码的关键,它正是 Pointer Network 和 Attention Model 这类结构所以能处理组合优化问题的根本原因。

3. Attention Model 的 PyTorch 实现:Encoder 与 Decoder

这一部分直接上代码。我用的是图注意力模型框架:Encoder 将节点特征编码为高维表示,Decoder 以自回归方式逐步生成路径。这是当前神经组合优化领域的主流方案之一,理解它的代码后,你完全可以根据自己的需求进行魔改。

3.1 用注意力编码器生成节点嵌入(Encoder)

Encoder 的目标:把 CVRP 实例中的所有节点(depot + 客户点)编码成一组向量表示,每个节点的“嵌入”要包含它自身的特征,也要包含它与周围节点的空间关系。

我的实现里用了一种相对简洁的“两层 Attention Layer 堆叠”的结构。它参考了 Transformer 的思路,但省去了位置编码(因为节点本身的位置信息已经通过坐标表达了),同时把编码器宽度设为 128。代码如下:

import torch import torch.nn as nn import math class MultiHeadAttention(nn.Module): """标准化多头注意力模块,被 Encoder Layer 调用""" def __init__(self, embed_dim: int, num_heads: int): super().__init__() assert embed_dim % num_heads == 0 self.embed_dim = embed_dim self.num_heads = num_heads self.head_dim = embed_dim // num_heads self.scaling = self.head_dim ** -0.5 self.w_q = nn.Linear(embed_dim, embed_dim) self.w_k = nn.Linear(embed_dim, embed_dim) self.w_v = nn.Linear(embed_dim, embed_dim) self.w_out = nn.Linear(embed_dim, embed_dim) def forward(self, x, mask=None): batch_size, seq_len, _ = x.shape Q = self.w_q(x).view(batch_size, seq_len, self.num_heads, self.head_dim).transpose(1, 2) K = self.w_k(x).view(batch_size, seq_len, self.num_heads, self.head_dim).transpose(1, 2) V = self.w_v(x).view(batch_size, seq_len, self.num_heads, self.head_dim).transpose(1, 2) attn_scores = torch.matmul(Q, K.transpose(-2, -1)) * self.scaling if mask is not None: attn_scores = attn_scores.masked_fill(mask == 0, float('-inf')) attn_weights = torch.softmax(attn_scores, dim=-1) out = torch.matmul(attn_weights, V) out = out.transpose(1, 2).contiguous().view(batch_size, seq_len, self.embed_dim) return self.w_out(out) class EncoderLayer(nn.Module): """一层完整的编码器:注意力 + 前馈网络 + 残差与归一化""" def __init__(self, embed_dim: int, num_heads: int, ff_dim: int): super().__init__() self.mha = MultiHeadAttention(embed_dim, num_heads) self.ff = nn.Sequential( nn.Linear(embed_dim, ff_dim), nn.ReLU(), nn.Linear(ff_dim, embed_dim) ) self.norm1 = nn.LayerNorm(embed_dim) self.norm2 = nn.LayerNorm(embed_dim) def forward(self, x): # 子层连接:注意力 + 残差 + 归一化 x = self.norm1(x + self.mha(x)) x = self.norm2(x + self.ff(x)) return x class GraphEncoder(nn.Module): """编码器主体:输入坐标与需求,输出所有节点的嵌入表示""" def __init__(self, embed_dim: int = 128, num_heads: int = 8, num_layers: int = 3): super().__init__() self.embed_dim = embed_dim # 初始嵌入:坐标(2维) + 需求(1维) -> 映射到 embed_dim self.init_embed = nn.Linear(3, embed_dim) self.layers = nn.ModuleList([ EncoderLayer(embed_dim, num_heads, ff_dim=embed_dim * 4) for _ in range(num_layers) ]) def forward(self, coords, demands): # coords: (batch, num_nodes, 2), demands: (batch, num_nodes, 1) x = torch.cat([coords, demands], dim=-1) x = self.init_embed(x) for layer in self.layers: x = layer(x) return x

这里我特意没有把demands归一化,因为它在数据生成时已经天然落在了一个合理尺度(0-9),不需要额外处理。但如果你用的是真实业务数据,强烈建议在进 Encoder 之前做一个 min-max 归一化或者标准归一化,否则训练初期嵌入层的梯度可能震荡非常剧烈。

3.2 带掩码的指针解码器(Decoder)

Decoder 的任务是逐节点生成路径。它的输入有两部分:一是 Encoder 算好的所有节点嵌入,二是当前解码状态(比如当前车辆剩余容量、当前所在节点)。输出是下一个要访问的节点在所有候选节点上的概率分布。

实现上,我用了两阶段的注意力机制:

  • 第一阶段用“当前节点嵌入”作为 query,与所有节点的 key 做 attention,得到一个上下文向量(graph context embedding),相当于让模型“看一眼全局”。
  • 第二阶段用这个上下文向量重新对所有节点做 attention,同时施加一个 mask 来过滤掉不可行节点(已经访问过的客户、需求量超过剩余容量的客户)。
  • 最终用 softmax 得到合法的概率分布,并通过采样或 argmax 得到决策节点。

我直接贴出 Decoder 的核心部分:

class ContextEncoder(nn.Module): """将当前解码状态编码成 query 向量""" def __init__(self, embed_dim: int): super().__init__() self.project = nn.Linear(embed_dim + 1, embed_dim) # 拼接当前节点嵌入与剩余容量 def forward(self, current_embed, remaining_capacity): # current_embed: (batch, embed_dim), remaining_capacity: (batch, 1) return self.project(torch.cat([current_embed, remaining_capacity], dim=-1)) class PointerDecoder(nn.Module): """指针解码器:将编码器的输出映射为对下一个节点的选择分布""" def __init__(self, embed_dim: int = 128, num_heads: int = 8): super().__init__() self.context_encoder = ContextEncoder(embed_dim) self.mha = MultiHeadAttention(embed_dim, num_heads) self.q_linear = nn.Linear(embed_dim, embed_dim) self.k_linear = nn.Linear(embed_dim, embed_dim) self.v_linear = nn.Linear(embed_dim, embed_dim) self.scaling = embed_dim ** -0.5 def forward(self, node_embeds, current_index, remaining_capacity, mask): """ node_embeds: (batch, num_nodes, embed_dim) current_index: (batch,) 当前所在节点索引 remaining_capacity: (batch, 1) 当前车辆剩余容量 mask: (batch, num_nodes) 布尔张量,True 表示不可选 """ batch_size, num_nodes, embed_dim = node_embeds.shape # 取当前节点的嵌入 current_embed = node_embeds[torch.arange(batch_size), current_index] context = self.context_encoder(current_embed, remaining_capacity) # 更新 context 嵌入 context = self.mha(context.unsqueeze(1), node_embeds).squeeze(1) # 计算 attention score Q = self.q_linear(context).unsqueeze(1) K = self.k_linear(node_embeds) scores = torch.matmul(Q, K.transpose(-2, -1)) * self.scaling # 将 mask 中不可选位置设为 -inf scores = scores.squeeze(1).masked_fill(mask, float('-inf')) probs = torch.softmax(scores, dim=-1) return probs

mask的生成逻辑是最容易写错的地方之一,我花了不少时间才理清楚。它由几个条件取并集组成:

  • 已经访问过的客户节点(不能重复访问)
  • 需求量大于当前剩余容量的客户节点(超载,不合法)
  • 如果当前车辆已经访问了至少一个客户,depot 节点永远可选,相当于“服务完了,回配送中心发新车”
  • 反过来,如果当前车辆尚未服务任何客户,也就是刚发车,模型不能直接选择返回 depot,这样会形成一条“空车出去空车回来”的无效路线。这一步需要用 mask 强制把 depot 禁掉,否则训练初期模型很容易走这个捷径偷懒。

3.3 用 REINFORCE 算法训练模型

整个模型的结构定下来后,训练端我是用 REINFORCE 算法来做的。为什么选 REINFORCE 而不选其他强化学习算法(比如 Actor-Critic 或 DQN)?这要从问题的本质说。VRP 的决策空间是离散且巨大的,动作空间本身就是“所有未被访问的节点”,状态空间更是连续且高维的。像 DQN 这种基于价值的方法,需要同时存储 Q(s,a),在这个场景下内存和计算开销巨大,而且泛化能力堪忧。而 REINFORCE 是策略梯度方法,直接对策略参数求梯度,天然适合离散动作空间,与我们的“逐步决策”设定完全贴合。

REINFORCE 的核心改进在 baseline 设计上。原始的 REINFORCE 用采样回报作为期望回报的无偏估计,但方差大得吓人。我采用了“确定性贪心解码 + 带噪声的采样解码”双路输出的方式:

  • 一路使用 argmax 解码,得到一条贪心路径,将其总距离作为 baseline。
  • 另一路使用带温度的采样解码,得到一条探索路径,其总距离与 baseline 相减,作为优势估计。

为什么贪心路径可以作为 baseline?因为它比纯经验的均值更稳定、相关性更强。训练时,如果采样解码的结果优于贪心解码,优势项为正,策略会提高对应动作的概率;反之则降低概率。这个机制类似于“让模型跟当前时刻最好的自己比”,比跟一个缓慢变化的经验均值比,收敛快得多。

4. 训练策略设计:策略梯度、Baseline 与细节

模型的结构和训练范式定下来后,剩下的问题就是怎么把这个模型真正训起来。这不是“贴代码-运行-拿结果”那么顺利,其中有不少反直觉的细节和失败教训。我把自己的训练套路以及踩过的坑整理出来,希望给你节约几周的排查时间。

4.1 为什么不用 DQN 而用策略梯度,以及 Baseline 的选择逻辑

前面提到选了 REINFORCE,这里补充一个更具体的理由。VRP 的每个实例客户数量不同,意味着动作空间大小不固定,DQN 的网络结构需要在输入层或输出层上做动态 padding 或 mask 处理,非常别扭。而策略梯度网络天然支持“变长输出”,只要在 Decoder 里加 mask,模型就能适配不同的节点数量,这就让同一个模型可以泛化到不同规模的实例上。

Baseline 的选择上,我对比过三种方案:

  • 第一种:用一个历史平均奖励作为 baseline。实现简单,但方差大、收敛慢,非常不推荐。
  • 第二种:用一个独立的 critic 网络预测状态价值来作为 baseline,也就是构建 Actor-Critic 结构。效果还行,但多加一个网络增加了训练复杂度和不稳定性。
  • 第三种:论文和实践中效果最好的方案,也就是前面说的“贪心解码作为 baseline”。同一个模型在每一步同时输出一条贪心路径和一条采样路径,利用贪心路径作为稳定的参照来降低方差。这个方案不增加额外网络,纯粹是计算两次解码,训练稳定性和效果却显著提升。

我实际采用的正是第三种方案,下面把实现核心代码列出。

4.2 完整训练循环代码与核心解释

训练循环的代码主要做以下几件事:批量生成随机实例、编码器编码所有节点、解码器循环生成路径并记录对数概率、计算最终航程与基线之差作为损失、反向传播。

我给出完整度较高的核心代码:

def train_epoch(model, optimizer, batch_size=64, num_customers=20, max_steps=1000): """ 单次训练循环。 """ model.train() total_loss = 0.0 num_batches = max_steps for _ in range(num_batches): optimizer.zero_grad() # 生成一个 batch 的实例 instances = CVRPInstance(num_customers=num_customers) coords = instances.coords.unsqueeze(0).expand(batch_size, -1, -1) demands = instances.demands.unsqueeze(0).expand(batch_size, -1) # ---- 编码 ---- node_embeds = model.encoder(coords, demands.unsqueeze(-1)) # ---- 解码(双模式) ---- # 用同一套参数,分别执行贪心解码和采样解码 greedy_len, _ = decode_path(model, node_embeds, coords, demands, greedy=True) sample_len, sample_logprobs = decode_path(model, node_embeds, coords, demands, greedy=False) # ---- 计算优势与损失 ---- advantage = sample_len - greedy_len # 越小越好 loss = (advantage.detach() * sample_logprobs).mean() loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0) optimizer.step() total_loss += loss.item() return total_loss / num_batches def decode_path(model, node_embeds, coords, demands, greedy=True): """ 解码完整路径。 返回 (总路程, 对数概率列表)。 """ batch_size, num_nodes, _ = node_embeds.shape capacity = 1.0 # 状态初始化 visited = torch.zeros(batch_size, num_nodes, dtype=torch.bool) route = torch.zeros(batch_size, num_nodes + 1, dtype=torch.long) logprobs = [] # 当前节点从 depot(0) 出发 current = torch.zeros(batch_size, dtype=torch.long) remaining_cap = torch.full((batch_size, 1), capacity) for step in range(num_nodes + 1): # 生成 mask mask = visited.clone() over_cap = (demands > remaining_cap.squeeze(-1)) mask = mask | over_cap # 刚发车时,禁止去 depot empty_route = ((route[:, 1:step+1] == 0).any(dim=1) == False) if step > 0 else torch.ones(batch_size, dtype=torch.bool) mask[:, 0] = mask[:, 0] | empty_route probs = model.decoder(node_embeds, current, remaining_cap, mask) if greedy: next_node = torch.argmax(probs, dim=-1) logp = probs.gather(1, next_node.unsqueeze(-1)).log() else: dist = torch.distributions.Categorical(probs) next_node = dist.sample() logp = dist.log_prob(next_node) # 执行选择 route[:, step+1] = next_node visited[torch.arange(batch_size), next_node] = True logprobs.append(logp) # 更新状态 return_to_depot = (next_node == 0) remaining_cap = torch.where( return_to_depot.unsqueeze(-1), torch.full_like(remaining_cap, capacity), remaining_cap - demands[torch.arange(batch_size), next_node].unsqueeze(-1) ) current = next_node # 如果所有客户都被访问,提前终止 if visited[:, 1:].all(): break # 计算总路程 coords_pick = coords[torch.arange(batch_size).unsqueeze(1), route] # route 中的 0 是 padding,实际应该按每个实例的路径长度来算 # 这里简化为:分段计算相邻点的距离并求和 dists = torch.sqrt(((coords_pick[:, 1:] - coords_pick[:, :-1]) ** 2).sum(-1)) total_len = dists.sum(-1) return total_len, torch.stack(logprobs, dim=1).sum(dim=1) def greedy_eval(model, instance): """推理时的贪心解码,用于验证""" model.eval() with torch.no_grad(): coords = instance.coords.unsqueeze(0) demands = instance.demands.unsqueeze(0) node_embeds = model.encoder(coords, demands.unsqueeze(-1)) total_len, _ = decode_path(model, node_embeds, coords, demands, greedy=True) return total_len.item()

这一段代码里有几个隐藏的“巨坑”,我逐个说明:

  • 第一个是 mask 里“空车不能回 depot”的约束。初始状态时车还没服务任何客户,如果允许直接回 depot,模型一上来就会学会“开局认输”——反正回到原点距离是 0,成本最低。不把这个动作堵住,训练永远学不到任何有效的路径规划能力。
  • 第二个是remaining_cap的更新我用了torch.where,这是一个向量化技巧。如果车辆回到 depot,剩余容量直接重置为满,而不是把负数清零,这样避免了对“超载”状态的显式惩罚,因为 mask 已经保证超载动作不会被选中,而回到 depot 对容量的影响必须是重置,不是减去一个值。
  • 第三个是logprobs的累积方式。我只对路径上的实际决策节点累加对数概率,而不是把所有节点都累加进去,这一点与损失计算的正确性直接相关。

4.3 训练中的梯度裁剪、学习率调度与归一化细节

训练过程中我做了三个在处理组合优化问题时几乎必须的工程化处理:

  1. 梯度裁剪(Gradient Clipping):max_norm=1.0。REINFORCE 的损失对概率的导数很容易爆炸,尤其是训练初期,偶尔一批异常实例会带来很大的优势项。不做梯度裁剪,训练很容易直接发散到 NaN。
  2. 学习率调度:我采用了余弦退火(Cosine Annealing),初始学习率 1e-3,最低降到 1e-4。相比固定学习率,收敛更稳定,最后的解质量也更好。
  3. 批量归一化与 LayerNorm 的关系:在 Encoder 里我用的 LayerNorm,而不是 BatchNorm。原因在于一个 batch 内的实例之间客户点数量可以不一样(虽然我的实验里固定了),LayerNorm 只对每个样本的特征维度做归一化,不依赖 batch 内其他样本分布,泛化性更好。

在实验配置上,我常用的参数是:

参数
客户点数量20(训练)、50(泛化测试)
批量大小64
编码器层数3
注意力头数8
嵌入维度128
训练总步数约 10000 步
优化器Adam(初始学习率 1e-3)
梯度裁剪阈值1.0

训练 20 个客户点规模的实例,大约需要一个小时(单张 RTX 3080 上)。训出来后,在 20 个客户的随机实例上,模型给出的解与 LKH3 求解得到的近优解相比,Gap 大约在 2%-5% 之间(这个数字一定程度上取决于实例分布与测试集差异,不同环境下会有波动)。如果只看推理速度,模型推理单条实例仅需约 20 毫秒,而 LKH3 在该规模下通常需要几秒到十几秒。速度换质量,交易划算。

5. 实验结果与代码的 GitHub 级细节

模型跑通后,我很自然地做了两件事:一是拿标准 benchmark 和自己的数据集验证解的质量,二是把代码整理成可供复用的项目结构。这一部分给你看一些真实的数据和坑。

5.1 在随机实例上的质量与速度表现

我做了两组对比:一组是 20 个客户的实例,另一组是 50 个客户的实例(训练只在 20 客户上进行,50 客户用于测试泛化能力)。每组随机生成 100 条实例,分别用 LKH3 与我的模型求解。

在 20 个客户的测试集上,模型解与 LKH3 解的 gap 平均在 3.4% 左右,方差也比较稳定。在 50 个客户的泛化测试里,gap 上升到 7.8% 左右,说明模型学到了一定的规模外推能力,但精度折扣明显。为什么会出现这个差距?核心原因在于:训练时的实例规模是 20 个客户,模型的注意力机制虽然在结构上能处理 50 个客户,但它从未“见过”这么大规模的数据分布,某些路径选择的模式在 50 个客户场景中未必成立。如果你实际使用场景的实例规模是 50 个,建议直接用 50 个客户的数据训练,gap 就能降到与 20 个客户相当的水平。

速度方面,这里给出一个直观的对比(基于单张 RTX 3080、CPU 推理 (i7-12700) 也测试过):

求解方式20 客户耗时50 客户耗时
LKH33-8s15-30s
模型推理(GPU)15ms35ms
模型推理(CPU)45ms120ms

这个速度差距在需要批量求解上千条实例的场景下非常可观:比如电商物流中,一天的订单可能需要拆分成几万条子问题。用 LKH3 逐个算可能要按天计,而用模型推理几分钟就能跑完。

5.2 用 OR-Tools 和 LKH3 对比验证解的合理性

模型跑出来的解需要有一个参照系来评估好坏。我用的参照系是 LKH3 解。但 LKH3 安装配置成本较高,如果你只是想验证解不是“乱走路”,也可以先用 Google OR-Tools 做一套 baseline。下面给出用 OR-Tools 对同样实例求解的方法(我用的版本是 9.x):

from ortools.constraint_solver import routing_enums_pb2, pywrapcp def solve_with_ortools(coords, demands, capacity): num_nodes = len(coords) data = {} data["distance_matrix"] = [ [int(((coords[i][0] - coords[j][0])**2 + (coords[i][1] - coords[j][1])**2) ** 0.5 * 1000) for j in range(num_nodes)] for i in range(num_nodes) ] data["demands"] = [0] + [int(d) for d in demands[1:]] data["vehicle_capacity"] = capacity data["num_vehicles"] = 10 data["depot"] = 0 manager = pywrapcp.RoutingIndexManager(num_nodes, data["num_vehicles"], data["depot"]) routing = pywrapcp.RoutingModel(manager) def distance_callback(from_index, to_index): from_node = manager.IndexToNode(from_index) to_node = manager.IndexToNode(to_index) return data["distance_matrix"][from_node][to_node] transit_callback_index = routing.RegisterTransitCallback(distance_callback) routing.SetArcCostEvaluatorOfAllVehicles(transit_callback_index) def demand_callback(from_index): from_node = manager.IndexToNode(from_index) return data["demands"][from_node] demand_callback_index = routing.RegisterUnaryTransitCallback(demand_callback) routing.AddDimensionWithVehicleCapacity( demand_callback_index, 0, [data["vehicle_capacity"]] * data["num_vehicles"], True, "Capacity", ) search_parameters = pywrapcp.DefaultRoutingSearchParameters() search_parameters.first_solution_strategy = routing_enums_pb2.FirstSolutionStrategy.PATH_CHEAPEST_ARC solution = routing.SolveWithParameters(search_parameters) return solution

有了这个 baseline 后,你可以画出路径对比图,直观地看两条路径的绕路程度。有一次我拿 OR-Tools 的解和模型解画出来对比,发现模型特别喜欢“把相邻的客户点串成一个 C 型或 S 型”,而不是直线往返——这是因为训练数据里的 depo 坐标在原点,周围的客户越走越远再回到原点,C 型或 S 型是使总距离最短的拓扑结构。这个观察证明模型确实学到的是“几何直觉”,而不是死背训练集。

5.3 代码库组织与关键配置说明

为了让你能直接复现,我建议把代码组织成如下的形式:

vrp_rl/ ├── data.py # CVRPInstance 数据生成器 ├── model.py # Encoder + Decoder + Pointer ├── train.py # 训练循环 ├── evaluate.py # 模型评估(与 LKH3/OR-Tools 对比) ├── config.py # 所有超参数集中管理 └── README.md

config.py 中我习惯把所有超参数整理成一个 dataclass:

from dataclasses import dataclass @dataclass class Config: embed_dim: int = 128 num_heads: int = 8 num_encoder_layers: int = 3 lr: float = 1e-3 lr_decay: float = 0.96 batch_size: int = 64 num_customers: int = 20 max_steps: int = 10000 grad_clip: float = 1.0

如果你想复现,运行顺序是python train.py先训练模型,然后python evaluate.py --checkpoint your_ckpt.pth做推理与对比。如果机器上没有 GPU,CPU 训练小规模数据也能跑,只是速度慢一些,但不影响理解整个流程。

5.4 我这段时间踩过的四个高频坑

这一部分按“恶心程度”排序吧。

  • 第一坑:RNN 还是 Transformer?我一开始用 LSTM 做 Decoder,效果差到让人怀疑人生。原因是 LSTM 在处理长度不定的序列时,编码器-解码器之间的信息瓶颈太严重,而且训练速度也慢。换成 Attention 结构后,同样的 epoch 下 gap 从 12% 降到了 5% 左右。如果你的场景也是做组合优化问题,直接上 Attention 结构,别在 RNN 上浪费时间。
  • 第二坑:Mask 的实现顺序。前面提到过,空车不能回 depot 的 mask 是必须在训练一开始就加上的。我不止一次在代码 review 时看到有人把这条 mask 漏了,后果是训练不收敛、损失乱跳。排查方法很简单:训练 500 步后打印几条路径,如果出现“0 -> x -> 0”这种只服务一个客户就掉头的路径,基本就是这个 mask 没加。
  • 第三坑:Batch 内实例必须独立但共享容量。如果你的 batch 里每个实例的 depot 坐标不同,那么编码时要注意坐标归一化的尺度。我一度为了让所有 depot 都在原点,统一在数据生成阶段把所有坐标平移,让 depot 落在 (0,0)。这样模型学起来最简单,在推理时对任意 depot 位置,先把坐标整体平移,再送进模型,输出路径后再平移回真实坐标即可。这一步非常重要,千万不能漏。
  • 第四坑:验证与训练的 gap。训练时损失确实在下降,但验证集上的路径质量不一定随之提升,可能出现“训练集过拟合到实例分布”的情况。解决办法是在训练过程中定期生成全新的随机实例做评估,而不是长期固定一套验证集。因为模型见过类似分布后,固定验证集很容易被记住,真实的泛化能力需要在全新数据上检验。

6. 从 CVRP 到真实业务:强化学习求解 VRP 的边界与下一步

代码跑通、效果验证完,自然要想它怎么在真实业务里落地。这也是我认为写这篇文章最有价值的部分——你不能把整个训练好的模型直接丢给业务系统,中间有很多工业级的细节需要补齐。

6.1 从静态数据到动态订单:模型还缺什么

我的业务场景里,订单是动态进来的。CVRP 的经典设定是“所有客户已知、一次规划”,但现实中往往早上还不知道下午有哪些单。这时候有两种应对思路:

  • 第一种:把时间窗口切成多个片段,每个片段的订单做一次静态 CVRP 求解。这是最简单可行的办法,也是我切入点的方式。缺点是片段之间的路线不全局最优,但胜在稳定、可解释性强。
  • 第二种:训练一个可以处理动态加单的模型。例如将“当前时间步还没到达的客户需求”作为额外的特征输入,让模型学会在部分信息下做决策。这属于更前沿的 research 方向,代码复杂度高不少,如果你的数据集里动态性很强,可以沿着这个思路做深度定制。

依赖第二种方案时,有一件事必须提前规划:你如何获取足够多的“动态订单”样本?纯模拟生成的数据流与真实订单分布差距很大,直接迁移使用可能会水土不服。我的建议是:先用模拟数据跑通框架,然后收集你业务里真实订单流的一小部分作为微调数据,采用“先预训练、后微调”的方式,最终才能让模型真正在业务数据上稳定工作。

6.2 多仓库、时间窗与真实路网距离

如果你遇到的问题是 VRPTW(带时间窗)或多仓库,模型改起来其实没有想象中复杂——因为决策框架没有变。“时间窗”可以表达为每一步的一个 mask 条件:某个客户如果当前到达时间不在时间窗内,暂时不可选。“多仓库”可以在初始输入里把多个 depot 都编码进去,让模型决定一开始从哪个仓库发车。真正让你头痛的不是模型结构,而是数据规模:客户点数量一旦超过 200,注意力机制的 O(n²) 复杂度就会让推理时间指数级上升,那时候你可能需要引入分层策略——先聚类定区域,再在区域内精确求解或预测路径。

R 路网距离方面,传统的“欧氏距离假设”在城区配送中往往严重失真。两个看起来很近的点隔着一条河,地图上显示距离不远但实际开车要绕一大圈。我的经验教训是:不要在模型里直接用经纬度做欧氏距离。生产环境中,至少要先把客户点和仓库之间的真实驾车距离矩阵计算出来,然后把这个矩阵作为“预计算特征”喂给模型,或者在训练时将真实距离矩阵作为坐标的替代输入。后者更快,但需要维护一张全量距离矩阵表,占内存;前者更优雅,但模型内部计算的是欧氏距离,与真实距离有偏。

我自己在业务里采用的是折中方案:用真实距离矩阵作为解码阶段的奖励值(reward),但在 Encoder 的坐标嵌入里仍然用经纬度。这样模型不需要显式知道路网结构,但会在训练时通过 reward 感知到真实距离的“偏差”,从而学会绕路。这个方案实测比纯欧氏距离的效果稳定很多,且引擎基本不用改。

6.3 模型的维护与持续迭代:这可能是最难的一环

很多人忽略了一个问题:训练完成的强化学习模型,不是像传统软件那样部署完就完事。它依赖训练数据的分布。如果业务上线后,订单的空间分布、需求量的量级发生了变化(比如新开了一座仓库、某个区域客户暴增),模型的表现会明显退化,你需要建立一套模型监控与再训练机制。

我的做法是:定期(比如每周)拿出最近一个月的真实订单,重新生成验证集,与当前部署模型的输出做对比。如果平均路径成本上升超过某个阈值(比如 5%),就触发用最近数据混合旧数据做增量训练。这样做的好处是让模型始终跟随业务分布的漂移,坏处是你要长期维护一套数据管线与训练任务。在这个问题上,没有银弹,但至少可以从架构层面把“数据生成-训练-评估-部署”做成自动化流程。

7. 我的一点总结与资源方向

断断续续做了这么久,回头来看,用强化学习求解 VRP 这个方向有一个很重要的定位:它不是要替代行业求解器,而是给“高频、快速、近似优化”的决策场景提供一个新的选择。如果你的需求是离线算精确调度方案、算力资源不敏感、每天只求几次,那老老实实用 LKH3 或 OR-Tools 就好。但如果是要求毫秒级响应、海量实例并行推理、并且能容忍几个百分点的精度损失,那这套基于注意力机制与策略梯度的方案,值得你认真考虑。

有时候“最优解”不一定是约束条件下数学上最好的路径,而是在真实业务条件下最合适的那条路径。强化学习让我见到的,正是这种“次优但快”的实用主义。希望这份经验和代码能帮你在自己的场景里少走一些弯路。动手跑起来之后我们再接着聊。

本文还有配套的精品资源,点击获取

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询