1. 从“数据”到“连续动作”到底在解决什么问题
连续动作空间的强化学习,一直是让不少做控制、机器人、量化甚至游戏AI的朋友头疼的方向。离散动作好办,输出几个概率选一个就行;但连续动作不一样,比如机械臂每个关节要给一个扭矩值,无人车要给油门和转向角,这些量都是实数域上的连续值。你没法用“选哪个”的思路去做,只能去拟合一个分布或者一个确定性映射。
而“基于模型的强化学习”(Model-Based RL,简称MBRL)又在这个基础上加了一层:它不满足于只学一个策略,而是想先学一个环境模型——也就是“我做了动作a之后,状态会怎么变”。有了这个模型,就可以在“想象”中 rollout,用更少的真实交互去训练策略。这个思路在样本效率上优势非常明显,尤其适合那些真实采样代价极高的场景,比如真实机器人调试、工业过程控制。
这个系列走到第四篇,核心落点就是“从数据到连续动作”的完整闭环。前三篇大概率已经把数据采集、模型学习、环境封装这些铺垫做完了,这一篇要解决的是:怎么把学到的模型真正用起来,让它输出可执行的连续动作,并且用一套顺手的Python工具箱把整条链路串起来。我会围绕一个可复现的工具箱结构来讲,包含数据管线、模型训练、策略优化、连续动作输出、评估与排查。适合已经了解RL基础概念、想往MBRL方向深入的同学,也适合做控制类项目、想找一套能直接抄作业的Python实现的人。
我自己的经验是,MBRL在连续动作上最容易翻车的不是算法本身,而是工程细节:数据归一化没做好、模型误差累积、动作边界处理不当、rollout长度选错。这些坑在后面会一个个拆开讲。
2. 工具箱整体设计与方案选型思路
2.1 为什么要有“工具箱”这个概念
很多人做MBRL是写一个大脚本,从采样到训练全塞在一起,跑通一次就再也不动了。但只要你想换一个环境、换一个模型结构、换一个策略优化器,整个脚本就得大改。工具箱的意义在于把可变的部分抽象成接口,把不变的部分沉淀成组件。
我设计的这套工具箱,核心分成四层:
- 数据层:负责真实环境交互、经验回放、数据归一化与批采样
- 模型层:负责动力学模型(dynamics model)的构建、训练、预测与不确定性估计
- 策略层:负责在模型里做rollout、计算策略梯度、输出连续动作
- 评估层:负责真实环境评估、模型误差监控、日志与可视化
这四层之间通过明确的接口通信,比如模型层只暴露predict(state, action) -> next_state, reward和sample(...)两个方法,策略层不关心模型是神经网络还是高斯过程。这样你换模型、换策略都不用动其他层。
2.2 连续动作场景下的关键选型
连续动作MBRL有几种主流路线,选型直接决定后面代码怎么写。
| 路线 | 代表思路 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|---|
| 概率动力学模型 + 策略梯度 | 学一个高斯输出模型,在模型里反传 | 实现直接,样本效率高 | 模型误差会被策略放大 | 中低维连续控制 |
| 确定性模型 + 模型预测控制 | 学确定性模型,用MPC在线优化动作 | 鲁棒性好,可加约束 | 每步优化开销大 | 实时性要求不极端的控制 |
| 集成模型 + 保守惩罚 | 多个模型集成,用方差惩罚策略 | 抗过拟合,探索更稳 | 计算量翻倍 | 高维、易过拟合场景 |
| 潜空间模型 | 先编码到低维潜空间再学动力学 | 高维观测友好 | 训练复杂 | 图像输入、高维状态 |
我这次选的是概率动力学模型 + 集成 + 策略梯度的组合。原因很实际:连续动作下策略梯度能直接输出动作,不需要在线求解优化问题,推理速度快;集成模型能给出不确定性,用来做保守估计,避免策略钻模型误差的空子。这套组合在中等维度(状态10到50维、动作3到10维)下表现最均衡。
2.3 数据流与训练循环的整体设计
整个训练循环是这样的:
- 用当前策略在真实环境采样一批轨迹,存入回放池
- 用回放池数据训练(或微调)集成动力学模型
- 在模型里做短程rollout,用策略梯度更新策略
- 每隔若干轮,用真实环境评估策略,记录回报和模型误差
- 根据模型误差决定是否继续用旧模型还是重新训练
这里有个关键设计:rollout长度不能太长。模型误差随步数指数累积,rollout到20步以上,想象出来的轨迹基本就偏离真实分布了。我一般把rollout长度控制在5到15步之间,具体看模型在验证集上的单步和多步预测误差。
提示:不要一上来就追求长rollout,短rollout配合高频模型更新,效果往往比长rollout更稳。
3. 核心细节解析与实操要点
3.1 数据采集与归一化的那些坑
连续动作场景下,状态和动作的量纲差异极大。比如机械臂,关节角度可能是0到3.14弧度,而末端力可能是几百牛顿。如果不做归一化,神经网络会被大量纲特征主导,小量纲特征几乎学不到东西。
我的做法是对状态和动作分别做标准化,用回放池里的均值和标准差:
import numpy as np class Normalizer: def __init__(self, shape, clip=5.0): self.mean = np.zeros(shape, dtype=np.float32) self.std = np.ones(shape, dtype=np.float32) self.count = 1e-4 self.clip = clip def update(self, x): batch_mean = x.mean(axis=0) batch_std = x.std(axis=0) batch_count = x.shape[0] # 增量更新均值和方差 delta = batch_mean - self.mean total = self.count + batch_count self.mean += delta * batch_count / total m_a = self.std ** 2 * self.count m_b = batch_std ** 2 * batch_count m2 = m_a + m_b + delta ** 2 * self.count * batch_count / total self.std = np.sqrt(m2 / total) self.count = total def normalize(self, x): x = (x - self.mean) / (self.std + 1e-8) return np.clip(x, -self.clip, self.clip) def denormalize(self, x): return x * (self.std + 1e-8) + self.mean这里有几个细节值得说。第一,clip设成5是经验值,防止个别异常值把归一化后的数据拉爆。第二,均值和方差要用增量更新,不能每次重新算,否则早期数据会被后期覆盖。第三,动作的归一化范围要和策略输出的激活函数匹配。如果动作归一化到[-1,1],策略输出用tanh就刚好;如果归一化到别的范围,策略输出还得再缩放。
注意:归一化器要跟着回放池一起保存和加载,评估时用的归一化参数必须和训练时一致,否则动作会完全错位。
3.2 动力学模型的结构与训练目标
动力学模型要预测的是(state, action) -> (delta_state, reward)。注意我预测的是状态增量而不是下一个状态本身。原因是增量通常数值更小、分布更集中,网络更容易学。预测增量后再加回原状态,得到下一个状态。
模型结构上,我用的是集成的高斯MLP:每个集成成员是一个多层感知机,输出均值和方差,用负对数似然做损失。
import torch import torch.nn as nn class GaussianMLP(nn.Module): def __init__(self, obs_dim, act_dim, hidden=256, num_layers=3): super().__init__() in_dim = obs_dim + act_dim layers = [] for _ in range(num_layers): layers += [nn.Linear(in_dim, hidden), nn.SiLU()] in_dim = hidden self.backbone = nn.Sequential(*layers) self.mean_head = nn.Linear(hidden, obs_dim + 1) # +1 是reward self.logstd_head = nn.Linear(hidden, obs_dim + 1) def forward(self, obs, act): x = torch.cat([obs, act], dim=-1) h = self.backbone(x) mean = self.mean_head(h) logstd = self.logstd_head(h).clamp(-10, 2) return mean, logstd def nll_loss(self, obs, act, target): mean, logstd = self.forward(obs, act) var = torch.exp(2 * logstd) nll = 0.5 * ((target - mean) ** 2 / var + 2 * logstd) return nll.sum(dim=-1).mean()logstd的clamp很关键。不clamp的话,训练早期方差可能爆炸或者塌缩到0,导致数值不稳定。clamp到[-10, 2]意味着标准差在约4.5e-5到7.4之间,覆盖了绝大多数连续控制场景。
集成成员数量我一般用5到7个。太少不确定性估计不准,太多训练和推理开销大。每个成员用不同的随机种子初始化,训练时用同一批数据但不同的mini-batch顺序,增加多样性。
3.3 策略网络与连续动作输出
策略网络输出连续动作,有两种常见形式:高斯策略和确定性策略。高斯策略输出均值和标准差,采样得到动作,适合需要探索的场景;确定性策略直接输出动作,配合外部噪声做探索,适合后期精细调优。
我通常用tanh-squashed高斯策略,这是连续动作里最稳的一种:
class TanhGaussianPolicy(nn.Module): def __init__(self, obs_dim, act_dim, hidden=256): super().__init__() self.net = nn.Sequential( nn.Linear(obs_dim, hidden), nn.SiLU(), nn.Linear(hidden, hidden), nn.SiLU(), ) self.mean_head = nn.Linear(hidden, act_dim) self.logstd_head = nn.Linear(hidden, act_dim) def forward(self, obs): h = self.net(obs) mean = self.mean_head(h) logstd = self.logstd_head(h).clamp(-5, 2) return mean, logstd def sample(self, obs): mean, logstd = self.forward(obs) std = torch.exp(logstd) eps = torch.randn_like(std) raw = mean + std * eps action = torch.tanh(raw) # 计算log_prob时要考虑tanh的雅可比修正 log_prob = -0.5 * ((raw - mean) / std) ** 2 \ - logstd - 0.5 * np.log(2 * np.pi) log_prob -= torch.log(1 - action ** 2 + 1e-6) return action, log_prob.sum(dim=-1, keepdim=True)tanh把动作压到[-1,1],正好对应归一化后的动作空间。但要注意log_prob的雅可比修正,很多人忘了这一步,导致策略梯度有偏。torch.log(1 - action**2 + 1e-6)就是修正项,1e-6是防止动作接近±1时log发散。
提示:如果你发现策略很快塌缩到确定性输出(std趋近0),检查一下logstd的clamp下界是不是太低了,或者熵正则系数是不是太小。
3.4 模型rollout与策略梯度的衔接
在模型里做rollout,核心是把模型预测的均值和方差都用上。只用均值会导致策略过拟合到模型的确定性部分,忽略不确定性;只用采样又会让梯度方差太大。我的做法是用均值做前向传播,用方差做保守惩罚。
具体来说,在模型里rollout时,每一步用模型预测的均值作为下一个状态,同时记录该步的预测方差。策略的奖励信号里加一项-lambda * variance,让策略倾向于选择模型预测更确定的动作。这个lambda一般取0.1到1.0之间,太大策略会变得过于保守,太小起不到作用。
def model_rollout(policy, models, obs, horizon, lambda_var=0.5): log_probs, rewards, variances = [], [], [] for _ in range(horizon): action, log_prob = policy.sample(obs) # 集成模型预测 means, vars = [], [] for model in models: m, logstd = model(obs, action) means.append(m) vars.append(torch.exp(2 * logstd)) mean = torch.stack(means).mean(dim=0) var = torch.stack(vars).mean(dim=0) delta_obs, reward = mean[:, :-1], mean[:, -1] obs = obs + delta_obs # 保守惩罚 penalty = lambda_var * var[:, :-1].sum(dim=-1) rewards.append(reward - penalty) log_probs.append(log_prob) variances.append(var) return log_probs, rewards, variances这里有个实操细节:obs的更新要用增量加回,而不是直接用模型输出的绝对状态。因为模型学的是增量,直接输出绝对状态的话,误差会更大。
4. 完整实操流程与关键环节实现
4.1 环境封装与数据采集循环
先定义一个统一的环境接口,不管底层是Gym还是自定义环境,都包装成reset() -> obs和step(action) -> obs, reward, done, info。
class EnvWrapper: def __init__(self, env, obs_normalizer, act_normalizer): self.env = env self.obs_norm = obs_normalizer self.act_norm = act_normalizer def reset(self): obs = self.env.reset() return self.obs_norm.normalize(obs) def step(self, action): # action是归一化后的,要反归一化再传给真实环境 real_action = self.act_norm.denormalize(action) obs, reward, done, info = self.env.step(real_action) return self.obs_norm.normalize(obs), reward, done, info数据采集循环里,我一般每轮采10到20条轨迹,每条轨迹长度200步左右。采完后立刻更新归一化器,然后用新数据训练模型。这里有个节奏问题:模型训练频率和策略更新频率要匹配。如果模型还没学好就频繁更新策略,策略会利用模型误差;如果模型学好了但策略不更新,样本效率上不去。我的经验是每采一轮数据,模型训练5到10个epoch,策略更新1到2轮。
4.2 模型训练的具体配置
模型训练的配置直接影响后面策略的质量。我列一下我常用的参数:
| 参数 | 取值 | 说明 |
|---|---|---|
| 集成成员数 | 5 | 平衡不确定性与开销 |
| 隐藏层宽度 | 256 | 中等维度够用 |
| 隐藏层数 | 3 | 太深容易过拟合 |
| 学习率 | 3e-4 | Adam默认附近 |
| batch size | 256 | 太小梯度噪声大 |
| 训练epoch | 8 | 每轮数据训练轮数 |
| 验证集比例 | 0.1 | 用于早停和误差监控 |
| 梯度裁剪 | 1.0 | 防止梯度爆炸 |
训练时我会把数据分成训练集和验证集,用验证集上的多步预测误差来决定是否早停。多步预测误差比单步更能反映模型在rollout中的表现。具体做法是:从验证集里取一批起始状态,用模型rollout 10步,和真实轨迹对比,算MSE。
def multi_step_val_error(models, val_data, horizon=10): obs, actions, next_obs = val_data pred_obs = obs.clone() total_error = 0.0 for t in range(horizon): act = actions[:, t] means = [] for model in models: m, _ = model(pred_obs, act) means.append(m[:, :-1]) delta = torch.stack(means).mean(dim=0) pred_obs = pred_obs + delta total_error += ((pred_obs - next_obs[:, t]) ** 2).mean().item() return total_error / horizon如果这个误差在训练过程中先降后升,说明过拟合了,要早停。如果一直降不下去,说明模型容量不够或者数据不够。
4.3 策略优化的完整实现
策略优化用模型rollout得到的轨迹做梯度上升。损失函数包含三部分:策略梯度损失、熵正则、保守惩罚。
def update_policy(policy, optimizer, models, obs_batch, horizon=10, lambda_var=0.5, entropy_coef=0.01): log_probs, rewards, _ = model_rollout( policy, models, obs_batch, horizon, lambda_var) # 计算折扣回报 returns = [] R = 0 gamma = 0.99 for r in reversed(rewards): R = r + gamma * R returns.insert(0, R) returns = torch.stack(returns).sum(dim=0) log_probs = torch.stack(log_probs).sum(dim=0) # 策略梯度损失 pg_loss = -(log_probs * returns.detach()).mean() # 熵正则 mean, logstd = policy(obs_batch) entropy = (logstd + 0.5 * np.log(2 * np.pi * np.e)).sum(dim=-1).mean() loss = pg_loss - entropy_coef * entropy optimizer.zero_grad() loss.backward() torch.nn.utils.clip_grad_norm_(policy.parameters(), 1.0) optimizer.step() return loss.item()这里returns的计算要注意,rollout里的奖励是逐步累加的,折扣因子0.99对应大约100步的有效视野。如果rollout只有10步,折扣因子可以设大一点比如0.995,让远端奖励也有足够权重。
熵正则系数0.01是个保守值,目的是防止策略过早塌缩。如果你发现策略探索不够,可以加到0.05;如果策略太随机,减到0.001。
4.4 真实环境评估与模型误差监控
每训练若干轮,必须回到真实环境评估。评估时策略用确定性输出(取均值再过tanh),不要采样,否则评估方差太大。
def evaluate(policy, env, num_episodes=5): total_rewards = [] for _ in range(num_episodes): obs = env.reset() done = False ep_reward = 0 while not done: with torch.no_grad(): mean, _ = policy(torch.FloatTensor(obs).unsqueeze(0)) action = torch.tanh(mean).squeeze(0).numpy() obs, reward, done, _ = env.step(action) ep_reward += reward total_rewards.append(ep_reward) return np.mean(total_rewards), np.std(total_rewards)同时要监控模型误差。我一般记录三个指标:单步预测MSE、10步rollout MSE、策略在模型里的回报和真实回报的差距。如果模型里的回报远高于真实回报,说明策略在钻模型误差的空子,这时候要么增加保守惩罚,要么重新训练模型。
5. 常见问题与排查技巧实录
5.1 策略在模型里表现好但真实环境崩了
这是MBRL最经典的问题,本质是模型误差被策略利用。排查思路:
- 先看模型的多步预测误差。如果10步误差已经很大,说明模型本身不够准,要加数据或改结构。
- 再看保守惩罚系数。如果lambda_var太小,策略会选高方差动作,加大到1.0试试。
- 最后看rollout长度。如果horizon太长,缩短到5步,配合更频繁的模型更新。
我踩过的一个坑是:模型在训练集上误差很小,但验证集上误差大,说明过拟合。这时候加数据比调参更有效。
5.2 动作输出总是贴着边界
tanh策略容易在训练后期把动作推到±1附近,因为边界处梯度小,策略觉得“安全”。但这会导致真实环境里动作饱和,控制效果差。
解决办法有两个:一是在奖励里加动作平滑惩罚,惩罚相邻步动作变化过大;二是在策略输出后加一个小的缩放,比如乘0.95,留出边界余量。我一般用第一种,因为它同时能改善控制平滑性。
5.3 训练不稳定,回报忽高忽低
连续动作MBRL的训练方差本来就大,但如果波动特别剧烈,通常是这几个原因:
- 归一化器更新太频繁:每采一批数据就更新归一化器,导致数据分布漂移。改成每几轮更新一次。
- 模型和策略更新节奏不匹配:模型还没收敛就更新策略。增加模型训练epoch。
- 学习率太大:策略学习率降到1e-4试试。
下面这张表是我整理的常见问题速查:
| 现象 | 可能原因 | 排查方法 | 解决方向 |
|---|---|---|---|
| 模型内回报高,真实低 | 模型误差被利用 | 对比模型与真实回报 | 加保守惩罚、缩短rollout |
| 动作贴边界 | tanh饱和 | 统计动作分布 | 加平滑惩罚、输出缩放 |
| 训练波动大 | 归一化漂移 | 检查归一化器更新频率 | 降低更新频率 |
| 策略不探索 | 熵塌缩 | 看logstd值 | 加熵正则系数 |
| 模型误差不降 | 容量不足/数据少 | 看训练验证误差 | 加宽网络、加数据 |
| 推理速度慢 | 集成成员太多 | 测单步推理耗时 | 减到3到5个成员 |
5.4 实操心得:几个让我少走弯路的习惯
第一个习惯是每次实验都保存归一化器和模型检查点。MBRL训练一次动辄几小时,不保存的话调参要从头来。我一般每10轮存一次,同时存归一化器的均值和方差。
第二个习惯是先用短rollout跑通再逐步加长。一开始horizon设3,确认整条链路没问题,再慢慢加到10。这样出问题容易定位。
第三个习惯是记录模型误差曲线和策略回报曲线在同一张图上。如果模型误差上升的同时策略回报也上升,基本可以确定策略在钻空子,这时候要立刻干预。
第四个习惯是动作空间维度超过10时,考虑用低秩分解或者分组建模。全连接网络在高维动作下参数太多,容易过拟合。我试过把动作分成几组分别预测,效果比一次性预测所有维度更稳。
6. 工具箱的扩展与复用建议
这套工具箱跑通之后,扩展性其实很好。换环境只需要改EnvWrapper里的底层env;换模型只需要实现同样的predict接口;换策略只需要保证有sample和forward方法。我自己后来把它用到了几个不同场景:一个是机械臂的关节控制,一个是无人车的横向控制,还有一个是化工过程的温度调节。每次迁移主要改的是归一化参数和rollout长度,核心代码几乎没动。
如果你想让工具箱更实用,我建议加两个东西。一个是模型误差的可视化面板,把单步误差、多步误差、策略回报画在一起,调参时一眼就能看出问题。另一个是自动rollout长度调节,根据当前模型的多步误差动态决定horizon,误差小就roll长一点,误差大就缩短。这个我试过,在训练早期能明显加速,后期能防止策略钻空子。
最后分享一个小技巧:如果你的真实环境采样特别贵,可以在模型训练时用优先经验回放,优先采样模型预测误差大的样本。这样模型会把容量集中在难学的区域,整体误差下降更快。这个改动不大,但在样本受限的场景下收益很明显。