在强化学习的浩瀚海洋中我们最熟悉的莫过于 Q-Learning 和 DQN。这类算法属于Value-based方法即先学习状态的价值再根据价值选动作。但在很多复杂场景如机器人控制、大型策略空间中直接学习策略往往更直观、更有效。这就是Policy-based方法的由来。今天我们就来聊聊其中的经典——Policy Gradient (策略梯度)并重点剖析两个让他“脱胎换骨”的技巧。目录什么是 Policy GradientPG 的痛点为什么需要优化关键技巧一Add a Baseline降低方差关键技巧二Assign Suitable Credit合理分配信用算法流程与代码实战一 什么是 Policy Gradient1.1 核心思想想象你在训练一只小狗。如果它做了一个正确的动作你给它零食正向奖励如果它做错了你批评它负向奖励。Policy Gradient 的逻辑也是如此如果一个动作执行后获得的奖励很高我们就增大该动作出现的概率。如果一个动作获得的奖励很低我们就减小该动作出现的概率。我们不直接评价“这个状态好不好”而是直接参数化策略通过调整参数 θ 来最大化期望回报。这里面的Policy对应一个神经网络,输入state,输出不同action对应的概率1.2 优化目标的数学形式化我们根据当前的策略跟环境交互生成一个轨迹该轨迹对应的累积奖赏为因为有不同的轨迹我们的目标函数是最大化期望奖励1.3 求解方法梯度上升根据梯度上升法我们需要计算 ∇θJ(θ)。经过一系列推导简单理解决定了往哪个方向更新参数增大还是减小概率。R(τ)权重。奖励越高拉力越大。二 PG 的痛点为什么需要优化如果你直接使用上面的公式训练你会发现模型很难收敛或者训练速度极慢。主要有两个核心问题方差太大即使是很差的策略也可能因为运气好得到正奖励反之亦然。采样引起的另一方面在某些任务中,reward总是正的信用分配不合理第 100 步的动作不应该为第 1 步的奖励负责。针对这两个问题我们引出今天的两个核心技巧。三 关键技巧一Add a Baseline降低方差1 Add a base line3.1 问题所有动作都是“好”的假设我们在玩一个游戏无论怎么玩每一局都能得 10 分以上。在朴素 PG 公式中由于 R(τ)0梯度会鼓励所有出现过的动作。虽然好动作的奖励比差动作高会被鼓励得更多但这种“全员鼓励”会导致训练不稳定收敛极慢。我们希望好的动作概率增加差的动作概率减小。3.2 解决方案引入基准线我们修改权重项将奖励减去一个基准值 b这里的 b 通常可以取为奖励的期望值即 bE[R]。效果如果 R(τ)b好于平均权重为正增加概率。如果 R(τ)b差于平均权重为负减小概率。这就是“去中心化”的思想只有比平均水平好的动作才值得被鼓励2 Add a suitable Credit解决方案3.3 数学证明不改变期望降低方差为什么这样做有效直观上减去 b 让权重有了正负之分减少了无意义的正向推动。数学上可以证明引入 b不会改变梯度的期望因为 ∇log⁡π 的期望为 0但能显著降低方差。低方差意味着梯度估计更准确训练更稳定注在 Actor-Critic 算法中这个 Baseline 实际上就是 Critic 网络输出的 V(s)利用优势函数 A(s,a)Q(s,a)−V(s) 来代替 R−b。四 关键技巧二Assign Suitable Credit合理分配信用4.1 问题当前动作要为历史负责吗在原始公式中我们使用整条轨迹的奖励和 R(τ)作为权重。这存在一个逻辑漏洞假设一个 Agent 在 t1 做了一个极好的动作在 t10做了一个极差的动作最终总分很低。按照原始公式t1 的好动作也会因为总分低被“打压”。这显然是不公平的第 t 步的动作只应该对 t 步之后发生的奖励负责而不应该为 t 步之前的奖励负责五 PG 总结尽管策略梯度方法有很多优点但它们也存在一些固有的挑战梯度估计方差较大策略梯度方法通过对概率分布进行采样来选择动作。实际上它们通过对轨迹进行采样来估计预期收益。由于采样过程本质上是随机的因此后续迭代中估计的收益可能存在较大的方差。这会导致智能体难以高效学习因为策略的更新在迭代之间可能会出现显著波动。训练期间的不稳定性策略梯度方法对学习率等超参数非常敏感。如果学习率过高策略参数的更新幅度可能过大导致训练无法找到最优参数。另一方面如果学习率过低收敛速度可能会很慢。策略梯度方法需要平衡探索和利用。如果智能体探索不足可能无法到达最优策略的邻域。反之如果探索过度则无法收敛到最优策略而是在动作空间中不断振荡。样本效率低下策略梯度方法通过执行每条策略直至终止并累加每一步的奖励来估计收益。因此它们需要与环境进行多次交互才能获得大量的样本轨迹。对于状态空间或动作空间较大的环境这种方法效率低下且成本高昂。稳定性解决方案由于策略梯度方法中稳定性问题较为普遍开发者们采用了多种解决方案来稳定训练过程。下面我们将介绍几种常用的策略梯度训练稳定性解决方案使用基线函数由于采样效率低下训练迭代过程中估计的收益梯度可能存在较大的方差导致训练不稳定且速度缓慢。一种常见的降低方差的方法是使用基线函数例如优势演员-评论家A2C方法。其核心思想是使用代理函数优势函数代替目标函数中的估计收益。优势值计算为采样轨迹的实际收益与给定初始状态的预期收益之差。这种方法使用值函数作为状态和状态-动作对的期望值。通过将损失表示为实际收益与预期收益之差而不是仅表示收益本身A2C 降低了损失函数的方差从而降低了梯度的方差使训练更加稳定。利用熵正则化在某些环境下例如奖励稀疏只有极少数状态提供奖励的情况下该策略会迅速采取确定性方法。它还会采取贪婪策略并利用已探索过的路径。这会阻碍进一步的探索并常常​​导致收敛到局部最优解和次优策略。解决方案是鼓励探索在策略过于确定时对其进行惩罚。这可以通过在目标函数中添加一个基于熵的项来实现。熵衡量策略中的随机性程度。熵越大智能体选择的动作就越随机。这个基于熵的项是熵系数与当前策略熵的乘积。将熵纳入目标函数有助于在开发和探索之间取得平衡。五 算法流程与代码实战Policy Gradient 很多简单的任务效果很好复杂的任务很难训练算法流程初始化策略网络 πθπθ​。循环 Epoch用当前策略采样 N 条轨迹 {τ1,τ2,...,τN}。对每条轨迹计算每一步的 Return Rt技巧二Credit Assignment。计算 Baseline bb通常取当前 Batch 的平均 Reward或训练一个 Value Net。计算优势估计 AtRt−b技巧一Baseline。计算损失函数 L−∑log⁡πθ(at∣st)×At。反向传播更新 θ。 基于策略梯度REINFORCE算法的强化学习实现用于解决OpenAI Gym的CartPole-v1环境。 该实现采用面向对象设计将智能体Agent、策略网络PolicyNetwork和训练流程分离 便于模块化扩展和实验。 环境说明CartPole-v1倒立摆/推车平衡 · 任务通过左右推动小车使竖立在车上的杆子尽可能长时间保持平衡不倒。 · State状态4维连续向量 obs[0] x 小车位置Cart Position范围 ≈ [-4.8, 4.8] obs[1] ẋ 小车速度Cart Velocity obs[2] θ 杆子与竖直方向的夹角Pole Angle弧度 ≈ [-0.418, 0.418] obs[3] θ̇ 杆子角速度Pole Angular Velocity 维度0、2用于判断“是否出界/倾倒”维度1、3反映运动趋势更适合预测下一步。 · Action动作2维离散 0 向左推小车push cart to the left 1 向右推小车push cart to the right · Reward奖励 每维持一个时间步杆子不倒获得 1 奖励。 episode 结束终止条件 ① 杆子倾斜超过 12°≈0.209 弧度—— 倾倒失败 ② 小车移出轨道|x| 2.4—— 脱轨失败 ③ 达到最大步数 500 —— 完美平衡正常结束。 因此“单局回报”≈ “杆子坚持的步数”最大值 500视为已解决。 import torch import torch.nn as nn import torch.optim as optim import torch.nn.functional as F import torch.distributions as distributions import numpy as np import matplotlib.pyplot as plt import gymnasium as gym from typing import Tuple, List, Dict, Any, Optional # # 1. 环境初始化 # # 创建CartPole环境该环境具有连续的状态空间4维和离散的动作空间2维左/右。 env gym.make(CartPole-v1) # # 2. 策略网络定义 # class PolicyNetwork(nn.Module): 策略网络Policy Network用于近似随机策略 π(a|s)。 该网络将状态映射为动作的对数概率未归一化通过Softmax函数输出动作概率分布。 参数: input_dim (int): 状态空间的维度。 hidden_dim (int): 隐藏层的神经元数量。 output_dim (int): 动作空间的维度。 dropout (float): Dropout层的丢弃率用于正则化。 def __init__(self, input_dim: int, hidden_dim: int, output_dim: int, dropout: float): super().__init__() # 第一层全连接层输入 - 隐藏层 self.layer1 nn.Linear(input_dim, hidden_dim) # 第二层全连接层隐藏层 - 输出层动作的未归一化分数 self.layer2 nn.Linear(hidden_dim, output_dim) # Dropout层用于在训练时随机丢弃部分神经元防止过拟合 self.dropout nn.Dropout(dropout) def forward(self, x: torch.Tensor) - torch.Tensor: 前向传播过程。 参数: x (torch.Tensor): 输入状态张量形状为 (batch_size, input_dim)。 返回: torch.Tensor: 动作的未归一化分数logits形状为 (batch_size, output_dim)。 x self.layer1(x) # 线性变换 x self.dropout(x) # 应用Dropout训练时生效 x F.relu(x) # ReLU激活函数引入非线性 x self.layer2(x) # 输出层得到动作的logits return x # # 3. 智能体Agent类定义 # class REINFORCEAgent: REINFORCE算法智能体。 该类封装了策略网络、优化器、超参数以及与环境交互、学习更新的完整流程。 参数: env (gym.Env): Gym环境实例。 hidden_dim (int): 策略网络隐藏层维度。 dropout (float): Dropout率。 learning_rate (float): 优化器学习率。 discount_factor (float): 折扣因子 γ。 def __init__( self, env: gym.Env, hidden_dim: int 128, dropout: float 0.5, learning_rate: float 0.01, discount_factor: float 0.99 ): self.env env self.discount_factor discount_factor # 获取状态和动作空间的维度 self.state_dim env.observation_space.shape[0] self.action_dim env.action_space.n # 创建策略网络 self.policy PolicyNetwork( input_dimself.state_dim, hidden_dimhidden_dim, output_dimself.action_dim, dropoutdropout ) # 创建优化器Adam self.optimizer optim.Adam(self.policy.parameters(), lrlearning_rate) # 用于存储当前episode轨迹的缓冲区 self.log_probs: List[torch.Tensor] [] # 动作对数概率 self.rewards: List[float] [] # 即时奖励 # 训练状态标志 self.is_training True def reset_trajectory(self) - None: 清空当前episode的轨迹缓冲区为采集新轨迹做准备。 self.log_probs [] self.rewards [] def select_action(self, state: np.ndarray) - int: 根据当前策略选择动作采样模式。 参数: state (np.ndarray): 当前环境状态。 返回: int: 选择的动作索引。 # 将numpy数组转换为张量并增加batch维度 state_tensor torch.FloatTensor(state).unsqueeze(0) # 前向传播得到动作logits action_logits self.policy(state_tensor) # 通过Softmax计算动作概率分布 action_probs F.softmax(action_logits, dim-1) # 创建分类分布 dist distributions.Categorical(action_probs) # 采样动作 action dist.sample() # 记录该动作的对数概率用于后续策略梯度更新 log_prob dist.log_prob(action) self.log_probs.append(log_prob) return action.item() def step(self, action: int) - Tuple[np.ndarray, float, bool, bool, Dict[str, Any]]: 执行动作并记录奖励。 参数: action (int): 要执行的动作。 返回: Tuple: (下一状态, 奖励, 终止标志, 截断标志, 额外信息) next_state, reward, terminated, truncated, info self.env.step(action) self.rewards.append(reward) return next_state, reward, terminated, truncated, info def collect_episode(self) - float: 采集一个完整的episode轨迹。 返回: float: 该episode的总奖励。 # 重置轨迹缓冲区 self.reset_trajectory() # 重置环境获取初始状态 state, _ self.env.reset() episode_return 0.0 done False # 设置网络为训练模式Dropout生效 self.policy.train() while not done: # 选择动作 action self.select_action(state) # 执行动作必须通过 self.step()其内部会记录奖励用于回报计算 next_state, reward, terminated, truncated, _ self.step(action) done terminated or truncated # 更新状态和累积奖励 state next_state episode_return reward return episode_return def compute_returns(self) - torch.Tensor: 计算标准化后的折扣回报。 返回: torch.Tensor: 标准化后的每步回报张量形状为 (T,)T为episode步数。 # 计算未标准化的折扣回报 returns [] R 0.0 for r in reversed(self.rewards): R r R * self.discount_factor returns.insert(0, R) returns_tensor torch.tensor(returns, dtypetorch.float32) # 标准化处理降低方差 mean returns_tensor.mean() std returns_tensor.std() if std 1e-8: # 避免除以零 normalized_returns (returns_tensor - mean) / std else: normalized_returns returns_tensor - mean # 如果标准差为0仅中心化 return normalized_returns def update_policy(self) - float: 使用收集到的轨迹数据更新策略网络。 返回: float: 更新后的损失值。 # 计算折扣回报 returns self.compute_returns() # 将回报从计算图中分离视为常数不进行梯度传播 returns returns.detach() # 将存储的对数概率拼接成张量 log_probs torch.cat(self.log_probs) # 计算策略梯度损失: L -Σ (G_t * log π(a_t|s_t)) loss -(returns * log_probs).sum() # 反向传播更新参数 self.optimizer.zero_grad() loss.backward() self.optimizer.step() return loss.item() def learn_from_episode(self) - Tuple[float, float]: 采集一个episode的轨迹并立即更新策略。 这是REINFORCE算法的主接口函数。 返回: Tuple[float, float]: (该episode总奖励, 损失值) # 1. 采集轨迹 episode_return self.collect_episode() # 2. 更新策略 loss_value self.update_policy() return episode_return, loss_value def set_train_mode(self, mode: bool True) - None: 设置训练/评估模式。 self.is_training mode self.policy.train(mode) def get_action_probs(self, state: np.ndarray) - np.ndarray: 获取给定状态下所有动作的概率分布用于评估或可视化。 参数: state (np.ndarray): 状态。 返回: np.ndarray: 动作概率数组。 self.policy.eval() # 评估模式关闭Dropout with torch.no_grad(): state_tensor torch.FloatTensor(state).unsqueeze(0) action_logits self.policy(state_tensor) action_probs F.softmax(action_logits, dim-1) return action_probs.squeeze().cpu().numpy() # # 4. 训练主程序 # def plot_training_curves(episode_returns: List[float], episode_losses: List[float], save_path: str training_curves.png) - None: 绘制训练曲线左侧为每回合回报(Reward/Return)右侧为每回合策略损失(Loss)。 将图形保存为 PNG 文件便于复盘训练过程与判断收敛情况。 episodes list(range(1, len(episode_returns) 1)) # 滑动平均让趋势更平滑、更易观察 def moving_average(data: List[float], window: int 25) - List[float]: if not data: return data window min(window, len(data)) return [sum(data[max(0, i - window 1): i 1]) / len(data[max(0, i - window 1): i 1]) for i in range(len(data))] fig, (ax1, ax2) plt.subplots(1, 2, figsize(14, 5)) # 左图每回合回报 ax1.plot(episodes, episode_returns, colorsteelblue, alpha0.35, linewidth0.8, labelReturn (per episode)) ax1.plot(episodes, moving_average(episode_returns), colorcrimson, linewidth2.0, labelReturn (moving avg)) ax1.set_xlabel(Episode) ax1.set_ylabel(Return (reward)) ax1.set_title(Return per Episode) ax1.legend() ax1.grid(alpha0.3) # 右图每回合策略损失 ax2.plot(episodes, episode_losses, colordarkorange, alpha0.35, linewidth0.8, labelLoss (per episode)) ax2.plot(episodes, moving_average(episode_losses), colorseagreen, linewidth2.0, labelLoss (moving avg)) ax2.set_xlabel(Episode) ax2.set_ylabel(Policy loss) ax2.set_title(Policy Loss per Episode) ax2.legend() ax2.grid(alpha0.3) fig.tight_layout() fig.savefig(save_path, dpi120) print(f 训练曲线已保存至: {save_path}) plt.close(fig) # 关闭图形释放内存避免阻塞/泄漏 def demo(agent: REINFORCEAgent, num_episodes: int 5) - None: 用训练好的策略进行可视化演示render_modehuman弹出窗口。 使用贪婪策略每个状态取概率最大的动作不看随机采样。 返回最佳模型训练过程中回报均值最高者的演示效果。 demo_env gym.make(CartPole-v1, render_modehuman) agent.set_train_mode(False) # 评估模式关闭 Dropout print( * 60) print( 使用最优模型进行可视化演示) print( * 60) for ep in range(1, num_episodes 1): state, _ demo_env.reset() done False total_reward 0.0 while not done: demo_env.render() with torch.no_grad(): action_probs agent.get_action_probs(state) action int(np.argmax(action_probs)) # 贪婪选择最有可能的动作 state, reward, terminated, truncated, _ demo_env.step(action) total_reward reward done terminated or truncated print(f演示回合 {ep}: 回报 {total_reward:.0f} | f{✅成功平衡(≥475) if total_reward 475 else ❌提前结束}) demo_env.close() print( * 60) print(演示结束关闭演示窗口) print( * 60) def main(): 主训练函数创建Agent并执行训练循环。 # ---------- 超参数设置 ---------- MAX_EPOCHS 500 # 最大训练回合数 DISCOUNT_FACTOR 0.99 # 折扣因子 γ N_TRIALS 25 # 用于计算平均回报的最近回合数 REWARD_THRESHOLD 475 # 成功阈值CartPole-v1 求解标准 475 PRINT_INTERVAL 10 # 打印训练信息的间隔 HIDDEN_DIM 128 # 隐藏层神经元数量 DROPOUT 0.5 # Dropout率 LEARNING_RATE 0.01 # 学习率 # ---------- 创建环境 ---------- env gym.make(CartPole-v1) # ---------- 创建智能体 ---------- agent REINFORCEAgent( envenv, hidden_dimHIDDEN_DIM, dropoutDROPOUT, learning_rateLEARNING_RATE, discount_factorDISCOUNT_FACTOR ) # ---------- 训练记录 ---------- episode_returns: List[float] [] episode_losses: List[float] [] # 记录训练过程中“最优模型”最近 N_TRIALS 平均回报最高者 best_mean_reward -float(inf) best_episode 0 best_state_dict None # ---------- 训练循环 ---------- print( * 60) print(开始训练 REINFORCE 智能体) print( * 60) for episode in range(1, MAX_EPOCHS 1): # Agent采集一个episode并学习更新 episode_return, loss_value agent.learn_from_episode() # 记录数据 episode_returns.append(episode_return) episode_losses.append(loss_value) # 计算最近 N_TRIALS 个episode的平均回报 mean_return np.mean(episode_returns[-N_TRIALS:]) # 若当前平均回报刷新纪录则快照此刻的“最优模型” if mean_return best_mean_reward: best_mean_reward mean_return best_episode episode best_state_dict {k: v.detach().clone() for k, v in agent.policy.state_dict().items()} # 定期打印训练进度 if episode % PRINT_INTERVAL 0: print( f| Episode: {episode:4d} | fMean Reward (last {N_TRIALS}): {mean_return:7.1f} | fLoss: {loss_value:8.4f} | fReturn: {episode_return:6.1f} | ) # 早停条件如果平均回报达到阈值则认为问题已解决 if mean_return REWARD_THRESHOLD: print( * 60) print(f 成功在 {episode} 个episode后达到奖励阈值 {REWARD_THRESHOLD}) print( * 60) break else: # 如果循环正常结束未触发break print( * 60) print(f训练完成。最大平均奖励: {np.mean(episode_returns[-N_TRIALS:]):.1f}) print( * 60) # ---------- 训练收尾保存曲线 最优模型演示 ---------- # 1) 绘制并保存训练曲线回报 损失 plot_training_curves(episode_returns, episode_losses) # 2) 载入训练过程中平均回报最高的“最优模型” if best_state_dict is not None: agent.policy.load_state_dict(best_state_dict) print(f 已载入最优模型Episode {best_episode}平均回报 {best_mean_reward:.1f}) else: print(ℹ️ 未找到更优模型使用最终策略进行演示) # 3) 用最优模型做可视化演示 demo(agent, num_episodes5) # 关闭环境释放资源 env.close() # # 5. 程序入口 # if __name__ __main__: main()参考Reinforcement-Learning/4. Polict Gradient Methods/main.py at main · cristianleoo/Reinforcement-Learning · GitHubhttps://www.bilibili.com/video/BV1XP4y1d7Bk?spm_id_from333.788.videopod.episodesvd_sourcea624c4a1aea4b867c580cc82f03c1745p4下面是第9课https://www.bilibili.com/video/BV1XP4y1d7Bk?spm_id_from333.788.videopod.episodesvd_sourcea624c4a1aea4b867c580cc82f03c1745p4