1. 项目概述为什么选择从GRPO开始复现强化学习算法如果你和我一样是个对强化学习Reinforcement Learning, RL既着迷又时常感到无从下手的实践者那么“复现经典算法”这条路大概率是你绕不开的必修课。市面上论文和开源代码汗牛充栋但真正能跑通、能理解、能为自己所用的却不多。我决定启动一个系列亲手复现10篇有代表性的强化学习算法而第一篇我选择了GRPO。GRPO全称Guided Policy Search with Reinforcement Learning from Observations这个名字听起来有点绕。简单来说它是一种“混合动力”算法。它巧妙地将传统的最优控制或轨迹优化方法与现代的深度强化学习结合了起来。为什么选它作为开篇原因有三第一它的思想非常直观像一位经验丰富的教练传统优化方法手把手地教一个新手学徒神经网络策略如何完成任务这种“引导式”的学习过程对于理解策略搜索的本质很有帮助。第二它的实现复杂度适中既有值得深究的理论细节又不至于像一些前沿的元强化学习或分布式算法那样让初学者在工程和理论的迷宫里彻底迷失。第三GRPO在机器人控制、连续动作空间任务上表现出的稳定性和样本效率使其至今仍有很强的实用参考价值绝非过时的“古董”。这个项目适合谁适合已经了解强化学习基本概念如马尔可夫决策过程MDP、策略、价值函数但对着论文里的公式和伪代码不知如何下手的同学。也适合那些想深入某个算法内部看看梯度究竟如何计算、网络究竟如何更新的硬核实践派。我们将不止于“跑通代码”更要拆解每一行代码背后的动机搞清楚“为什么这么做”以及“不这么做会怎样”。2. GRPO核心思想与算法框架拆解在深入代码之前我们必须把GRPO的“设计图纸”看明白。很多复现失败问题都出在对算法框架的一知半解上。2.1 传统优化与深度学习的“双轨制”GRPO的核心洞察在于传统基于模型的轨迹优化方法如iLQR, DDP在局部搜索上非常高效和稳定但它们通常依赖于精确的动力学模型并且得到的“策略”往往是一个关于时间和状态的复杂函数难以泛化。而深度强化学习如TRPO, PPO能学习出强大的、可泛化的神经网络策略但在训练初期由于探索效率低下可能非常不稳定需要海量的交互样本。GRPO的想法很聪明为什么不让他们俩合作呢让传统的轨迹优化器充当“引导者”Guider在每次迭代中基于当前可能不精确的动力学模型为当前任务规划出一条或多条高质量的轨迹。然后让神经网络策略这个“学习者”Learner去模仿这些高质量的轨迹。通过反复迭代动力学模型会利用神经网络策略收集的新数据变得越来越准而神经网络策略也会在越来越准的模型生成的优质轨迹引导下变得越来越好。这就形成了一个正向循环更好的策略 - 收集更高质量的数据 - 训练出更精确的模型 - 生成更优的引导轨迹 - 训练出更好的策略。2.2 算法流程的三阶段循环一个完整的GRPO迭代周期通常包含三个阶段理解这个循环是看懂代码的关键轨迹优化阶段Trajectory Optimization利用当前估计的动力学模型和当前策略通过迭代LQR线性二次调节器等方法优化生成一系列期望的轨迹状态-动作序列(\tau^*)。这些轨迹在当前模型下是“最优”或“次优”的。这里的一个关键技巧是轨迹优化并不是天马行空地规划它会被约束在靠近当前策略行为分布的区域内以保证生成的轨迹是“可学”的不会离当前策略能力太远导致模仿失败。监督学习阶段Supervised Learning将上一步优化得到的目标轨迹 (\tau^) 作为监督信号来训练神经网络策略 (\pi_\theta(a|s))。这本质上是一个监督学习问题给定状态 (s)希望策略输出的动作 (a) 尽可能接近轨迹优化器给出的目标动作 (a^)。常用的损失函数是均方误差MSE。这一步让策略快速“学会”那些被证明是好的行为模式。模型学习阶段Model Learning使用神经网络策略在真实环境或仿真环境中交互收集新的状态-动作-下一状态数据对 ((s, a, s))。用这些数据来更新训练动力学模型 (f_\phi(s|s, a))。模型通常也是一个神经网络学习预测在状态(s)下执行动作(a)后下一个状态(s)的概率分布或确定性值。这三个阶段循环往复直到策略性能收敛。这里有一个至关重要的细节在第一阶段我们并不是用学到的动力学模型从头开始规划而是以当前策略产生的轨迹为“初始解”在其基础上进行局部优化。这保证了优化的起点是可行的提高了优化效率和稳定性。注意很多初学者会混淆“模型”的概念。在GRPO中我们同时维护两个模型一个是用于预测状态转移的动力学模型Dynamics Model另一个是我们要学习的策略模型Policy Model。它们是不同的网络承担不同的任务。3. 代码结构设计与关键模块解析现在我们进入实战环节。一个清晰、模块化的代码结构是复现成功的一半。下面是我实现的GRPO代码的核心目录结构及每个模块的职责。grpo_implementation/ ├── agents/ │ ├── __init__.py │ └── grpo_agent.py # GRPO算法主逻辑类 ├── models/ │ ├── __init__.py │ ├── policy_network.py # 策略神经网络 │ └── dynamics_model.py # 动力学模型神经网络 ├── optimizers/ │ ├── __init__.py │ └── trajectory_optimizer.py # 轨迹优化器基于iLQR/DDP ├── envs/ │ └── dummy_env.py # 示例环境如MuJoCo的HalfCheetah ├── utils/ │ ├── replay_buffer.py # 经验回放池 │ └── logger.py # 训练日志记录 ├── config.yaml # 超参数配置文件 └── train.py # 主训练脚本3.1 策略网络与动力学模型的设计要点在models/目录下我们定义了两个核心网络。策略网络policy_network.py通常采用一个简单的多层感知机MLP。对于连续动作空间输出一般是动作的均值向量同时还有一个独立输出的对数标准差向量或一个状态无关的对数标准差参数。这意味着我们学习的是一个高斯策略。import torch import torch.nn as nn import torch.nn.functional as F class GaussianPolicyMLP(nn.Module): def __init__(self, obs_dim, act_dim, hidden_sizes[256, 256]): super().__init__() layers [] prev_size obs_dim for size in hidden_sizes: layers.append(nn.Linear(prev_size, size)) layers.append(nn.ReLU()) prev_size size self.shared_layers nn.Sequential(*layers) self.mean_layer nn.Linear(prev_size, act_dim) # 对数标准差可以作为一个独立的可学习参数不依赖于状态 self.log_std nn.Parameter(torch.zeros(1, act_dim)) def forward(self, obs, deterministicFalse): features self.shared_layers(obs) mean self.mean_layer(features) log_std self.log_std.expand_as(mean) # 扩展到batch维度 if deterministic: return mean else: std torch.exp(log_std) normal_dist torch.distributions.Normal(mean, std) action normal_dist.rsample() # 使用rsample以支持重参数化 log_prob normal_dist.log_prob(action).sum(dim-1) return action, log_prob, mean, log_std动力学模型dynamics_model.py同样是一个MLP但其输入是当前状态和动作的拼接输出是下一状态的预测差值(\Delta s s - s)。预测差值通常比直接预测下一状态更稳定、更容易学习。class DeterministicDynamicsModel(nn.Module): def __init__(self, obs_dim, act_dim, hidden_sizes[200, 200]): super().__init__() self.net nn.Sequential( nn.Linear(obs_dim act_dim, hidden_sizes[0]), nn.ReLU(), nn.Linear(hidden_sizes[0], hidden_sizes[1]), nn.ReLU(), nn.Linear(hidden_sizes[1], obs_dim) # 预测状态差值 ) def forward(self, obs, act): x torch.cat([obs, act], dim-1) delta self.net(x) return delta实操心得在训练动力学模型时一定要对输入状态、动作和输出状态差值进行标准化。使用运行均值running mean和标准差running std进行标准化可以极大地稳定训练这是实践中非常关键但容易被忽略的一步。可以将标准化模块集成在DynamicsModel类内部。3.2 轨迹优化器的实现核心trajectory_optimizer.py是整个GRPO的“大脑”也是实现难度最高的部分。这里我们实现一个简化版的迭代LQRiLQR。iLQR的核心是反向传播和前向滚推。给定一条初始轨迹通常由当前策略生成和一个动力学模型iLQR通过以下步骤进行局部优化前向滚推Forward Rollout使用当前开环控制序列动作和动力学模型模拟出整条轨迹。反向传播Backward Pass从轨迹末端开始反向计算价值函数的二次近似即Q函数关于状态和动作的二次展开的系数并由此得到最优控制律的线性反馈项和开环项。前向更新Forward Pass使用新的控制律结合了反馈和开环项重新进行前向滚推得到一条新的、期望成本更低的轨迹。这个过程可以迭代多次。在我们的实现中动力学模型是神经网络因此计算梯度需要借助自动微分。我们使用torch的优化器来执行这个“优化轨迹”的过程将其视为一个关于动作序列的优化问题并利用模型的一阶、二阶信息通过自动微分计算Hessian向量积近似来指导搜索。class iLQROptimizer: def __init__(self, dynamics_model, cost_fn, horizon30, num_iter5, lr0.01): self.dynamics dynamics_model self.cost_fn cost_fn # 成本函数输入(s, a)返回标量成本 self.horizon horizon self.num_iter num_iter self.lr lr def optimize(self, init_states, init_actions): init_states: [horizon1, state_dim] init_actions: [horizon, action_dim] 返回优化后的动作序列 [horizon, action_dim] actions init_actions.clone().requires_grad_(True) # 将动作序列作为可优化参数 optimizer torch.optim.Adam([actions], lrself.lr) for i in range(self.num_iter): optimizer.zero_grad() states [init_states[0]] total_cost 0.0 # 前向滚推计算总成本 for t in range(self.horizon): next_state_pred states[t] self.dynamics(states[t], actions[t]) cost self.cost_fn(states[t], actions[t]) total_cost cost states.append(next_state_pred) # 终端成本 total_cost self.cost_fn(states[-1], torch.zeros_like(actions[-1])) total_cost.backward() optimizer.step() # 可选在这里加入线搜索确保成本确实下降 return actions.detach()注意事项上述是一个高度简化的版本真正的iLQR需要计算反馈增益矩阵K和开环项k并在前向滚推中应用它们。完整的实现涉及大量的矩阵运算和微分方程求解。为了首次复现的简洁性我们可以先用这个基于梯度的简化优化器它虽然不如标准的iLQR高效但足以阐明思想并让算法运行起来。后续可以将其替换为更高效的实现。4. GRPO智能体主循环的完整实现grpo_agent.py是粘合所有模块的胶水。它的update方法实现了第2章所述的三阶段循环。让我们一步步拆解。4.1 数据收集与缓冲区管理首先我们需要一个经验回放池ReplayBuffer来存储策略与环境交互的数据(s, a, s, r, done)。GRPO中这些数据有两个用途1) 用于训练动力学模型2) 为轨迹优化阶段提供初始策略轨迹。# 在agent的初始化中 self.replay_buffer ReplayBuffer(capacity1e6, obs_dimobs_dim, act_dimact_dim) # 收集数据的方法 def collect_rollouts(self, env, num_steps): obs, _ env.reset() for step in range(num_steps): with torch.no_grad(): obs_tensor torch.FloatTensor(obs).unsqueeze(0).to(self.device) action, _, _, _ self.policy_network(obs_tensor, deterministicFalse) action action.squeeze(0).cpu().numpy() next_obs, reward, terminated, truncated, _ env.step(action) done terminated or truncated self.replay_buffer.store(obs, action, next_obs, reward, done) obs next_obs if not done else env.reset()[0]4.2 核心训练循环拆解以下是update函数的核心逻辑def update(self, iteration): # 阶段一从缓冲区采样数据用于模型学习和生成初始轨迹 batch self.replay_buffer.sample(batch_sizeself.batch_size) obs, acts, next_obs, rewards, dones batch # --- 阶段3可并行更新动力学模型 --- dynamics_loss self.update_dynamics_model(obs, acts, next_obs) # --- 阶段1轨迹优化 --- # 1.1 使用当前策略从当前状态分布中采样多条初始轨迹 init_trajectories self.sample_trajectories_from_policy(batch_obsobs, horizonself.optimization_horizon) # 1.2 使用动力学模型和iLQR优化器对这些初始轨迹进行优化 optimized_actions_seq self.trajectory_optimizer.optimize( init_statesinit_trajectories[states], # [num_traj, horizon1, state_dim] init_actionsinit_trajectories[actions] # [num_traj, horizon, action_dim] ) # 返回 [num_traj, horizon, action_dim] # --- 阶段2监督学习更新策略 --- # 2.1 准备监督学习数据将优化后的动作作为目标 # 我们有多条轨迹将其展平 sup_states init_trajectories[states][:, :-1, :].reshape(-1, self.state_dim) # 排除最后一个状态 sup_actions_target optimized_actions_seq.reshape(-1, self.action_dim) # 2.2 计算策略的负对数似然损失对于高斯策略即MSE # 策略网络输出动作的分布参数 _, log_probs, means, log_stds self.policy_network(sup_states) # 计算目标动作在策略分布下的概率最大化它 dist torch.distributions.Normal(means, torch.exp(log_stds)) log_prob_target dist.log_prob(sup_actions_target).sum(dim-1) policy_loss -log_prob_target.mean() # 负号因为我们要最大化对数似然 # 2.3 更新策略网络参数 self.policy_optimizer.zero_grad() policy_loss.backward() torch.nn.utils.clip_grad_norm_(self.policy_network.parameters(), self.max_grad_norm) self.policy_optimizer.step() # 记录日志 self.logger.log({ iteration: iteration, dynamics_loss: dynamics_loss.item(), policy_loss: policy_loss.item(), avg_reward: np.mean(self.replay_buffer.rewards[-1000:]) if len(self.replay_buffer.rewards) 0 else 0 })4.3 超参数配置的深层逻辑config.yaml文件不是简单的参数堆砌每个值背后都有考量# config.yaml algorithm: grpo env_name: HalfCheetah-v4 # MuJoCo环境连续控制基准 # 网络结构 policy_hidden_sizes: [256, 256] dynamics_hidden_sizes: [200, 200] # 训练循环 total_timesteps: 1e6 rollout_steps_per_update: 1000 # 每次更新前用当前策略收集多少步数据 batch_size: 256 # 从缓冲区采样用于更新的批量大小 optimization_horizon: 30 # 轨迹优化的步长视野 # 优化器参数 policy_lr: 3e-4 # Adam优化器学习率RL中的常用起点 dynamics_lr: 1e-3 # 模型学习通常可以用稍大的学习率 ilqr_iterations: 5 # 每个轨迹优化周期内iLQR的迭代次数 ilqr_lr: 0.01 # 轨迹优化中梯度下降的学习率 # 正则化与稳定性 max_grad_norm: 0.5 # 梯度裁剪防止策略更新步长过大 dynamics_train_freq: 1 # 每轮迭代都训练动力学模型 policy_train_freq: 1 # 每轮迭代都训练策略 # 探索 init_log_std: -0.5 # 策略初始对数标准差控制初始探索强度关键参数解读optimization_horizon这是一个权衡。太短优化器看不到长远的收益太长优化计算量剧增且模型预测误差会随着滚推步长累积。对于HalfCheetah猎豹奔跑这类需要一定步数才能体现任务特性的环境30-50是一个合理的起点。policy_lrvsdynamics_lr策略网络需要稳定地学习因此学习率通常较低3e-4。动力学模型学习的是一个相对“平滑”的回归问题可以使用稍大的学习率1e-3以加速收敛。init_log_std初始探索强度。设为负值意味着初始策略的随机性较小倾向于利用轨迹优化器给出的“指导”。随着学习进行我们可以让这个参数也参与学习或者固定为一个小的随机性。5. 训练流程、调试与可视化实战有了所有组件我们在train.py中编写主训练循环。5.1 主训练脚本的搭建import yaml import torch import numpy as np from envs import make_env from agents import GRPOAgent from utils.logger import Logger def main(): # 加载配置 with open(config.yaml, r) as f: cfg yaml.safe_load(f) # 创建环境、智能体、日志器 env make_env(cfg[env_name]) agent GRPOAgent( obs_dimenv.observation_space.shape[0], act_dimenv.action_space.shape[0], configcfg ) logger Logger() # 初始数据收集用随机策略填充一部分缓冲区 print(Warming up replay buffer...) agent.collect_rollouts(env, num_stepscfg[batch_size]*5) # 主训练循环 global_step 0 while global_step cfg[total_timesteps]: # 1. 用当前策略收集数据 agent.collect_rollouts(env, num_stepscfg[rollout_steps_per_update]) global_step cfg[rollout_steps_per_update] # 2. 执行一次GRPO更新包含轨迹优化、策略学习、模型学习 agent.update(iterationlogger.epoch) # 3. 定期评估策略 if logger.epoch % cfg[eval_freq] 0: eval_reward evaluate_policy(agent.policy_network, env, n_episodes5) logger.log({eval_avg_reward: eval_reward}) print(fIter {logger.epoch}, Step {global_step}, Eval Reward: {eval_reward:.2f}) logger.epoch 1 # 保存最终模型 torch.save(agent.policy_network.state_dict(), final_policy.pth) env.close() def evaluate_policy(policy, env, n_episodes10): total_reward 0 for _ in range(n_episodes): obs, _ env.reset() done False episode_reward 0 while not done: with torch.no_grad(): obs_tensor torch.FloatTensor(obs).unsqueeze(0).to(agent.device) action, _, _, _ policy(obs_tensor, deterministicTrue) # 评估时使用确定性策略 action action.squeeze(0).cpu().numpy() obs, reward, terminated, truncated, _ env.step(action) done terminated or truncated episode_reward reward total_reward episode_reward return total_reward / n_episodes if __name__ __main__: main()5.2 训练过程中的关键监控指标训练时不能只看最终奖励以下几个指标的走势能帮你诊断算法状态动力学模型损失dynamics_loss这个损失应该随着数据积累逐渐下降并最终稳定在一个较低的值。如果损失一直很高或剧烈波动说明模型学习失败可能是网络结构不合适、学习率太高、或数据没有标准化。策略监督损失policy_loss在GRPO中这个损失是策略输出动作与优化器目标动作的负对数似然。理想情况下它应该随着迭代下降表明策略在成功模仿优化轨迹。如果它不降反升可能是优化轨迹质量太差模型不准或策略学习能力不足网络太小、学习率不当。评估回报eval_avg_reward这是终极指标。在HalfCheetah环境中随机策略的回报约为-1000一个训练良好的策略能达到2000以上。GRPO的典型学习曲线是初期由于模型不准、优化轨迹差回报增长缓慢甚至徘徊随着模型变准优化轨迹质量提升回报会进入快速上升期最后逐渐收敛。可视化使用TensorBoard或Weights Biases记录上述指标。特别有用的一个可视化是将优化器规划出的轨迹状态序列与策略实际执行产生的轨迹进行对比。初期它们可能相差甚远后期应该越来越吻合这直观地展示了“引导”的成功。6. 常见陷阱、调试技巧与进阶优化即使代码逻辑正确第一次运行也难免遇到问题。下面是我在复现过程中踩过的坑和总结的调试技巧。6.1 动力学模型学习失败症状dynamics_loss居高不下或爆炸导致轨迹优化器基于一个垃圾模型做规划产生毫无意义的轨迹策略学不到任何东西。排查与解决检查数据标准化这是最常见的原因。确保在将数据输入动力学模型前对状态、动作和状态差值进行了标准化。使用运行统计量均值、标准差进行标准化并在模型前向传播时应用。降低模型复杂度如果环境本身不太复杂但用了很深的网络容易过拟合早期少量数据。尝试减小网络层数和宽度。添加正则化在动力学模型的损失函数中加入L2权重衰减。检查梯度在训练动力学模型时打印出梯度的范数。如果梯度爆炸使用梯度裁剪。验证模型预测定期进行一个简单的测试从缓冲区取一批数据用训练后的动力学模型做单步预测计算预测值与真实值的平均误差。这个误差应该远小于状态本身的量级。6.2 策略无法模仿优化轨迹症状policy_loss不下降评估回报停滞。优化器可能规划出了好的轨迹但策略网络学不会。排查与解决策略网络容量增加策略网络的隐藏层大小。策略需要拟合一个可能相当复杂的映射。监督学习强度增加策略网络在每次迭代中的训练步数policy_train_freq或增加内循环epoch。模仿学习可能需要更多的梯度步。优化轨迹的可信度如果动力学模型还不准优化轨迹可能是“虚假的好”。可以尝试在早期迭代中给优化轨迹的目标动作加入噪声或者降低优化轨迹的权重让策略更多地从真实交互中学习类似于DAgger算法。探索与利用的平衡检查策略输出的标准差。如果它变得太小探索不足策略可能会陷入局部最优。可以设置一个最小标准差或者使用像PPO中那样的熵正则项来鼓励探索。6.3 训练不稳定或性能突然崩溃症状回报曲线出现断崖式下跌。排查与解决经验回放池的覆盖确保缓冲区足够大并且包含了策略不同阶段的数据。如果缓冲区只保留最新数据可能会遗忘早期学到的有用经验。可以考虑使用一个大的、均匀采样的缓冲区。模型误差累积在轨迹优化的前向滚推中模型误差会一步步累积。限制优化视野horizon可以缓解。也可以使用集成模型训练多个动力学模型用它们的平均或不确定性来规划来提高鲁棒性这是MBRL基于模型的强化学习中的常用技巧。策略更新步长过大即使使用了监督学习过大的策略更新也可能破坏已学到的知识。确保使用了梯度裁剪并可以尝试降低策略学习率。6.4 性能天花板与进阶优化方向当你的GRPO实现能够在HalfCheetah上取得不错成绩例如回报1500后可以考虑以下进阶优化集成动力学模型Ensemble Dynamics训练N个动力学模型在轨迹优化时使用它们的平均预测或者选择悲观预测worst-case来鼓励鲁棒性。这能显著缓解模型偏差问题。不确定性感知的轨迹优化不仅使用模型的预测均值还利用其预测的不确定性如集成模型输出的方差来进行规划倾向于选择模型确信度高的区域。替换更强大的轨迹优化器将我们简化的梯度优化器替换为真正的二阶方法如iLQR/DDP或者使用基于采样的优化方法如CEM交叉熵方法。引入价值函数原始的GRPO主要依赖成本函数进行优化。可以引入一个价值函数Critic来更好地评估长期回报并用于调整轨迹优化的目标这更接近现代的Actor-Critic框架与模型规划的融合。复现GRPO只是一个起点。通过这个项目你不仅得到了一个可工作的算法实现更重要的是你深入理解了基于模型的强化学习MBRL中“规划”与“学习”如何协同工作的核心范式。这种理解将为你后续复现更复杂的RL算法如PETS, MBPO, Dreamer等打下坚实的基础。在接下来的系列文章中我们将挑战样本效率更高、或处理更复杂状态如图像的算法。