MAPPO多智能体强化学习实战：共享Critic与独立Actor设计

简介：本资源是面向强化学习研究者与算法工程师的MAPPO（多智能体近端策略优化）完整实现示例，聚焦多智能体协作与竞争场景下的分布式训练实践，解决单智能体PPO难以扩展至多智能体系统的核心挑战。压缩包共2000个文件，含9个核心Python脚本（含训练主逻辑、环境封装与策略网络定义）、1984张训练过程可视化PNG图（如loss曲线、奖励收敛图、策略热力图等），以及readme.md、配置说明文档和挑战杯试题JSON等辅助材料，整体45.12MB，结构清晰便于分模块复现与调试。已有168人学习下载，适合具备PyTorch与OpenAI Gym基础的中高级学习者深入理解MAPPO的策略协同机制、价值函数设计、经验回放调度及多线程并行采样实现细节。

1. MAPPO 不是“多个 PPO 拼起来”：它用共享价值函数+独立策略+分布式采样，把 OpenAI Gym 多智能体环境跑通了，且能复现论文级协作/竞争行为

你试过在simple_spread或simple_adversary里堆 3 个独立 PPO agent 吗？大概率会崩——agent 互相干扰、梯度爆炸、训练曲线像心电图。MAPPO（Multi-Agent Proximal Policy Optimization）不是简单套壳，它用一个关键设计破局：所有 agent 共享 critic 网络（全局价值函数），但各自维护独立的 actor 网络（本地策略）。这个结构让 critic 能看到全局状态（state + all agents' obs），从而评估联合动作的价值；而每个 actor 只基于自己的观测做决策，保留策略的去中心化特性。项目包里这套实现，完整覆盖了从 Gym 环境封装、多线程 rollout、经验回放池构建、到 MAPPO 特有的 critic 共享训练逻辑——不是玩具 demo，而是能跑通MPE（Multi-Agent Particle Environment）全系列任务、支持 2~8 个 agent 并行采样的生产级代码。如果你正卡在多智能体训练收敛慢、reward 爆炸、或者 agent 互相“内卷”上，这份资源就是你该拆开的第一份真实工程包。

2. 环境准备与核心模块解耦：为什么必须重写 Gym 的 multi-agent wrapper？

MAPPO 对环境输入有硬性要求：它需要同时提供全局 state（用于 critic）和各 agent 的局部 observation（用于 actor），而标准 OpenAI Gym 的step()返回的是单 agent 的(obs, reward, done, info)。原生 Gym 不支持多智能体同步 step，更不提供全局 state。所以第一步不是 pip install，而是重构环境接口。

2.1 从 MPE 入手：加载并验证 multi-agent 环境的正确性

项目使用的是multiagent-particle-envs（MPE）库，这是 MAPPO 论文（2021）的基准环境。注意：不要用 pip install multiagent—— 官方 PyPI 包已废弃且不兼容新 Gym API。必须 clone 官方 GitHub 仓库并手动安装：

git clone https://github.com/shariqiqbal/multiagent-particle-envs.git cd multiagent-particle-envs pip install -e .

验证是否装对，运行以下脚本检查关键字段：

# test_mpe_env.py from multiagent.environment import MultiAgentEnv from multiagent.policy import InteractivePolicy import numpy as np # 加载 simple_spread 环境（4 个 agent，2 个 landmark） env = MultiAgentEnv( scenario_name="simple_spread", num_agents=4, num_landmarks=2, seed=42 ) print("Environment created successfully") print(f"Observation space for agent 0: {env.observation_space[0]}") print(f"Action space for agent 0: {env.action_space[0]}") print(f"Global state dim: {env.state().shape}") # 必须存在！输出应为 (16,) 或类似 print(f"Num agents: {env.n}") # 测试 step 返回结构 obs_n, reward_n, done_n, info_n = env.step([np.array([0,0,0,0]) for _ in range(env.n)]) print(f"obs_n length: {len(obs_n)}, each obs shape: {obs_n[0].shape}") print(f"reward_n: {reward_n}") print(f"done_n: {done_n}")

提示：env.state()是 MAPPO 的命脉。如果报错AttributeError: 'MultiAgentEnv' object has no attribute 'state'，说明你装的是旧版或 fork 版。必须确保multiagent/environment.py中MultiAgentEnv类定义了state(self)方法，返回np.concatenate([x for x in self._get_state()])。这是 critic 输入的唯一来源。

2.2 构建 MAPPO 专用 Wrapper：统一 state/obs 接口

原生 MPE 的step()返回obs_n（list of arrays），但 MAPPO 需要：

• obs_batch: shape(batch_size, n_agents, obs_dim)，供 actor 网络批量前向；
• state_batch: shape(batch_size, state_dim)，供 critic 网络输入；
• act_batch: shape(batch_size, n_agents, act_dim)，记录所有 agent 动作；
• rew_batch: shape(batch_size, n_agents)，每个 agent 的 reward；
• done_batch: shape(batch_size, n_agents)，done 标志（注意：MPE 中done_n是 list of bool，需转为 array）。

项目中的env_wrapper.py实现了这个转换。核心逻辑是重写step()和reset()：

# env_wrapper.py 关键片段 class MAPPOEnvWrapper: def __init__(self, env): self.env = env self.n_agents = env.n self.obs_dims = [env.observation_space[i].shape[0] for i in range(self.n_agents)] self.act_dims = [env.action_space[i].n for i in range(self.n_agents)] # 离散动作空间 def reset(self): obs_n = self.env.reset() # obs_n 是 list，转为 (n_agents, obs_dim) 数组 obs_array = np.stack(obs_n, axis=0) state = self.env.state() # 全局 state return obs_array, state def step(self, actions): # actions: (n_agents,) int array or (n_agents, act_dim) one-hot obs_n, rew_n, done_n, info_n = self.env.step(actions) obs_array = np.stack(obs_n, axis=0) state = self.env.state() rew_array = np.array(rew_n) done_array = np.array(done_n) return obs_array, state, rew_array, done_array, info_n

这个 wrapper 是后续所有数据 pipeline 的基石。没有它，rollout函数根本无法组织 batch 数据，critic 会因缺少state输入而报维度错误。

2.3 分布式 rollout 的线程安全设计：为什么不能用 multiprocessing.Pool？

MAPPO 要求高吞吐采样——单个进程 rollout 太慢，但 naive 的multiprocessing会导致：

• Gym 环境内部的随机种子冲突（多个进程同时调用np.random）；
• env.state()在 fork 后可能引用失效内存；
• 进程间无法共享 replay buffer，导致经验回放数据碎片化。

项目采用threading.Thread+queue.Queue实现线程安全 rollout：

# rollout_worker.py import threading import queue import time class RolloutWorker(threading.Thread): def __init__(self, env_wrapper, policy, rollout_queue, max_steps=25): super().__init__() self.env_wrapper = env_wrapper self.policy = policy # 共享的 actor 网络（只读） self.rollout_queue = rollout_queue self.max_steps = max_steps self.daemon = True # 主进程退出时自动结束 def run(self): while True: obs, state = self.env_wrapper.reset() episode_buffer = [] for t in range(self.max_steps): # actor 前向：输入 obs -> 输出 action logits -> sample with torch.no_grad(): logits = self.policy.actor(obs.unsqueeze(0)) # (1, n_agents, act_dim) actions = torch.distributions.Categorical(logits=logits).sample().squeeze(0) next_obs, next_state, rewards, dones, _ = self.env_wrapper.step(actions.numpy()) # 存储 transition：(obs, state, act, rew, next_obs, next_state, done) episode_buffer.append(( obs.numpy(), state.numpy(), actions.numpy(), rewards.numpy(), next_obs.numpy(), next_state.numpy(), dones.numpy() )) obs, state = next_obs, next_state if dones.any(): # 任一 agent done 即终止 episode break # 放入队列，由主进程统一收集 self.rollout_queue.put(episode_buffer) time.sleep(0.001) # 防止忙等

注意：self.policy.actor是主线程传入的模型，worker 线程只做前向推理（torch.no_grad()），不更新参数。所有 worker 共享同一份 actor 参数，保证采样策略一致性。rollout_queue是queue.Queue(maxsize=10)，避免内存爆炸。

3. MAPPO 核心网络架构与训练流程：共享 critic 如何避免梯度污染？

MAPPO 的灵魂在于 critic 的设计：它必须接收全局 state，输出每个 agent 的 Q 值（或 V 值），但反向传播时不能让 critic 梯度污染 actor 的独立策略。项目采用Centralized Critic结构，而非Decentralized Critic，这是论文标准做法。

3.1 Actor-Critic 网络定义：分离参数 + 共享 critic 输入

网络定义在networks.py中，关键点：

• Actor：每个 agent 一个独立网络，输入obs_i，输出logits_i（action distribution）；
• Critic：一个全局网络，输入state（concatenated global state），输出values（shape(n_agents,)），即每个 agent 的 V 值估计；
• Actor和Critic的 hidden layers 都是 MLP，但Critic 的输入层维度 =state_dim，Actor 的输入层维度 =obs_dim_i。

# networks.py import torch import torch.nn as nn class Actor(nn.Module): def __init__(self, obs_dim, act_dim, hidden_dim=64): super().__init__() self.net = nn.Sequential( nn.Linear(obs_dim, hidden_dim), nn.Tanh(), nn.Linear(hidden_dim, hidden_dim), nn.Tanh(), nn.Linear(hidden_dim, act_dim) ) def forward(self, obs): # obs: (batch, obs_dim) -> logits: (batch, act_dim) return self.net(obs) class Critic(nn.Module): def __init__(self, state_dim, n_agents, hidden_dim=128): super().__init__() self.net = nn.Sequential( nn.Linear(state_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, n_agents) # 输出每个 agent 的 V 值 ) def forward(self, state): # state: (batch, state_dim) -> values: (batch, n_agents) return self.net(state)

为什么 critic 输出(batch, n_agents)而不是(batch,)？因为 MAPPO 的 advantage 计算需要 per-agent 的 baseline。advantage_i = reward_i + gamma * next_value_i - value_i，所以 critic 必须为每个 agent 输出独立的 V 值。这和单 agent PPO 的 critic（输出 scalar）有本质区别。

3.2 MAPPO 的 PPO loss：如何构造 per-agent advantage？

标准 PPO 的advantage = GAE，但 MAPPO 的 GAE 必须基于critic 对每个 agent 的 V 值预测。项目中compute_gae函数严格按论文公式实现：

def compute_gae(rewards, values, dones, next_values, gamma=0.99, lam=0.95): """ rewards: (T, n_agents) values: (T, n_agents) # 当前 timestep 的 V(s_t) dones: (T, n_agents) # done mask next_values: (T, n_agents) # V(s_{t+1})，最后一行用 0 填充 Returns: advantages (T, n_agents), returns (T, n_agents) """ T, n = rewards.shape advantages = torch.zeros_like(rewards) lastgaelam = 0 # 逆序计算 GAE，对每个 agent 独立 for t in reversed(range(T)): # delta = r_t + gamma * V(s_{t+1}) * (1-done_{t+1}) - V(s_t) delta = rewards[t] + gamma * next_values[t] * (1 - dones[t]) - values[t] advantages[t] = delta + gamma * lam * (1 - dones[t]) * lastgaelam lastgaelam = advantages[t] returns = advantages + values return advantages, returns

这个函数是 MAPPO 正确性的核心。如果next_values维度不对（比如误用(T,)而非(T, n_agents)），advantage 会广播错误，导致某些 agent 的梯度爆炸。

3.3 Critic 更新：为什么必须 detach actor 的 log_prob？

MAPPO 的 critic loss 是 MSE：L_critic = mean((returns - values)^2)。但 actor loss 更复杂，涉及 ratio clipping：

# ppo_trainer.py def ppo_update(self, data): obs_batch = data['obs'] # (B, n, obs_dim) act_batch = data['acts'] # (B, n) old_logp_batch = data['logp'] # (B, n) adv_batch = data['adv'] # (B, n) ret_batch = data['ret'] # (B, n) # Critic update values = self.critic(data['state']) # (B, n) critic_loss = ((values - ret_batch) ** 2).mean() self.critic_opt.zero_grad() critic_loss.backward() torch.nn.utils.clip_grad_norm_(self.critic.parameters(), 0.5) self.critic_opt.step() # Actor update # 重新计算当前策略的 logp（避免重复计算） logits = self.actor(obs_batch.view(-1, obs_batch.size(-1))) # (B*n, act_dim) dist = torch.distributions.Categorical(logits=logits) new_logp = dist.log_prob(act_batch.view(-1)).view_as(old_logp_batch) # (B, n) # Ratio = exp(new_logp - old_logp) ratio = torch.exp(new_logp - old_logp_batch.detach()) # 关键：old_logp 必须 detach！ surr1 = ratio * adv_batch surr2 = torch.clamp(ratio, 1.0 - 0.2, 1.0 + 0.2) * adv_batch actor_loss = -torch.min(surr1, surr2).mean() self.actor_opt.zero_grad() actor_loss.backward() torch.nn.utils.clip_grad_norm_(self.actor.parameters(), 0.5) self.actor_opt.step()

注意old_logp_batch.detach()：这是防止 critic 更新时意外影响 actor 参数。old_logp来自 rollout 时的旧策略，其计算图必须切断，否则ratio的梯度会反向流到 actor，破坏 PPO 的稳定性。这是 MAPPO 训练中最容易被忽略的细节之一。

4. 经验回放与多线程协同：如何避免 replay buffer 成为性能瓶颈？

MAPPO 不像 DQN 那样强制用 replay buffer，但项目实现了on-policy + episodic replay，目的是：

• 缓冲多线程 rollout 的数据洪峰；
• 支持 multiple epochs per batch（PPO 标准做法）；
• 避免单次 rollout 数据量不足导致训练震荡。

4.1 Episodic Replay Buffer 设计：按 episode 切片，非 flat storage

项目不用deque或np.array存 raw transition，而是存Episode对象：

# replay_buffer.py class Episode: def __init__(self, obs, state, acts, rews, dones, next_obs, next_state): self.obs = obs # (T, n, obs_dim) self.state = state # (T, state_dim) self.acts = acts # (T, n) self.rews = rews # (T, n) self.dones = dones # (T, n) self.next_obs = next_obs # (T, n, obs_dim) self.next_state = next_state # (T, state_dim) class ReplayBuffer: def __init__(self, max_size=1000): self.buffer = [] self.max_size = max_size def add(self, episode): if len(self.buffer) >= self.max_size: self.buffer.pop(0) self.buffer.append(episode) def sample(self, batch_size=32): # 随机选 batch_size 个 episode，然后从每个 episode 中随机截取一段 episodes = random.sample(self.buffer, min(batch_size, len(self.buffer))) batch = {'obs': [], 'state': [], 'acts': [], 'rews': [], 'dones': [], 'next_obs': [], 'next_state': []} for ep in episodes: T = ep.obs.shape[0] if T < 10: continue # 跳过太短的 episode start = random.randint(0, T-10) end = start + 10 batch['obs'].append(ep.obs[start:end]) batch['state'].append(ep.state[start:end]) batch['acts'].append(ep.acts[start:end]) batch['rews'].append(ep.rews[start:end]) batch['dones'].append(ep.dones[start:end]) batch['next_obs'].append(ep.next_obs[start:end]) batch['next_state'].append(ep.next_state[start:end]) # stack to (B*T, ...) for k in batch: batch[k] = torch.cat(batch[k], dim=0) return batch

为什么按 episode 存？因为 MAPPO 的 GAE 计算依赖时间连续性。如果 flat 存 transition，GAE 的next_values[t]会跨 episode 错位，导致优势估计崩溃。sample()中的start/end截取保证了每个 batch 内 transition 的时间局部性。

4.2 多线程写入 buffer 的锁机制：Queue vs Lock 的取舍

Rollout worker 将 episode 放入queue.Queue，主训练线程消费。但ReplayBuffer.add()是线程安全的吗？项目没用threading.Lock，而是依赖queue.Queue的原子性：

# main_train.py rollout_queue = queue.Queue(maxsize=50) # 启动 4 个 worker for _ in range(4): worker = RolloutWorker(env_wrapper, policy, rollout_queue) worker.start() # 主线程循环 replay_buffer = ReplayBuffer(max_size=200) while not training_done: try: episode = rollout_queue.get(timeout=1) # block until data replay_buffer.add(episode) # 这里是主线程调用，无并发 rollout_queue.task_done() except queue.Empty: continue if len(replay_buffer.buffer) > 50: batch = replay_buffer.sample(batch_size=256) trainer.update(batch)

关键点：rollout_queue.get()由主线程独占调用，replay_buffer.add()在主线程执行，因此无需额外锁。queue.Queue的put()和get()本身是线程安全的，这是 Python 标准库的保证。强行加Lock反而降低吞吐。

4.3 Batch 组织的陷阱：obs/state/act 的维度对齐

最常翻车的 bug 是obs和state的 batch 维度不一致。例如：

• obs:(B, n, obs_dim)→ actor 输入；
• state:(B, state_dim)→ critic 输入；
• acts:(B, n)→ 用于 logp 计算。

如果sample()返回的state是(B, n, state_dim)（错误地按 agent 维度 broadcast），critic 会报matmul维度不匹配。项目中ReplayBuffer.sample()显式保证：

# 在 sample() 中 batch['state'] = torch.cat([ep.state[start:end] for ep in episodes], dim=0) # (B*T, state_dim) # 而不是 # batch['state'] = torch.cat([ep.state[start:end] for ep in episodes], dim=1) # 错！

血泪经验：每次修改sample()逻辑后，必须用print(batch['obs'].shape, batch['state'].shape, batch['acts'].shape)验证三者 batch size 一致（即B*T）。不一致的维度是 silent bug，训练 loss 看似下降，但 agent 行为完全随机。

5. 避坑指南：MAPPO 训练中五个让你重启实验的致命问题

MAPPO 表面是 PPO 的多智能体扩展，实则处处是暗礁。以下问题均来自真实复现过程，每一条都对应一次 12 小时以上的 debug。

5.1 现象：训练初期 reward 爆炸（+1e5 或 -1e5），随后归零

原因：Critic初始化权重过大，导致values输出极值，GAE 计算中delta = r + gamma*next_v - v产生巨大误差，advantage 失控。
解决：在Critic的最后一层 Linear 后加nn.init.orthogonal_，并 scale 输出：

# networks.py class Critic(nn.Module): def __init__(self, state_dim, n_agents, hidden_dim=128): super().__init__() self.net = nn.Sequential( nn.Linear(state_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, n_agents) ) # 关键修复：初始化缩放 nn.init.orthogonal_(self.net[-1].weight, gain=0.01) # 默认 gain=1.0 太大 nn.init.constant_(self.net[-1].bias, 0)

5.2 现象：所有 agent 的 policy 完全相同（输出 action 总是一样）

原因：Actor网络的输入obs_i维度错误，实际输入了state或其他 agent 的 obs，导致所有 actor 看到相同输入。
解决：在rollout中打印obs_n[0].shape,obs_n[1].shape，确认每个 agent 的 obs 维度独立（如simple_spread中 agent 0 obs 是 18-dim，agent 1 是 18-dim，但内容不同）。检查env_wrapper.reset()是否正确stack(obs_n, axis=0)，而非concatenate。

5.3 现象：ValueError: Expected input batch_size (128) to match target batch_size (32)

原因：batch_size在sample()和update()中不一致。例如sample(batch_size=256)但update()中obs被 reshape 为(32, 8, obs_dim)，而state是(256, state_dim)，维度错位。
解决：统一用B = obs.shape[0]动态推导 batch size，禁止硬编码：

def update(self, batch): B = batch['obs'].shape[0] # 动态获取 n = batch['obs'].shape[1] # agent 数 # ... 后续所有 tensor 操作基于 B, n

5.4 现象：训练 loss 下降，但环境 reward 不升反降，agent 互相碰撞

原因：done信号处理错误。MPE 中done_n是 list of bool，但dones被转为(T, n)array 后，若未正确 mask，GAE 会将 terminal state 的next_v计入（实际应为 0）。
解决：在compute_gae中，next_values[t]必须乘(1 - dones[t])，且next_values的最后一行必须设为 0（而非用 rollout 最后一步的v）：

# 在 compute_gae 前 next_values = torch.cat([values[1:], torch.zeros(1, n)], dim=0) # 手动 pad zero # 而不是 # next_values = values[1:] # 错！少一行

5.5 现象：多线程 rollout 吞吐极低（< 100 steps/sec），CPU 占用 100%

原因：env.step()内部有time.sleep()或pygame渲染阻塞（MPE 默认开启 render）。
解决：在env_wrapper.__init__()中强制关闭渲染：

def __init__(self, env): self.env = env # 关键：禁用 render if hasattr(self.env, 'render_mode'): self.env.render_mode = None elif hasattr(self.env, 'render'): # monkey patch self.env.render = lambda: None

6. 验证与调试技巧：用 reward decomposition 和 gradient norm 监控 MAPPO 健康度

MAPPO 的训练不像单 agent 那样看一条 reward 曲线就完事。你需要至少三个视角交叉验证：per-agent reward 分解、critic gradient norm、actor entropy decay。这是我从第 7 次失败中学到的铁律。

6.1 Per-agent reward 分解：识别协作失效的早期信号

在simple_spread中，理想行为是 agent 均匀分布包围 landmarks。如果 reward 曲线整体上升但某个 agent 的 reward 持续为负，说明它被“牺牲”了——这不是协作，是内卷。项目在logger.py中添加了 agent-level reward tracking：

# logger.py def log_episode_metrics(self, rewards, infos): # rewards: (n_agents,) for i, r in enumerate(rewards): self.writer.add_scalar(f'reward/agent_{i}', r, self.global_step) # 计算 spread score：agent 到最近 landmark 的距离 std if 'n_collisions' in infos: self.writer.add_scalar('metric/collisions', infos['n_collisions'], self.global_step) # 关键指标：reward variance reward_var = np.var(rewards) self.writer.add_scalar('metric/reward_variance', reward_var, self.global_step) # variance < 0.1 表示 reward 分配均衡；> 1.0 表示 agent 分化严重

从那以后我每次启动训练，第一件事就是打开 TensorBoard 看reward/agent_0到reward/agent_3四条线是否同步上升。如果某条线长期低于均值 50%，立刻暂停，检查它的 obs 是否被遮挡（MPE 中 agent 视野有限），或 critic 是否给它分配了过低的 baseline。

6.2 Critic gradient norm：诊断 value collapse 的后悔药

Critic 的 gradient norm 应该稳定在0.1 ~ 1.0。如果突然降到1e-4以下，说明 critic 已经“躺平”，输出恒定值，advantage 全为 0，actor 停止学习。项目在trainer.update()中插入监控：

def update(self, batch): # ... critic update ... critic_loss.backward() grad_norm = torch.nn.utils.clip_grad_norm_(self.critic.parameters(), 0.5) self.writer.add_scalar('grad_norm/critic', grad_norm, self.global_step) # ... actor update ... actor_loss.backward() grad_norm = torch.nn.utils.clip_grad_norm_(self.actor.parameters(), 0.5) self.writer.add_scalar('grad_norm/actor', grad_norm, self.global_step)

我一般会设置一个 early stop 条件：如果critic grad_norm < 0.01持续 10 个 epoch，自动 reload 上一个 checkpoint 并降低 critic learning rate（lr *= 0.5）。这比等 24 小时后发现 reward 归零强得多。

6.3 Actor entropy：判断 exploration 是否死亡

MAPPO 的 actor entropy 应该缓慢下降（从 ~1.5 到 ~0.3），如果第 100 epoch 就降到 0.05，说明策略过早收敛到次优解。项目在ppo_update中计算：

# 在 actor update 后 with torch.no_grad(): logits = self.actor(obs_batch.view(-1, obs_batch.size(-1))) dist = torch.distributions.Categorical(logits=logits) entropy = dist.entropy().mean() self.writer.add_scalar('metric/entropy', entropy, self.global_step)

表格：entropy 健康区间参考（simple_spread, 4 agents）

从那以后我每次保存 checkpoint，都会把entropy、reward_variance、critic_grad_norm三个值写入metadata.json。下次复现时，直接grep "entropy.*0.02"就能定位到 collapse 的 checkpoint，省下半天时间。

希望帮到你。

本文还有配套的精品资源，点击获取