摘要

近端策略优化(Proximal Policy Optimization, PPO)是深度强化学习领域最主流的算法之一,其在机器人控制、大语言模型对齐等场景中应用广泛。然而,PPO 的实现细节繁多,超参数敏感,初学者常遭遇“代码跑通了但模型不收敛”的困境。本文提供一份完整的 PPO PyTorch 复现指南,从 Actor-Critic 网络设计、GAE 优势估计到 PPO-Clip 损失函数,逐模块拆解实现细节。同时,针对训练中的调参与收敛难题,给出可操作的监控指标与调试策略。目标读者为具备一定 RL 基础的开发者,目标是“手把手教你写一版稳定的 PPO”。

1、引言

PPO 由 OpenAI 在 2017 年提出,其核心思想极为简洁:通过一个 Clipping 机制限制策略更新的幅度,既保留了 TRPO 的稳定性优势,又避免了对 KL 散度二阶近似的复杂计算。理论上,PPO 的损失函数只需几行公式就能概括;但工程上,一个完整的 PPO 实现涉及采样、优势估计、多轮更新、价值网络协同等多个环节,任何一处细节出错都可能导致训练崩溃或策略退化。

本文将复现目标设定为经典控制环境(如 CartPole、Lunar Lander),完整覆盖 PPO 的 Actor-Critic 架构。文章结构如下:第 2 节建立问题建模与符号体系;第 3 节分模块完成 PyTorch 实现;第 4 节解析 Clip 机制;第 5 节给出调参方案;第 6 节介绍训练监控方法;第 7 节总结。

2、基础概念与问题建模

2.1 策略梯度与 Actor-Critic

强化学习的目标是最大化期望累积奖励:

$$ J(\theta) = \mathbb{E}_{\tau \sim \pi_\theta} \left[ \sum_{t=0}^T \gamma^t r_t \right] $$

策略梯度方法通过梯度上升优化 $J(\theta)$。PPO 采用 Actor-Critic 架构:

  • Actor(策略网络):输入状态 $s$,输出动作概率分布 $\pi_\theta(a|s)$。
  • Critic(价值网络):输入状态 $s$,输出状态价值 $V_\phi(s)$ 的估计。

Critic 的存在使得我们可以计算 优势函数(Advantage),衡量某个动作比“平均水平”好多少:

$$ A_t = Q(s_t, a_t) - V(s_t) $$

优势函数是 PPO 损失的核心信号。

2.2 重要性采样与概率比率

PPO 属于 同策略(On-Policy) 方法,使用当前策略采样的数据更新网络。为复用旧策略采样的数据,引入重要性采样比率:

$$ r_t(\theta) = \frac{\pi_\theta(a_t|s_t)}{\pi_{\theta_{old}}(a_t|s_t)} $$

当 $r_t > 1$ 时,新策略比旧策略更倾向于选择该动作;当 $r_t < 1$ 时则相反。PPO 的核心约束正是作用在这个比率上。

3 PPO 算法的 PyTorch 完整实现

3.1 依赖库与超参数配置

import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np
import gymnasium as gym
from torch.distributions import Categorical

# 超参数配置
class Config:
    env_name = "CartPole-v1"
    lr_actor = 3e-4          # Actor 学习率
    lr_critic = 1e-3         # Critic 学习率(通常更大)
    gamma = 0.99             # 折扣因子
    gae_lambda = 0.95        # GAE 参数
    clip_epsilon = 0.2       # PPO Clipping 范围
    ppo_epochs = 10          # 每批数据的更新轮次
    batch_size = 64          # mini-batch 大小
    rollout_steps = 2048     # 每次采样步数
    entropy_coef = 0.01      # 熵正则化系数
    value_coef = 0.5         # 价值损失系数
    max_grad_norm = 0.5      # 梯度裁剪阈值

cfg = Config()

选择依据:Critic 的学习率通常设为 Actor 的 3-10 倍,因为价值网络是回归任务需要更快收敛。clip_epsilon=0.2 是论文推荐值,在大多数任务中表现稳定。

3.2 环境初始化

env = gym.make(cfg.env_name)
state_dim = env.observation_space.shape[0]
action_dim = env.action_space.n

print(f"State dim: {state_dim}, Action dim: {action_dim}")

我们选择 CartPole-v1 作为示例环境,状态为 4 维连续向量,动作为 2 个离散值(左/右)。

3.3 Actor-Critic 网络结构

class ActorCritic(nn.Module):
    """共享主干的 Actor-Critic 网络"""
    def __init__(self, state_dim, action_dim, hidden_dim=64):
        super().__init__()
        self.shared = nn.Sequential(
            nn.Linear(state_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, hidden_dim),
            nn.ReLU()
        )
        self.actor = nn.Linear(hidden_dim, action_dim)
        self.critic = nn.Linear(hidden_dim, 1)

    def forward(self, state):
        x = self.shared(state)
        action_logits = self.actor(x)
        value = self.critic(x)
        return action_logits, value

    def get_action(self, state, deterministic=False):
        """从策略中采样动作"""
        logits, value = self.forward(state)
        probs = torch.softmax(logits, dim=-1)
        dist = Categorical(probs)
        if deterministic:
            action = torch.argmax(probs, dim=-1)
        else:
            action = dist.sample()
        log_prob = dist.log_prob(action)
        return action, log_prob, value, dist.entropy()

设计要点:

  • 共享前两层可减少参数量,但 Actor 和 Critic 的优化方向可能冲突。对于简单任务共享是可行的,复杂任务建议分离。
  • get_action 返回 entropy(),用于后续熵正则化以鼓励探索。

3.4 经验收集(Rollout)

class RolloutBuffer:
    """存储交互数据"""
    def __init__(self):
        self.states = []
        self.actions = []
        self.log_probs = []
        self.rewards = []
        self.values = []
        self.dones = []

    def add(self, state, action, log_prob, reward, value, done):
        self.states.append(state)
        self.actions.append(action)
        self.log_probs.append(log_prob)
        self.rewards.append(reward)
        self.values.append(value)
        self.dones.append(done)

    def clear(self):
        for attr in ['states', 'actions', 'log_probs',
                     'rewards', 'values', 'dones']:
            getattr(self, attr).clear()

    def get_tensors(self):
        """将所有数据转换为 PyTorch 张量"""
        return (
            torch.FloatTensor(np.array(self.states)),
            torch.LongTensor(np.array(self.actions)),
            torch.FloatTensor(np.array(self.log_probs)),
            torch.FloatTensor(np.array(self.rewards)),
            torch.FloatTensor(np.array(self.values)),
            torch.FloatTensor(np.array(self.dones))
        )

3.5 GAE 优势估计

广义优势估计(Generalized Advantage Estimation, GAE)是 PPO 的标配,它通过 λ 参数在偏差和方差之间取得平衡:

def compute_gae(rewards, values, dones, gamma=0.99, lam=0.95):
    """
    计算 GAE 优势和 returns
    returns = advantage + value
    """
    advantages = []
    gae = 0
    # 从后往前遍历
    for t in reversed(range(len(rewards))):
        next_value = values[t+1] if t < len(rewards)-1 else 0
        next_done = dones[t+1] if t < len(rewards)-1 else 1
        delta = rewards[t] + gamma * next_value * (1 - next_done) - values[t]
        gae = delta + gamma * lam * (1 - next_done) * gae
        advantages.insert(0, gae)

    advantages = torch.FloatTensor(advantages)
    returns = advantages + values
    # 归一化优势(稳定训练)
    advantages = (advantages - advantages.mean()) / (advantages.std() + 1e-8)
    return advantages, returns

关键细节:优势归一化是可选的,但实践中能显著降低梯度方差,加快收敛。dones 用于处理 episode 终止边界,防止优势跨越 episode 传播。

3.6 PPO 更新模块

这是 PPO 最核心的部分——Clipped Surrogate Loss:

def ppo_update(actor_critic, optimizer, buffer, cfg):
    """执行 PPO 更新"""
    states, actions, old_log_probs, rewards, old_values, dones = buffer.get_tensors()

    # 1. 计算优势和 returns
    advantages, returns = compute_gae(
        rewards, old_values, dones, cfg.gamma, cfg.gae_lambda
    )

    # 2. 多轮更新
    total_actor_loss = 0
    total_critic_loss = 0
    dataset_size = len(states)

    for epoch in range(cfg.ppo_epochs):
        # 随机打乱数据
        indices = torch.randperm(dataset_size)
        for start in range(0, dataset_size, cfg.batch_size):
            end = start + cfg.batch_size
            batch_indices = indices[start:end]

            batch_states = states[batch_indices]
            batch_actions = actions[batch_indices]
            batch_old_log_probs = old_log_probs[batch_indices]
            batch_advantages = advantages[batch_indices]
            batch_returns = returns[batch_indices]

            # --- Actor Loss ---
            logits, values = actor_critic(batch_states)
            probs = torch.softmax(logits, dim=-1)
            dist = Categorical(probs)
            new_log_probs = dist.log_prob(batch_actions)
            entropy = dist.entropy().mean()

            # 概率比率
            ratio = torch.exp(new_log_probs - batch_old_log_probs)

            # Clipped Surrogate Loss
            surr1 = ratio * batch_advantages
            surr2 = torch.clamp(ratio, 1 - cfg.clip_epsilon,
                                1 + cfg.clip_epsilon) * batch_advantages
            actor_loss = -torch.min(surr1, surr2).mean()

            # --- Critic Loss ---
            critic_loss = nn.MSELoss()(values.squeeze(-1), batch_returns)

            # --- 总损失 ---
            loss = (actor_loss
                    + cfg.value_coef * critic_loss
                    - cfg.entropy_coef * entropy)

            optimizer.zero_grad()
            loss.backward()
            nn.utils.clip_grad_norm_(actor_critic.parameters(), cfg.max_grad_norm)
            optimizer.step()

            total_actor_loss += actor_loss.item()
            total_critic_loss += critic_loss.item()

    return total_actor_loss, total_critic_loss

模块解析:

  • Clipping:torch.clamp(ratio, 1-ε, 1+ε) 将概率比率限制在 [0.8, 1.2] 区间,torch.min 取悲观的损失估计。
  • 熵正则化:entropy_coef * entropy 为负数放入总损失,即最大化熵以鼓励探索。
  • 梯度裁剪:clip_grad_norm_ 防止单步梯度过大,是稳定 PPO 训练的必备操作。

4 PPO 损失函数设计详解

4.1 Clip 机制的工作原理

PPO 的 Clipped Surrogate Loss 乍看有些费解,但逻辑其实很简洁:

当优势 A > 0(好动作):我们希望增大该动作的概率(ratio > 1)。但如果 ratio 已超过 1+ε,说明概率增加得太多,clamp 会限制在 1+ε,min 会选择被限制后的项,阻止进一步增大。

当优势 A < 0(坏动作):我们希望减小概率(ratio < 1)。但如果 ratio 已低于 1-ε,clamp 会限制在 1-ε,min 会选择被限制后的项,阻止进一步减小。

4.2 与 TRPO 的比较

特性TRPOPPO
约束方式KL 散度二阶近似Clip 或 Adaptive KL
优化方法共轭梯度SGD/Adam
计算开销
实现复杂度

PPO 的精妙之处在于:用一阶方法近似了 TRPO 的信任域约束,性能几乎不输 TRPO,但实现简单得多。

4.3 价值网络损失的配合

价值网络拟合的准确性直接影响优势估计的质量。实践中常用 SmoothL1Loss(Huber Loss)替代 MSE,以减少异常值的影响:

critic_loss = nn.SmoothL1Loss()(values.squeeze(-1), batch_returns)

5 调参方案与收敛问题

5.1 五个核心超参数及其调参策略

以下是 PPO 最关键的超参数及其敏感度分析:

① 裁剪系数 ε(敏感度:★★★★★)

作用:控制策略更新的最大幅度

常见值:0.1 ~ 0.3,论文推荐 0.2

调参方向:训练不稳定时减小 ε(如 0.1);训练太慢时增大 ε(如 0.3)

进阶:动态 ε——训练初期设为 0.3 鼓励探索,逐步衰减至 0.1

② 学习率(敏感度:★★★★☆)

Actor 学习率:3e-4 ~ 1e-4

Critic 学习率:应为 Actor 的 3-10 倍

关键经验:Actor 依赖 Critic 的价值估计,Critic 必须先收敛

③ 熵正则化系数(敏感度:★★★★☆)

作用:平衡探索与利用

常见值:0.01 ~ 0.001

调参方向:动作空间大、奖励稀疏时增大;训练后期逐步衰减

④ GAE λ(敏感度:★★★☆☆)

作用:平衡优势估计的偏差(低 λ)与方差(高 λ)

常见值:0.9 ~ 0.98

λ=0 相当于单步 TD 误差;λ=1 相当于蒙特卡洛估计

⑤ 价值损失系数(敏感度:★★★★☆)

常见值:0.5 ~ 1.0

过大:价值损失主导,策略更新缓慢

过小:优势估计不准,策略优化失准

5.2 常见不收敛问题与解决方案

症状 1:Reward 持续震荡、无法稳定

可能原因:ε 过大或学习率过高

排查:检查 KL 散度是否过大

方案:减小 ε 至 0.1,降低学习率,增加 clip_grad_norm

症状 2:Reward 短时间飙升后崩溃

可能原因:Clipping 失效或策略“一步跨太大”

排查:观察 ratio 分布,是否大量样本被 clip

方案:减小 PPO epochs,增大 batch size

症状 3:模型输出越来越短(LLM 场景)

可能原因:Reward 模型无意中惩罚了长度

方案:检查 reward 与长度的相关性,考虑长度归一化

症状 4:模型几乎不动

可能原因:KL 约束太强或学习率太小

方案:检查 KL 是否远小于目标值,适当增大学习率或减小 ε

5.3 训练前期加速策略

PPO 收敛较慢是公认的。以下策略可加速早期训练:

先用 DQN/SAC 预训练:简单任务可先用收敛快的算法训练基础策略,再切换到 PPO 精细优化。

多 episode 再更新:单个 episode 更新方差大,累积 2-4 个 episode 后一次更新更稳定。

状态归一化:对输入状态做标准化(Zero-Mean, Unit-Variance),平滑优化地形。

6. 训练监控与评估

6.1 关键监控指标

成熟的 PPO 训练不应只看 Reward 曲线。以下指标能更早暴露问题:

指标健康范围预警信号
KL 散度0.001 ~ 0.01> 0.03:策略变化过大
Clip 比率10% ~ 30%> 50%:ε 过大或学习率过高
Reward 均值持续上升持续不变或剧烈波动
价值损失稳定下降持续增大:价值网络欠拟合
缓慢下降快速趋零:策略过拟合

如果只能选一个重点盯——盯 KL 散度。

6.2 训练循环完整代码

def train():
    actor_critic = ActorCritic(state_dim, action_dim)
    optimizer = optim.Adam(actor_critic.parameters(), lr=cfg.lr_actor)
    buffer = RolloutBuffer()
    episode_rewards = []

    state, _ = env.reset()
    episode_reward = 0
    steps = 0

    for step in range(100000):  # 总训练步数
        state_tensor = torch.FloatTensor(state).unsqueeze(0)
        action, log_prob, value, _ = actor_critic.get_action(state_tensor)

        next_state, reward, terminated, truncated, _ = env.step(action.item())
        done = terminated or truncated

        buffer.add(state, action, log_prob.item(), reward, value.item(), done)

        state = next_state if not done else env.reset()[0]
        episode_reward += reward
        steps += 1

        if done:
            episode_rewards.append(episode_reward)
            episode_reward = 0

        # 收集足够数据后更新
        if steps >= cfg.rollout_steps:
            actor_loss, critic_loss = ppo_update(actor_critic, optimizer, buffer, cfg)
            buffer.clear()
            steps = 0
            avg_reward = np.mean(episode_rewards[-10:]) if episode_rewards else 0
            print(f"Step {step} | Avg Reward: {avg_reward:.1f} | "
                  f"Actor Loss: {actor_loss:.4f} | Critic Loss: {critic_loss:.4f}")

    return actor_critic, episode_rewards

# 开始训练
trained_model, rewards = train()

注意:实际运行时建议引入评估模式(deterministic=True),定期记录无噪声的测试 Reward。

7. 总结与展望

本文完成了 PPO 算法的完整 PyTorch 复现,核心要点回顾:

Actor-Critic 架构是 PPO 的基础,Actor 输出动作,Critic 输出价值;

GAE 优势估计平衡偏差与方差,是稳定训练的关键;

Clipped Surrogate Loss 通过 [1-ε, 1+ε] 的比率约束实现信任域更新;

调参是核心技能——ε、学习率、熵系数、λ 需要按任务调整。

进阶方向包括:引入 Value Clipping、自适应 KL 系数、分布式并行采样、以及对大语言模型的 RLHF 扩展。

PPO 的真正难点不在算法本身,而在于工程细节的把控。不要期待第一版就跑出完美曲线——先稳定,再优化。