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

3. 为什么 PPO 很强？

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

4. PPO 的直观类比

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

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

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

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

1. 算法目标

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

目标是优化以下期望：

$$J(\theta) = \mathbb{E}_{\pi\theta} \left[ R \right]$$

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

2. PPO 的概率比率

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

$$r_t(\theta) = \frac{\pi_\theta(a_t | s_t)}{\pi_{\theta_{\text{old}}}(a_t | s_t)}$$

• $\pi_{\theta_{\text{old}}}(a_t | s_t)$：旧策略对动作 $a_t$ 的概率。
• $\pi_\theta(a_t | s_t)$：新策略对动作 $a_t$ 的概率。

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

3. 优化目标

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

$$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]$$

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

4. 值函数优化

PPO 不仅优化策略，还同时更新值函数 $V(s_t)$，通过最小化均方误差来更新：

$$L^{VF}(\theta) = \mathbb{E}_t \left[ \left( V(s_t; \theta) - R_t \right)^2 \right]$$

• $R_t = \sum_{k=0}^n \gamma^k r_{t+k}$：累计回报。
• $V(s_t; \theta)$：当前状态的值函数预测。

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

5. 策略熵正则化

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

$$L^{ENT}(\theta) = \mathbb{E}t \left[ H(\pi\theta(s_t)) \right]$$

• 增加熵可以防止策略过早收敛到局部最优。
• $H(\pi_\theta(s_t))$：策略的熵，表示策略分布的不确定性。

6. 总损失函数

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$：权重系数，用于平衡策略优化、值函数更新和熵正则化。

7. PPO 算法流程

PPO 可以简化为以下步骤：

1. 重复以上步骤：通过多轮迭代，使策略逐步优化，直到收敛。
2. 值函数更新：用以下损失函数优化值函数 $V(s_t)$： $$L^{VF}(\theta) = \mathbb{E}_t \left[ \left( V(s_t) - R_t \right)^2 \right]$$
3. 策略更新：如果更新过大（超出剪辑范围 $1-\epsilon$ 到 $1+\epsilon$），会被惩罚。保证更新幅度适中，既不太保守，也不太激进。
4. 计算概率比率：比较新策略和旧策略对动作 $a_t$ 的选择概率。
5. 计算优势函数：评估某个动作 $a_t$ 在状态 $s_t$ 下相对于平均表现的优劣（优势函数 $A_t$）。利用 $A_t$ 引导策略改进。
6. 采样：使用当前策略 $\pi_\theta$ 与环境交互，收集状态 $s_t$、动作 $a_t$、奖励 $r_t$。

8. PPO 算法的关键

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

• 探索与稳定性平衡：通过 $L^{ENT}(\theta)$，鼓励策略探索。
• 同时优化值函数：通过 $L^{VF}(\theta)$，提高 Critic 的预测精度。
• 限制更新幅度：通过剪辑函数 $\text{clip}()$，避免策略更新过大导致不稳定。
• 核心目标：优化策略，使 $r_t(\theta) A_t$ 的改进在限制范围内。

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

Python PPO 算法的代码实现

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

"""《PPO 算法的代码》 时间：2024.12 环境：gym 作者：不去幼儿园 """
import torch # Import PyTorch, a popular machine learning library
import torch.nn as nn # Import the neural network module
import torch.optim as optim # Import optimization algorithms
from torch.distributions import Categorical # Import Categorical for probabilistic action sampling
import numpy as np # Import NumPy for numerical computations
import gym # Import OpenAI Gym for environment simulation

逐行解释 PPO 代码和公式

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

1. Actor-Critic 神经网络

# Define Actor-Critic Network
class ActorCritic(nn.Module): 
    # Define the Actor-Critic model
    def __init__(self, state_dim, action_dim): 
        # Initialize with state and action dimensions
        super(ActorCritic, self).__init__() # Call parent class constructor
        self.shared_layer = nn.Sequential( # Shared network layers for feature extraction
            nn.Linear(state_dim, 128), # Fully connected layer with 128 neurons
            nn.ReLU() # ReLU activation function
        )
        self.actor = nn.Sequential( # Define the actor (policy) network
            nn.Linear(128, action_dim), # Fully connected layer to output action probabilities
            nn.Softmax(dim=-1) # Softmax to ensure output is a probability distribution
        )
        self.critic = nn.Linear(128, 1) # Define the critic (value) network to output state value

    def forward(self, state): 
        # Forward pass for the model
        shared = self.shared_layer(state) # Pass state through shared layers
        action_probs = self.actor(shared) # Get action probabilities from actor network
        state_value = self.critic(shared) # Get state value from critic network
        return action_probs, state_value # Return action probabilities and state value

• shared_layer：将状态 $s$ 映射到一个隐层表示，使用 ReLU 激活函数。
• critic：输出状态值函数 $V(s)$，表示在状态 $s$ 下的预期累计奖励。
• actor：输出策略 $\pi_\theta(a|s)$，表示在状态 $s$ 下选择动作 $a$ 的概率分布。使用 Softmax 确保输出是概率。

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

其中 $f(s)$ 是共享层的输出，$W_a$ 和 $W_v$ 是权重矩阵。

2. Memory 类

# Memory to store experiences
class Memory: 
    # Class to store agent's experience
    def __init__(self): 
        # Initialize memory
        self.states = [] # List to store states
        self.actions = [] # List to store actions
        self.logprobs = [] # List to store log probabilities of actions
        self.rewards = [] # List to store rewards
        self.is_terminals = [] # List to store terminal state flags

    def clear(self): 
        # Clear memory after an update
        self.states = [] # Clear stored states
        self.actions = [] # Clear stored actions
        self.logprobs = [] # Clear stored log probabilities
        self.rewards = [] # Clear stored rewards
        self.is_terminals = [] # Clear terminal state flags

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

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

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

3. PPO 初始化

# PPO Agent class
class PPO: 
    # Define the PPO agent
    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) # Initialize the Actor-Critic model
        self.optimizer = optim.Adam(self.policy.parameters(), lr=lr) # Adam optimizer for parameter updates
        self.policy_old = ActorCritic(state_dim, action_dim).to(device) # Copy of the policy for stability
        self.policy_old.load_state_dict(self.policy.state_dict()) # Synchronize parameters
        self.MseLoss = nn.MSELoss() # Mean Squared Error loss for critic updates
        self.gamma = gamma # Discount factor for rewards
        self.eps_clip = eps_clip # Clipping parameter for PPO
        self.K_epochs = K_epochs # Number of epochs for optimization

• policy：当前策略网络，用于输出动作概率和状态值。
• gamma：折扣因子，用于奖励的时间衰减。
• eps_clip：剪辑阈值 $\epsilon$，用于限制策略更新幅度。
• policy_old：旧策略网络，用于计算概率比率 $r_t(\theta)$。

4. 动作选择

def select_action(self, state, memory):
    state = torch.FloatTensor(state).to(device) # Convert state to PyTorch tensor
    action_probs, _ = self.policy_old(state) # Get action probabilities from old policy
    dist = Categorical(action_probs) # Create a categorical distribution
    action = dist.sample() # Sample an action from the distribution
    memory.states.append(state) # Store state in memory
    memory.actions.append(action) # Store action in memory
    memory.logprobs.append(dist.log_prob(action)) # Store log probability of the action
    return action.item() # Return action as a scalar value

• log_prob(action)：记录动作的对数概率 $\log \pi_\theta(a_t|s_t)$。
• dist.sample()：按照概率分布采样动作 $a_t$。
• action_probs：当前策略下的动作概率 $\pi_\theta(a|s)$。

5. 策略更新

def update(self, memory): 
    # Convert memory to tensors
    old_states = torch.stack(memory.states).to(device).detach() # Convert states to tensor
    old_actions = torch.stack(memory.actions).to(device).detach() # Convert actions to tensor
    old_logprobs = torch.stack(memory.logprobs).to(device).detach() # Convert log probabilities to tensor
    
    # Monte Carlo rewards
    rewards = [] # Initialize rewards list
    discounted_reward = 0 # Initialize discounted reward
    for reward, is_terminal in zip(reversed(memory.rewards), reversed(memory.is_terminals)): 
        if is_terminal: # If the state is terminal
            discounted_reward = 0 # Reset discounted reward
        discounted_reward = reward + (self.gamma * discounted_reward) # Compute discounted reward
        rewards.insert(0, discounted_reward) # Insert at the beginning of the list
    rewards = torch.tensor(rewards, dtype=torch.float32).to(device) # Convert rewards to tensor
    rewards = (rewards - rewards.mean()) / (rewards.std() + 1e-7) # Normalize rewards

6. Surrogate Loss

# Update for K epochs
for _ in range(self.K_epochs): 
    # Get action probabilities and state values
    action_probs, state_values = self.policy(old_states) # Get action probabilities and state values
    dist = Categorical(action_probs) # Create a categorical distribution
    new_logprobs = dist.log_prob(old_actions) # Compute new log probabilities of actions
    entropy = dist.entropy() # Compute entropy for exploration
    
    # Calculate ratios
    ratios = torch.exp(new_logprobs - old_logprobs.detach()) # Compute probability ratios
    
    # Advantages
    advantages = rewards - state_values.detach().squeeze() # Compute advantages
    
    # Surrogate loss
    surr1 = ratios * advantages # Surrogate loss 1
    surr2 = torch.clamp(ratios, 1 - self.eps_clip, 1 + self.eps_clip) * advantages # Clipped loss
    loss_actor = -torch.min(surr1, surr2).mean() # Actor loss
    
    # Critic loss
    loss_critic = self.MseLoss(state_values.squeeze(), rewards) # Critic loss
    
    # Total loss
    loss = loss_actor + 0.5 * loss_critic - 0.01 * entropy.mean() # Combined loss
    
    # Update policy
    self.optimizer.zero_grad() # Zero the gradient buffers
    loss.backward() # Backpropagate loss
    self.optimizer.step() # Perform a parameter update
    
    # Update old policy
    self.policy_old.load_state_dict(self.policy.state_dict()) # Copy new policy parameters to old policy

• 熵正则化：$H(\pi_\theta(s_t)) = - \sum_a \pi_\theta(a|s_t) \log \pi_\theta(a|s_t)$
• 价值函数损失：$L^{VF}(\theta) = \left( V(s_t) - R_t \right)^2$
• 策略损失：$L^{CLIP}(\theta) = \min \left( r_t(\theta) A_t, \text{clip}(r_t(\theta), 1-\epsilon, 1+\epsilon) A_t \right)$
• 比率：$r_t(\theta) = \frac{\pi_\theta(a_t|s_t)}{\pi_{\text{old}}(a_t|s_t)}$

7. 主程序

# Hyperparameters
device = torch.device("cuda" if torch.cuda.is_available() else "cpu") # Use GPU if available
env = gym.make("CartPole-v1") # Initialize CartPole environment
state_dim = env.observation_space.shape[0] # Dimension of state space
action_dim = env.action_space.n # Number of possible actions
lr = 0.002 # Learning rate
gamma = 0.99 # Discount factor
eps_clip = 0.2 # Clipping parameter
K_epochs = 4 # Number of epochs for policy update
max_episodes = 1000 # Maximum number of episodes
max_timesteps = 300 # Maximum timesteps per episode

# PPO Training
ppo = PPO(state_dim, action_dim, lr, gamma, eps_clip, K_epochs) # Initialize PPO agent
memory = Memory() # Initialize memory

for episode in range(1, max_episodes + 1): 
    # Loop over episodes
    state = env.reset() # Reset environment
    total_reward = 0 # Initialize total reward
    for t in range(max_timesteps): 
        # Loop over timesteps
        action = ppo.select_action(state, memory) # Select action using PPO
        state, reward, done, _ = env.step(action) # Take action and observe results
        memory.rewards.append(reward) # Store reward in memory
        memory.is_terminals.append(done) # Store terminal state flag in memory
        total_reward += reward # Accumulate total reward
        if done: # If episode is done
            break # Exit loop
    ppo.update(memory) # Update PPO agent
    memory.clear() # Clear memory
    print(f"Episode {episode}, Total Reward: {total_reward}") # Print episode statistics

env.close() # Close the environment

代码解释

• Actor-Critic 网络结构：ActorCritic 模型通过共享层生成动作概率和状态值。
• PPO 的优化目标：使用裁剪的目标函数限制策略更新幅度，避免过度更新。
• State Values：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 在资源受限的场景下表现优异。