共计 3365 个字符,预计需要花费 9 分钟才能阅读完成。
1. 多智能体强化学习的应用背景
多智能体强化学习 (Multi-Agent Reinforcement Learning, MARL) 在近年来的快速发展,主要得益于其在复杂协同任务中的出色表现。无论是游戏 AI 中的团队对战,还是现实世界中的机器人协作,MARL 都展现出了强大的潜力。例如,在 MOBA 类游戏中,多个英雄需要相互配合才能取得胜利;在仓储物流中,多台机器人需要协调路径以避免碰撞。这些场景的共同特点是:每个智能体的决策会相互影响,且整体收益取决于所有智能体的联合行为。

2. 多智能体学习方案对比
在 MARL 中,最直观的方法是让每个智能体独立运行 Q -learning 算法(Independent Q-Learning, IQL)。这种方法实现简单,但存在明显缺陷:
- 环境对每个智能体来说是非平稳的(因为其他智能体也在学习)
- 无法显式建模智能体间的交互关系
MADDPG(Multi-Agent Deep Deterministic Policy Gradient)通过集中式训练、分布式执行的框架部分解决了这些问题。但其 critic 网络需要知道所有智能体的动作,这在某些实际场景中可能不现实。相比之下,actor-critic 结构提供了更灵活的框架:
- actor(策略网络)是分布式的,每个智能体独立决策
- critic(价值网络)可以采用不同架构(集中式 / 分布式)来平衡性能与可行性
3. Actor-Critic 核心实现
3.1 分布式 critic 网络架构
我们设计了一个混合架构的 critic 网络:
- 局部观察编码层:每个智能体通过 MLP 处理自己的观察 $o_i$
- 联合特征提取层:所有智能体的编码特征被拼接后输入到共享的 LSTM 网络
- 价值输出层:输出联合动作价值 $Q^\pi(s,a_1,…,a_N)$
这种设计既考虑了单个智能体的局部信息,又通过共享网络捕获了智能体间的交互模式。
3.2 PyTorch 实现关键代码
import torch
import torch.nn as nn
import torch.nn.functional as F
class CentralizedCritic(nn.Module):
"""集中式 critic 网络实现"""
def __init__(self, obs_dim, act_dim, num_agents, hidden_size=128):
super().__init__()
# 局部观察编码器(每个智能体独立)self.obs_encoder = nn.ModuleList([nn.Linear(obs_dim, hidden_size) for _ in range(num_agents)]
)
# 联合动作处理层
self.joint_action_layer = nn.Linear(num_agents*act_dim, hidden_size)
# LSTM 时序建模
self.lstm = nn.LSTM(2*hidden_size, hidden_size)
# Q 值输出
self.q_out = nn.Linear(hidden_size, 1)
def forward(self, obs_list, actions):
# obs_list: [num_agents, batch_size, obs_dim]
# actions: [num_agents, batch_size, act_dim]
encoded_obs = [F.relu(enc(o)) for enc, o in zip(self.obs_encoder, obs_list)]
h_obs = torch.stack(encoded_obs, dim=0).mean(0) # 平均池化
# 处理联合动作
flat_actions = torch.cat([a for a in actions], dim=-1)
h_act = F.relu(self.joint_action_layer(flat_actions))
# LSTM 时序建模
lstm_in = torch.cat([h_obs, h_act], dim=-1).unsqueeze(0)
_, (h_n, _) = self.lstm(lstm_in)
return self.q_out(h_n.squeeze(0))
class PolicyNetwork(nn.Module):
"""actor 策略网络"""
def __init__(self, obs_dim, act_dim, hidden_size=64):
super().__init__()
self.fc1 = nn.Linear(obs_dim, hidden_size)
self.fc2 = nn.Linear(hidden_size, hidden_size)
self.act_out = nn.Linear(hidden_size, act_dim)
def forward(self, obs):
x = F.relu(self.fc1(obs))
x = F.relu(self.fc2(x))
return torch.tanh(self.act_out(x)) # 假设动作空间在[-1,1]
3.3 策略梯度更新
采用带 baseline 的策略梯度方法更新 actor 参数:
$$
\nabla_\theta J(\theta) = \mathbb{E}\left[\nabla_\theta \log \pi_\theta(a_i|o_i) \cdot (Q^\pi(s,a_1,…,a_N) – b(s))\right]
$$
其中 baseline $b(s)$ 通常取状态价值 $V^\pi(s)$ 以减少方差。
3.4 经验回放设计
多智能体场景需要存储完整的联合状态转移:
from collections import deque
import random
class MARLReplayBuffer:
def __init__(self, capacity):
self.buffer = deque(maxlen=capacity)
def push(self, transition):
"""transition 格式: (obs_list, actions, rewards, next_obs_list, done)"""
self.buffer.append(transition)
def sample(self, batch_size):
batch = random.sample(self.buffer, batch_size)
# 解压 batch 数据
obs_list, actions, rewards, next_obs_list, dones = zip(*batch)
return (torch.stack(obs_list, dim=1), # [batch, num_agents, obs_dim]
torch.stack(actions, dim=1), # [batch, num_agents, act_dim]
torch.tensor(rewards, dtype=torch.float32),
torch.stack(next_obs_list, dim=1),
torch.tensor(dones, dtype=torch.float32)
)
4. 训练优化实践
4.1 常见不稳定因素
- 探索不足:智能体过早收敛到局部最优策略
- 解决方案:采用参数空间噪声(parameter space noise)
- 信用分配 (Credit Assignment) 不当:难以区分单个智能体对团队奖励的贡献
- 解决方案:使用 counterfactual baseline
4.2 超参数调优经验
| 参数 | 推荐范围 | 影响说明 |
|---|---|---|
| 学习率(actor) | 1e-4 ~ 3e-4 | 策略更新需要更保守 |
| 学习率(critic) | 3e-4 ~ 1e-3 | 价值函数收敛更快 |
| 折扣因子 γ | 0.9 ~ 0.99 | 长期任务需要更高 γ |
| 回放缓冲大小 | 1e5 ~ 1e6 | 取决于任务复杂度 |
5. 生产环境建议
- 分布式训练框架:
- 使用 Ray 的 RLlib 实现并行采样
-
每个 worker 运行环境副本,定期同步模型参数
-
信用分配监控:
- 记录每个智能体的单独贡献度
- 可视化团队协作模式的变化过程
6. 延伸思考
在部分可观测环境中,critic 网络可能无法获取全局状态信息。如何改进网络架构以处理这种情况?可能的思路包括:
- 使用注意力机制动态聚焦关键信息
- 引入基于记忆的模块来维护历史上下文
- 采用图神经网络显式建模智能体间的关系
这些改进方向都值得在实际项目中验证其效果。
