ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

PyTorch实现HER算法:解决稀疏奖励下的目标达成强化学习

PyTorch实现HER算法:解决稀疏奖励下的目标达成强化学习 目标导向的强化学习这几年一直是很多朋友卡住的一道门槛。大家跑熟了CartPole、HalfCheetah这类简单连续控制任务之后总会遇到一个绕不开的问题真实任务里哪来那么多密集奖励机械臂抓取东西抓空了就是没奖励碰一下就结束模型根本学不起来。这个场景下想要从代码层面真正理解深度强化学习怎么解决“稀疏奖励下的目标达成”问题HERHindsight Experience Replay事后经验回放算法就是最实用、最值得吃透的一个切入点。这篇文章就用PyTorch把这个算法从原理到代码一步步过一遍适合对DDPG、Actor-Critic这类基础结构已经有些了解、但一直不知道怎么处理稀疏奖励任务的读者。我会把HER在目标条件化强化学习里的设计思路、代码实现要点、训练时的采样细节全部展开顺便讲讲我在实际调试中踩过的坑。不管你是准备做机械臂控制、机器人导航还是路径规划这套用“失败经验”反向训练的思路都值得你彻底搞明白。1. 内容整体设计与思路拆解1.1 目标导向的强化学习为什么难普通的深度强化学习任务比如让一个智能体走迷宫环境给它的奖励通常是“越接近终点分数越高”这样每一步都有反馈模型很容易通过试错学到策略。但目标导向的任务不一样它的环境很少给你中间信号机械臂要把方块推到指定位置推到一半没到目标就是没有任何奖励小车要导航到某个坐标前半段路程也没有任何反馈。这种**二值奖励到达目标1其它0**的训练信号极其稀疏如果只用随机探索模型几乎不可能在有限步数内碰到成功状态梯度信号基本为零整个学习过程直接僵住。目标导向的强化学习其实就是把问题统一建模成“给定一个目标g智能体要学会从任意状态s出发做出能达成g的动作”。这里的关键词是“目标条件化”状态空间和动作空间里都要加入目标的信息模型不再是单纯学一套固定动作而是学一个“以目标为输入”的映射关系。这样一来即使当前任务的具体目标变了只要模型学会了“根据目标调整动作”的规律就能泛化到新的目标上。1.2 HER算法如何解决稀疏奖励HER提出的时候想法非常简单粗暴我头一次看懂的时候觉得这思路简直是白给我们经常要求智能体实现一个目标但训练过程中它总是失败。失败就不给学习信号了吗换个角度看——虽然它没有完成任务但它确实到达了一个“现实中的状态”。如果我们把这个已经到达的状态当作一个“替代目标”那这条经验就变成了一次成功经验可以给模型提供正向的引导。举个例子。机械臂的目标是把积木推到桌面的A点实际推了十次九次推到B点。按传统RL这十次都是全零奖励没有意义。但HER会把“推到B点”这件事重新解释为一个新目标已知目标是“推到B点”智能体成功做到了奖励1然后把这个“原始目标A失败轨迹”和“替代目标B成功标记”一起存进经验池。这样一来原本全是零奖励的记忆里不断产生出真实的成功样本模型就有梯度可学了慢慢就能从“推到哪里算哪里”进化成“推到我想要的地方”。从数学上看HER其实是在重新构造转移样本的四元组(s, a, r, s)或者五元组(s, a, r, s, g)把目标g替换成轨迹中实际到达的状态。关键点在于这是数据层面上的增广不修改环境、不改变奖励函数、不动网络结构纯粹通过“换目标”制造出更多正样本。这是它最优雅的地方也是为什么它可以和任意off-policy算法搭配的原因。1.3 为什么选PyTorch实现HERPyTorch在实现这类算法时确实顺手很多。HER的核心工作是操作经验池里的transition需要频繁地对目标项做替换、拼接和重新计算奖励PyTorch的张量操作和Dataset接口能够非常自然地做成动态回放缓冲区。另外DDPG、TD3这类连续控制算法在PyTorch里都有很成熟的实现参考网络结构、损失函数和梯度更新都很好改。做实验的时候torch.save一键保存模型后续做评估、可视化也比较方便。总之用PyTorch版本跟着代码一步步走比看那些大而全的强化学习框架更容易理解算法本身的逻辑。2. 核心细节解析与实操要点2.1 HER经验池的数据结构设计实现HER之前首先要搞清楚一条经验里要存什么。普通off-policy算法存的是(s, a, r, s, done)HER因为需要做目标替换所以每条经验里必须额外带上当前轨迹对应的目标g。我用的经验池每条存储在dict里transition { obs: obs, # 当前状态shape取决于环境 action: action, # 动作 reward: reward, # 原始奖励 next_obs: next_obs, # 下一状态 goal: goal, # 这条经验对应的目标 done: done, # 是否到达终止状态 }要注意的是obs和next_obs建议直接存储“拼接后的完整观测”也就是把原始状态和goal拼好再存还是分开存在实际跑实验时我建议在环境返回之后立刻把achieved goal实际到达的状态和desired goal期望目标存下来obs里可以只存原始状态后续需要时再算差值或拼接。但如果不做额外特征工程直接把原始状态和goal拼成一个向量丢给网络那最省事的做法是数据进buffer之前就拼接。这个选择会影响到后续奖励重计算的复杂度文章后面会给代码示例。HER里还有一个关键概念叫“achieved goal”也就是轨迹里每个时刻智能体实际到达的目标状态。在FetchReach这类环境里achieved goal就是机械臂末端的三维坐标desired goal是目标位置的坐标。自己写代码时一定要让环境把这两个量都返回出来否则HER无从做起。2.2 目标替换策略future、final和episodeHER论文里提了4种从轨迹中采样替代目标的方式分别是final、future、episode、random。原理不多说直接给结论和代码量化的建议。final只用轨迹最后一步到达的状态作为替代目标最简单但样本量太少。future从当前时间步t之后的状态中随机采样一个状态作为替代目标这也是论文作者最推荐的因为它在时间上保持了因果关系智能体在t时刻的动作确实对后续状态有影响。episode从整条轨迹中随机采样一个状态不管时间先后。random从过往所有经验中随机采样一个已经发生过的状态。我在复现的时候绝大多数场景直接用future采样比例K设置在4左右也就是每条原始轨迹额外生成4条替换目标后的经验。实践经验是K太大内存涨得快、训练速度下降K太小又不够充分4是一个平衡点。future策略的采样代码很简单def sample_her_transitions(episode, t, replay_buffer, her_ratio4): # episode是一个列表存了整条轨迹的transition transitions [] for transition in episode[:t1]: transitions.append(transition) if len(episode) - 1 t: for _ in range(her_ratio): future_state episode[np.random.randint(t 1, len(episode))] # 用future_state里的achieved_goal替换原transition里的goal new_transition replace_goal(transition, future_state[achieved_goal]) transitions.append(new_transition) return transitions在替换目标之后所有原始经验里的reward和done都要重新计算。这也是HER比较容易出错的地方。哪怕状态转移数据完全没变只要目标变了那条经验到底是成功还是失败就不一样了必须重新用环境的奖励函数通常是一个阈值判断函数判断一次。2.3 算法选型为什么用DDPG做HER的基础网络HER只是一个经验回放层面的技巧它不规定必须用哪个RL算法。但大部分PyTorch版教学代码都会把它和DDPG或TD3搭配原因很简单HER面向的是连续动作空间而且需要一个离线的、基于经验的算法。DQN做不了连续动作PPO是on-policyHER的“替换目标生成新经验”在on-policy下意义会削弱很多。DDPG结构简单Actor和Critic两个网络便于读者理解所以是入门最佳选择。DDPG的四个网络分别是Actor当前网络、Actor目标网络、Critic当前网络、Critic目标网络PyTorch里可以用nn.Module类快速定义。Actor网络输入状态含目标输出动作Critic网络输入状态含目标加上动作输出Q值。网络参数更新时Actor用DDPG的确定性策略梯度更新Critic用TD误差更新目标网络用软更新soft updatefor target_param, param in zip(target_net.parameters(), net.parameters()): target_param.data.copy_(target_param.data * tau param.data * (1.0 - tau))这里有个我自己调代码时总结的要点HER对Critic的依赖比普通RL更大因为样本目标被替换过Critic必须准确地评价“在这个替代目标下这个状态动作值有多好”如果Critic本身拟合能力不足策略很容易被误导。所以网络不要贪深两层256或三层256就够了我试过加宽到512反而更不稳定。2.4 整体代码架构怎么组织我用PyTorch实现HER时把工程分成了这么几个模块环境封装处理observation和goal的转换、经验池包含原始回放和HER增强后的样本、网络结构Actor/Critic、DDPG更新器负责目标网络软更新和策略更新、训练主循环采样、存储、增强、更新。这样拆的好处是每个模块都能单独调试比如经验池的替换逻辑可以单独写两个test用例验证不必每次都跑到整个训练流程里去排错。训练循环的伪码结构大概是这样for epoch in range(n_epochs): for cycle in range(n_cycles): # 收集一条轨迹 obs env.reset() episode [] for t in range(max_episode_steps): action actor.act(obs_with_goal, noise_scale) next_obs, reward, done, achieved_goal env.step(action) episode.append(...) if done: break # 将轨迹存入replay buffer并用HER做增强 store_episode(episode, her_ratio) # 从replay buffer采样batch更新网络 for _ in range(n_batches): batch replay_buffer.sample(batch_size) update_actor_and_critic(batch) # 每个epoch结束评估一次 evaluate(actor, env)代码看起来不复杂难点其实全在细节里比如reward函数归一化、goal距离阈值、动作噪声的衰减策略、环境reset时随机生成目标的范围。下面我分别展开。3. 实操过程与核心环节实现3.1 环境准备从个人造一个简单环境开始如果直接用OpenAI的FetchReach环境或者gym Robotics套件里面自带fetch机器人视觉上很炫但对机器要求高安装还要依赖mujoco。新手很容易被环境和包依赖劝退。我自己调试的时候干脆写了一个极简的2D推球环境用一个质点代表机械臂末端目标是把它推到一个随机生成的目的地奖励只有到达目标0其它-1。这样做的好处是所有数据和状态都是自己控制的方便观察HER到底有没有生效。简单环境的观测与目标设计# 状态二维坐标动作二维速度 state_dim 2 action_dim 2 goal_dim 2动作是连续值限制在[-1, 1]。目标坐标在单位正方形内随机生成。机械臂每走一步判断当前位置和目标位置的距离是否小于阈值0.05小于则doneTruereward0否则reward-1。这个环境虽然简陋但它完整保留了稀疏奖励的性质而且训练速度极快跑几万个step就能看到效果是理解HER的完美试验场。3.2 完整PyTorch代码实现核心部分下面是网络定义和模型初始化的代码。我习惯把所有输入原始状态和目标拼成一个向量所以输入维度是state_dim goal_dim。import torch import torch.nn as nn import torch.nn.functional as F import numpy as np import random from collections import deque class Actor(nn.Module): def __init__(self, state_dim, action_dim, hidden256): super().__init__() self.fc1 nn.Linear(state_dim, hidden) self.fc2 nn.Linear(hidden, hidden) self.fc3 nn.Linear(hidden, action_dim) def forward(self, x): x F.relu(self.fc1(x)) x F.relu(self.fc2(x)) # 连续动作一般用tanh把输出限制在[-1,1] return torch.tanh(self.fc3(x)) class Critic(nn.Module): def __init__(self, state_dim, action_dim, hidden256): super().__init__() self.fc1 nn.Linear(state_dim action_dim, hidden) self.fc2 nn.Linear(hidden, hidden) self.fc3 nn.Linear(hidden, 1) def forward(self, x, action): x torch.cat([x, action], dim1) x F.relu(self.fc1(x)) x F.relu(self.fc2(x)) return self.fc3(x)经验池和HER增强的部分这里只给出核心逻辑。为了简化代码我把每条transition都存成dict采样时再批量转成tensor。class HindsightReplayBuffer: def __init__(self, capacity, her_ratio4): self.capacity capacity self.her_ratio her_ratio self.buffer deque(maxlencapacity) def store_episode(self, episode): # episode包含原始轨迹每个元素是一个transition dict for t, trans in enumerate(episode): # 存原始经验 self.buffer.append(trans) # 对每个时间步做HER增强 if len(episode) - 1 t: for _ in range(self.her_ratio): # 在t之后随机挑一个状态作为替代目标 future episode[np.random.randint(t 1, len(episode))] # 复制当前transition替换goal her_trans trans.copy() her_trans[goal] future[achieved_goal] # 重新计算奖励和done her_trans[reward], her_trans[done] self.env_compute_reward( future[achieved_goal], her_trans[goal] ) self.buffer.append(her_trans) staticmethod def env_compute_reward(achieved_goal, desired_goal): # 简单距离判断 dist np.linalg.norm(achieved_goal - desired_goal) return (0.0 if dist 0.05 else -1.0), (dist 0.05)注意一点在store_episode时要传一个奖励计算函数避免环境类被耦合进buffer。上面代码里我为了展示写成了静态方法实际项目里建议通过构造函数传入一个callable。DDPG更新部分的代码相对标准这里重点说HER带来的一个注意点采样batch后batch里的每个transition的goal各不相同所以用网络前向计算时不能把整批数据当成同一个goal要让输入中包含每个样本自己的goal。def update(actor, critic, target_actor, target_critic, replay_buffer, optim_actor, optim_critic, gamma0.98, tau0.05, batch_size128): batch random.sample(replay_buffer.buffer, batch_size) obs torch.FloatTensor(np.array([t[obs] for t in batch])) goal torch.FloatTensor(np.array([t[goal] for t in batch])) action torch.FloatTensor(np.array([t[action] for t in batch])) next_obs torch.FloatTensor(np.array([t[next_obs] for t in batch])) reward torch.FloatTensor(np.array([t[reward] for t in batch])).unsqueeze(1) done torch.FloatTensor(np.array([t[done] for t in batch])).unsqueeze(1) # 拼接输入 state_input torch.cat([obs, goal], dim1) next_state_input torch.cat([next_obs, goal], dim1) target_action target_actor(next_state_input) target_q target_critic(next_state_input, target_action) y reward gamma * (1 - done) * target_q current_q critic(state_input, action) critic_loss F.mse_loss(current_q, y.detach()) optim_critic.zero_grad() critic_loss.backward() optim_critic.step() actor_loss -critic(state_input, actor(state_input)).mean() optim_actor.zero_grad() actor_loss.backward() optim_actor.step() for target_param, param in zip(target_critic.parameters(), critic.parameters()): target_param.data.copy_(tau * param.data (1.0 - tau) * target_param.data) for target_param, param in zip(target_actor.parameters(), actor.parameters()): target_param.data.copy_(tau * param.data (1.0 - tau) * target_param.data)3.3 训练流程与参数设置经验我实际运行的时候会把整个训练过程分成多个epoch每个epoch内先做交互采集数据然后集中更新网络。这个“交互-更新”分离的模式比每步都更新要稳定得多。参考参数如下max_episode_steps: 50 n_epochs: 100 n_cycles_per_epoch: 20 batch_size: 128 replay_capacity: 100000 her_ratio: 4 actor_lr: 0.001 critic_lr: 0.001 gamma: 0.98 tau: 0.05 noise_scale: 0.2 noise_decay: 0.995训练时动作噪声建议开始时大一点0.2~0.3保证探索充分然后按epoch衰减否则后期策略收敛之后还在乱动。我自己试过一上来噪声就设0.05结果HER增强样本再多也没用因为探索完全不足替代目标覆盖范围太小训练效率极低。训练效果的实时观察指标有两个一是平均原始reward基本为负二是训练过程中成功率。我在代码里会在每个epoch结束跑一次固定seed的评估统计50个随机目标里的成功率这个数字比loss更能反映问题。3.4 评估展示怎么证明HER真的有效我用自己写的2D环境做对比实验一组开HER一组关掉HER也就是只存原始经验。训练同样的epoch数每100步记录一次平均成功率。结果非常明显关掉HER的成功率在50个epoch后仍然是0而开HER的组大概在20个epoch之后成功率能到50%以上后面稳定在80%以上。这极大震撼了我也让我彻底理解了HER的价值。有一个可视化小技巧训练完保存actor网络随便采样几个目标把智能体每一步的坐标画出来这样能看到它从最初来回乱撞到后来几乎直线冲向目标的过程。用matplotlib逐帧保存成gif发朋友圈或者以后写文章都很直观。4. 常见问题与排查技巧实录4.1 训练不收敛成功率一直为0这个是新手遇到最多的状况。先别急着调网络结构大概率是以下几个原因按顺序排查。第一HER目标替换之后奖励是否重新计算了很多人会把原来的reward直接复制给替换后的transition这等于让模型学一个矛盾的标签同一个动作一会是成功一会是失败梯度直接对冲。解决方法是强制在store_episode时用奖励函数重建一次reward和done。第二目标范围是否定义合理如果目标坐标超出了智能体能到达的范围比如目标在(10,10)而你机械臂只能动到单位正方形里那替换目标再怎么替换也没有意义因为轨迹里根本到不了那个地方。检查一下achieved_goal的覆盖范围是不是和目标采样范围一致。第三done标记是否写对了HER里替换目标后如果done仍然按原始轨迹记录有时会出现“这个时刻其实没完成替代目标但done1”的矛盾。我建议所有HER样本的done都根据替代目标重新计算不要沿用原done。第四探索噪声是否过度衰减如果是跑长训练噪声减到0之后模型基本不做探索成功率容易卡住。4.2 HER经验池的重复和偏差问题HER增强倍数K如果设得太高经验池里几乎全是替换出来的“人造成功样本”会导致训练偏差智能体的行为会显得过于激进因为它见过太多成功轨迹但真实环境里根本没那么容易成功。我个人的经验是K不要超过8最好在4左右。另一个问题是经验池容量有限时HER生成的样本会把原始失败经验挤掉。这会削弱模型对真实失败状态的理解。一个比较实用的做法是优先保原始经验增强样本可以设置一个单独的maxlen或者使用优先级的buffer。开源社区里不少实现都直接把所有经验塞进同一个deque简单但并非最优。4.3 PyTorch常见实现bug张量维度不匹配是最常见的问题。特别是从dict批量构造tensor时如果环境返回的goal是list直接np.array后可能是object类型转成torch.FloatTensor就会报错。统一的所有观测、目标、动作维度在环境封装层就把它全部转成np.ndarray并且dtypefloat32。还有一个坑是actor的输出层激活。如果动作范围是[-1,1]最后用tanh没问题但如果你的环境动作范围是[0,1]或更广一定记得在环境封装层做动作缩放不要直接让网络输出后硬塞进env.step。动量很大的话训练一开始就会因为非法动作报错。软更新时PyTorch里如果用.data.copy_可以跳过梯度记录但要注意参数的类型保持一致。有些实现里混用了param.data和param.clone()会造成目标网络参数反复横跳。稳妥写法是统一用target_net.load_state_dict({ k: tau * v (1 - tau) * target_net.state_dict()[k] for k, v in net.state_dict().items() })或者直接写for循环别怕慢反正也就每步更新一次。4.4 调试技巧如何快速判断HER是否正常工作当你第一次写好HER代码不想直接跑几百个epoch才看到结果可以做一个“单轨迹测试”。固定环境seed手动让智能体执行一些随机动作这样一定产生一条失败轨迹。然后用HER的store_episode处理打印buffer里的transition数量应该是原始步数*(1her_ratio)。再随机抽几条增强样本验证它们的goal确实来自后续状态reward也确实等于0。这一步只要过了基本就证明你的HER逻辑没写错。接着做一个更进阶的诊断训练最开始几个epoch计算一下buffer里正样本reward0的占比。如果占比长期低于5%说明探索性不足或者替代目标生成太少需要调大noise或者增大her_ratio如果占比突然高到50%以上说明模型已经开始学会“故意走到某个状态再停下来”这时候要检查是不是目标阈值太宽松或者环境本身太简单。4.5 传统RL经验迁移到HER时的其他坑有些人之前跑过普通DDPG习惯在每步交互后马上更新网络。但HER的实现里经验池需要存储完整轨迹才能做future采样如果边交互边更新往往一条轨迹没结束就更新了这样future策略可能采样不到完整的后续状态。稳妥做法是一个episode结束后再把整条轨迹送入buffer然后集中采样更新。另外目标网络的更新频率和软更新系数tau在HER里可以适当增加比如每更新一次critic就更新一次target而不是等几步再更新。因为HER生成的样本目标变化比较剧烈target网络如果不跟上会让Q值更新滞后训练容易震荡。5. 从代码到实战一些我自己的心得HER这个算法表面上看就是“换一个目标”真正写代码跑起来之后才会体会到里面很多细节决定成败。我自己在复现PyTorch版时最大的感触是代码里最难的不是网络和loss而是经验池里的数据流。目标替换、奖励重算、done重算、目标范围对齐每个环节都可能埋雷任何一个地方偷懒都会导致训练结果出现“好像有问题又好像没问题”的诡异状态。如果你准备做机械臂抓取、移动机器人导航这类真实项目我强烈建议不要一上来就套用现成的gym环境先在一个你自己能完全控制的仿真环境里把HER跑通。等到你亲手写出“无HER成功率0有HER成功率80%”的实验结果你对这个算法的理解会远超任何教程带给你的。后续可以做的扩展方向很多比如把HER和TD3结合处理更复杂的动作空间或者改成优先级采样让那些替换目标离原目标更远的样本有更高权重又或者把goal编码成向量配合图像输入做端到端的视觉目标导向强化学习。不过这些都是后话先把PyTorch版的HER彻底吃透你手里的这一把钥匙能开的门已经很多了。
返回列表