AC强化学习实战：从算法原理到工业级实现

1次阅读

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

传统的 Q -Learning 和 DQN 在离散动作空间中表现良好，但在连续控制任务（如机器人控制、自动驾驶等）中存在明显局限性。主要原因包括：

动作空间离散化 ：连续动作需要被离散化处理，导致维度爆炸问题
高方差估计 ：Q 值估计在连续空间中波动剧烈，收敛困难
策略单一 ：无法生成随机策略，探索效率低下

AC（Actor-Critic）框架通过分离策略网络（Actor）和价值网络（Critic），完美解决了这些问题。Actor 直接输出连续动作，Critic 评估状态价值，二者协同工作大幅提升了算法稳定性。

算法	适用场景	计算开销	关键改进
A2C	单机环境	中等	同步更新
A3C	分布式 CPU	较高	异步梯度
SAC	高维动作	高	熵最大化

import torch
import torch.nn as nn

# Actor 网络（策略网络）class PolicyNetwork(nn.Module):
    def __init__(self, state_dim, action_dim, hidden_size=256):
        super().__init__()
        self.fc1 = nn.Linear(state_dim, hidden_size)
        self.fc2 = nn.Linear(hidden_size, action_dim)  # 输出动作均值
        self.log_std = nn.Parameter(torch.zeros(action_dim))  # 可训练的动作标准差

    def forward(self, state):
        x = torch.relu(self.fc1(state))
        mean = torch.tanh(self.fc2(x))  # [-1,1] 范围输出
        return torch.distributions.Normal(mean, self.log_std.exp())

# Critic 网络（价值网络）class ValueNetwork(nn.Module):
    def __init__(self, state_dim, hidden_size=256):
        super().__init__()
        self.fc1 = nn.Linear(state_dim, hidden_size)
        self.fc2 = nn.Linear(hidden_size, 1)  # 输出状态价值

    def forward(self, state):
        x = torch.relu(self.fc1(state))
        return self.fc2(x)

优势函数计算 ：

def compute_advantages(rewards, values, masks, gamma=0.99, tau=0.95):
    returns = torch.zeros_like(rewards)
    advantages = torch.zeros_like(rewards)
    next_value = 0

    for t in reversed(range(len(rewards))):
        returns[t] = rewards[t] + gamma * masks[t] * next_value
        next_value = values[t]

    advantages = returns - values
    advantages = (advantages - advantages.mean()) / (advantages.std() + 1e-8)
    return advantages

策略梯度更新 ：

def update_policy(actor, critic, optimizer, states, actions, advantages):
    dist = actor(states)
    log_probs = dist.log_prob(actions).sum(-1)

    # 策略梯度损失
    policy_loss = -(log_probs * advantages).mean()

    # 价值函数损失
    value_loss = F.mse_loss(critic(states), returns)

    # 熵正则化
    entropy = dist.entropy().mean()

    optimizer.zero_grad()
    (policy_loss + 0.5*value_loss - 0.01*entropy).backward()
    torch.nn.utils.clip_grad_norm_(actor.parameters(), 0.5)
    optimizer.step()

graph TD
    A[Parameter Server] -->| 同步参数 | B[Worker1]
    A -->| 同步参数 | C[Worker2]
    B -->| 上传梯度 | A
    C -->| 上传梯度 | A

每个 Worker 独立与环境交互，定期从参数服务器拉取最新网络参数，计算梯度后上传更新。

from torch.cuda.amp import autocast, GradScaler

scaler = GradScaler()

with autocast():
    dist = actor(states)
    log_probs = dist.log_prob(actions)
    loss = -(log_probs * advantages).mean()

scaler.scale(loss).backward()
scaler.step(optimizer)
scaler.update()