Deep RL learning[2026/8] Policy Gradient：从原理到优化技巧
·
在强化学习的浩瀚海洋中，我们最熟悉的莫过于 Q-Learning 和 DQN。这类算法属于 Value-based 方法，即先学习状态的价值，再根据价值选动作。
但在很多复杂场景（如机器人控制、大型策略空间）中，直接学习策略往往更直观、更有效。这就是 Policy-based 方法的由来。今天我们就来聊聊其中的经典—— Policy Gradient (策略梯度)，并重点剖析两个让他“脱胎换骨”的技巧。
目录：
- 什么是 Policy Gradient
- PG 的痛点：为什么需要优化？
- 关键技巧一：Add a Baseline（降低方差）
- 关键技巧二：Assign Suitable Credit（合理分配信用）
- 算法流程与代码实战
一 什么是 Policy Gradient
1.1 核心思想
想象你在训练一只小狗。如果它做了一个正确的动作，你给它零食（正向奖励）；如果它做错了，你批评它（负向奖励）。
Policy Gradient 的逻辑也是如此：
- 如果一个动作执行后，获得的奖励很高，我们就增大该动作出现的概率。
- 如果一个动作获得的奖励很低，我们就减小该动作出现的概率。
我们不直接评价“这个状态好不好”，而是直接参数化策略
，通过调整参数 θ 来最大化期望回报。
这里面的Policy对应一个神经网络,输入state,输出不同action对应的概率
1.2 优化目标的数学形式化
我们根据当前的策略跟环境交互生成一个轨迹
该轨迹对应的累积奖赏为
因为有不同的轨迹，我们的目标函数是最大化期望奖励：
1.3 求解方法梯度上升
根据梯度上升法，我们需要计算 ∇θJ(θ)。经过一系列推导：
简单理解：
：决定了往哪个方向更新参数（增大还是减小概率）。
R(τ)：权重。奖励越高，拉力越大。
二 PG 的痛点：为什么需要优化
如果你直接使用上面的公式训练，你会发现模型很难收敛，或者训练速度极慢。主要有两个核心问题：
- 方差太大： 即使是很差的策略，也可能因为运气好得到正奖励；反之亦然。（采样引起的，另一方面在某些任务中,reward总是正的）
- 信用分配不合理： 第 100 步的动作，不应该为第 1 步的奖励负责。
针对这两个问题，我们引出今天的两个核心技巧。
三 关键技巧一：Add a Baseline（降低方差）
1： Add a base line
3.1 问题：所有动作都是“好”的？
假设我们在玩一个游戏，无论怎么玩，每一局都能得 10 分以上。
在朴素 PG 公式中，由于 R(τ)>0，梯度会鼓励所有出现过的动作。虽然好动作的奖励比差动作高，会被鼓励得更多，但这种“全员鼓励”会导致训练不稳定，收敛极慢。
我们希望：好的动作概率增加，差的动作概率减小。
3.2 解决方案：引入基准线
我们修改权重项，将奖励减去一个基准值 b：
这里的 b 通常可以取为奖励的期望值，即 b=E[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 在 t=1 做了一个极好的动作，
在 t=10做了一个极差的动作，最终总分很低。
按照原始公式，t=1 的好动作也会因为总分低被“打压”。这显然是不公平的！
第 t 步的动作，只应该对 t 步之后发生的奖励负责，而不应该为 t 步之前的奖励负责
五 PG 总结
尽管策略梯度方法有很多优点，但它们也存在一些固有的挑战：
- 梯度估计方差较大：策略梯度方法通过对概率分布进行采样来选择动作。实际上，它们通过对轨迹进行采样来估计预期收益。由于采样过程本质上是随机的，因此后续迭代中估计的收益可能存在较大的方差。这会导致智能体难以高效学习，因为策略的更新在迭代之间可能会出现显著波动。
- 训练期间的不稳定性：
- 策略梯度方法对学习率等超参数非常敏感。如果学习率过高，策略参数的更新幅度可能过大，导致训练无法找到最优参数。另一方面，如果学习率过低，收敛速度可能会很慢。
- 策略梯度方法需要平衡探索和利用。如果智能体探索不足，可能无法到达最优策略的邻域。反之，如果探索过度，则无法收敛到最优策略，而是在动作空间中不断振荡。
- 样本效率低下：策略梯度方法通过执行每条策略直至终止并累加每一步的奖励来估计收益。因此，它们需要与环境进行多次交互才能获得大量的样本轨迹。对于状态空间或动作空间较大的环境，这种方法效率低下且成本高昂。
稳定性解决方案
由于策略梯度方法中稳定性问题较为普遍，开发者们采用了多种解决方案来稳定训练过程。下面，我们将介绍几种常用的策略梯度训练稳定性解决方案：
使用基线函数
由于采样效率低下，训练迭代过程中估计的收益梯度可能存在较大的方差，导致训练不稳定且速度缓慢。一种常见的降低方差的方法是使用基线函数，例如优势演员-评论家（A2C）方法。其核心思想是使用代理函数（优势函数）代替目标函数中的估计收益。
优势值计算为采样轨迹的实际收益与给定初始状态的预期收益之差。这种方法使用值函数作为状态和状态-动作对的期望值。通过将损失表示为实际收益与预期收益之差，而不是仅表示收益本身，A2C 降低了损失函数的方差，从而降低了梯度的方差，使训练更加稳定。
利用熵正则化
在某些环境下，例如奖励稀疏（只有极少数状态提供奖励）的情况下，该策略会迅速采取确定性方法。它还会采取贪婪策略，并利用已探索过的路径。这会阻碍进一步的探索，并常常导致收敛到局部最优解和次优策略。
解决方案是鼓励探索在策略过于确定时对其进行惩罚。这可以通过在目标函数中添加一个基于熵的项来实现。熵衡量策略中的随机性程度。熵越大，智能体选择的动作就越随机。这个基于熵的项是熵系数与当前策略熵的乘积。
将熵纳入目标函数有助于在开发和探索之间取得平衡。
五 算法流程与代码实战
Policy Gradient 很多简单的任务效果很好，复杂的任务很难训练
算法流程：
- 初始化策略网络 πθπθ。
- 循环 Epoch：
- 用当前策略采样 N 条轨迹 {τ1,τ2,...,τN}。
- 对每条轨迹计算每一步的 Return Rt（技巧二：Credit Assignment）。
- 计算 Baseline bb（通常取当前 Batch 的平均 Reward，或训练一个 Value Net）。
- 计算优势估计 At=Rt−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_dim=self.state_dim, hidden_dim=hidden_dim, output_dim=self.action_dim, dropout=dropout ) # 创建优化器（Adam） self.optimizer = optim.Adam(self.policy.parameters(), lr=learning_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, dtype=torch.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, color="steelblue", alpha=0.35, linewidth=0.8, label="Return (per episode)") ax1.plot(episodes, moving_average(episode_returns), color="crimson", linewidth=2.0, label="Return (moving avg)") ax1.set_xlabel("Episode") ax1.set_ylabel("Return (reward)") ax1.set_title("Return per Episode") ax1.legend() ax1.grid(alpha=0.3) # 右图：每回合策略损失 ax2.plot(episodes, episode_losses, color="darkorange", alpha=0.35, linewidth=0.8, label="Loss (per episode)") ax2.plot(episodes, moving_average(episode_losses), color="seagreen", linewidth=2.0, label="Loss (moving avg)") ax2.set_xlabel("Episode") ax2.set_ylabel("Policy loss") ax2.set_title("Policy Loss per Episode") ax2.legend() ax2.grid(alpha=0.3) fig.tight_layout() fig.savefig(save_path, dpi=120) print(f"📈 训练曲线已保存至: {save_path}") plt.close(fig) # 关闭图形释放内存，避免阻塞/泄漏 def demo(agent: 'REINFORCEAgent', num_episodes: int = 5) -> None: """ 用训练好的策略进行可视化演示（render_mode='human'，弹出窗口）。 使用贪婪策略：每个状态取概率最大的动作，不看随机采样。 返回最佳模型（训练过程中回报均值最高者）的演示效果。 """ demo_env = gym.make('CartPole-v1', render_mode='human') 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( env=env, hidden_dim=HIDDEN_DIM, dropout=DROPOUT, learning_rate=LEARNING_RATE, discount_factor=DISCOUNT_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} | " f"Mean Reward (last {N_TRIALS}): {mean_return:7.1f} | " f"Loss: {loss_value:8.4f} | " f"Return: {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_episodes=5) # 关闭环境，释放资源 env.close() # ==================== # 5. 程序入口 # ==================== if __name__ == "__main__": main()
参考：
下面是第9课
DAMO开发者矩阵，由阿里巴巴达摩院和中国互联网协会联合发起，致力于探讨最前沿的技术趋势与应用成果，搭建高质量的交流与分享平台，推动技术创新与产业应用链接，围绕“人工智能与新型计算”构建开放共享的开发者生态。
更多推荐
4
0- 0
已为社区贡献13条内容
所有评论(0)