【Agentic RL 智体强化学习】第 07 篇:PPO 算法详解 — RLHF 的基石

本系列定位:从强化学习数学基础出发,系统讲解 LLM 对齐(RLHF/DPO/GRPO)到 Agentic RL(多轮智能体强化学习)的全链路,以 OpenRLHF 框架为实战主线。


本篇你将学到

  • PPO 算法的核心思想:从 TRPO 到 PPO 的演进
  • PPO-Clip 目标函数的推导与直觉
  • 裁剪机制(Clipping)为什么有效——限制策略更新幅度
  • 多轮训练与 Mini-batch 的工程实践
  • GAE(广义优势估计)与 PPO 的配合
  • 用 PyTorch 从零实现 PPO 并在 LunarLander 上训练

学完本篇,你将彻底理解 PPO——这个算法是 ChatGPT RLHF、DeepSeek R1 训练、以及 OpenRLHF 的核心基石。


一、为什么需要 PPO

第 05 篇的 REINFORCE 有一个致命问题:更新步长不可控。有时候梯度更新太大,策略直接崩溃;有时候更新太小,训练极慢。

REINFORCE 的问题

梯度更新

太大

太小

恰好

策略 θ_old

策略 θ_new

步长合适?

❌ 策略崩溃
性能暴跌且无法恢复

⏰ 收敛极慢
需要海量样本

✅ 稳定提升

核心矛盾:策略梯度方法的性能提升高度依赖学习率,但最优学习率会随着训练过程变化——前期需要大步探索,后期需要小步精调。

1.1 TRPO 的解决思路

Trust Region Policy Optimization(TRPO,2015)的思路是:限制新旧策略之间的差异。

用什么衡量"策略差异"?KL 散度

D ˉ K L ( π o l d ∥ π n e w ) ≤ δ \bar{D}_{KL}(\pi_{old} \| \pi_{new}) \leq \delta DˉKL(πoldπnew)δ

TRPO 保证每次更新后,新旧策略的 KL 散度不超过阈值 δ \delta δ。这是一个二次约束优化问题,需要计算 Fisher 信息矩阵和共轭梯度——计算复杂度极高。

1.2 PPO 的洞察

PPO(Proximal Policy Optimization,2017)的核心洞察:用简单的裁剪(Clipping)来近似 TRPO 的信任区域约束

不需要复杂的二阶优化,只需要在目标函数中加一个裁剪机制,就能达到类似的效果。这种简洁性使 PPO 成为最广泛使用的 RL 算法。


二、PPO 目标函数推导

2.1 从策略梯度到重要性采样

REINFORCE 直接使用 ∇ θ log ⁡ π θ ( a ∣ s ) ⋅ G t \nabla_\theta \log \pi_\theta(a|s) \cdot G_t θlogπθ(as)Gt。PPO 从另一个角度出发——使用重要性采样(Importance Sampling)来复用旧策略收集的数据。

重要性采样的比值(Probability Ratio):

r t ( θ ) = π θ ( a t ∣ s t ) π θ o l d ( a t ∣ s t ) r_t(\theta) = \frac{\pi_\theta(a_t|s_t)}{\pi_{\theta_{old}}(a_t|s_t)} rt(θ)=πθold(atst)πθ(atst)

它衡量的是"在新策略 θ \theta θ 下选这个动作的概率"与"在旧策略 θ o l d \theta_{old} θold 下选这个动作的概率"之比。

  • r t = 1 r_t = 1 rt=1:新旧策略对这个动作的偏好相同
  • r t > 1 r_t > 1 rt>1:新策略更喜欢这个动作
  • r t < 1 r_t < 1 rt<1:新策略更不喜欢这个动作

2.2 PPO-Clip 目标函数

PPO 的目标函数(被称为 “Clipped Surrogate Objective”):

L C L I P ( θ ) = E t [ min ⁡ ( r t ( θ ) A ^ t ,    clip ( r t ( θ ) , 1 − ϵ , 1 + ϵ ) A ^ t ) ] L^{CLIP}(\theta) = \mathbb{E}_t \left[ \min \left( r_t(\theta) \hat{A}_t, \; \text{clip}(r_t(\theta), 1-\epsilon, 1+\epsilon) \hat{A}_t \right) \right] LCLIP(θ)=Et[min(rt(θ)A^t,clip(rt(θ),1ϵ,1+ϵ)A^t)]

看起来复杂,让我们逐项拆解:

L^CLIP = E[min(r_t · A_hat, clip(r_t, 1-ε, 1+ε) · A_hat)]

情况1: A_hat > 0(好动作)

鼓励 r_t 增大

但 clip 限制 r_t ≤ 1+ε

效果:好动作被强化
但有上限,防止过度强化

情况2: A_hat < 0(坏动作)

鼓励 r_t 减小

但 clip 限制 r_t ≥ 1-ε

效果:坏动作被削弱
但有下限,防止过度惩罚

A ^ t > 0 \hat{A}_t > 0 A^t>0(好动作,应该增大其概率)

目标函数鼓励 r t r_t rt 增大(增大好动作的概率)。但 clip 限制了 r t ≤ 1 + ϵ r_t \leq 1 + \epsilon rt1+ϵ

  • 如果 r t < 1 + ϵ r_t < 1 + \epsilon rt<1+ϵ:目标函数 = r t A ^ t = r_t \hat{A}_t =rtA^t,正常梯度上升
  • 如果 r t ≥ 1 + ϵ r_t \geq 1 + \epsilon rt1+ϵ:目标函数 = ( 1 + ϵ ) A ^ t = (1+\epsilon) \hat{A}_t =(1+ϵ)A^t,梯度为零——不再继续增大

效果:好动作被强化,但有上限。

A ^ t < 0 \hat{A}_t < 0 A^t<0(坏动作,应该减小其概率)

目标函数鼓励 r t r_t rt 减小。但 clip 限制了 r t ≥ 1 − ϵ r_t \geq 1 - \epsilon rt1ϵ

  • 如果 r t > 1 − ϵ r_t > 1 - \epsilon rt>1ϵ:目标函数 = r t A ^ t = r_t \hat{A}_t =rtA^t,正常梯度下降(因为 A ^ t \hat{A}_t A^t 为负)
  • 如果 r t ≤ 1 − ϵ r_t \leq 1 - \epsilon rt1ϵ:目标函数 = ( 1 − ϵ ) A ^ t = (1-\epsilon) \hat{A}_t =(1ϵ)A^t,梯度为零——不再继续减小

效果:坏动作被削弱,但有下限。

2.3 min 操作的作用

为什么要取 min ⁡ \min min?这是 PPO 的精髓——悲观估计(Pessimistic Bound)。

对于好动作( A ^ > 0 \hat{A} > 0 A^>0),取 min ⁡ \min min 确保我们不会高估收益;对于坏动作( A ^ < 0 \hat{A} < 0 A^<0),取 min ⁡ \min min 确保我们不会低估惩罚。总之, min ⁡ \min min 让目标函数成为真实目标的一个下界,保证更新是保守的。

2.4 ε 的含义

ϵ \epsilon ϵ(通常设为 0.1-0.3)控制了每次更新的最大步长:

ε 值效果
0.1保守,每次更新步幅小,训练稳定但慢
0.2平衡(PPO 论文默认值)
0.3激进,步幅大,可能不稳定
0.05非常保守,适合精调阶段

三、完整 PPO 目标函数

PPO 的完整目标函数包含三项:

L P P O ( θ ) = L C L I P ( θ ) − c 1 L V F ( θ ) + c 2 S [ π θ ] ( s ) L^{PPO}(\theta) = L^{CLIP}(\theta) - c_1 L^{VF}(\theta) + c_2 S[\pi_\theta](s) LPPO(θ)=LCLIP(θ)c1LVF(θ)+c2S[πθ](s)

含义系数
L C L I P L^{CLIP} LCLIP策略裁剪目标(最大化)1
L V F L^{VF} LVF价值函数损失(最小化,用 MSE) c 1 ≈ 0.5 c_1 \approx 0.5 c10.5
S [ π θ ] S[\pi_\theta] S[πθ]策略熵(最大化,鼓励探索) c 2 ≈ 0.01 c_2 \approx 0.01 c20.01

熵正则化的作用:鼓励策略保持一定随机性。如果策略过早收敛到确定性策略(熵 → 0),它将停止探索,陷入局部最优。


四、GAE:广义优势估计

第 06 篇介绍了优势函数 A t = G t − V ( s t ) A_t = G_t - V(s_t) At=GtV(st)。但如何估计 A t A_t At 有多种方法,各有优劣。PPO 使用 **GAE(Generalized Advantage Estimation)**来平衡偏差和方差。

4.1 从 TD 误差到 GAE

TD 误差本身就是优势的一种估计:

A ^ t ( 1 ) = δ t = r t + 1 + γ V ( s t + 1 ) − V ( s t ) \hat{A}_t^{(1)} = \delta_t = r_{t+1} + \gamma V(s_{t+1}) - V(s_t) A^t(1)=δt=rt+1+γV(st+1)V(st)

上标 (1) 表示 1 步估计。类似地,n 步优势估计:

A ^ t ( n ) = ∑ k = 0 n − 1 ( γ λ ) k δ t + k \hat{A}_t^{(n)} = \sum_{k=0}^{n-1} (\gamma\lambda)^k \delta_{t+k} A^t(n)=k=0n1(γλ)kδt+k

等等,这混入了 λ。让我们更精确地推导。

GAE 定义

A ^ t G A E ( γ , λ ) = ∑ k = 0 ∞ ( γ λ ) k δ t + k + 1 \hat{A}_t^{GAE(\gamma,\lambda)} = \sum_{k=0}^{\infty} (\gamma\lambda)^k \delta_{t+k+1} A^tGAE(γ,λ)=k=0(γλ)kδt+k+1

其中 δ t + 1 = r t + 1 + γ V ( s t + 1 ) − V ( s t ) \delta_{t+1} = r_{t+1} + \gamma V(s_{t+1}) - V(s_t) δt+1=rt+1+γV(st+1)V(st) 是 TD 误差。

λ 值GAE 退化为何特点
λ=0 A ^ t = δ t \hat{A}_t = \delta_t A^t=δt(1 步 TD)高偏差,低方差
λ=1 A ^ t = G t − V ( s t ) \hat{A}_t = G_t - V(s_t) A^t=GtV(st)(MC)无偏差,高方差
0<λ<1偏差-方差的折中实践中最常用

λ=0
纯TD
高偏差/低方差

λ=0.5
折中

λ=0.95
实践中常用

λ=1
纯MC
无偏差/高方差

4.2 GAE 在 LLM RLHF 中的设置

在 LLM RLHF 中,通常设置:

参数典型值原因
γ(折扣因子)1.0整个回答视为一个 Episode,不折扣
λ(GAE 参数)1.0退化为 MC 优势 G t − V ( s t ) G_t - V(s_t) GtV(st)
ε(PPO clip)0.2OpenAI 原始设定

为什么 LLM RLHF 常用 γ=1, λ=1?因为一次回答的长度有限(几百到几千 token),不会像无限 Episode 那样导致回报发散。


五、PPO 完整训练流程

收集经验阶段

用当前策略 π_θ
与环境交互 N 步

计算每步的优势 Â_t
用 GAE

存储 (s,a,r,s',done,logπ_old,Â) 到 buffer

优化阶段
K 个 epoch

从 buffer 随机取 mini-batch

计算 r_t = π_θ(a|s) / π_θ_old(a|s)

计算 L^CLIP = min(r_t·Â, clip(r_t,1-ε,1+ε)·Â)

加上 Value Loss 和 Entropy

梯度上升更新 θ

epoch 还没完?

更新旧策略: θ_old ← θ

关键设计:

  1. 收集 N 步经验后做 K 轮更新(通常 K=4-10),提高数据利用率
  2. 每次更新用 mini-batch,像监督学习一样
  3. 更新完成后,旧策略 θ o l d \theta_{old} θold 同步为新的 θ \theta θ

六、PPO 代码实现

6.1 完整 PPO 智能体

"""
AgentRL-Lab v0.1: PPO 算法完整实现
在 CartPole 和 LunarLander 上训练
"""
import torch
import torch.nn as nn
import torch.optim as optim
from torch.distributions import Categorical
import numpy as np


class PPONetwork(nn.Module):
    """Actor-Critic 共享主干网络"""

    def __init__(self, state_dim, action_dim, hidden_dim=64):
        super().__init__()
        # 共享特征提取层
        self.feature = nn.Sequential(
            nn.Linear(state_dim, hidden_dim),
            nn.Tanh(),
            nn.Linear(hidden_dim, hidden_dim),
            nn.Tanh()
        )
        # Actor 头:输出动作 logits
        self.actor_head = nn.Linear(hidden_dim, action_dim)
        # Critic 头:输出状态价值
        self.critic_head = nn.Linear(hidden_dim, 1)

    def forward(self, x):
        features = self.feature(x)
        action_logits = self.actor_head(features)
        state_value = self.critic_head(features)
        return action_logits, state_value


class PPOAgent:
    """PPO 智能体"""

    def __init__(self, state_dim, action_dim, lr=3e-4, gamma=0.99,
                 gae_lambda=0.95, eps_clip=0.2, k_epochs=10,
                 entropy_coef=0.01, value_coef=0.5):
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

        self.network = PPONetwork(state_dim, action_dim).to(self.device)
        self.optimizer = optim.Adam(self.network.parameters(), lr=lr)

        self.gamma = gamma
        self.gae_lambda = gae_lambda
        self.eps_clip = eps_clip
        self.k_epochs = k_epochs
        self.entropy_coef = entropy_coef
        self.value_coef = value_coef

        # 经验缓冲区
        self.states = []
        self.actions = []
        self.rewards = []
        self.dones = []
        self.log_probs = []
        self.values = []

    def select_action(self, state, deterministic=False):
        """选择动作"""
        state_t = torch.FloatTensor(state).to(self.device)
        logits, value = self.network(state_t)
        dist = Categorical(logits=logits)

        if deterministic:
            action = torch.argmax(logits)
        else:
            action = dist.sample()

        return action.item(), dist.log_prob(action).item(), value.item()

    def compute_gae(self, rewards, values, dones, last_value):
        """
        计算广义优势估计 GAE
        """
        advantages = []
        gae = 0

        # 从后往前递推
        values_with_last = values + [last_value]
        for t in reversed(range(len(rewards))):
            delta = rewards[t] + self.gamma * values_with_last[t + 1] * (1 - dones[t]) - values_with_last[t]
            gae = delta + self.gamma * self.gae_lambda * (1 - dones[t]) * gae
            advantages.insert(0, gae)

        advantages = torch.FloatTensor(advantages).to(self.device)
        returns = advantages + torch.FloatTensor(values).to(self.device)
        return advantages, returns

    def update(self, last_value):
        """PPO 策略更新"""
        # 转换为 tensor
        states = torch.FloatTensor(np.array(self.states)).to(self.device)
        actions = torch.LongTensor(self.actions).to(self.device)
        old_log_probs = torch.FloatTensor(self.log_probs).to(self.device)

        # 计算 GAE
        advantages, returns = self.compute_gae(
            self.rewards, self.values, self.dones, last_value
        )

        # 优势归一化
        advantages = (advantages - advantages.mean()) / (advantages.std() + 1e-8)

        # K 轮 epoch 更新
        loss_history = []
        for _ in range(self.k_epochs):
            # 前向传播
            logits, values = self.network(states)
            dist = Categorical(logits=logits)
            new_log_probs = dist.log_prob(actions)

            # 计算概率比值 r_t(θ)
            ratio = torch.exp(new_log_probs - old_log_probs)

            # PPO Clip 目标
            surr1 = ratio * advantages
            surr2 = torch.clamp(ratio, 1.0 - self.eps_clip, 1.0 + self.eps_clip) * advantages
            policy_loss = -torch.min(surr1, surr2).mean()

            # 价值函数损失
            value_loss = nn.functional.mse_loss(values.squeeze(-1), returns)

            # 策略熵(鼓励探索)
            entropy = dist.entropy().mean()

            # 总损失
            loss = policy_loss + self.value_coef * value_loss - self.entropy_coef * entropy

            # 反向传播
            self.optimizer.zero_grad()
            loss.backward()
            # 梯度裁剪
            nn.utils.clip_grad_norm_(self.network.parameters(), max_norm=0.5)
            self.optimizer.step()

            loss_history.append({
                'policy_loss': policy_loss.item(),
                'value_loss': value_loss.item(),
                'entropy': entropy.item()
            })

        # 清空缓冲区
        self.states = []
        self.actions = []
        self.rewards = []
        self.dones = []
        self.log_probs = []
        self.values = []

        return loss_history[-1]


if __name__ == "__main__":
    import gymnasium as gym

    # 使用 LunarLander(比 CartPole 更有挑战性)
    env = gym.make("CartPole-v1")  # 先用 CartPole 快速验证
    agent = PPOAgent(
        state_dim=env.observation_space.shape[0],
        action_dim=env.action_space.n,
        lr=3e-4,
        gamma=0.99,
        gae_lambda=0.95,
        eps_clip=0.2,
        k_epochs=10
    )

    n_episodes = 300
    update_interval = 2000  # 每 2000 步更新一次
    reward_history = []
    total_steps = 0

    print("=== PPO 训练 CartPole ===\n")

    for ep in range(n_episodes):
        state, _ = env.reset(seed=ep)
        total_reward = 0

        for step in range(500):
            action, log_prob, value = agent.select_action(state)
            next_state, reward, terminated, truncated, _ = env.step(action)
            done = terminated or truncated

            # 存储经验
            agent.states.append(state)
            agent.actions.append(action)
            agent.rewards.append(reward)
            agent.dones.append(float(done))
            agent.log_probs.append(log_prob)
            agent.values.append(value)

            state = next_state
            total_reward += reward
            total_steps += 1

            # 每 update_interval 步执行一次更新
            if total_steps % update_interval == 0:
                _, _, last_value = agent.select_action(state)
                loss_info = agent.update(last_value)

            if done:
                break

        reward_history.append(total_reward)

        if (ep + 1) % 50 == 0:
            avg_r = np.mean(reward_history[-50:])
            print(f"Episode {ep+1:3d} | 平均奖励(近50轮): {avg_r:6.1f} | 总步数: {total_steps}")

    env.close()

    # 最终评估
    print("\n=== 最终评估(确定性策略)===")
    eval_rewards = []
    for ep in range(10):
        state, _ = env.reset(seed=1000 + ep)
        total_reward = 0
        for step in range(500):
            action, _, _ = agent.select_action(state, deterministic=True)
            state, reward, terminated, truncated, _ = env.step(action)
            total_reward += reward
            if terminated or truncated:
                break
        eval_rewards.append(total_reward)

    print(f"评估平均奖励: {np.mean(eval_rewards):.1f} ± {np.std(eval_rewards):.1f}")
    print("\n💡 PPO 对比 REINFORCE 的优势:")
    print("   - 通过 clip 机制保证稳定更新")
    print("   - 多轮 epoch 更新提高数据利用率")
    print("   - GAE 平衡偏差和方差")
    print("   - 这就是 ChatGPT RLHF 使用的核心算法!")

6.2 代码解读

PPO 实现的几个关键点值得仔细理解:

关键点一:Ratio 计算

ratio = torch.exp(new_log_probs - old_log_probs)

这就是 r t ( θ ) = π θ ( a ∣ s ) / π θ o l d ( a ∣ s ) r_t(\theta) = \pi_\theta(a|s) / \pi_{\theta_{old}}(a|s) rt(θ)=πθ(as)/πθold(as)。用对数减法代替除法,数值更稳定。

关键点二:Clipped Surrogate

surr1 = ratio * advantages
surr2 = torch.clamp(ratio, 1-eps, 1+eps) * advantages
policy_loss = -torch.min(surr1, surr2).mean()

取负号是因为 PyTorch 做梯度下降,而我们要最大化目标。

关键点三:多轮更新

for _ in range(self.k_epochs):
    ...

同一批数据被使用 K 次。这是 PPO 比 REINFORCE 样本效率高的关键。但 K 不能太大(通常 4-10),否则 π θ \pi_\theta πθ π θ o l d \pi_{\theta_{old}} πθold 差距变大,clip 频繁触发。

关键点四:梯度裁剪

nn.utils.clip_grad_norm_(self.network.parameters(), max_norm=0.5)

限制梯度范数,防止梯度爆炸。这在 LLM RLHF 中尤为重要。


七、PPO 超参数速查

超参数经典 RL 典型值LLM RLHF 典型值说明
ε(clip range)0.20.2PPO 裁剪范围
γ(折扣因子)0.991.0LLM 中通常不折扣
λ(GAE)0.951.0LLM 中常用纯 MC 优势
K(epochs)4-101-4LLM 中通常更少
学习率3e-41e-6 ~ 5e-7LLM 精调用极小学习率
熵系数0.010.0-0.001LLM 不需要太多探索
Value 系数0.50.5-1.0价值函数损失权重
梯度裁剪0.51.0最大梯度范数
batch size64-2048128-1024mini-batch 大小

💡 注意:LLM RLHF 中的学习率(1e-6 ~ 5e-7)比经典 RL 小 2-3 个数量级!这是因为 LLM 已经是预训练好的强策略,大幅更新会破坏已有能力。


本篇小结

知识点核心内容
PPO 核心思想用裁剪机制近似 TRPO 的信任区域约束
概率比值 r t ( θ ) = π θ / π o l d r_t(\theta) = \pi_\theta / \pi_{old} rt(θ)=πθ/πold,衡量新旧策略差异
PPO-Clip 目标 min ⁡ ( r t A ^ , clip ( r t , 1 − ϵ , 1 + ϵ ) A ^ ) \min(r_t \hat{A}, \text{clip}(r_t, 1-\epsilon, 1+\epsilon) \hat{A}) min(rtA^,clip(rt,1ϵ,1+ϵ)A^)
裁剪机制限制策略更新幅度,好/坏动作都不过度
GAE A ^ G A E ( γ , λ ) = ∑ ( γ λ ) k δ t + k \hat{A}^{GAE(\gamma,\lambda)} = \sum (\gamma\lambda)^k \delta_{t+k} A^GAE(γ,λ)=(γλ)kδt+k,平衡偏差方差
多轮更新同一批数据训练 K 个 epoch,提高数据利用率
PPO 完整损失策略损失 + 价值损失 + 熵正则化

下篇预告

第 08 篇:模块一总结 — 从 RL 基础到 LLM 对齐的桥梁
回顾模块一全部知识,整理 AgentRL-Lab v0.1,建立从经典 RL 到 LLM 对齐的认知桥梁。


如果本篇内容对你有帮助,欢迎点赞收藏!有任何疑问,欢迎在评论区交流。

Logo

AtomGit AI 社区提供模型库、数据集、Agent、Token等资源

更多推荐