近端策略优化算法 (PPO) 详解
近端策略优化（Proximal Policy Optimization, PPO）是一种强化学习算法，旨在复杂任务中兼顾性能提升、训练稳定性与效率。相比传统策略梯度方法，PPO 通过限制策略更新幅度，有效防止模型因参数过大更新而崩溃。
1. 核心思想与背景
PPO 由 OpenAI 在 2017 年提出，其核心目标是简化训练过程，克服 TRPO（Trust Region Policy Optimization）的计算复杂性。在强化学习中，直接优化策略往往导致不稳定的训练。PPO 的解决方案是引入概率比率和剪辑机制，确保每一步训练都不会偏离当前策略太多，同时高效利用采样数据。
1.1 概率比率
PPO 使用概率比率 $r_t(\theta)$ 来衡量新旧策略的差异：
$$r_t(\theta) = \frac{\pi_\theta(a_t | s_t)}{\pi_{\theta_{\text{old}}}(a_t | s_t)}$$
其中 $\pi_{\theta_{\text{old}}}$ 为旧策略，$\pi_\theta$ 为新策略。该比率表示新策略在相同状态下选择动作的概率变化程度。
1.2 优势函数
为了评价某个动作的相对好坏，PPO 引入了优势函数 $A_t$：
$$A_t = Q(s_t, a_t) - V(s_t)$$
或者使用广义优势估计（GAE）进行近似。优势函数引导策略向更优方向改进。
2. 优化目标与损失函数
PPO 的目标是在保持改进的同时防止策略变化过大，主要通过以下损失函数组合实现：
2.1 裁剪策略损失
为了防止策略更新过度，PPO 采用裁剪操作将概率比率限制在区间 $[1-\epsilon, 1+\epsilon]$ 内：
$$L^{CLIP}(\theta) = \mathbb{E}_t \left[ \min \left( r_t(\theta) A_t, \text{clip}(r_t(\theta), 1-\epsilon, 1+\epsilon) A_t \right) \right]$$
这一机制相当于给策略更新设定了'安全边界'，既允许进步，又避免激进更新导致的性能崩塌。
2.2 值函数损失
Critic 网络负责估计状态价值 $V(s_t)$，通过最小化均方误差进行更新：
$$L^{VF}(\theta) = \mathbb{E}_t \left[ \left( V(s_t; \theta) - R_t \right)^2 \right]$$
其中 $R_t$ 为累计回报。这有助于 Critic 更准确地评估当前状态的价值。
2.3 熵正则化
为了鼓励探索，防止策略过早收敛到局部最优，加入熵正则化项：
$$L^{ENT}(\theta) = \mathbb{E}t \left[ H(\pi\theta(s_t)) \right]$$
2.4 总损失函数
综合上述三项，PPO 的总损失函数为：
$$L(\theta) = \mathbb{E}_t \left[ L^{CLIP}(\theta) - c_1 L^{VF}(\theta) + c_2 L^{ENT}(\theta) \right]$$
其中 $c_1$ 和 $c_2$ 为权重系数，用于平衡策略优化、值函数更新和探索能力。
3. PyTorch 代码实现
以下是基于 PyTorch 的完整 PPO 实现，包含 Actor-Critic 网络、经验存储及训练循环。代码逻辑清晰，便于理解各模块的作用。
import torch
import torch.nn as nn
import torch.optim as optim
from torch.distributions import Categorical
import numpy as np
import gym
# 设备配置
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
# Actor-Critic 神经网络
class ActorCritic(nn.Module):
def __init__(self, state_dim, action_dim):
super(ActorCritic, self).__init__()
# 共享层：提取状态特征
self.shared_layer = nn.Sequential(
nn.Linear(state_dim, 128),
nn.ReLU()
)
# Actor：输出动作概率分布
self.actor = nn.Sequential(
nn.Linear(128, action_dim),
nn.Softmax(dim=-1)
)
# Critic：输出状态价值
self.critic = nn.Linear(128, 1)
def forward(self, state):
shared = self.shared_layer(state)
action_probs = self.actor(shared)
state_value = self.critic(shared)
return action_probs, state_value
# 经验存储类
class Memory:
def __init__(self):
self.states = []
self.actions = []
self.logprobs = []
self.rewards = []
self.is_terminals = []
def clear(self):
self.states = []
self.actions = []
self.logprobs = []
self.rewards = []
self.is_terminals = []
# PPO Agent 类
class PPO:
def __init__(self, state_dim, action_dim, lr=0.002, gamma=0.99, eps_clip=0.2, K_epochs=4):
self.policy = ActorCritic(state_dim, action_dim).to(device)
self.optimizer = optim.Adam(self.policy.parameters(), lr=lr)
self.policy_old = ActorCritic(state_dim, action_dim).to(device)
self.policy_old.load_state_dict(self.policy.state_dict())
self.MseLoss = nn.MSELoss()
self.gamma = gamma
self.eps_clip = eps_clip
self.K_epochs = K_epochs
def select_action(self, state, memory):
state = torch.FloatTensor(state).to(device)
action_probs, _ = self.policy_old(state)
dist = Categorical(action_probs)
action = dist.sample()
memory.states.append(state)
memory.actions.append(action)
memory.logprobs.append(dist.log_prob(action))
return action.item()
def update(self, memory):
old_states = torch.stack(memory.states).to(device).detach()
old_actions = torch.stack(memory.actions).to(device).detach()
old_logprobs = torch.stack(memory.logprobs).to(device).detach()
# 计算折扣奖励
rewards = []
discounted_reward = 0
for reward, is_terminal in zip(reversed(memory.rewards), reversed(memory.is_terminals)):
if is_terminal:
discounted_reward = 0
discounted_reward = reward + (self.gamma * discounted_reward)
rewards.insert(0, discounted_reward)
rewards = torch.tensor(rewards, dtype=torch.float32).to(device)
# 奖励归一化
rewards = (rewards - rewards.mean()) / (rewards.std() + 1e-7)
# 多轮迭代更新
for _ in range(self.K_epochs):
action_probs, state_values = self.policy(old_states)
dist = Categorical(action_probs)
new_logprobs = dist.log_prob(old_actions)
entropy = dist.entropy()
# 概率比率
ratios = torch.exp(new_logprobs - old_logprobs.detach())
advantages = rewards - state_values.detach().squeeze()
# 裁剪损失
surr1 = ratios * advantages
surr2 = torch.clamp(ratios, 1 - self.eps_clip, 1 + self.eps_clip) * advantages
loss_actor = -torch.min(surr1, surr2).mean()
# 值函数损失
loss_critic = self.MseLoss(state_values.squeeze(), rewards)
# 总损失
loss = loss_actor + 0.5 * loss_critic - 0.01 * entropy.mean()
self.optimizer.zero_grad()
loss.backward()
self.optimizer.step()
self.policy_old.load_state_dict(self.policy.state_dict())
# 主程序
if __name__ == "__main__":
env = gym.make("CartPole-v1")
state_dim = env.observation_space.shape[0]
action_dim = env.action_space.n
ppo = PPO(state_dim, action_dim, lr=0.002, gamma=0.99, eps_clip=0.2, K_epochs=4)
memory = Memory()
max_episodes = 1000
max_timesteps = 300
for episode in range(1, max_episodes + 1):
state = env.reset()
total_reward = 0
for t in range(max_timesteps):
action = ppo.select_action(state, memory)
state, reward, done, _ = env.step(action)
memory.rewards.append(reward)
memory.is_terminals.append(done)
total_reward += reward
if done:
break
ppo.update(memory)
memory.clear()
print(f"Episode {episode}, Total Reward: {total_reward}")
env.close()
4. 关键实现细节
- Actor-Critic 结构：共享底层特征提取层，分别输出动作概率和状态价值，减少参数量并提高训练效率。
- 经验回放管理：
Memory类存储单条 Episode 的状态、动作、对数概率及奖励，更新后清空以释放内存。 - 奖励归一化：对累积奖励进行标准化处理，加速收敛并稳定训练过程。
- 多轮迭代：每次采样后对数据进行多次 Epoch 更新（K_epochs），提高样本利用率。
5. 算法对比：PPO vs TRPO vs A3C
| 特性 | PPO | TRPO | A3C |
|---|---|---|---|
| 核心思想 | 裁剪目标函数，限制更新幅度 | 信任区域约束，二次优化 | 异步多线程并行采样 |
| 优化目标 | 引入剪辑机制 | KL 散度限制 | 策略梯度优化 |
| 更新方式 | 同步更新，支持多轮迭代 | 同步更新，严格步幅限制 | 异步更新，线程独立运行 |
| 计算复杂度 | 低，无需二次优化 | 高，涉及二次规划 | 较低，依赖并行计算 |
| 样本利用率 | 高，可重复利用数据 | 高，严格优化目标 | 较低，存在数据冗余 |
| 稳定性 | 高，裁剪机制防震荡 | 极高，理论保证强 | 较低，异步可能冲突 |
| 应用场景 | 广泛，主流算法 | 高稳定性要求场景 | 资源受限或快速实验 |
总结：PPO 作为 TRPO 的改进版，用简单的裁剪机制替代了复杂的二次优化，显著降低了实现难度，同时保持了良好的稳定性和效率。A3C 则侧重于并行加速。在实际工程中，PPO 因其简单、稳定、高效的特点，已成为强化学习领域的首选算法之一。
注：代码示例基于 Gym 环境，实际项目中需根据具体任务调整超参数及网络结构。