PPO解决了什么痛点为什么PPO提高了数据的利用率总结传统 PG 不能多次利用因为它是“死脑筋”只能吃最新鲜的数据。数据一旦导致了脑子升级旧数据就立刻和新脑子八字不合强行用会导致网络崩溃。PPO 能多次利用因为它自带了一个“数据折算翻译官重要性采样”并且给自己戴上了“绝对不瞎改Clip 截断”的锁链。这使得旧数据在一段时间内保鲜从而让采样效率瞬间翻了近 10 倍PPO算法的执行流程要彻底弄懂 PPO最核心的就是要理清它“采样玩游戏”和“学习更新参数”这两步是如何交替进行的。传统策略梯度算法是“玩一把学一次扔掉数据”而 PPO 的执行流程可以被形象地概括为“用旧脑子玩一堆游戏存下来然后关起门来用这批录像把新脑子反反复复打磨好几遍最后拿新脑子去顶替旧脑子。”PPO 算法的完整执行流程分为以下 4 个标准步骤代码实现import gym # 导入 OpenAI Gym 库用于构建强化学习的物理或游戏交互环境 import torch # 导入 PyTorch 深度学习框架的核心库 import torch.nn.functional as F # 导入 PyTorch 的函数式接口包含激活函数 ReLU, 损失函数 MSE 等 import numpy as np # 导入 NumPy 库用于高效的数组和矩阵运算 import matplotlib.pyplot as plt # 导入 Matplotlib用于训练后的数据可视化与画图 import rl_utils # 导入自定义的强化学习工具包包含计算广义优势估计 GAE 等辅助函数 class PolicyNet(torch.nn.Module): 策略网络 (Actor)根据当前状态输出每个离散动作的概率分布 def __init__(self, state_dim, hidden_dim, action_dim): super(PolicyNet, self).__init__() # 调用父类 nn.Module 的初始化方法 self.fc1 torch.nn.Linear(state_dim, hidden_dim) # 定义第一层全连接层输入为状态维度输出为隐藏层维度 self.fc2 torch.nn.Linear(hidden_dim, action_dim) # 定义第二层全连接层输入为隐藏层维度输出为动作空间维度 def forward(self, x): x F.relu(self.fc1(x)) # 输入状态 x 经过第一层全连接层后使用 ReLU 激活函数增加非线性表达能力 # 经过第二层后使用 softmax 函数在动作维度dim1上进行归一化输出各个动作的概率所有动作概率之和为1 return F.softmax(self.fc2(x), dim1) class ValueNet(torch.nn.Module): 价值网络 (Critic)评估当前状态有多好输出一个标量状态价值 V(s) def __init__(self, state_dim, hidden_dim): super(ValueNet, self).__init__() # 调用父类 nn.Module 的初始化方法 self.fc1 torch.nn.Linear(state_dim, hidden_dim) # 定义第一层全连接层输入为状态维度输出为隐藏层维度 self.fc2 torch.nn.Linear(hidden_dim, 1) # 定义第二层全连接层输出维度固定为 1因为只需输出一个标量评分 V(s) def forward(self, x): x F.relu(self.fc1(x)) # 输入状态 x 经过第一层后使用 ReLU 激活函数 return self.fc2(x) # 直接输出第二层的计算结果不需要激活函数因为价值可以是任意实数正负皆可 class PPO: PPO算法核心 (Proximal Policy Optimization)采用 Clip (截断) 方式限制策略更新幅度 def __init__(self, state_dim, hidden_dim, action_dim, actor_lr, critic_lr, lmbda, epochs, eps, gamma, device): # 初始化策略网络 (Actor)并将其部署到指定的计算设备如 CPU 或 GPU self.actor PolicyNet(state_dim, hidden_dim, action_dim).to(device) # 初始化价值网络 (Critic)并将其部署到指定的计算设备 self.critic ValueNet(state_dim, hidden_dim).to(device) # 为策略网络定义 Adam 优化器专门负责更新 Actor 的参数学习率为 actor_lr self.actor_optimizer torch.optim.Adam(self.actor.parameters(), lractor_lr) # 为价值网络定义 Adam 优化器专门负责更新 Critic 的参数学习率为 critic_lr self.critic_optimizer torch.optim.Adam(self.critic.parameters(), lrcritic_lr) self.gamma gamma # 奖励的折扣因子决定对未来长远奖励的重视程度 self.lmbda lmbda # GAE (广义优势估计) 的平滑衰减参数用于平衡方差与偏差 self.epochs epochs # 魔法机制PPO 允许同一批采集到的序列数据被重复送入网络训练的轮数 self.eps eps # PPO 截断范围参数 (通常设为 0.2)控制新策略偏离旧策略的最大安全幅度 self.device device # 记录当前使用的计算设备 def take_action(self, state): 根据当前状态通过策略网络 (Actor) 采样选择一个动作 # 将传入的 state (通常是 numpy 数组) 转换为 PyTorch 的 float 张量增加 batch 维度 [1, state_dim]并送入设备 state torch.tensor([state], dtypetorch.float).to(self.device) probs self.actor(state) # 状态进行前向传播获取当前状态下所有动作的概率分布例如 [0.8, 0.2] action_dist torch.distributions.Categorical(probs) # 根据概率分布构建一个类别分布生成器 action action_dist.sample() # 按概率“掷骰子”进行采样选出具体执行的动作概率越大的越容易被选中保证了探索性 return action.item() # 将 PyTorch 张量格式的动作提取为 Python 标准数值例如 0 或 1返回给环境执行 def update(self, transition_dict): 核心训练逻辑使用收集到的一批数据更新 Actor 和 Critic 的参数 # 将这批字典里的状态数据批量转换为 float 张量并送入设备形状如 [batch_size, state_dim] states torch.tensor(transition_dict[states], dtypetorch.float).to(self.device) # 将动作数据转换为张量并通过 view(-1, 1) 强行转换成列向量 [batch_size, 1]方便后续做索引操作 actions torch.tensor(transition_dict[actions]).view(-1, 1).to( self.device) # 将奖励数据转换为 float 张量的列向量 [batch_size, 1] rewards torch.tensor(transition_dict[rewards], dtypetorch.float).view(-1, 1).to(self.device) # 将下一状态数据批量转换为 float 张量 [batch_size, state_dim] next_states torch.tensor(transition_dict[next_states], dtypetorch.float).to(self.device) # 将游戏结束标志转换为 float 张量的列向量 [batch_size, 1]游戏结束为 1未结束为 0 dones torch.tensor(transition_dict[dones], dtypetorch.float).view(-1, 1).to(self.device) # 计算 TD 目标值 (TD Target)眼前真实奖励 折扣因子 * Critic对下一步的价值预测。 # 绝妙细节*(1 - dones) 保证了如果游戏在当前步结束未来价值期望会被强制清零仅剩眼前奖励。 td_target rewards self.gamma * self.critic(next_states) * (1 - dones) # 计算单步 TD 误差用拿到手的 TD 目标值 减去 Critic 之前对当前状态瞎猜的价值 td_delta td_target - self.critic(states) # 利用 rl_utils 提供的工具计算 GAE 优势函数Advantage通过 lmbda 综合多步 TD 误差使打分更平滑、方差更低 advantage rl_utils.compute_advantage(self.gamma, self.lmbda, td_delta.cpu()).to(self.device) # 获取当时采集数据时“旧策略 (Behavior Policy)” 选择那些动作的对数概率 # 必须加 .detach() 锁定为纯常数它作为参考基准绝不能参与后面的梯度反向传播 old_log_probs torch.log(self.actor(states).gather(1, actions)).detach() # PPO 魔法开始让这同一批旧数据在网络里反复迭代学习 epochs 次传统策略梯度这么干会直接崩溃 for _ in range(self.epochs): # 获取不断更新的“新策略”在同样状态下选择同样动作的对数概率 log_probs torch.log(self.actor(states).gather(1, actions)) # 计算重要性采样比率 (Ratio) 新策略概率 / 旧策略概率。 # 数学技巧e^(ln(A) - ln(B)) A/B。如果 ratio 1说明新策略更喜欢这个动作了。 ratio torch.exp(log_probs - old_log_probs) # PPO 损失的第一部分未截断的原始目标 (新旧比率 * 优势) surr1 ratio * advantage # PPO 损失的第二部分截断目标。使用 clamp 将比率强行锁死在 [1-eps, 1eps] (如 0.8~1.2) 之间然后再乘以优势 surr2 torch.clamp(ratio, 1 - self.eps, 1 self.eps) * advantage # 截断 # 计算 Actor 的最终 Loss取 surr1 和 surr2 中的较小值取悲观下界加负号转化为让 PyTorch 最小化的 Loss。 # 这保证了如果新策略偏离旧策略太远收益会被切断从而保护策略的稳定性不“翻车”。 actor_loss torch.mean(-torch.min(surr1, surr2)) # PPO损失函数 # 计算 Critic 的 Loss预测的价值与绝对真理 (TD 目标) 之间的 MSE 均方误差。 # td_target 作为靶子必须加 .detach() 锁定防止梯度错误地向回传导。 critic_loss torch.mean( F.mse_loss(self.critic(states), td_target.detach())) # 在反向传播前必须清空 Actor 优化器中上一步残留的梯度垃圾 self.actor_optimizer.zero_grad() # 清空 Critic 优化器中上一步残留的梯度垃圾 self.critic_optimizer.zero_grad() # 误差反向传播自动计算出 Actor 网络里每一个参数权重对 Actor Loss 的梯度责任分锅 actor_loss.backward() # 误差反向传播自动计算出 Critic 网络参数对 Critic Loss 的梯度 critic_loss.backward() # 优化器执行刀斧手操作沿着刚才算出的梯度方向切实修改 Actor 网络的参数 self.actor_optimizer.step() # 切实修改 Critic 网络的参数 self.critic_optimizer.step()
五分钟入门强化学习PPO(Proximal Policy Optimization)
PPO解决了什么痛点为什么PPO提高了数据的利用率总结传统 PG 不能多次利用因为它是“死脑筋”只能吃最新鲜的数据。数据一旦导致了脑子升级旧数据就立刻和新脑子八字不合强行用会导致网络崩溃。PPO 能多次利用因为它自带了一个“数据折算翻译官重要性采样”并且给自己戴上了“绝对不瞎改Clip 截断”的锁链。这使得旧数据在一段时间内保鲜从而让采样效率瞬间翻了近 10 倍PPO算法的执行流程要彻底弄懂 PPO最核心的就是要理清它“采样玩游戏”和“学习更新参数”这两步是如何交替进行的。传统策略梯度算法是“玩一把学一次扔掉数据”而 PPO 的执行流程可以被形象地概括为“用旧脑子玩一堆游戏存下来然后关起门来用这批录像把新脑子反反复复打磨好几遍最后拿新脑子去顶替旧脑子。”PPO 算法的完整执行流程分为以下 4 个标准步骤代码实现import gym # 导入 OpenAI Gym 库用于构建强化学习的物理或游戏交互环境 import torch # 导入 PyTorch 深度学习框架的核心库 import torch.nn.functional as F # 导入 PyTorch 的函数式接口包含激活函数 ReLU, 损失函数 MSE 等 import numpy as np # 导入 NumPy 库用于高效的数组和矩阵运算 import matplotlib.pyplot as plt # 导入 Matplotlib用于训练后的数据可视化与画图 import rl_utils # 导入自定义的强化学习工具包包含计算广义优势估计 GAE 等辅助函数 class PolicyNet(torch.nn.Module): 策略网络 (Actor)根据当前状态输出每个离散动作的概率分布 def __init__(self, state_dim, hidden_dim, action_dim): super(PolicyNet, self).__init__() # 调用父类 nn.Module 的初始化方法 self.fc1 torch.nn.Linear(state_dim, hidden_dim) # 定义第一层全连接层输入为状态维度输出为隐藏层维度 self.fc2 torch.nn.Linear(hidden_dim, action_dim) # 定义第二层全连接层输入为隐藏层维度输出为动作空间维度 def forward(self, x): x F.relu(self.fc1(x)) # 输入状态 x 经过第一层全连接层后使用 ReLU 激活函数增加非线性表达能力 # 经过第二层后使用 softmax 函数在动作维度dim1上进行归一化输出各个动作的概率所有动作概率之和为1 return F.softmax(self.fc2(x), dim1) class ValueNet(torch.nn.Module): 价值网络 (Critic)评估当前状态有多好输出一个标量状态价值 V(s) def __init__(self, state_dim, hidden_dim): super(ValueNet, self).__init__() # 调用父类 nn.Module 的初始化方法 self.fc1 torch.nn.Linear(state_dim, hidden_dim) # 定义第一层全连接层输入为状态维度输出为隐藏层维度 self.fc2 torch.nn.Linear(hidden_dim, 1) # 定义第二层全连接层输出维度固定为 1因为只需输出一个标量评分 V(s) def forward(self, x): x F.relu(self.fc1(x)) # 输入状态 x 经过第一层后使用 ReLU 激活函数 return self.fc2(x) # 直接输出第二层的计算结果不需要激活函数因为价值可以是任意实数正负皆可 class PPO: PPO算法核心 (Proximal Policy Optimization)采用 Clip (截断) 方式限制策略更新幅度 def __init__(self, state_dim, hidden_dim, action_dim, actor_lr, critic_lr, lmbda, epochs, eps, gamma, device): # 初始化策略网络 (Actor)并将其部署到指定的计算设备如 CPU 或 GPU self.actor PolicyNet(state_dim, hidden_dim, action_dim).to(device) # 初始化价值网络 (Critic)并将其部署到指定的计算设备 self.critic ValueNet(state_dim, hidden_dim).to(device) # 为策略网络定义 Adam 优化器专门负责更新 Actor 的参数学习率为 actor_lr self.actor_optimizer torch.optim.Adam(self.actor.parameters(), lractor_lr) # 为价值网络定义 Adam 优化器专门负责更新 Critic 的参数学习率为 critic_lr self.critic_optimizer torch.optim.Adam(self.critic.parameters(), lrcritic_lr) self.gamma gamma # 奖励的折扣因子决定对未来长远奖励的重视程度 self.lmbda lmbda # GAE (广义优势估计) 的平滑衰减参数用于平衡方差与偏差 self.epochs epochs # 魔法机制PPO 允许同一批采集到的序列数据被重复送入网络训练的轮数 self.eps eps # PPO 截断范围参数 (通常设为 0.2)控制新策略偏离旧策略的最大安全幅度 self.device device # 记录当前使用的计算设备 def take_action(self, state): 根据当前状态通过策略网络 (Actor) 采样选择一个动作 # 将传入的 state (通常是 numpy 数组) 转换为 PyTorch 的 float 张量增加 batch 维度 [1, state_dim]并送入设备 state torch.tensor([state], dtypetorch.float).to(self.device) probs self.actor(state) # 状态进行前向传播获取当前状态下所有动作的概率分布例如 [0.8, 0.2] action_dist torch.distributions.Categorical(probs) # 根据概率分布构建一个类别分布生成器 action action_dist.sample() # 按概率“掷骰子”进行采样选出具体执行的动作概率越大的越容易被选中保证了探索性 return action.item() # 将 PyTorch 张量格式的动作提取为 Python 标准数值例如 0 或 1返回给环境执行 def update(self, transition_dict): 核心训练逻辑使用收集到的一批数据更新 Actor 和 Critic 的参数 # 将这批字典里的状态数据批量转换为 float 张量并送入设备形状如 [batch_size, state_dim] states torch.tensor(transition_dict[states], dtypetorch.float).to(self.device) # 将动作数据转换为张量并通过 view(-1, 1) 强行转换成列向量 [batch_size, 1]方便后续做索引操作 actions torch.tensor(transition_dict[actions]).view(-1, 1).to( self.device) # 将奖励数据转换为 float 张量的列向量 [batch_size, 1] rewards torch.tensor(transition_dict[rewards], dtypetorch.float).view(-1, 1).to(self.device) # 将下一状态数据批量转换为 float 张量 [batch_size, state_dim] next_states torch.tensor(transition_dict[next_states], dtypetorch.float).to(self.device) # 将游戏结束标志转换为 float 张量的列向量 [batch_size, 1]游戏结束为 1未结束为 0 dones torch.tensor(transition_dict[dones], dtypetorch.float).view(-1, 1).to(self.device) # 计算 TD 目标值 (TD Target)眼前真实奖励 折扣因子 * Critic对下一步的价值预测。 # 绝妙细节*(1 - dones) 保证了如果游戏在当前步结束未来价值期望会被强制清零仅剩眼前奖励。 td_target rewards self.gamma * self.critic(next_states) * (1 - dones) # 计算单步 TD 误差用拿到手的 TD 目标值 减去 Critic 之前对当前状态瞎猜的价值 td_delta td_target - self.critic(states) # 利用 rl_utils 提供的工具计算 GAE 优势函数Advantage通过 lmbda 综合多步 TD 误差使打分更平滑、方差更低 advantage rl_utils.compute_advantage(self.gamma, self.lmbda, td_delta.cpu()).to(self.device) # 获取当时采集数据时“旧策略 (Behavior Policy)” 选择那些动作的对数概率 # 必须加 .detach() 锁定为纯常数它作为参考基准绝不能参与后面的梯度反向传播 old_log_probs torch.log(self.actor(states).gather(1, actions)).detach() # PPO 魔法开始让这同一批旧数据在网络里反复迭代学习 epochs 次传统策略梯度这么干会直接崩溃 for _ in range(self.epochs): # 获取不断更新的“新策略”在同样状态下选择同样动作的对数概率 log_probs torch.log(self.actor(states).gather(1, actions)) # 计算重要性采样比率 (Ratio) 新策略概率 / 旧策略概率。 # 数学技巧e^(ln(A) - ln(B)) A/B。如果 ratio 1说明新策略更喜欢这个动作了。 ratio torch.exp(log_probs - old_log_probs) # PPO 损失的第一部分未截断的原始目标 (新旧比率 * 优势) surr1 ratio * advantage # PPO 损失的第二部分截断目标。使用 clamp 将比率强行锁死在 [1-eps, 1eps] (如 0.8~1.2) 之间然后再乘以优势 surr2 torch.clamp(ratio, 1 - self.eps, 1 self.eps) * advantage # 截断 # 计算 Actor 的最终 Loss取 surr1 和 surr2 中的较小值取悲观下界加负号转化为让 PyTorch 最小化的 Loss。 # 这保证了如果新策略偏离旧策略太远收益会被切断从而保护策略的稳定性不“翻车”。 actor_loss torch.mean(-torch.min(surr1, surr2)) # PPO损失函数 # 计算 Critic 的 Loss预测的价值与绝对真理 (TD 目标) 之间的 MSE 均方误差。 # td_target 作为靶子必须加 .detach() 锁定防止梯度错误地向回传导。 critic_loss torch.mean( F.mse_loss(self.critic(states), td_target.detach())) # 在反向传播前必须清空 Actor 优化器中上一步残留的梯度垃圾 self.actor_optimizer.zero_grad() # 清空 Critic 优化器中上一步残留的梯度垃圾 self.critic_optimizer.zero_grad() # 误差反向传播自动计算出 Actor 网络里每一个参数权重对 Actor Loss 的梯度责任分锅 actor_loss.backward() # 误差反向传播自动计算出 Critic 网络参数对 Critic Loss 的梯度 critic_loss.backward() # 优化器执行刀斧手操作沿着刚才算出的梯度方向切实修改 Actor 网络的参数 self.actor_optimizer.step() # 切实修改 Critic 网络的参数 self.critic_optimizer.step()
