主题
PPO:近端策略优化
PPO(Proximal Policy Optimization,Schulman et al., 2017)是目前最常用的策略梯度算法之一。它保留了 TRPO “每次更新不要让新策略离旧策略太远”的思想,但不再求解带约束的二阶优化问题,而是对新旧策略的概率比进行裁剪(clip),只用一阶优化(Adam、小批量 SGD)就能实现,并且可以在同一批数据上做多轮更新。PPO 实现简单、稳定、通用,被广泛用于游戏(OpenAI Five)、机器人控制,也是 InstructGPT 等工作中用 RLHF 微调大语言模型的算法(见 RLHF 与大模型中的 RL)。
动机:同一批数据能不能多用几次
A2C 等同策略算法每批数据只更新一次就丢弃,样本效率低。如果直接在同一批数据上做多轮梯度更新,策略会偏离采样时的旧策略越来越远,梯度估计失效,训练常常崩溃。TRPO 用 KL 约束解决了这个问题,但实现复杂。PPO 的问题是:能否用简单的一阶方法达到类似的效果?
裁剪目标函数
概率比
定义新旧策略在采样动作上的概率比:
TRPO 的代理目标为
PPO-Clip
PPO 的核心目标:
clip 机制的直观解释
分两种情况看:
(好动作):我们想增大 ,即增大 。但当 后,目标变为常数 ,梯度为 0——不再继续奖励过度的增大。 (坏动作):我们想减小 。当 后目标变为常数,梯度为 0——不再继续惩罚。
取
| 情况 | 目标 | 梯度 | |
|---|---|---|---|
| 有 | |||
| 0 | |||
| 有 | |||
| 0 |
KL 惩罚版本
论文还提出了另一个变体,把 KL 作为惩罚项并自适应调节系数:
每轮更新后,若实际 KL 大于目标值的 1.5 倍则
价值函数损失与熵奖励
使用 Actor-Critic 结构时,完整的损失为:
其中优势
算法流程:
text
对每次迭代:
1. 用 π_θold 在 N 个并行环境中各运行 T 步,收集 N×T 条数据
2. 用 GAE 计算优势 Â_t 和回报目标 R̂_t
3. 重复 K 个 epoch:
把数据打乱,分成 M 个小批量
对每个小批量,最小化 −L^CLIP + c1·L^VF − c2·熵
4. θold ← θ1
2
3
4
5
6
7
2
3
4
5
6
7
实现细节(Implementation Details)
PPO 的效果在很大程度上依赖实现细节。Engstrom 等人(2020)与 Huang 等人(2022, The 37 Implementation Details of PPO)系统地研究了这些细节。最重要的几条:
- 优势标准化:在每个小批量内把
标准化。 - 正交初始化:隐藏层增益
,策略输出层 0.01,价值输出层 1。 - 学习率线性退火:从初始值线性降到 0。
- 梯度范数裁剪:0.5。
- Adam 的 eps 设为
。 - 价值函数裁剪(可选,研究表明并不总是有益)。
- 观测和奖励标准化(在 MuJoCo 等连续控制任务中通常很重要;但并非总是有益,见下面 Pendulum 的实验)。
- 正确处理截断:被时间上限截断的回合要自举最终状态的价值。
代码:PPO 在 CartPole 上的实现
下面的实现参考了 CleanRL 的单文件风格,包含上面列出的大部分细节(4 个并行环境,每次采样 128 步,GAE
py
"""PPO-Clip:CartPole-v1,4 个并行环境(参考 CleanRL 的单文件实现风格)。"""
import gymnasium as gym
import numpy as np
import torch
import torch.nn as nn
from torch.distributions import Categorical
N_ENVS, N_STEPS, TOTAL_STEPS = 4, 128, 200_000
GAMMA, LAM, LR = 0.99, 0.95, 2.5e-4
CLIP, EPOCHS, N_MINIBATCH, ENT_COEF, VF_COEF, MAX_GRAD = 0.2, 10, 4, 0.01, 0.5, 0.5
torch.manual_seed(0)
envs = gym.vector.SyncVectorEnv([lambda: gym.make("CartPole-v1") for _ in range(N_ENVS)],
autoreset_mode=gym.vector.AutoresetMode.SAME_STEP)
envs = gym.wrappers.vector.RecordEpisodeStatistics(envs)
obs_dim, n_actions = envs.single_observation_space.shape[0], envs.single_action_space.n
def layer(i, o, std=np.sqrt(2)):
l = nn.Linear(i, o)
nn.init.orthogonal_(l.weight, std); nn.init.zeros_(l.bias) # 正交初始化
return l
actor = nn.Sequential(layer(obs_dim, 64), nn.Tanh(), layer(64, 64), nn.Tanh(), layer(64, n_actions, 0.01))
critic = nn.Sequential(layer(obs_dim, 64), nn.Tanh(), layer(64, 64), nn.Tanh(), layer(64, 1, 1.0))
params = list(actor.parameters()) + list(critic.parameters())
opt = torch.optim.Adam(params, lr=LR, eps=1e-5)
buf_obs = torch.zeros(N_STEPS, N_ENVS, obs_dim)
buf_act, buf_logp, buf_rew, buf_done, buf_val = (torch.zeros(N_STEPS, N_ENVS) for _ in range(5))
obs = torch.as_tensor(envs.reset(seed=0)[0], dtype=torch.float32)
ep_returns, n_updates = [], TOTAL_STEPS // (N_ENVS * N_STEPS)
for update in range(n_updates):
for g in opt.param_groups: # 学习率线性退火
g["lr"] = LR * (1 - update / n_updates)
# ---------- 1) 用旧策略采样 ----------
for t in range(N_STEPS):
with torch.no_grad():
dist = Categorical(logits=actor(obs))
a = dist.sample()
buf_val[t] = critic(obs).squeeze(-1)
next_obs, r, term, trunc, info = envs.step(a.numpy())
r = torch.as_tensor(r, dtype=torch.float32)
if trunc.any(): # 截断时自举最终状态价值
for i in np.flatnonzero(trunc):
with torch.no_grad():
r[i] += GAMMA * critic(torch.as_tensor(info["final_obs"][i], dtype=torch.float32)).item()
if "episode" in info:
ep_returns += list(info["episode"]["r"][info["_episode"]])
buf_obs[t], buf_act[t], buf_logp[t] = obs, a, dist.log_prob(a)
buf_rew[t], buf_done[t] = r, torch.as_tensor(term | trunc, dtype=torch.float32)
obs = torch.as_tensor(next_obs, dtype=torch.float32)
# ---------- 2) GAE ----------
with torch.no_grad():
next_v = critic(obs).squeeze(-1)
adv, gae = torch.zeros(N_STEPS, N_ENVS), 0
for t in reversed(range(N_STEPS)):
v_next = next_v if t == N_STEPS - 1 else buf_val[t + 1]
delta = buf_rew[t] + GAMMA * v_next * (1 - buf_done[t]) - buf_val[t]
gae = delta + GAMMA * LAM * (1 - buf_done[t]) * gae
adv[t] = gae
ret = adv + buf_val
# ---------- 3) 多轮小批量优化裁剪目标 ----------
b_obs, b_act, b_logp = buf_obs.reshape(-1, obs_dim), buf_act.reshape(-1), buf_logp.reshape(-1)
b_adv, b_ret = adv.reshape(-1), ret.reshape(-1)
batch = N_ENVS * N_STEPS
mb = batch // N_MINIBATCH
for _ in range(EPOCHS):
perm = torch.randperm(batch)
for start in range(0, batch, mb):
idx = perm[start:start + mb]
dist = Categorical(logits=actor(b_obs[idx]))
logp = dist.log_prob(b_act[idx])
ratio = (logp - b_logp[idx]).exp() # r_t(θ)
a_mb = (b_adv[idx] - b_adv[idx].mean()) / (b_adv[idx].std() + 1e-8)
pg_loss = -torch.min(ratio * a_mb, ratio.clamp(1 - CLIP, 1 + CLIP) * a_mb).mean()
v_loss = ((critic(b_obs[idx]).squeeze(-1) - b_ret[idx]) ** 2).mean()
loss = pg_loss + VF_COEF * v_loss - ENT_COEF * dist.entropy().mean()
opt.zero_grad(); loss.backward()
nn.utils.clip_grad_norm_(params, MAX_GRAD)
opt.step()
steps = (update + 1) * batch
if (update + 1) % 40 == 0 or update == n_updates - 1:
print(f"steps {steps:6d} | 最近20回合平均回报 {np.mean(ep_returns[-20:]):6.1f}")
# ---------- 训练结束:用确定性(argmax)策略评估 10 个回合 ----------
eval_env = gym.make("CartPole-v1")
scores = []
for i in range(10):
o, _ = eval_env.reset(seed=10_000 + i)
done, total = False, 0.0
while not done:
with torch.no_grad():
act = int(actor(torch.as_tensor(o, dtype=torch.float32)).argmax())
o, r, te, tr, _ = eval_env.step(act)
total += r; done = te or tr
scores.append(total)
print(f"贪心评估(10 回合)平均回报: {np.mean(scores):.1f}")1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
实际运行输出:
text
steps 20480 | 最近20回合平均回报 201.6
steps 40960 | 最近20回合平均回报 360.8
steps 61440 | 最近20回合平均回报 327.6
steps 81920 | 最近20回合平均回报 451.6
steps 102400 | 最近20回合平均回报 345.0
steps 122880 | 最近20回合平均回报 500.0
steps 143360 | 最近20回合平均回报 500.0
steps 163840 | 最近20回合平均回报 500.0
steps 184320 | 最近20回合平均回报 500.0
steps 199680 | 最近20回合平均回报 500.0
贪心评估(10 回合)平均回报: 500.01
2
3
4
5
6
7
8
9
10
11
2
3
4
5
6
7
8
9
10
11
一个真实的调参经历:我们最初使用 CleanRL 默认的 EPOCHS = 4,在 3 个种子上训练 20 万步后,训练回报分别停留在约 190、245、221,曾在 10 万步附近达到 429 后又回落。改为 EPOCHS = 10(Stable-Baselines3 的默认值)后,种子 0 和种子 1 都在约 12 万步时稳定达到 500,贪心评估均为满分。这说明 PPO 对超参数相当敏感,“同一个算法”在不同默认参数下表现可能差异很大;而 CleanRL 的 CartPole 默认配置训练的是 50 万步,我们的预算更短。
连续动作:Pendulum 上的 PPO
把 PPO 用于连续动作,只需要换掉策略分布:
- 对角高斯策略:策略网络输出动作均值
,另设一个与状态无关的可学习参数 (初始 )。动作的对数概率是各维度之和: 。 - 动作裁剪:Pendulum 的扭矩范围是
。缓冲区里保存未裁剪的采样动作,用它计算概率比;只在送入环境时裁剪。否则概率比会算错。 - 截断自举:Pendulum 每回合固定 200 步后被截断(truncated),没有真正的终止。截断时要把
加回奖励,而不能把截断当作价值为 0 的终止。 - 较小的
:Pendulum 的奖励很密集, 学得更快。 - 不加熵奖励:连续控制中通常不加,探索由可学习的
负责,训练中 会从 1 逐渐减小到约 0.3。
py
"""连续动作 PPO-Clip:Pendulum-v1(对角高斯策略,状态无关的可学习 log σ)。
与离散版 ppo_cartpole.py 相比,主要变化:高斯策略、动作裁剪、截断时的价值自举、较小的 γ 和较小的小批量。
用法:python ppo_pendulum.py [种子];DQN_QUICK=1 只跑很短的版本。"""
import os
import sys
import gymnasium as gym
import numpy as np
import torch
import torch.nn as nn
from torch.distributions import Normal
QUICK = os.environ.get("DQN_QUICK") == "1"
torch.set_num_threads(1)
N_ENVS, N_STEPS = 4, 1024
TOTAL_STEPS = 40_960 if QUICK else 204_800
GAMMA, LAM, LR = 0.9, 0.95, 1e-3 # Pendulum 的奖励很密集,较小的 γ 学得更快
CLIP, EPOCHS, MINIBATCH, VF_COEF, MAX_GRAD = 0.2, 10, 64, 0.5, 0.5
def layer(i, o, std=np.sqrt(2)):
l = nn.Linear(i, o); nn.init.orthogonal_(l.weight, std); nn.init.zeros_(l.bias); return l
def run(seed):
torch.manual_seed(seed); np.random.seed(seed)
envs = gym.vector.SyncVectorEnv([lambda: gym.make("Pendulum-v1") for _ in range(N_ENVS)],
autoreset_mode=gym.vector.AutoresetMode.SAME_STEP)
envs = gym.wrappers.vector.RecordEpisodeStatistics(envs)
obs_dim, act_dim = envs.single_observation_space.shape[0], envs.single_action_space.shape[0]
a_low, a_high = envs.single_action_space.low, envs.single_action_space.high
actor_mean = nn.Sequential(layer(obs_dim, 64), nn.Tanh(), layer(64, 64), nn.Tanh(), layer(64, act_dim, 0.01))
actor_logstd = nn.Parameter(torch.zeros(act_dim)) # 与状态无关的 log σ,初始 σ = 1
critic = nn.Sequential(layer(obs_dim, 64), nn.Tanh(), layer(64, 64), nn.Tanh(), layer(64, 1, 1.0))
params = list(actor_mean.parameters()) + [actor_logstd] + list(critic.parameters())
opt = torch.optim.Adam(params, lr=LR, eps=1e-5)
def dist(o):
return Normal(actor_mean(o), actor_logstd.exp().expand(o.shape[0], act_dim))
obs_buf = torch.zeros(N_STEPS, N_ENVS, obs_dim); act_buf = torch.zeros(N_STEPS, N_ENVS, act_dim)
logp_buf, rew_buf, done_buf, val_buf = (torch.zeros(N_STEPS, N_ENVS) for _ in range(4))
o, _ = envs.reset(seed=seed); o = torch.as_tensor(o, dtype=torch.float32)
returns, n_iters = [], TOTAL_STEPS // (N_ENVS * N_STEPS)
for it in range(1, n_iters + 1):
for g in opt.param_groups:
g["lr"] = LR * (1 - (it - 1) / n_iters) # 学习率线性退火
for t in range(N_STEPS):
with torch.no_grad():
d = dist(o); a = d.sample(); val_buf[t] = critic(o).squeeze(1)
obs_buf[t], act_buf[t], logp_buf[t] = o, a, d.log_prob(a).sum(1) # 存未裁剪的动作,保证概率比正确
o2, r, term, trunc, info = envs.step(np.clip(a.numpy(), a_low, a_high))
rew_buf[t] = torch.as_tensor(r, dtype=torch.float32)
# Pendulum 只有时间截断(200 步):被截断时把 γ·V(s_T) 加进奖励,而不是当作“价值为 0 的终止”
if trunc.any():
final = torch.as_tensor(np.stack([info["final_obs"][i] if trunc[i] else o2[i] for i in range(N_ENVS)]), dtype=torch.float32)
with torch.no_grad():
rew_buf[t] += GAMMA * critic(final).squeeze(1) * torch.as_tensor(trunc, dtype=torch.float32)
done_buf[t] = torch.as_tensor(term | trunc, dtype=torch.float32)
if "episode" in info:
returns += list(info["episode"]["r"][info["_episode"]])
o = torch.as_tensor(o2, dtype=torch.float32)
with torch.no_grad(): # GAE
next_v = critic(o).squeeze(1); adv = torch.zeros_like(rew_buf); last = 0
for t in reversed(range(N_STEPS)):
nv = next_v if t == N_STEPS - 1 else val_buf[t + 1]
delta = rew_buf[t] + GAMMA * nv * (1 - done_buf[t]) - val_buf[t]
last = delta + GAMMA * LAM * (1 - done_buf[t]) * last; adv[t] = last
ret = adv + val_buf
b = [x.reshape(N_STEPS * N_ENVS, *x.shape[2:]) for x in (obs_buf, act_buf, logp_buf, adv, ret)]
for _ in range(EPOCHS):
perm = torch.randperm(N_STEPS * N_ENVS)
for i in range(0, len(perm), MINIBATCH):
j = perm[i:i + MINIBATCH]
s, a, lp_old, A, R = (x[j] for x in b)
A = (A - A.mean()) / (A.std() + 1e-8)
d = dist(s); ratio = (d.log_prob(a).sum(1) - lp_old).exp()
pg = -torch.min(ratio * A, ratio.clamp(1 - CLIP, 1 + CLIP) * A).mean()
v_loss = ((critic(s).squeeze(1) - R) ** 2).mean()
loss = pg + VF_COEF * v_loss # 连续控制中通常不加熵奖励
opt.zero_grad(); loss.backward(); nn.utils.clip_grad_norm_(params, MAX_GRAD); opt.step()
if it % (2 if QUICK else 10) == 0:
print(f" 种子{seed} 步数 {it * N_ENVS * N_STEPS:7d} | 最近20回合平均回报 {np.mean(returns[-20:]):8.1f} | σ={actor_logstd.exp().mean().item():.2f}", flush=True)
# 评估:用高斯策略的均值作为确定性动作
test, scores = gym.make("Pendulum-v1"), []
for k in range(10):
x, _ = test.reset(seed=1000 + k); done, G = False, 0.0
while not done:
with torch.no_grad():
a = actor_mean(torch.as_tensor(x, dtype=torch.float32)[None])[0].numpy()
x, r, te, tr, _ = test.step(np.clip(a, a_low, a_high)); G += r; done = te or tr
scores.append(G)
return float(np.mean(scores))
if __name__ == "__main__":
seeds = [int(sys.argv[1])] if len(sys.argv) > 1 else ([0] if QUICK else [0, 1, 2])
for sd in seeds:
print(f"种子{sd}: 确定性策略评估(10 回合)平均回报 {run(sd):.1f}", flush=True)1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
设置:4 个并行环境,每次采样 1024 步,GAE
| 种子 | 4.1 万步 | 8.2 万步 | 12.3 万步 | 16.4 万步 | 20.5 万步 | 确定性评估(10 回合) |
|---|---|---|---|---|---|---|
| 0 | −502.6 | −241.5 | −185.8 | −155.2 | −217.3 | −206.3 |
| 1 | −252.0 | −201.3 | −152.1 | −149.7 | −180.5 | −202.8 |
| 2 | −459.7 | −222.7 | −173.5 | −174.0 | −176.6 | −226.6 |
3 个种子都在约 8 万步内学会把摆立起来。作为对照,SAC 和 MBPO 页 中的 SAC 在 4000–5000 步时就达到约 −125:同策略的 PPO 每批数据只用一次,样本效率低一到两个数量级,这是 PPO 在连续控制中的典型特点。
又一个真实的调参经历:我们最初照搬了 MuJoCo 上常用的做法——每个环境各自做运行均值方差的观测标准化和奖励标准化,小批量 256。结果 30 万步后确定性评估只有 −1096;之后尝试了延长到 100 万步以上、调整
| 观测标准化 | 奖励标准化 | 确定性评估回报 |
|---|---|---|
| 有 | 有 | −1124.2 |
| 有 | 无 | −1066.1 |
| 无 | 有 | −204.2 |
| 无 | 无 | −203.5 |
问题出在观测标准化上。我们的推测(未进一步验证)是:训练初期策略接近随机,摆大部分时间在底部附近晃动,
常见调参经验
| 超参数 | 常用范围 | 说明 |
|---|---|---|
| 裁剪系数 | 0.1–0.3 | 0.2 最常用 |
| 学习率 | 配合线性退火 | |
| 每次采样步数 | 512 – 8192 | 连续控制常用 2048 |
| epoch 数 | 3–10 | 太大会导致策略偏离过远 |
| 小批量数 | 4–32 | |
| GAE | 0.9–0.99 | 0.95 最常用 |
| 0.99 | 长时域任务可用 0.995–0.999 | |
| 熵系数 | 0–0.01 | 离散动作常用 0.01 |
监控指标:近似 KL(每次更新后新旧策略的 KL,通常应在 0.01–0.05)和裁剪比例(被裁剪的样本占比,通常 0.1–0.3)。KL 过大说明更新太激进,可以减小学习率或 epoch 数,或者设置 KL 早停。
常见误区
- 旧策略的 log 概率没有 detach:
b_logp必须是采样时记录的常数。 - 每个 epoch 重新计算优势:优势应该在采样后计算一次,整个更新过程中保持不变(有的实现会重新计算,但要清楚自己在做什么)。
- 以为裁剪能严格限制 KL:裁剪只是让超出范围的样本没有梯度,多个 epoch 后策略仍可能偏离很远,所以需要监控 KL。
- 连续控制不做观测标准化:MuJoCo 任务中这会严重影响效果。但反过来也不要无条件照搬:本页 Pendulum 实验中,逐环境的观测标准化反而让 PPO 学不会。
- 把裁剪后的动作存进缓冲区:概率比要用实际采样出的(未裁剪)动作计算。
小结
- PPO 用裁剪的概率比构造悲观代理目标,防止策略更新过大。
- 只用一阶优化,同一批数据可以多轮小批量更新,实现简单。
- 实现细节(优势标准化、正交初始化、学习率退火、梯度裁剪等)对效果影响很大。
- PPO 是同策略算法中最常用的基线,也是 RLHF 的经典算法。
练习题
- 把
改为 0.05 和 0.5,比较学习曲线。 - 在代码中记录每次更新的近似 KL
和裁剪比例。 - 去掉学习率退火,结果有何变化?
- 实现 KL 惩罚版本的 PPO,与裁剪版本比较。
- 在
ppo_pendulum.py中把与状态无关的 改为由网络输出(状态相关),比较学习曲线和最终 。 - 在
ppo_pendulum.py中加上观测标准化,但只用所有并行环境共享的统计量,并在前 1 万步之后冻结,看能否避免上面的失败。
延伸阅读
论文速递:PPO 解读;交互工具:折扣回报与 GAE 计算器
参考文献
- Schulman, J., Wolski, F., Dhariwal, P., Radford, A., & Klimov, O. (2017). Proximal Policy Optimization Algorithms. arXiv:1707.06347
- Schulman, J. et al. (2016). High-Dimensional Continuous Control Using Generalized Advantage Estimation. ICLR. arXiv:1506.02438
- Engstrom, L. et al. (2020). Implementation Matters in Deep Policy Gradients: A Case Study on PPO and TRPO. ICLR. arXiv:2005.12729
- Andrychowicz, M. et al. (2021). What Matters in On-Policy Reinforcement Learning? A Large-Scale Empirical Study. ICLR. arXiv:2006.05990
- Huang, S. et al. (2022). The 37 Implementation Details of Proximal Policy Optimization. ICLR Blog Track. ICLR 博客原文