近端策略优化算法 PPO 详解与代码实现

：优势函数 通过以下公式计算：

或者用广义优势估计（GAE）的方法近似。

：概率比率

它表示新策略和旧策略在同一状态下选择动作的概率比值。

3. 为什么 PPO 很强？

1. 简洁性： 比 TRPO（Trust Region Policy Optimization）更简单，无需二次优化。
2. 稳定性： 使用剪辑机制防止策略更新过度。
3. 高效性： 利用采样数据多次训练，提高样本利用率。

4. PPO 的直观类比

假设你是一个篮球教练，训练球员投篮：

• 如果每次训练完全改变投篮动作，球员可能会表现失常（类似于策略更新过度）。
• 如果每次训练动作变化太小，可能很难进步（类似于更新不足）。
• PPO 的剪辑机制就像一个'适度改进'的规则，告诉球员在合理范围内调整投篮动作，同时评估每次投篮的表现是否优于平均水平。

PPO 算法的流程推导及数学公式

PPO（Proximal Policy Optimization）也是一种策略优化算法，它的核心思想是对策略更新进行限制，使训练更加稳定，同时保持效率。以下是其数学公式推导和整体流程：

1. 算法目标

强化学习的核心目标是优化策略

，最大化累积奖励 R。策略梯度方法（如 REINFORCE）直接优化策略，但更新过大可能导致不稳定。为了解决这个问题，PPO 通过引入限制更新幅度的机制，保证策略的稳定性。

目标是优化以下期望：

通过梯度上升法更新策略。

2. PPO 的概率比率

PPO 在优化过程中引入了概率比率，用于衡量新旧策略的差异：

：旧策略对动作

的概率。

：新策略对动作

的概率。

这个比率表示策略变化的程度。

3. 优化目标

为了限制策略的更新幅度，PPO 引入了剪辑目标函数：

PPO 的目标是找到一个折中：在保持改进的同时防止策略变化过大。

4. 值函数优化

PPO 不仅优化策略，还同时更新值函数

，通过最小化均方误差来更新：

：累计回报。

：当前状态的值函数预测。

这个损失函数使得 Critic 能够更准确地估计状态值。

5. 策略熵正则化

为了鼓励策略的探索，PPO 引入了熵正则化项：

• 增加熵可以防止策略过早收敛到局部最优。

：策略的熵，表示策略分布的不确定性。

6. 总损失函数

PPO 结合策略损失、值函数损失和熵正则化项，形成总损失函数：

和

：权重系数，用于平衡策略优化、值函数更新和熵正则化。

7. PPO 算法流程

PPO 可以简化为以下步骤：

1. 重复以上步骤： 通过多轮迭代，使策略逐步优化，直到收敛。

值函数更新： 用以下损失函数优化值函数

：

**策略更新：**如果更新过大（超出剪辑范围

到

），会被惩罚。 保证更新幅度适中，既不太保守，也不太激进。

计算概率比率

： 比较新策略和旧策略对动作

的选择概率。

**计算优势函数：**评估某个动作

在状态

下相对于平均表现的优劣（优势函数

）。 利用

引导策略改进。

采样： 使用当前策略

与环境交互，收集状态

、动作

、奖励

。

8. PPO 算法的关键

PPO 的关键公式和目标可以概括如下：

探索与稳定性平衡： 通过

，鼓励策略探索。

同时优化值函数： 通过

，提高 Critic 的预测精度。

限制更新幅度： 通过剪辑函数

，避免策略更新过大导致不稳定。

核心目标： 优化策略，使

的改进在限制范围内。

这种设计使得 PPO 在训练过程中高效且稳定，是目前强化学习中的常用算法之一。

Python PPO 算法的代码实现

以下是使用 PyTorch 实现 PPO（Proximal Policy Optimization）算法的完整代码

"""PPO 算法代码示例"""
import torch
import torch.nn as nn
import torch.optim as optim
from torch.distributions import Categorical
import numpy as np
import gym

逐行解释 PPO 代码和公式

以下是对实现的 PyTorch PPO 算法代码的详细解释，逐行结合公式解析：

1. Actor-Critic 神经网络

# Define Actor-Critic Network
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()
        )
        self.actor = nn.Sequential(
            nn.Linear(128, action_dim),
            nn.Softmax(dim=-1)
        )
        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

• shared_layer: 将状态 s 映射到一个隐层表示，使用 ReLU 激活函数。
• critic: 输出状态值函数 V(s)，表示在状态 s 下的预期累计奖励。
• actor: 输出策略

，表示在状态 s 下选择动作 a 的概率分布。使用 Softmax 确保输出是概率。

![\pi_\theta(a|s) = \text{Softmax}(W_a \cdot f(s)) ] V(s) = W_v \cdot f(s)

其中

是共享层的输出，

和

是权重矩阵。

2. Memory 类

# Memory to store experiences
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 = []

用于存储一个 episode 的经验数据：

states: 状态、actions: 动作、logprobs: 动作的对数概率、rewards: 即时奖励、is_terminals: 是否为终止状态（布尔值）

作用：为后续策略更新提供样本数据。

3. PPO 初始化

# PPO Agent class
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

• policy: 当前策略网络，用于输出动作概率和状态值。
• gamma: 折扣因子，用于奖励的时间衰减。
• eps_clip: 剪辑阈值

，用于限制策略更新幅度。

• policy_old: 旧策略网络，用于计算概率比率

。

4. 动作选择

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

log_prob(action): 记录动作的对数概率

。

dist.sample(): 按照概率分布采样动作

。

action_probs: 当前策略下的动作概率

。

5. 策略更新

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)

6. Surrogate Loss

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

熵正则化：

价值函数损失：

策略损失：

比率：

7. 主程序

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
env = gym.make("CartPole-v1")
state_dim = env.observation_space.shape[0]
action_dim = env.action_space.n
lr = 0.002
gamma = 0.99
eps_clip = 0.2
K_epochs = 4
max_episodes = 1000
max_timesteps = 300

ppo = PPO(state_dim, action_dim, lr, gamma, eps_clip, K_epochs)
memory = Memory()

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

代码说明

Actor-Critic 网络结构：

ActorCritic 模型通过共享层生成动作概率和状态值。

PPO 的优化目标：

使用裁剪的目标函数限制策略更新幅度，避免过度更新。

state_values 通过 Critic 网络提供状态价值的估计。

内存存储：

Memory 用于存储每一轮的状态、动作、奖励和终止标志。

策略更新：

通过多个 epoch 更新，计算优势函数和裁剪后的策略梯度。

奖励归一化：

使用标准化方法对奖励进行处理，以加快收敛。

训练循环：

PPO 从环境中采样，更新策略，打印每一集的总奖励。

# 环境配置
Python 3.11.5
torch 2.1.0
torchvision 0.16.0
gym 0.26.2

由于博文主要为了介绍相关算法的原理和应用的方法，缺乏对于实际效果的关注，算法可能在上述环境中的效果不佳或者无法运行，一是算法不适配上述环境，二是算法未调参和优化，三是没有呈现完整的代码，四是等等。上述代码用于了解和学习算法足够了，但若是想直接将上面代码应用于实际项目中，还需要进行修改。

总结

PPO 的关键是通过限制策略的变化范围（剪辑），让优化更加稳定，同时通过优势函数引导策略改进，充分利用采样数据。这种平衡使得 PPO 成为许多强化学习任务的默认算法。

PPO 算法、TRPO 算法 和 A3C 算法对比

以下是 PPO 算法、TRPO 算法 和 A3C 算法的区别分析：

三种算法的对比总结：

1. PPO 是 TRPO 的改进版： PPO 使用简单的裁剪机制代替了 TRPO 的二次优化，显著降低了实现复杂度，同时保持了良好的稳定性和效率。
2. A3C 的并行化设计： A3C 的核心是通过多线程异步更新提升效率，但其稳定性略低于 PPO 和 TRPO。
3. 实用性： PPO 因其简单、稳定、高效的特点，已成为强化学习领域的主流算法；TRPO 更适合需要极高策略稳定性的任务；A3C 在资源受限的场景下表现优异。