图解强化学习 |Policy Gradients（策略梯度）

欢迎来到图解强化学习的世界博客主页卿云阁欢迎关注点赞收藏⭐️留言首发时间2026年4月5日✉️希望可以和大家一起完成进阶之路作者水平很有限如果发现错误请留言轰炸哦万分感谢目录什么是策略梯度算法PG 的核心驱动力——蒙特卡洛Monte Carlo与 G 值计算代码实现什么是策略梯度算法不通过分析奖励值, 直接输出行为的方法输出预测的动作对比起以值为基础的方法, Policy Gradients 直接输出动作的最大好处就是,它能在一个连续区间内挑选动作, 而基于值的, 比如 Q-learning, 它如果在无穷多的动作中计算价值, 从而选择行为, 这,它可吃不消。DQN 家族的“瓶颈”在 DQN 中AI 的每一步都在疯狂计算公式Q(s, a)到底等于几分如果是玩贪吃蛇只有上下左右 4 个动作DQN 游刃有余。但如果我们在教机械臂抓取物体或者控制自动驾驶汽车的油门和方向盘动作是连续且无穷无尽的比如方向盘可以打 12.5 度也可以打 12.51 度。这时候如果让 DQN 去算出每一个微小角度的 Q 值然后再用 argmax 挑最大的电脑的算力当场就会崩溃。Policy Gradients 的“降维打击”PG 算法极其简单粗暴我不管这个动作值多少分我只管此时此刻我该以多大的“概率”去做这个动作它绕过了“估值”这个中间商。我们将画面输入进神经网络网络直接在输出层吐出一个概率分布 pi(a|s)比如60% 概率向左40% 概率向右。动作怎么选 直接根据这个概率分布掷骰子采样。连续动作怎么搞网络可以直接输出高斯分布的“均值”和“方差”在无穷的连续空间里极其优雅地完成动作选择。PG 的核心驱动力——蒙特卡洛Monte Carlo与 G 值计算为什么在 DQN 里小车每走一步网络就在疯狂学习而在 PG 里AI 必须强忍着直到小车彻底死掉或者通关游戏结束才敢去更新大脑⏳延迟奖励问题回合制更新在 DQN 中我们有 Q 值对未来的预估作为缓冲所以每走一步都能算出一个误差。但PG 抛弃了预估直接输出概率它手头没有任何用来判断对错的参考答案。你前 90 步可能都在送人头、丢阵地每步的即时 Reward 都是 0 甚至是负数但就在第100 步你突然一招反杀赢下了整局比赛拿到最终 Reward 1。如果在第 10 步就急着去评判刚才那个动作好不好AI 绝对会得出错误的结论。PG 必须以“整局游戏Episode”为单位走到大结局拿到那个盖棺定论的最终得分然后再顺藤摸瓜地回溯去评价沿途每一个动作的真实价值。这就叫做蒙特卡洛方法Monte Carlo Method。 严禁打乱的记忆回放区只记录 (s, a, r) 三要素严格按照时间顺序DQN 的错题本把所有经历打碎成独立的五元组 (s, a, r, s, done)扔进池子里然后均匀随机抽样。时间顺序彻底被抹杀了。PG 的记事本绝对不能打乱我们只记录 (s, a, r) 三要素并且必须像写日记一样严格按照时间顺序一笔一笔地记下来。因为如果打乱了顺序我们就根本无法从大结局往回倒推 G 值了这些记录用完一次就会被立刻清空。 G 值的逆向推演假设小车只存活了 3 步就结束了游戏每一步拿到的即时奖励是ep_rs [0.0, 0.0, 1.0]我们设定衰减系数gamma 0.9。我们要计算的是每个时刻的真实回报G值它的核心递归公式非常优雅ep_rs [0.0, 0.0, 1.0]倒数第 1 步 (t 2, 大结局)即时奖励 R_2 为 1.0。running_add 0 *0.91.0 1.0所以G_2 1.0。(大结局的这一步实打实地拿到了 1 分)倒数第 2 步 (t 1)即时奖励 R_1 为 0.0。running_add 1.0 *0.9 0.0 0.9所以G_1 0.9。(虽然这一步没得分但因为它促成了大结局的胜利所以沾光拿到了 0.9 分)倒数第 3 步 (t 0, 开局)即时奖励 R_0 为 0.0。running_add 0.9 * 0.9 0.0 0.81所以G_0 0.81。(开局的第一步同样因为最终的胜利被赋予了 0.81 的高权重)最终计算出来的 G 值列表是[0.81, 0.9, 1.0]。️ 数据归一化在拿到 [0.81, 0.9, 1.0] 后我们在实际写代码时必定会做一步操作减去均值除以标准差。如果一个游戏的奖励永远都是正数比如存活越久得分越多那么所有的 G 值算出来全都是正的。如果全是正数网络就会盲目增加所有动作的概率学得极慢。做了归一化之后原本的[0.81, 0.9, 1.0] 可能会变成 [-1.2, 0.0, 1.2]。这样立刻就有了“好与坏”的绝对对比表现低于平均水平的动作负数会被狠狠打压表现好的动作正数才会被鼓励我们在实际写代码时必定会做一步操作减去均值除以标准差。模块三神奇的反向传播没有标签怎么算 Loss在传统的深度学习比如图像识别中算 Loss 是一件理所当然的事网络预测是猫真实标签Label是狗两者一减算出误差搞定。但在强化学习的迷雾里AI 面临一个绝境走到十字路口左转还是右转上帝环境根本没有给标准答案没有真实标签PyTorch 怎么loss.backward() 第一步假装自己是对的 (Fake it till you make it)既然没有标准答案PG 算法做了一个极其大胆的决定AI 这次碰巧选了什么动作我们就把这个动作当成绝对正确的“临时标签”。假设网络输出的概率是 [向左 30%, 向右 70%]。AI 掷骰子选了“向左”。 在代码中我们直接调用分类任务中最常用的交叉熵损失函数Cross Entropy强行让网络向着“向左”去优化。交叉熵的数学本质就是计算负对数概率危险警告如果只算到这一步就去更新参数AI 就会陷入盲目自信不管做的好坏都会强行增加刚才那个动作的概率。这就需要下一步来救场⚖️ 第二步G 值的无情审判 (Reward-Guided Loss)这就是我们在模块二千辛万苦算出来的 归一化 G 值 闪亮登场的时刻我们把刚才算出来的交叉熵直接乘以这局游戏大结局给出的 G 值代码实现策略模型给定状态生成各个动作的概率PolicyModel# 策略模型给定状态生成各个动作的概率 class PolicyModel(nn.Module): def __init__(self, input_dim, output_dim): super(PolicyModel, self).__init__() # 使用全连接层构建一个简单的神经网络ReLU作为激活函数 # 最后加一个Softmax层使得输出可以看作是概率分布 self.fc nn.Sequential( nn.Linear(input_dim, 128), nn.ReLU(), nn.Linear(128, 128), nn.ReLU(), nn.Linear(128, output_dim), ) # 定义前向传播输出动作概率 def forward(self, x): action_prob torch.sigmoid(self.fc(x)) # 动作空间只有0和1因此采用sigmoid函数直接输出动作概率 return action_prob输入维度 (input_dim 4)小车位置、小车速度、木棍角度、木棍角速度。输出维度 (output_dim 1)因为只有“向左0”和“向右1”两个动作我们不需要输出两个数字只需要输出 1 个维度的数字代表“执行动作 1向右的概率”。x tensor([[-0.05, -0.5, -0.15, -0.8]])self.fc(x)-2.5action_prob 0.075 执行动作 1向右的概率是7.5%。定义Policy Gradient Methods类(PGM)构造函数(__init__)参数包含环境学习率折扣因子经验回放缓冲区大小目标网络更新频率使用策略模型生成动作概率分布并采样# 使用策略模型生成动作概率分布并采样 def choose_action(self, state): # 将状态转换为tensor输入模型 state torch.FloatTensor(np.array([state])).to(self.device) probs self.policy_model(state.to(self.device)) # 生成分布后采样返回动作 m torch.distributions.Bernoulli(probs) # 伯努利分布 action int(m.sample().item()) # 转为标量 return actionstate tensor([[ 0.01, 0.15, -0.05, -0.22]])probs tensor([[0.85]])action1 小车向右定义Policy Gradient Methods类剧本设定小车的生死 3 步假设在这局游戏里小车只存活了 3 步就倒了gamma 0。在这 3 步里我们记录下的核心数据如下actions (真实动作)[1, 0, 1] 向右向左向右rewards (即时奖励)[1.0, 1.0, 1.0] 小车活了 3 步每步环境都给 1 分action_prob (事后网络对当时动作 1 的概率预测)[0.6, 0.3, 0.8]⏳ 计算 G 值running_add 0 for i in reversed(range(len(rewards))): # ... (假设进入 else 分支) running_add running_add * self.gamma rewards[i] rewards[i] running_add代码从最后一步i 2开始往回倒推第 3 步 (i 2)running_add 0 *0.9 1.0 1.0。 rewards[2] 1.0第 2 步 (i 1)running_add 1.0 *0.9 1.0 1.9。 rewards[1] 1.9第 1 步 (i 0)running_add 1.9 * 0.9 1.0 2.71。 rewards[0] 2.71推演结果原本清一色的 [1.0, 1.0, 1.0]经过时间发酵变成了 [2.71, 1.9, 1.0]。⚖️ 奖励标准化reward_mean np.mean(rewards) reward_std np.std(rewards) for i in range(len(rewards)): rewards[i] (rewards[i] - reward_mean) / reward_stdrewards[1.20, 0.04, -1.24]# i 0 的循环 action_prob self.policy_model(state) # 输出 0.5 c torch.distributions.Bernoulli(action_prob) loss -c.log_prob(action.to(self.device)) * reward # action1, reward2.0 loss.backward()完整代码​ # 模型更新 def update(self, buffer): states, actions, rewards, next_states, dones zip(*buffer) states, actions, rewards list(states), list(actions), list(rewards) # 对奖励进行修正考虑未来并加入衰减因子 running_add 0 for i in reversed(range(len(rewards))): if rewards[i] 0: running_add 0 else: running_add running_add * self.gamma rewards[i] rewards[i] running_add reward_mean np.mean(rewards) # 求奖励均值 reward_std np.std(rewards) # 求奖励标准差 for i in range(len(rewards)): # 标准化奖励 rewards[i] (rewards[i] - reward_mean) / reward_std # 梯度下降 self.optimizer.zero_grad() loss 0 # 初始化损失 for i in range(len(rewards)): state states[i] action torch.FloatTensor([actions[i]]) reward rewards[i] state torch.FloatTensor(np.array([state])).to(self.device) action_prob self.policy_model(state) c torch.distributions.Bernoulli(action_prob) # 加权(reward)损失函数加负号(将最大化问题转化为最小化问题) loss -c.log_prob(action.to(self.device)) * reward loss.backward() self.optimizer.step() ​主函数# 定义超参数 max_episodes 2000 # 训练episode数 max_steps 500 # 每个回合的最大步数 # batch_size 10 # 采样数 # 创建DQN对象 agent PGM(env) # 定义保存每个回合奖励的列表 episode_rewards [] # 开始循环tqdm用于显示进度条并评估任务时间开销 for episode in tqdm(range(max_episodes), filesys.stdout): # 重置环境并获取初始状态 state, _ env.reset() # 当前回合的奖励 episode_reward 0 # 记录每个episode的信息 buffer [] # 循环进行每一步操作 for step in range(max_steps): # 根据当前状态选择动作 action agent.choose_action(state) # 执行动作获取新的信息 next_state, reward, terminated, truncated, info env.step(action) # 判断是否达到终止状态 done terminated or truncated # 将这个五元组加到buffer中 buffer.append((state, action, reward, next_state, done)) # 累计奖励 episode_reward reward # 更新当前状态 state next_state if done: break # 更新策略 agent.update(buffer) # 记录当前回合奖励值 episode_rewards.append(episode_reward) # 打印中间值 if episode % (max_episodes // 10) 0: tqdm.write(Episode str(episode) : str(episode_reward))