MADDPG实战：用Python从零搭建多智能体协同与竞争环境（附避坑指南）

引言：多智能体强化学习的独特挑战

在单智能体强化学习中，我们通常假设环境是静态的——即状态转移概率和奖励函数不会随时间改变。但当多个智能体同时学习时，这个假设就被打破了。想象一下足球训练场景：如果你每天都面对相同的防守策略，学习射门会很简单；但如果防守队员也在不断改进策略，你的学习过程就会变得异常困难。这就是多智能体系统中著名的"非平稳性"问题。

MADDPG（Multi-Agent Deep Deterministic Policy Gradient）通过创新的"集中训练-分散执行"框架解决了这一核心挑战。其核心洞见在于：如果知道所有智能体的策略，环境就重新变得平稳。这就像在足球训练中，虽然对手的策略会变，但如果你能准确预测他们的防守动作，就能制定出稳定的进攻策略。

1. 环境搭建：多智能体粒子世界

1.1 安装依赖环境

首先需要配置Python 3.7+环境和必要依赖：

# 创建conda环境（推荐）
conda create -n maddpg python=3.8
conda activate maddpg

# 安装核心依赖
pip install torch==1.12.1+cu113 torchvision==0.13.1+cu113 --extra-index-url https://download.pytorch.org/whl/cu113
pip install gym==0.21.0 pettingzoo==1.22.0 

1.2 构建自定义环境

OpenAI的multiagent-particle-envs提供了基础环境模板。我们扩展其功能：

from gym.utils import seeding
from multiagent.environment import MultiAgentEnv
from multiagent.scenarios import BaseScenario

class CustomScenario(BaseScenario):
    def make_world(self):
        world = World()
        # 添加3个合作型智能体和2个对抗型智能体
        world.agents = [Agent() for i in range(5)]
        world.landmarks = [Landmark() for i in range(3)]
        # 设置物理引擎参数
        world.dim_c = 2  # 通信频道维度
        world.collaborative = False  # 混合合作-竞争模式
        return world

    def reset_world(self, world):
        # 重置智能体位置和目标点
        for i, agent in enumerate(world.agents):
            agent.color = np.array([0.35, 0.35, 0.85] if i < 3 else [0.85, 0.35, 0.35])
            agent.state.p_pos = np.random.uniform(-1, +1, world.dim_p)
            agent.state.p_vel = np.zeros(world.dim_p)

关键参数调优经验：

• dim_c（通信维度）建议设置在2-4之间，过高会导致训练不稳定
• 碰撞惩罚系数建议从0.1开始逐步调整
• 智能体感知半径设置为环境对角线长度的1/3

2. 网络架构设计：Actor-Critic的进阶实现

2.1 集中式Critic网络

Critic网络需要接收所有智能体的联合观测和动作：

import torch.nn as nn

class CentralizedCritic(nn.Module):
    def __init__(self, obs_dim, act_dim, num_agents, hidden_size=128):
        super().__init__()
        input_dim = (obs_dim + act_dim) * num_agents
        self.net = nn.Sequential(
            nn.Linear(input_dim, hidden_size),
            nn.ReLU(),
            nn.Linear(hidden_size, hidden_size),
            nn.ReLU(),
            nn.Linear(hidden_size, 1)
        )
    
    def forward(self, obs_list, act_list):
        # 拼接所有智能体的观测和动作
        joint_input = torch.cat(obs_list + act_list, dim=-1)
        return self.net(joint_input)

2.2 分散式Actor网络

每个智能体的Actor只处理自己的观测：

class DecentralizedActor(nn.Module):
    def __init__(self, obs_dim, act_dim, hidden_size=128):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(obs_dim, hidden_size),
            nn.ReLU(),
            nn.Linear(hidden_size, hidden_size),
            nn.ReLU(),
            nn.Linear(hidden_size, act_dim),
            nn.Tanh()  # 输出在[-1,1]范围内
        )
    
    def forward(self, obs):
        return self.net(obs)

网络设计技巧：

• 最后一层使用Tanh激活限制动作范围
• 中间层建议使用ReLU而非LeakyReLU（实验表明训练更稳定）
• 对Critic使用比Actor更大的网络（如256 vs 128单元）

3. 经验回放缓冲区的特殊处理

多智能体场景需要存储完整的联合状态转移：

class MultiAgentReplayBuffer:
    def __init__(self, capacity, obs_dims, act_dims, num_agents):
        self.capacity = capacity
        self.obs_buffs = [np.zeros((capacity, dim)) for dim in obs_dims]
        self.act_buffs = [np.zeros((capacity, dim)) for dim in act_dims]
        self.rew_buffs = [np.zeros(capacity) for _ in range(num_agents)]
        self.next_obs_buffs = [np.zeros((capacity, dim)) for dim in obs_dims]
        self.done_buffs = np.zeros(capacity)
        self.idx = 0
        self.size = 0

    def store(self, obs_list, act_list, rew_list, next_obs_list, done):
        for i in range(len(obs_list)):
            self.obs_buffs[i][self.idx] = obs_list[i]
            self.act_buffs[i][self.idx] = act_list[i]
            self.next_obs_buffs[i][self.idx] = next_obs_list[i]
        for i in range(len(rew_list)):
            self.rew_buffs[i][self.idx] = rew_list[i]
        self.done_buffs[self.idx] = done
        self.idx = (self.idx + 1) % self.capacity
        self.size = min(self.size + 1, self.capacity)

缓冲区使用要点：

• 批大小建议设置在1024-4096之间
• 当环境有部分可观测性时，建议存储最近3-5步的历史观测
• 优先经验回放（PER）在多智能体场景效果有限，不建议使用

4. 训练流程与策略振荡解决方案

4.1 核心训练循环

def train(maddpg, env, episodes=5000):
    for ep in range(episodes):
        obs_list = env.reset()
        episode_rewards = [0] * maddpg.num_agents
        
        for t in range(1000):  # 最大步数
            act_list = maddpg.act(obs_list, noise_scale=0.1)  # 添加探索噪声
            next_obs_list, rew_list, done_list, _ = env.step(act_list)
            maddpg.store(obs_list, act_list, rew_list, next_obs_list, done_list[0])
            
            if maddpg.replay_buffer.size >= BATCH_SIZE:
                maddpg.update()
            
            obs_list = next_obs_list
            for i in range(maddpg.num_agents):
                episode_rewards[i] += rew_list[i]
            
            if done_list[0]: break

4.2 解决策略振荡的三大技巧

技巧1：策略集合（Policy Ensembles）

class MADDPG:
    def __init__(self, num_agents, obs_dims, act_dims, num_policies=3):
        self.actors = [[DecentralizedActor(obs_dims[i], act_dims[i]) 
                       for _ in range(num_policies)] for i in range(num_agents)]
        # 每个episode随机选择一个策略子集
        self.current_policy_idx = [np.random.randint(num_policies) 
                                  for _ in range(num_agents)]

技巧2：延迟策略更新

if total_steps % POLICY_UPDATE_FREQ == 0:  # 通常设为critic更新频率的2-5倍
    for agent in agents:
        agent.update_actor()

技巧3：对手建模

class OpponentModel(nn.Module):
    """学习其他智能体策略的近似模型"""
    def __init__(self, obs_dim, act_dim):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(obs_dim, 64),
            nn.ReLU(),
            nn.Linear(64, act_dim)
        )
    
    def forward(self, obs):
        return torch.softmax(self.net(obs), dim=-1)

5. 实战案例：物理欺骗与合作导航

5.1 物理欺骗（Physical Deception）

配置3个合作智能体和1个对抗智能体：

scenario = {
    'n_good': 3,  # 合作智能体
    'n_adversaries': 1,  # 对抗智能体
    'n_landmarks': 3,  # 地标数量
    'reward_scale': 1.0,
    'collision_penalty': 0.5  # 碰撞惩罚系数
}

训练曲线分析：

• 前1000episode：智能体随机移动
• 1000-3000episode：开始形成基本协作
• 3000episode后：稳定欺骗策略形成

5.2 合作导航（Cooperative Navigation）

关键指标监控表：

指标	初期值	中期值	收敛值
平均覆盖率	0.25	0.63	0.92
碰撞次数/episode	8.7	3.2	0.5
到达时间(步)	150+	75	45

参数敏感度测试：

• 学习率：Critic建议1e-3，Actor建议5e-4
• 折扣因子γ：0.95-0.99表现稳定
• 目标网络更新率τ：0.01效果最佳

6. 高级调优与Debug技巧

6.1 梯度裁剪的特别处理

# 对Critic使用更严格的梯度裁剪
torch.nn.utils.clip_grad_norm_(critic.parameters(), 0.5)
# 对Actor使用稍大的裁剪阈值
torch.nn.utils.clip_grad_norm_(actor.parameters(), 1.0)

6.2 探索噪声的退火策略

def get_noise_scale(episode):
    initial_noise = 0.2
    min_noise = 0.01
    decay_rate = 0.995
    return max(min_noise, initial_noise * (decay_rate ** episode))

6.3 常见问题排查表

现象	可能原因	解决方案
奖励不增长	探索不足	增大初始噪声
策略振荡	学习率过高	降低Actor学习率
Critic损失爆炸	梯度裁剪不足	减小裁剪阈值
收敛速度慢	批大小太小	增大到2048+

在实际项目中，我发现最影响性能的三个因素是：(1)Critic网络容量不足会导致Q值低估，(2)过小的回放缓冲区会使训练不稳定，(3)不恰当的探索噪声衰减会陷入局部最优。经过多次实验，最终采用的噪声退火方案是在前2000episode线性衰减，之后保持最小噪声。