A3C强化学习实战：从原理到分布式训练优化

1次阅读

没有评论

共计 2465 个字符，预计需要花费 7 分钟才能阅读完成。

A3C（Asynchronous Advantage Actor-Critic）是强化学习领域的重要算法，它建立在 Actor-Critic 框架之上。要理解 A3C，首先需要掌握两个关键概念：

Actor：负责根据当前状态选择动作的策略函数
Critic：评估状态或动作的价值，指导 Actor 的更新

A3C 的创新点在于引入 异步优势函数 和多线程并行训练。优势函数（Advantage）的计算公式为：

$$A(s,a) = Q(s,a) – V(s)$$

其中 $Q(s,a)$ 是动作价值函数，$V(s)$ 是状态价值函数。这个设计让算法能更准确地评估动作的相对好坏。

样本效率：A3C 通过多线程并行收集经验，比 DQN 的单一经验回放更高效
策略类型：DQN 只能处理离散动作空间，而 A3C 的 Actor 可以输出连续动作
收敛速度：A3C 的异步更新通常比 DQN 收敛更快

并行度：A3C 天然支持分布式，PPO 通常需要额外设计
稳定性：PPO 有 clip 机制保证训练稳定，A3C 需要更精细的超参调节
实现复杂度：A3C 的异步逻辑实现起来比 PPO 稍复杂

import torch
import torch.nn as nn
import torch.optim as optim

class ActorCritic(nn.Module):
    def __init__(self, input_dim, output_dim):
        super(ActorCritic, self).__init__()
        # 共享的特征提取层
        self.feature = nn.Sequential(nn.Linear(input_dim, 128),
            nn.ReLU())
        # Actor 分支 - 输出动作概率
        self.actor = nn.Linear(128, output_dim)
        # Critic 分支 - 输出状态价值
        self.critic = nn.Linear(128, 1)

    def forward(self, x):
        x = self.feature(x)
        policy = torch.softmax(self.actor(x), dim=-1)
        value = self.critic(x)
        return policy, value

import threading

def train(global_model, optimizer, thread_id):
    # 每个线程有自己的局部模型
    local_model = ActorCritic(input_dim, output_dim)
    local_model.load_state_dict(global_model.state_dict())

    while True:
        # 1. 收集经验
        states, actions, rewards = collect_experience(local_model)

        # 2. 计算优势
        advantages = compute_advantages(local_model, states, rewards)

        # 3. 计算损失
        policy, value = local_model(states)
        policy_loss = -torch.log(policy[range(len(actions)), actions]) * advantages
        value_loss = F.mse_loss(value.squeeze(), rewards)
        loss = policy_loss.mean() + 0.5 * value_loss

        # 4. 异步更新全局模型
        optimizer.zero_grad()
        loss.backward()
        # 梯度裁剪防止爆炸
        torch.nn.utils.clip_grad_norm_(local_model.parameters(), 0.5)
        # 将局部梯度累加到全局模型
        for global_param, local_param in zip(global_model.parameters(), 
                                           local_model.parameters()):
            if global_param.grad is not None:
                break
            global_param._grad = local_param.grad
        optimizer.step()

        # 5. 同步最新全局参数
        local_model.load_state_dict(global_model.state_dict())