V · 2月21日

近端策略优化(PPO)算法的理论基础与 PyTorch 代码详解

image.png

近端策略优化(Proximal Policy Optimization, PPO)算法作为一种高效的策略优化方法,在深度强化学习领域获得了广泛应用。特别是在大语言模型(LLM)的人类反馈强化学习(RLHF)过程中,PPO 扮演着核心角色。本文将深入探讨 PPO 的基本原理和实现细节。

PPO 属于在线策略梯度方法的范畴。其基础形式可以用带有优势函数的策略梯度表达式来描述:

策略梯度的基础表达式(包含优势函数)。

这个表达式实际上构成了优势演员-评论家(Advantage Actor-Critic)方法的基础目标函数。PPO 算法可以视为对该方法的一种改进和优化。

PPO 算法的损失函数设计

PPO 通过引入策略更新约束机制来提升训练稳定性。这种机制很好地平衡了更新幅度:过大的策略更新可能导致训练偏离优化方向,而过小的更新则可能降低训练效率。为此,PPO 采用了一个特殊的替代目标函数,该函数由裁剪项和非裁剪项组成,并取两者的最小值。

PPO 的损失函数结构。

替代损失函数的非裁剪部分分析

损失函数中的非裁剪部分示意图。

在 PPO 中,比率函数定义为在状态 st 下执行动作 at 时,当前策略与旧策略的概率比值。

策略概率比率 r(θ)的定义。

这个比率函数 r(θ)为我们提供了一个度量新旧策略差异的有效工具,它可以替代传统策略梯度目标函数中的对数概率项。非裁剪部分的损失通过将此比率与优势函数相乘得到。

非裁剪部分损失计算示意图。

替代损失函数的裁剪机制

为了防止过大的策略更新,PPO 引入了裁剪机制,将策略比率 r(θ)限制在[1-ϵ, 1+ϵ]的区间内。其中 ϵ 是一个重要的超参数,在 PPO 的原始论文中设定为 0.2。这样,我们可以得到完整的 PPO 目标函数:

PPO 完整目标函数,包含非裁剪项和裁剪项。

PPO 的最终优化目标是在这两部分中取较小值,从而实现稳定的策略优化。

算法实现流程

 1. 系统初始化
   a. 设置随机种子
   b. 初始化演员网络与评论家网络的优化器
   c. 配置损失追踪器和奖励记录器
   d. 加载超参数配置

 2. 训练回合迭代
   a. 环境重置
   b. 回合内循环:
     i. 通过演员网络预测动作概率分布并采样
     ii. 记录动作的对数概率(作为old policy的参考)
     iii. 执行环境交互,获取转移数据
   c. 计算衰减回报
   d. 存储回合经验数据
   e. 按更新频率执行网络优化:
     i. 评估状态价值
     ii. 计算优势估计
     iii. 构建PPO损失函数
     iv. 执行梯度优化

 3. 训练监控
   a. 记录并可视化平均损失指标

PyTorch 实现详解

1、初始化**

 torch.manual_seed(self.cfg['train']['random_seed'])
 actor_optim = optim.Adam(self.actor.parameters(), lr=self.cfg['train']['lr'], betas=self.cfg['train']['betas'])
 critic_optim = optim.Adam(self.critic.parameters(), lr=self.cfg['train']['lr'], betas=self.cfg['train']['betas'])
 avg_actor_losses = []
 avg_critic_losses = []
 actor_losses = []
 critic_losses = []
 eps = np.finfo(np.float32).eps.item()
 batch_data = []

2、回合循环

2.1 重置环境

 for episode in range(self.cfg['train']['n_epidode']):
     rewards = []
     log_probs = []
     actions = []
     states = []
     state_values = []
     self.actor.train()
     self.critic.train()
     terminated, truncated = False, False ## 初始化终止和截断标志
     state, info = self.env.reset()
     ## 转换为张量
     state = torch.FloatTensor(state).to(self.device)

2.2 当回合未结束时:

 timesteps = 0
 ## 遍历时间步
 while not terminated and not truncated:
     timesteps += 1

演员预测动作概率并从分布中采样动作。

 ## 演员层输出动作概率,因为演员神经网络在输出层有softmax
 action_prob = self.actor(state)

 ## 我们知道我们不直接使用分类交叉熵损失函数,而是手动构建以获得更多控制。
 ## PyTorch中的分类交叉熵损失函数使用softmax将logits转换为概率到分类分布,
 ## 然后计算损失。所以通常不需要显式地将softmax函数添加到神经网络中。在这项工作中,
 ## 我们在神经网络上添加softmax层并计算分类分布。

 ## categorical函数可以从softmax概率或从logits(输出中没有softmax层)生成分类分布,
 ## 将logits作为属性
 action_dist= Categorical(action_prob)

 ## 采样动作
 action = action_dist.sample()
 actions.append(action)

获取对数概率,这被视为比率的 log pi_theta_old

 ## 获取对数概率以得到log pi_theta_old(a|s)并保存到列表中
 log_probs.append(action_dist.log_prob(action))

在环境中执行动作,获取下一个状态、奖励和终止标志。

 ## 动作必须从张量转换为numpy以供环境处理
 next_state, reward, terminated, truncated, info = self.env.step(action.item())
 rewards.append(reward)

 ## 将下一个状态分配为当前状态
 state = torch.FloatTensor(next_state).to(self.device)

2.3 计算折扣回报

 R = 0
 returns = [] ## 用于保存真实值的列表

 ## 使用环境在回合中返回的奖励计算每个回合的回报
 for r in rewards[::-1]:
     ## 计算折扣值
     R = r + self.cfg['train']['gamma'] * R
     returns.insert(0, R)

 returns = torch.tensor(returns).to(self.device)
 returns = (returns - returns.mean()) / (returns.std() + eps)

2.4 将每个回合的经验存储在 batch_data 中

 ## 存储数据
 batch_data.append([states, actions, returns, log_probs, state_values])

2.5 每 update_freq 回合更新网络:

 if episode != 0 and episode%self.cfg['train']['update_freq'] == 0:
   ## 这是我们更新网络一些n个epoch的循环。这个额外的for循环
   ## 改善了训练效果
   for _ in range(5):
       for states_b, actions_b, returns_b, old_log_probs, old_state_values in batch_data:
         ## 将列表转换为张量
         old_states = torch.stack(states_b, dim=0).detach()
         old_actions = torch.stack(actions_b, dim=0).detach()
         old_log_probs = torch.stack(old_log_probs, dim=0).detach()

计算状态值

 state_values = self.critic(old_states)

计算优势。

 ## 计算优势
 advantages = returns_b.detach() - state_values.detach()
 ## 规范化优势在理论上不是必需的,但在实践中它降低了我们优势的方差,
 ## 使收敛更加稳定和快速。我添加这个是因为
 ## 解决一些环境问题如果没有它会太不稳定。
 advantages = (advantages - advantages.mean()) / (advantages.std() + 1e-10)

计算演员和评论家的 PPO 损失。

 ## 现在我们需要计算比率(pi_theta / pi_theta__old)。为了做到这一点,
 ## 我们需要从存储的状态中获取所采取动作的旧策略,并计算相同动作的新策略
 ## 演员层输出动作概率,因为演员神经网络在输出层有softmax
 action_probs = self.actor(old_states)
 dist = Categorical(action_probs)
 new_log_probs = dist.log_prob(old_actions)
 ## 因为我们取对数,所以我们可以用减法代替除法。然后取指数将得到与除法相同的结果
 ratios = torch.exp(new_log_probs - old_log_probs)

 ## 替代损失函数的非裁剪部分
 surr1 = ratios * advantages

 ## 替代损失函数的裁剪部分
 surr2 = torch.clamp(ratios, 1 - self.cfg['train']['clip_param'], 1 + self.cfg['train']['clip_param']) * advantages

 ## 更新演员网络:loss = min(surr1, surr2)
 actor_loss = -torch.min(surr1, surr2).mean()
 actor_losses.append(actor_loss.item())

 ## 使用Huber损失计算评论家(价值)损失
 ## Huber损失对数据中的异常值比平方误差损失更不敏感。在基于价值的RL设置中,
 ## 推荐使用Huber损失。
 ## Smooth L1损失与HuberLoss密切相关
 critic_loss =  F.smooth_l1_loss(state_values, returns_b.unsqueeze(1)) #F.huber_loss(state_value, torch.tensor([R]))
 critic_losses.append(critic_loss.item())

使用梯度下降更新网络

 actor_optim.zero_grad()
 critic_optim.zero_grad()

 ## 执行反向传播
 actor_loss.backward()
 critic_loss.backward()

 ## 执行优化
 actor_optim.step()
 critic_optim.step()

完整代码

https://avoid.overfit.cn/post/ff4d892c414a4b9c9391a1812690eceb

推荐阅读
关注数
4213
内容数
956
SegmentFault 思否旗下人工智能领域产业媒体,专注技术与产业,一起探索人工智能。
目录
极术微信服务号
关注极术微信号
实时接收点赞提醒和评论通知
安谋科技学堂公众号
关注安谋科技学堂
实时获取安谋科技及 Arm 教学资源
安谋科技招聘公众号
关注安谋科技招聘
实时获取安谋科技中国职位信息