“仅到终点”导航子技能训练：PPO、奖励整形与checkpoint复用

简介：腾讯开悟-重返秘境模型（仅到终点）是一份面向深度强化学习学习者与算法实践者的模型工程包，聚焦“重返秘境”任务中智能体从起点抵达终点的策略实现，适合已具备Python与强化学习基础、希望研究DQN类算法落地细节的读者。包内共56个文件，以31个py源码、11个pyc编译文件、6个toml与2个yaml配置为主，并含1个pkl检查点、1个json及sign等元信息，压缩包约4.69MB；目录中可见target_dqn、dqn、diy等模块与ckpt、conf分层，配置侧涉及actor、learner、aisrv、client等运行角色，便于对照网络定义、训练流程、智能体实现和参数配置。该模型平均得分约800分，可作为任务性能参照，已有2594人学习下载，热度从侧面说明其在同类赛题实践中的参考价值。读者可从中了解DQN目标网络、检查点保存与配置调参的组织方式，并借助完整工程结构复盘训练与复现思路，也可将状态、动作与工作流模块作为拆分学习入口，梳理从环境交互到策略更新的链路。

1. 一个只会「走到终点」的模型，为什么反而值得先拆

把重返秘境模型往评测环境里一丢，回放画面往往挺滑稽：它能贴着墙根绕过障碍、从窄门里侧身挤过去，稳稳踩到终点格上，然后就停在那儿不动了。奖励曲线在最后三分之一几乎是平的，因为它根本没学终点之后那段交互。这版「仅到终点」的 checkpoint 就是干这个的：它并非能通关的完整策略，它的价值在于把「导航到终点」这段子技能练得足够稳，属于深度学习里典型的分段训练产物。适合三类人：想在多智能体赛题环境里先跑通一条完整链路的、想把导航段拆出来做技能复用的、以及想把遥测和评测流程摸清楚的。下面按我拆包时的顺序，从工程骨架讲到怎么改。

2. 工程骨架：从平台环境到本地训练进程的三层结构

先把整套东西的分层想清楚，后面改代码才不会乱。这类赛题工程基本都是三层：最底下是平台侧的环境进程，负责推进物理、渲染和规则判定；中间是本地跑的训练进程，负责收观测、出动作、攒样本、更参数；最上面是模型产物，也就是save_model吐出来的权重加上一堆与之绑定的配置。三层之间只有两个契约：观测的字段结构，和动作的编码格式。这两个契约一旦对不上，后面所有调参都是玄学。

2.1 目录逐项过一遍：谁管配置、谁管算法

拿到包先别急着pip install，把目录摊开看一眼，能省掉后面一半的排查时间。这类工程的目录命名高度一致，大致长这样：

kaiwu-mijing/ ├── agent.py # 智能体入口：reset / predict / learn / save_model / load_model ├── algorithm/ │ ├── model.py # Actor-Critic 网络定义 │ ├── feature_process.py # 观测摊平、归一化、帧堆叠 │ └── algorithm.py # PPO 更新：采样缓冲 + loss 计算 ├── conf/ │ ├── conf.py # 配置加载器，把 toml 读成对象 │ ├── train_env_conf.toml # 训练环境参数：地图、随机种子、最大步数 │ ├── eval_env_conf.toml # 评估环境参数，跑回放走这份 │ └── episode_conf.toml # episode 与 workflow 的调度参数 ├── train.py # 训练入口 ├── predict.py # 单局对局 / 回放入口 └── workflow.py # 把 train 和 predict 串成多阶段调度

agent.py是唯一被平台回调的文件，其余都是它 import 进来的。这一点很关键：平台只认入口文件里的那几个固定方法名，你把算法写得多花都行，方法名改了就等着评测时一片空。

配置那四个文件的分工，我用一张表记，比翻代码快：

注意：conf.py里的默认值会覆盖 toml 里没写的字段，两者不一致时以代码里的默认值为准。我踩过一次，toml 里改了max_step没生效，查了半小时才发现被默认值兜住了。

2.2 把环境跑起来：依赖、冒烟、第一轮训练

依赖这块不要自作聪明。平台镜像里的 Python 版本和 CUDA 版本是压过的，本地建虚拟环境时对齐主版本号，别顺手升到最新。常见做法是：

# 1) 建环境，Python 主版本跟平台镜像保持一致 python -m venv .venv && source .venv/bin/activate pip install -r requirements.txt # 2) 先验证配置能解析，路径写错在这一步就暴露 python -c "from conf.conf import get_conf; print(get_conf().train_env)" # 3) 正式训练前，把 episode 数调到 1、并行环境调到 1，跑一轮冒烟 python train.py

第三步的冒烟测试是我每次必做的。冒烟不等于「跑完不报错」，而是要在日志里确认三件事：观测向量维度是不是打印成了你预期的那个数、动作采样出来的值域有没有超界、learn有没有被真正调用到（很多 bug 是采样了但没进更新）。我一般会在learn里临时加一句打印 buffer 长度，确认它在涨。

跑通之后再把并行环境数和 episode 数调回去。第一轮正式训练别用默认学习率硬拉，先跑个几万步看熵值和 value loss 的量级：熵掉得飞快说明初始化太激进或者奖励尺度不对，value loss 剧烈震荡说明回报没归一化。这两个信号比最终胜率出现得早得多。

2.3 workflow 的调度：训练和评估不是两件事

workflow.py容易被当成可有可无的脚手架，其实它决定了 checkpoint 什么时候落盘、评估用哪份配置。我习惯把它读成一条时间线：先按episode_conf里的节奏跑 N 轮训练，每轮结束触发一次模型保存，保存后拉起predict.py用eval_env_conf跑若干局做评估，评估结果写日志但不参与梯度。这条线的好处是评估环境和训练环境彻底隔离，避免训练期的随机化开关污染评估结论。

如果只是想验证「仅到终点」这段子技能稳不稳，可以把 workflow 简化成只跑评估：加载已有权重，固定种子跑 20 局，统计到达终点的局数和平均剩余步数。这个统计比总奖励有用得多，因为它直接对齐了这版模型的设计目标。

3. 特征与网络：从原始观测到动作分布的那一段

这一段是整份代码里最值得逐行读的部分，因为 90% 的维度不匹配错误都发生在这里。观测从环境出来是个 dict，键名和平台配置里的观测定义一一对应；进去网络之前必须摊成定长向量。摊平的方式、归一化的统计量、帧堆叠的长度，三样东西必须和网络第一层的输入维度严丝合缝，任何一个改了另外两个都要跟着改。

3.1 观测摊平：字段顺序就是契约

import numpy as np # 顺序必须与 conf 里观测字段的定义顺序一致，改顺序等于换了一套特征 OBS_KEYS = ["hero_state", "enemy_state", "map_local", "progress"] class FeatureProcess: def __init__(self, frame_stack=4): self.frame_stack = frame_stack self.stack = [] # 每个字段的量纲差异很大，按字段给缩放系数，别用全局一个数 self.scale = { "hero_state": 1.0, "enemy_state": 1.0, "map_local": 1.0, "progress": 100.0, # 进度类是 0~100 的整数，压到 0~1 } self.obs_mean = None self.obs_std = None def reset(self): self.stack = [] def _flatten(self, obs: dict) -> np.ndarray: parts = [] for k in OBS_KEYS: v = np.asarray(obs[k], dtype=np.float32).reshape(-1) parts.append(v / max(self.scale.get(k, 1.0), 1e-8)) return np.concatenate(parts, axis=0) def process(self, obs: dict, legal_mask=None) -> np.ndarray: vec = self._flatten(obs) if self.obs_mean is not None: vec = (vec - self.obs_mean) / (self.obs_std + 1e-8) self.stack.append(vec) if len(self.stack) > self.frame_stack: self.stack.pop(0) # 开局前几帧用零向量补齐，保证输出维度恒定 while len(self.stack) < self.frame_stack: self.stack.insert(0, np.zeros_like(vec)) out = np.concatenate(self.stack, axis=0) if legal_mask is not None: out = np.concatenate([out, legal_mask.astype(np.float32)], axis=0) return out.astype(np.float32)

逐段说：OBS_KEYS的顺序定义了一次隐式契约，_flatten按这个顺序拼，网络输入维度就是sum(各字段展平长度) * frame_stack (+ 掩码长度)。scale按字段给而不是全局给，是因为进度这种 0~100 的量纲会把其他 0~1 的特征压成噪声。obs_mean/obs_std留成可注入的成员，是为了让它在训练中统计、随模型一起存盘——这点在后面的避坑章还会提。

legal_mask拼接在最后，是为了在动作头上做非法动作屏蔽。赛题环境里往往有「当前帧不能做某些动作」的约束，不屏蔽的话策略会把概率分给不可能执行的动作，采样出来的动作被环境丢弃，表现出来就是智能体站着发呆。

3.2 网络结构：共享躯干加两个头

import torch import torch.nn as nn class ActorCritic(nn.Module): def __init__(self, obs_dim, act_dim, hidden=256): super().__init__() self.trunk = nn.Sequential( nn.Linear(obs_dim, hidden), nn.ReLU(), nn.Linear(hidden, hidden), nn.ReLU(), ) self.actor = nn.Linear(hidden, act_dim) self.critic = nn.Linear(hidden, 1) # 输出层用小增益初始化，开局策略别太自信 for m in (self.actor, self.critic): nn.init.orthogonal_(m.weight, gain=0.01) nn.init.zeros_(m.bias) def forward(self, x, legal_mask=None): h = self.trunk(x) logits = self.actor(h) if legal_mask is not None: logits = logits.masked_fill(legal_mask < 0.5, -1e9) value = self.critic(h).squeeze(-1) return logits, value

obs_dim由上一节的FeatureProcess决定，act_dim由动作空间决定，这两个数我是直接用一行print打出来再填进去的，不手算。躯干两层的 256 是够用的起点，重返秘境这类场景的观测维度通常在小几百，再宽收益递减而显存吃紧。

两个头的分工要理解清楚：actor输出每个离散动作的 logit，经过softmax变成分布；critic输出状态价值，只参与 loss 不参与决策。gain=0.01的正交初始化是老经验，能让开局的动作分布接近均匀，熵不会一上来就塌掉。掩码用-1e9而不是-inf，是为了避免softmax出 NaN——这个细节在调试期能救你一次。

如果时序依赖明显（比如需要记住几帧前敌人的位置），躯干后面接一层 GRU 会更好，代价是采样缓冲要按序列组织，PPO 的 batch 划分也得改成按 episode 切。我的建议是先用纯 MLP 把链路跑通，确认「仅到终点」能稳定收敛，再考虑上 RNN。

3.3 agent.py：平台回调的五个方法

import torch from torch.distributions import Categorical from algorithm.model import ActorCritic from algorithm.feature_process import FeatureProcess class Agent: def __init__(self, model_path=None, device="cpu"): self.device = torch.device(device) self.fp = FeatureProcess(frame_stack=4) self.model = ActorCritic(obs_dim=OBS_DIM, act_dim=ACT_DIM).to(self.device) self.training = True self.traj = [] if model_path: self.load_model(model_path) def reset(self): self.fp.reset() self.traj = [] def set_training_mode(self, mode: bool): self.training = mode self.model.train(mode) @torch.no_grad() def predict(self, obs, legal_mask=None): x = self.fp.process(obs, legal_mask) logits, value = self.model( torch.as_tensor(x, device=self.device).unsqueeze(0), None if legal_mask is None else torch.as_tensor(legal_mask, device=self.device).unsqueeze(0), ) dist = Categorical(logits=logits) # 训练期采样保探索，评估期取 argmax 保稳定 act = dist.sample() if self.training else logits.argmax(dim=-1) if self.training: self.traj.append({ "obs": x, "act": int(act.item()), "logp": float(dist.log_prob(act).item()), "value": float(value.item()), }) return int(act.item())

predict是热路径，每帧都调，所以里面只做前向，不做任何写盘、日志落文件、梯度相关的事。reset里清空帧堆叠和轨迹缓冲，是因为新一局开始观测序列断了，不清会把上一局的尾巴带进这一局，表现出来就是开局动作莫名其妙。

set_training_mode这个钩子必须实现，评估阶段如果还在采样，同一份权重的成绩波动会大到没法比较。我一般把探索率、采样/贪心切换、是否记录轨迹都挂在这个开关上，一处控制，不留后门。

4. 训练循环：「仅到终点」的奖励整形与终止判定

这版模型之所以叫「仅到终点」，核心就在这一章。环境给的奖励是稀疏的，只在关键事件上给分，导航过程中的中间状态没有任何信号。直接把稀疏奖励丢给 PPO，几万步之内策略基本不动。解决办法是在 agent 侧做奖励整形，用观测里已有的位置信息自己算一个稠密的接近信号，叠在环境奖励上，同时把「到达终点」这一个事件当作唯一的终止条件来对齐。

4.1 奖励整形：距离差分加时间惩罚

import math GOAL_RADIUS = 0.8 # 与终点判定体的半径对齐，别自己拍 STEP_PENALTY = -0.001 GOAL_BONUS = 10.0 def shaping_reward(prev_pos, cur_pos, goal_pos, env_reward, done, truncated): """在 agent 侧叠加稠密信号，env_reward 仍以平台返回的为准。""" d_prev = math.dist(prev_pos, goal_pos) d_cur = math.dist(cur_pos, goal_pos) r = env_reward r += 0.1 * (d_prev - d_cur) # 靠近终点给正，远离给负 r += STEP_PENALTY # 每步微惩罚，压制绕圈 if math.dist(cur_pos, goal_pos) <= GOAL_RADIUS: r += GOAL_BONUS done = True # 到达终点即终止，这是本模型唯一的成功条件 if truncated: done = True # 超时截断也要标 done，但它不算成功 return r, done

三个参数值得单独说。GOAL_RADIUS必须和环境的终点碰撞半径一致，取大了智能体在门外就被判定成功、学不到最后那一步的精确对位；取小了永远触发不了，奖励一直是零。这个数我是从回放里量出来的，不是猜的。0.1这个系数控制稠密信号和稀疏奖励的相对权重，太大策略会为了刷接近奖励在终点附近来回磨蹭，太小则起不到引导作用。STEP_PENALTY存在的意义是压住「原地不动也安全」这种局部最优，量级必须远小于GOAL_BONUS，否则智能体会为了省步数提前终止。

done和truncated要分开处理。真实到达终点是done=True的正向终止，超时是truncated=True的中性截断。两者在 GAE 里的自举行为不同：前者下一步价值按 0 算，后者应该用最后一步的 value 自举。混在一起写会让价值函数学出一个偏低的基线，表现出来就是训练很久 critic loss 都下不去。

4.2 GAE 与 PPO 更新

import numpy as np import torch import torch.nn.functional as F def compute_gae(rewards, values, dones, gamma=0.99, lam=0.95): n = len(rewards) adv = np.zeros(n, dtype=np.float32) last = 0.0 for t in reversed(range(n)): next_v = values[t + 1] if t + 1 < n else 0.0 mask = 1.0 - float(dones[t]) delta = rewards[t] + gamma * next_v * mask - values[t] last = delta + gamma * lam * mask * last adv[t] = last returns = adv + np.asarray(values, dtype=np.float32) return adv, returns def ppo_update(model, optimizer, batch, clip=0.2, vf_coef=0.5, ent_coef=0.01): obs = torch.as_tensor(batch["obs"]) acts = torch.as_tensor(batch["act"]).long() old_logp = torch.as_tensor(batch["logp"]) adv = torch.as_tensor(batch["adv"]) ret = torch.as_tensor(batch["ret"]) adv = (adv - adv.mean()) / (adv.std() + 1e-8) # 优势归一化，稳定训练 logits, value = model(obs, torch.as_tensor(batch["mask"])) dist = torch.distributions.Categorical(logits=logits) logp = dist.log_prob(acts) ratio = torch.exp(logp - old_logp) surr1 = ratio * adv surr2 = torch.clamp(ratio, 1.0 - clip, 1.0 + clip) * adv policy_loss = -torch.min(surr1, surr2).mean() value_loss = F.mse_loss(value, ret) entropy = dist.entropy().mean() loss = policy_loss + vf_coef * value_loss - ent_coef * entropy optimizer.zero_grad() loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), 0.5) optimizer.step() return float(policy_loss), float(value_loss), float(entropy)

compute_gae从后往前扫，last是累积的优势估计。mask在终止步把自举掐掉，这是终止判定能生效的关键。返回的returns是优势加价值，直接当 critic 的回归目标用，比单独再算一遍折扣回报省事。

ppo_update里几个系数：clip=0.2是默认值，策略更新幅度超过这个比例就不给梯度；vf_coef=0.5平衡价值损失，如果 critic loss 明显压过 policy loss，可以调小；ent_coef=0.01是熵奖励，探索不足就加，策略太散就减。优势归一化那一行不能省，赛题奖励尺度经常在几百上下浮动，不归一化的话 ratio 会炸。

外层训练循环里，我一般把采样和更新做成「攒够 N 步或凑满 M 局就更新一次」，更新时把 buffer 打乱切成小 batch 跑 4 个 epoch。epoch 别贪多，同一批数据反复过十几次，策略会过拟合到那批分布上，下一轮采样直接崩。

4.3 终点判定的三种失败形态

判定写对了训练才有意义，我把见过的失败形态归成三类。第一类是「假成功」：GOAL_RADIUS太大，智能体还在门外就触发终止，最后权重学出一套犹豫不前的策略，回放里看它在门口反复横跳。第二类是「永不终止」：半径太小或坐标基准不一致（一个是局部坐标、一个是世界坐标），episode 每次都靠超时截断结束，训练日志里 episode 长度恒定等于max_step，看到这个数就该去查坐标。第三类是「截断当成功」：done和truncated混用，价值函数把超时也当成好事，智能体会主动往超时方向走，表现是后期策略越来越消极。

排查手法很朴素：把每局的终止原因打成一列，统计「到达 / 超时 / 其他」三类各占多少。正常收敛过程里到达占比应该单调往上走，如果它长期是 0，问题一定在坐标或半径上，不在算法。

5. 避坑与排查：五个把我卡住的现场

5.1 现象：跑几十步就报维度不匹配

原因：FeatureProcess里写死的输入维度和conf里的观测字段不一致，最常见的是加了帧堆叠但忘了把obs_dim乘以frame_stack，或者legal_mask长度变了没同步。

解决：不要在模型定义里手填维度。在Agent.__init__里先构造一个假观测过一遍fp.process，用返回的实际长度当obs_dim，维度问题当场归零。

5.2 现象：奖励恒为零，智能体原地不动

原因：动作格式没对上。平台对动作的编码有约定，返回了 numpy 的整型或者浮点，被环境当成非法动作丢弃，每步都是无效操作，自然不会产生任何状态变化。

解决：predict的返回值强转成 Python 原生int，并且在冒烟测试里打印前一帧和当前帧的位置差，差值恒为 0 就是动作没生效，别去调学习率。

5.3 现象：本地训练正常，评测超时或成绩为零

原因：predict路径里混进了训练期才该做的事——写日志文件、保存 checkpoint、甚至算梯度。评测环境的单局时间预算比训练期紧得多，这些操作会把时间吃干净。

解决：用set_training_mode(False)把训练相关逻辑全部关掉，predict里只保留一次前向。我习惯在评估前跑一遍torch.no_grad()的单元检查，确认没有任何.backward()路径可达。

5.4 现象：加载权重后行为像随机策略

原因：只存了state_dict，没存与之绑定的东西——观测归一化的obs_mean/obs_std、网络超参、frame_stack长度。加载时结构勉强对得上，但归一化统计量是空的，输入分布和训练期完全不同。

解决：存盘时打包成一个 dict，把权重、统计量、超参、特征配置一起塞进去，加载时先按超参重建网络再灌权重。这是我吃过最大的一次亏，从那以后权重文件里不带配置信息我一律不认。

5.5 现象：智能体在终点前反复抖动

原因：稠密奖励的接近项和每步时间惩罚在终点附近打平了——再走一步的收益和停一步的成本差不多，策略在两种动作之间反复采样。也可能是终点附近有碰撞体，智能体的实际可达位置永远差一点点。

解决：先把STEP_PENALTY调小一个量级看抖动是否消失，若消失就是奖励设计的锅；若还在，就在回放里量一下智能体到终点的最小距离，确认它物理上能不能进到GOAL_RADIUS内部。这一步不做，改多少超参都是白费。

6. 二次开发：把「仅到终点」当成技能基座接上去

这版权重的正确用法不是拿去通关，而是当技能库里的一个原子技能。我的做法是在它上面加一层管理器，把整局拆成「导航到终点」和「终点后交互」两段，前者直接复用这份 checkpoint，后者单独训一个小策略，管理器只负责在两段之间做切换。

class SkillRouter: def __init__(self, nav_agent, interact_agent, switch_radius=1.0, goal_pos=None): self.nav = nav_agent self.interact = interact_agent self.switch_radius = switch_radius self.goal_pos = goal_pos def act(self, obs, legal_mask=None): # 用观测里的位置字段判断当前该由哪个技能接管 dist = math.dist(obs["hero_state"][:2], self.goal_pos) if dist > self.switch_radius: return self.nav.predict(obs, legal_mask) return self.interact.predict(obs, legal_mask) def reset(self): self.nav.reset() self.interact.reset()

switch_radius要比GOAL_RADIUS略大一点，留出交接余量，否则两段技能会在边界上互相抢控制权、来回切；太大则导航技能提前退出，智能体停在半路。交接瞬间还有个细节：两个技能的帧堆叠是各自维护的，切换时新技能看到的前几帧是零填充，容易做出突兀动作。稳妥做法是切换后强制让新技能空转两帧不输出，或者把导航技能的最后一帧观测热启动给交互技能。

验证这块我有个固定套路，比看总奖励靠谱：固定种子跑 50 局，把每局拆成「到达终点耗时」「切换次数」「切换后成功率」三个指标分开统计。如果到达耗时分布的双峰很明显，说明导航技能在某些地图上不稳；如果切换次数远大于 1，说明两个技能的边界在打架。这套指标跑一遍只要几分钟，但能定位到具体是哪一段出了问题，比对着一条总奖励曲线猜半天强得多。

想把导航段单独再练一练也有办法：把环境的终止条件改成「到达终点即成功」并冻结后续交互，等于把这份模型的目标函数原样复现一遍，然后用更大的并行环境数和更长的训练窗口继续推。注意这时GOAL_BONUS要保留，不然策略会退化成「尽量慢点走」以延长接近奖励的累积。

血泪经验是：改这类分段结构之前，先给原版权重留一份只读备份，并且把训练和评估的随机种子分开设。我上次图省事两边共用一个种子，评估集和训练集踩到同一批地图，指标漂亮得离谱，换种子一测直接翻车。从那以后我每次动分段逻辑之前都强制走一遍「备份权重 + 换种子评估 + 三指标分开看」，不省这三步。希望帮到你。

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