在上一篇文章中，我们介绍了 Actor-Critic 算法，并使用它解决了 CartPole 问题。本文将深入探讨 Proximal Policy Optimization (PPO) 算法，这是一种更稳定、更高效的策略优化方法。我们将使用 PyTorch 实现 PPO 算法，并应用于经典的 CartPole 问题。
一、PPO 算法基础
PPO 是 OpenAI 提出的一种强化学习算法，旨在解决策略梯度方法中的训练不稳定问题。PPO 通过限制策略更新的幅度，确保每次更新不会偏离当前策略太远，从而提高训练的稳定性。
1. PPO 的核心思想
重要性采样：
PPO 使用重要性采样（Importance Sampling）来估计新策略的期望回报，从而避免重新采样。
裁剪目标函数：
PPO 通过裁剪策略更新的幅度，确保新策略不会偏离旧策略太远。
优势函数：
PPO 使用优势函数（Advantage Function）来评估动作的好坏，从而更高效地更新策略。
2. PPO 的优势
训练稳定：
通过限制策略更新的幅度，PPO 避免了策略梯度方法中的训练不稳定问题。
高效采样：
PPO 可以重复使用旧策略的采样数据，从而提高数据利用率。
适用范围广：
PPO 可以应用于连续动作空间和离散动作空间的问题。
3. PPO 的算法流程
使用当前策略采样一批数据。
计算优势函数和旧策略的概率。
通过裁剪目标函数更新策略。
重复上述过程，直到策略收敛。
二、CartPole 问题实战
我们将使用 PyTorch 实现 PPO 算法，并应用于 CartPole 问题。目标是控制小车使其上的杆子保持直立。
1. 问题描述
CartPole 环境的状态空间包括小车的位置、速度、杆子的角度和角速度。动作空间包括向左或向右移动小车。智能体每保持杆子直立一步，就会获得 +1 的奖励，当杆子倾斜超过一定角度或小车移动超出范围时，游戏结束。
2. 实现步骤
安装并导入必要的库。
定义策略网络和价值网络。
定义 PPO 训练过程。
测试模型并评估性能。
3. 代码实现
以下是完整的代码实现：
import gym
import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F
import numpy as np
import random
import matplotlib.pyplot as plt
from torch.optim.lr_scheduler import StepLR
# 可视化设置
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False
# 环境初始化
env = gym.make('CartPole-v1')
state_dim = env.observation_space.shape[0]
action_dim = env.action_space.n
# 随机种子设置
SEED = 42
# env.seed(SEED)
torch.manual_seed(SEED)
np.random.seed(SEED)
random.seed(SEED)
# 增强型策略网络
class PolicyNetwork(nn.Module):
def __init__(self, state_size, action_size):
super().__init__()
self.net = nn.Sequential(
nn.Linear(state_size, 128),
nn.ReLU(),
nn.Linear(128, 128),
nn.ReLU(),
nn.Linear(128, 64),
nn.ReLU(),
nn.Linear(64, action_size)
)
self._initialize_weights()
def _initialize_weights(self):
for m in self.modules():
if isinstance(m, nn.Linear):
nn.init.orthogonal_(m.weight, gain=np.sqrt(2))
nn.init.constant_(m.bias, 0.0)
def forward(self, x):
return self.net(x)
# 增强型价值网络
class ValueNetwork(nn.Module):
def __init__(self, state_size):
super().__init__()
self.net = nn.Sequential(
nn.Linear(state_size, 128),
nn.ReLU(),
nn.Linear(128, 128),
nn.ReLU(),
nn.Linear(128, 64),
nn.ReLU(),
nn.Linear(64, 1)
)
self._initialize_weights()
def _initialize_weights(self):
for m in self.modules():
if isinstance(m, nn.Linear):
nn.init.orthogonal_(m.weight, gain=1.0)
nn.init.constant_(m.bias, 0.0)
def forward(self, x):
return self.net(x)
class PPO:
def __init__(self, state_dim, action_dim):
# 超参数设置
self.gamma = 0.995
self.gae_lambda = 0.97
self.clip_ratio = 0.15
self.update_epochs = 4
self.norm_adv = True
# 网络初始化
self.actor = PolicyNetwork(state_dim, action_dim)
self.critic = ValueNetwork(state_dim)
# 优化器设置
self.actor_optim = optim.Adam(self.actor.parameters(), lr=2e-4)
self.critic_optim = optim.Adam(self.critic.parameters(), lr=8e-4)
# 学习率调度
self.actor_scheduler = StepLR(self.actor_optim, step_size=100, gamma=0.95)
self.critic_scheduler = StepLR(self.critic_optim, step_size=100, gamma=0.95)
def get_action(self, state):
# state_tensor = torch.FloatTensor(state) # 如果使用 Gym <0.26
state_tensor = torch.FloatTensor(state).unsqueeze(0) # 如果使用 Gym >=0.26
with torch.no_grad():
logits = self.actor(state_tensor)
probs = torch.softmax(logits, dim=-1)
action = torch.multinomial(probs, 1).item()
return action
def compute_gae(self, rewards, values, next_values, dones):
advantages = np.zeros_like(rewards)
last_gae = 0
for t in reversed(range(len(rewards))):
if dones[t]:
delta = rewards[t] - values[t]
last_gae = delta
else:
delta = rewards[t] + self.gamma * next_values[t] - values[t]
last_gae = delta + self.gamma * self.gae_lambda * last_gae
advantages[t] = last_gae
return torch.FloatTensor(advantages)
def update(self, states, actions, rewards, next_states, dones):
# 转换为张量
states = torch.FloatTensor(np.array(states))
actions = torch.LongTensor(np.array(actions))
rewards = torch.FloatTensor(np.array(rewards))
next_states = torch.FloatTensor(np.array(next_states))
dones = torch.BoolTensor(np.array(dones))
# 计算价值估计
with torch.no_grad():
current_values = self.critic(states).squeeze()
next_values = self.critic(next_states).squeeze()
next_values[dones] = 0.0
# 计算GAE
advantages = self.compute_gae(rewards.numpy(),
current_values.numpy(),
next_values.numpy(),
dones.numpy())
# 标准化优势
if self.norm_adv:
advantages = (advantages - advantages.mean()) / (advantages.std() + 1e-8)
# 获取旧策略概率
with torch.no_grad():
old_logits = self.actor(states)
old_probs = torch.softmax(old_logits, dim=-1)
old_log_probs = torch.log(old_probs.gather(1, actions.unsqueeze(1))).squeeze()
# 策略优化
for _ in range(self.update_epochs):
logits = self.actor(states)
probs = torch.softmax(logits, dim=-1)
new_log_probs = torch.log(probs.gather(1, actions.unsqueeze(1))).squeeze()
ratios = torch.exp(new_log_probs - old_log_probs)
clipped_ratios = torch.clamp(ratios, 1 - self.clip_ratio, 1 + self.clip_ratio)
policy_loss = -torch.min(ratios * advantages, clipped_ratios * advantages).mean()
self.actor_optim.zero_grad()
policy_loss.backward()
torch.nn.utils.clip_grad_norm_(self.actor.parameters(), 0.6)
self.actor_optim.step()
# 价值优化
for _ in range(self.update_epochs):
current_values = self.critic(states).squeeze()
target_values = rewards + self.gamma * next_values
value_loss = F.mse_loss(current_values, target_values)
self.critic_optim.zero_grad()
value_loss.backward()
torch.nn.utils.clip_grad_norm_(self.critic.parameters(), 0.8)
self.critic_optim.step()
# 更新学习率
self.actor_scheduler.step()
self.critic_scheduler.step()
# 训练流程
def train_ppo(env, agent, episodes=800, early_stop=30):
rewards_history = []
moving_avg = []
best_score = -np.inf
no_improve = 0
for ep in range(episodes):
# 重置环境时确保正确处理返回值
# state = env.reset() # 如果使用 Gym <0.26
state, _ = env.reset() # 如果使用 Gym >=0.26
episode_data = {
'states': [],
'actions': [],
'rewards': [],
'next_states': [],
'dones': []
}
total_reward = 0
done = False
while not done:
action = agent.get_action(state)
# next_state, reward, done, _ = env.step(action) # 如果使用 Gym <0.26
next_state, reward, done, _,_ = env.step(action) # 如果使用 Gym >=0.26
# 处理环境提前终止
if env._elapsed_steps >= env.spec.max_episode_steps:
done = True
reward = 0
# 存储轨迹
episode_data['states'].append(state)
episode_data['actions'].append(action)
episode_data['rewards'].append(reward)
episode_data['next_states'].append(next_state)
episode_data['dones'].append(done)
state = next_state
total_reward += reward
# 更新模型
agent.update(**episode_data)
# 记录训练进度
rewards_history.append(total_reward)
current_avg = np.mean(rewards_history[-50:])
moving_avg.append(current_avg)
# 早停机制
if current_avg > best_score:
best_score = current_avg
no_improve = 0
else:
no_improve += 1
if no_improve >= early_stop:
print(f"早停触发于第{ep + 1}轮，最佳平均奖励: {best_score:.1f}")
break
# 进度报告
if (ep + 1) % 50 == 0:
print(f"Episode {ep + 1:3d} | 当前奖励: {total_reward:4.0f} | 平均奖励: {current_avg:6.1f}")
return moving_avg, rewards_history
# 训练启动
ppo_agent = PPO(state_dim, action_dim)
moving_avg, rewards_history = train_ppo(env, ppo_agent)
# 可视化结果
plt.figure(figsize=(12, 6))
plt.plot(rewards_history, alpha=0.4, label='单轮奖励')
plt.plot(moving_avg, 'r-', linewidth=2, label='滑动平均（50轮）')
plt.xlabel('训练轮次')
plt.ylabel('奖励')
plt.title('PPO训练进程')
plt.legend()
plt.grid(True)
plt.show()三、代码解析
策略网络和价值网络：
策略网络输出动作的 logits（未归一化的概率），通过
F.softmax转换为概率分布。价值网络输出状态的价值估计。
PPO 训练过程：
使用当前策略采样一批数据。
计算优势函数和旧策略的概率。
通过裁剪目标函数更新策略。
训练过程：
在训练过程中，每 50 个 episode 打印一次平均奖励。
训练结束后，绘制训练过程中的总奖励曲线。
四、运行结果
运行上述代码后，你将看到以下输出：
训练过程中每 50 个 episode 打印一次平均奖励。
训练结束后，绘制训练过程中的总奖励曲线。
Episode 50 | 当前奖励: 29 | 平均奖励: 20.8 Episode 100 | 当前奖励: 132 | 平均奖励: 69.1 Episode 150 | 当前奖励: 499 | 平均奖励: 261.6 Episode 200 | 当前奖励: 295 | 平均奖励: 412.3 Episode 250 | 当前奖励: 499 | 平均奖励: 481.8 早停触发于第290轮，最佳平均奖励: 497.7
五、总结
本文介绍了 PPO 算法的基本原理，并使用 PyTorch 实现了一个简单的 PPO 模型来解决 CartPole 问题。通过这个例子，我们学习了如何使用 PPO 算法进行策略优化。
在下一篇文章中，我们将探讨更高级的强化学习算法，如 Deep Deterministic Policy Gradient (DDPG)。敬请期待！
代码实例说明：
本文代码可以直接在 Jupyter Notebook 或 Python 脚本中运行。
如果你有 GPU，可以将模型和数据移动到 GPU 上运行，例如：
actor = actor.to('cuda')，state = state.to('cuda')。
希望这篇文章能帮助你更好地理解 PPO 算法！如果有任何问题，欢迎在评论区留言讨论。