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

近端策略优化算法 (PPO) 详解与代码实现 | 极客日志

"""《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

# 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

# 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

# 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

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

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

# 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

# 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

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

特性	PPO (Proximal Policy Optimization)	TRPO (Trust Region Policy Optimization)	A3C (Asynchronous Advantage Actor-Critic)
核心思想	使用裁剪的目标函数，限制策略更新幅度，保持稳定性和效率。	限制策略更新的步幅（Trust Region），通过二次约束优化确保稳定性。	通过异步多线程运行环境并行采样和训练，降低方差并加快收敛速度。
优化目标函数	引入剪辑机制	通过 KL 散度限制策略更新	优化策略梯度
更新方式	同步更新，支持多轮迭代更新样本数据以提高效率。	同步更新，通过优化约束的目标函数严格限制更新步长。	异步更新，多个线程独立采样和更新全局模型。
计算复杂度	低，计算简单，使用裁剪避免复杂的二次优化问题。	高，涉及二次优化问题，计算复杂，资源需求较大。	较低，依赖异步线程并行计算，资源利用率高。
样本利用率	高效，可重复利用采样数据进行多轮梯度更新。	高效，严格优化目标，提升了样本效率。	较低，因为每个线程独立运行，可能导致数据重复和冗余。
实现难度	中等，使用简单的裁剪方法，适合大多数场景。	高，涉及复杂的约束优化和实现细节。	较低，直接异步实现，简单易用。
收敛速度	快，因裁剪机制限制更新幅度，能快速稳定收敛。	慢，因严格的步幅限制，收敛稳定但需要较多训练迭代。	快，因多线程并行采样，能够显著减少训练时间。
稳定性	高，裁剪机制限制过大更新，避免不稳定行为。	高，严格限制更新步幅，保证策略稳定改进。	较低，异步更新可能导致收敛不稳定（如策略冲突）。
应用场景	广泛使用，适合大规模环境或复杂问题。	适合需要极高稳定性的场景，如机器人控制等。	适合资源受限的场景或需要快速实验的任务，如强化学习基准测试。
优点	简单易实现，收敛快，稳定性高，是主流强化学习算法。	理论支持强，更新步幅严格受控，策略非常稳定。	异步更新高效，能够充分利用多线程资源，加速训练。
缺点	理论支持弱于 TRPO，可能过于保守。	实现复杂，计算资源需求高，更新速度慢。	异步更新可能导致训练不稳定，样本利用率较低。
论文来源	Schulman et al., "Proximal Policy Optimization Algorithms" (2017)	Schulman et al., "Trust Region Policy Optimization" (2015)	Mnih et al., "Asynchronous Methods for Deep Reinforcement Learning" (2016)

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

近端策略优化算法 (PPO) 详解

PPO 算法介绍

1. 背景

2. PPO 的核心思想

优化目标

3. 为什么 PPO 很强？

4. PPO 的直观类比

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

1. 算法目标

2. PPO 的概率比率

3. 优化目标

4. 值函数优化

5. 策略熵正则化

6. 总损失函数

7. PPO 算法流程

8. PPO 算法的关键

Python PPO 算法的代码实现

逐行解释 PPO 代码和公式

1. Actor-Critic 神经网络

2. Memory 类

3. PPO 初始化

4. 动作选择

5. 策略更新

6. Surrogate Loss

7. 主程序

代码解释

总结

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

三种算法的对比总结：

更多推荐文章

相关免费在线工具

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

近端策略优化算法 (PPO) 详解

PPO 算法介绍

1. 背景

2. PPO 的核心思想

优化目标

3. 为什么 PPO 很强？

4. PPO 的直观类比

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

1. 算法目标

2. PPO 的概率比率

3. 优化目标

4. 值函数优化

5. 策略熵正则化

6. 总损失函数

7. PPO 算法流程

8. PPO 算法的关键

Python PPO 算法的代码实现

逐行解释 PPO 代码和公式

1. Actor-Critic 神经网络

2. Memory 类

3. PPO 初始化

4. 动作选择

5. 策略更新

6. Surrogate Loss

7. 主程序

代码解释

总结

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

三种算法的对比总结：

微信扫一扫，关注极客日志

更多推荐文章

相关免费在线工具

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