PyTorch 深度学习实战(13):Proximal Policy Optimization (PPO) 算法

发布于:2025-03-16 ⋅ 阅读:(13) ⋅ 点赞:(0)

在上一篇文章中,我们介绍了 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 的算法流程

  1. 使用当前策略采样一批数据。

  2. 计算优势函数和旧策略的概率。

  3. 通过裁剪目标函数更新策略。

  4. 重复上述过程,直到策略收敛。


二、CartPole 问题实战

我们将使用 PyTorch 实现 PPO 算法,并应用于 CartPole 问题。目标是控制小车使其上的杆子保持直立。

1. 问题描述

CartPole 环境的状态空间包括小车的位置、速度、杆子的角度和角速度。动作空间包括向左或向右移动小车。智能体每保持杆子直立一步,就会获得 +1 的奖励,当杆子倾斜超过一定角度或小车移动超出范围时,游戏结束。

2. 实现步骤

  1. 安装并导入必要的库。

  2. 定义策略网络和价值网络。

  3. 定义 PPO 训练过程。

  4. 测试模型并评估性能。

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()

三、代码解析

  1. 策略网络和价值网络

    • 策略网络输出动作的 logits(未归一化的概率),通过 F.softmax 转换为概率分布。

    • 价值网络输出状态的价值估计。

  2. PPO 训练过程

    • 使用当前策略采样一批数据。

    • 计算优势函数和旧策略的概率。

    • 通过裁剪目标函数更新策略。

  3. 训练过程

    • 在训练过程中,每 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 算法!如果有任何问题,欢迎在评论区留言讨论。