简介:MADDPG(多智能体深度确定性策略梯度)算法资源包,面向强化学习方向的学习者与研究者,用于理解并复现混合协作—竞争环境下的多智能体决策方法。该算法由2017年论文《Multi-Agent Actor-Critic for Mixed Cooperative-Competitive Environments》提出,在DDPG基础上扩展出集中训练、分散执行的Actor-Critic框架,是入门多智能体强化学习(MADRL)的经典范例。压缩包共21个文件,约24KB,以6个Python源码文件为核心,涵盖MADDPG与MATD3主程序、网络结构、经验回放缓冲区及MPE环境封装;另有5个XML与5个pyc等工程配置和字节码文件,以及若干txt说明文档,便于直接运行与二次修改。目前已有1312人学习下载。读者可借此掌握多智能体策略梯度算法的代码组织方式、训练流程与调试思路,并对照MATD3实现理解算法改进方向,适合具备一定Python与强化学习基础的中高级学习者。
1. 从一场「各自为战」的仿真翻车说起
如果你正在做多机器人协同、无人机编队、交通信号联控或者对抗博弈这类课题,大概率绕不开一个名字:MADRL多智能体深度确定性策略梯度(MADDPG)算法。我第一次把它塞进一个多车协同避障的仿真里时,满心以为把单智能体的 DDPG 复制 N 份就能跑通,结果训练曲线像心电图一样乱跳,奖励不升反降,智能体学会了「集体摆烂」——谁都不动,因为动就撞。这个坑让我意识到,多智能体不是简单的「单机乘 N」,环境非平稳性会把每个智能体的学习目标搅成一锅粥。
MADDPG 解决的正是这个问题:它用「集中训练、分散执行」的思路,让每个智能体在训练时能看到全局信息(包括其他智能体的动作),执行时又只依赖自己的局部观测。这套算法适合谁?做多智能体强化学习入门、需要一套能复现的基线、又不想一上来就啃复杂通信架构的从业者。下面我按「原理立住 → 环境搭起来 → 代码跑通 → 参数调明白 → 坑绕过去」的顺序,把这条路径拆开讲。
2. MADDPG 到底改了什么:从 DDPG 到集中训练分散执行
2.1 单智能体 DDPG 的三个零件
要理解 MADDPG,得先把 DDPG 拆开看。DDPG 是确定性策略梯度算法,核心是四个网络:Actor 当前网络、Actor 目标网络、Critic 当前网络、Critic 目标网络。Actor 负责根据状态输出一个确定的动作,Critic 负责评估这个动作值不值。目标网络用软更新(soft update)缓慢跟随当前网络,避免训练震荡。
它的动作空间是连续的,所以输出的是一个实数向量,而不是概率分布。经验回放池存的是(state, action, reward, next_state, done)五元组。训练时从池子里采样一批,用 Critic 的 TD 误差更新 Critic,用 Critic 对动作的梯度更新 Actor。这套结构在单智能体场景下很稳,但一旦环境里多了几个也在学习的家伙,每个智能体眼里的「环境转移概率」就变了——因为别人的策略在变。
2.2 非平稳性:多智能体最要命的敌人
假设两个智能体在同一个环境里,智能体 A 的策略更新了,那么对智能体 B 来说,同样的状态和动作,下一时刻的奖励分布变了。B 的 Critic 在拟合一个不断漂移的目标,这就是非平稳性。Q-learning 类方法的收敛性假设环境是平稳的,多智能体直接打破了这个假设。
常见做法有两种:一种是完全独立学习(Independent DDPG),每个智能体把别人当环境的一部分,简单但容易不收敛;另一种就是 MADDPG,把全局信息喂给 Critic,让每个 Critic 评估的是「在全局状态下,我采取这个动作,其他人采取他们的动作,能拿多少分」。这样每个 Critic 的目标相对稳定,因为全局状态和联合动作包含了所有智能体的信息。
2.3 集中训练分散执行的具体形式
MADDPG 的「集中」体现在 Critic 的输入上:第 i 个智能体的 Critic 输入是全局状态s(所有智能体观测的拼接)和联合动作a = (a_1, ..., a_N)。输出是一个标量 Q 值。而 Actor 只输入自己的局部观测o_i,输出自己的动作a_i。执行阶段,每个 Actor 只用自己的观测就能决策,不需要通信。
损失函数上,Critic 的更新目标为:
y = r_i + gamma * Q_i'(s', a_1', ..., a_N') | a_j' = mu_j'(o_j')其中mu_j'是第 j 个智能体的目标 Actor。Actor 的更新则是最大化 Critic 对自身动作的评分:
grad = E[ grad_a Q_i(s, a_1,...,a_N) * grad_theta mu_i(o_i) ]这里有个关键细节:更新 Actor 时,其他智能体的动作是固定的,只对当前智能体的动作求梯度。这就是「集中评判、各自改进」的数学表达。
2.4 为什么它适合做基线
MADDPG 的代码结构清晰,扩展性强。你可以在 Critic 里加通信模块、加注意力机制、加图神经网络,都是在这个框架上做加法。对于刚入门的从业者,先跑通原始版本,再改结构,比一上来就搞复杂架构要踏实得多。而且它的论文实验覆盖了合作、竞争、混合场景,参考价值高。
3. 把仿真环境搭起来:从零写一个多智能体粒子世界
3.1 环境设计的最小闭环
为了复现 MADDPG,我一般会先写一个极简的多智能体环境,不依赖任何外部仿真器。这个环境包含 N 个智能体,每个智能体有位置和速度,观测是自己的位置、速度以及目标点位置。动作是二维加速度。奖励设计为:离目标越近奖励越高,碰撞则惩罚。
这个环境虽然简单,但包含了多智能体学习的核心要素:局部观测、连续动作、共享奖励或独立奖励、智能体间交互。下面是一个用 Python 写的环境骨架。
import numpy as np class SimpleMultiAgentEnv: def __init__(self, num_agents=3, world_size=1.0): self.num_agents = num_agents self.world_size = world_size # 每个智能体观测维度:自身位置(2) + 自身速度(2) + 目标位置(2) self.obs_dim = 6 # 动作维度:二维加速度 self.act_dim = 2 self.reset() def reset(self): # 随机初始化位置和速度 self.positions = np.random.uniform(-0.5, 0.5, (self.num_agents, 2)) self.velocities = np.zeros((self.num_agents, 2)) # 每个智能体有一个随机目标点 self.goals = np.random.uniform(-0.5, 0.5, (self.num_agents, 2)) return self._get_obs() def _get_obs(self): obs = [] for i in range(self.num_agents): obs_i = np.concatenate([ self.positions[i], self.velocities[i], self.goals[i] ]) obs.append(obs_i) return np.array(obs) def step(self, actions): # actions: (num_agents, 2) actions = np.clip(actions, -1.0, 1.0) self.velocities += actions * 0.1 self.velocities *= 0.95 # 阻尼 self.positions += self.velocities * 0.1 # 边界反弹 self.positions = np.clip(self.positions, -self.world_size/2, self.world_size/2) rewards = [] for i in range(self.num_agents): dist = np.linalg.norm(self.positions[i] - self.goals[i]) reward = -dist # 越近奖励越大 # 碰撞惩罚 for j in range(self.num_agents): if i != j: if np.linalg.norm(self.positions[i] - self.positions[j]) < 0.05: reward -= 1.0 rewards.append(reward) done = False # 所有智能体都接近目标则结束 if all(np.linalg.norm(self.positions[i] - self.goals[i]) < 0.05 for i in range(self.num_agents)): done = True return self._get_obs(), np.array(rewards), done, {}这段代码里,obs_dim=6是每个智能体的局部观测维度,act_dim=2是动作维度。step函数里先更新速度再更新位置,加阻尼是为了让运动更平滑。碰撞惩罚用的是距离阈值 0.05,这个值可以根据世界大小调整。奖励是负距离,所以智能体会被激励靠近目标。
3.2 观测归一化与奖励缩放
上面这个环境有个问题:位置范围是 [-0.5, 0.5],速度可能到几,奖励是负距离,量级在 0 到 1 之间。如果不做归一化,神经网络训练会很不稳定。我一般会把观测归一化到 [-1, 1] 或 [0, 1],奖励乘以一个缩放系数。
def normalize_obs(obs, world_size=1.0): # 位置和速度分别归一化 obs_norm = obs.copy() obs_norm[:, 0:2] /= (world_size / 2) # 位置 obs_norm[:, 2:4] /= 2.0 # 速度假设最大为2 obs_norm[:, 4:6] /= (world_size / 2) # 目标 return np.clip(obs_norm, -1.0, 1.0)奖励缩放可以在环境里做,也可以在训练循环里做。我习惯在环境里把奖励乘以 0.1,这样 TD 误差不会太大。注意,奖励缩放会改变最优策略的值函数尺度,但不会改变策略本身,所以是安全的。
3.3 经验回放池的存储结构
MADDPG 的经验回放池和单智能体不同,它需要存储全局状态和联合动作。全局状态是所有智能体观测的拼接,联合动作是所有智能体动作的拼接。每条经验包含:全局状态、每个智能体的局部观测、联合动作、每个智能体的奖励、下一个全局状态、下一个局部观测、done 标志。
class ReplayBuffer: def __init__(self, capacity, num_agents, obs_dim, act_dim): self.capacity = capacity self.num_agents = num_agents self.obs_dim = obs_dim self.act_dim = act_dim self.ptr = 0 self.size = 0 # 全局状态维度 = num_agents * obs_dim self.global_obs_dim = num_agents * obs_dim self.global_act_dim = num_agents * act_dim self.buffer = { 'global_obs': np.zeros((capacity, self.global_obs_dim)), 'obs': np.zeros((capacity, num_agents, obs_dim)), 'actions': np.zeros((capacity, num_agents, act_dim)), 'rewards': np.zeros((capacity, num_agents)), 'next_global_obs': np.zeros((capacity, self.global_obs_dim)), 'next_obs': np.zeros((capacity, num_agents, obs_dim)), 'dones': np.zeros((capacity, 1)) } def push(self, global_obs, obs, actions, rewards, next_global_obs, next_obs, done): idx = self.ptr self.buffer['global_obs'][idx] = global_obs self.buffer['obs'][idx] = obs self.buffer['actions'][idx] = actions self.buffer['rewards'][idx] = rewards self.buffer['next_global_obs'][idx] = next_global_obs self.buffer['next_obs'][idx] = next_obs self.buffer['dones'][idx] = done self.ptr = (self.ptr + 1) % self.capacity self.size = min(self.size + 1, self.capacity) def sample(self, batch_size): idxs = np.random.randint(0, self.size, size=batch_size) batch = {k: v[idxs] for k, v in self.buffer.items()} return batch这里global_obs是num_agents * obs_dim维的向量,actions是(num_agents, act_dim)的数组。采样时返回的 batch 里,每个键的第一维都是 batch_size。这种存储方式比把全局状态和局部观测分开存更省内存,因为局部观测可以从全局状态里切片得到,但为了代码清晰,我还是分开存了。
4. 用 PyTorch 实现 MADDPG:Actor、Critic 与训练循环
4.1 Actor 网络:只吃局部观测
Actor 网络的结构很简单:输入局部观测,经过两层全连接,输出动作。输出层用 tanh 激活,把动作限制在 [-1, 1]。这是因为大多数连续控制环境的动作范围是 [-1, 1],如果你的环境动作范围不同,可以乘一个缩放系数。
import torch import torch.nn as nn import torch.nn.functional as F class Actor(nn.Module): def __init__(self, obs_dim, act_dim, hidden_dim=64): super().__init__() self.fc1 = nn.Linear(obs_dim, hidden_dim) self.fc2 = nn.Linear(hidden_dim, hidden_dim) self.fc3 = nn.Linear(hidden_dim, act_dim) def forward(self, obs): x = F.relu(self.fc1(obs)) x = F.relu(self.fc2(x)) return torch.tanh(self.fc3(x))hidden_dim=64对于简单环境足够了。如果环境复杂,可以加到 128 或 256。注意,Actor 的输入维度是obs_dim,不是全局状态维度。这是 MADDPG 分散执行的关键:每个 Actor 只依赖自己的观测。
4.2 Critic 网络:吃全局状态和联合动作
Critic 网络输入全局状态和联合动作,输出一个 Q 值。全局状态维度是num_agents * obs_dim,联合动作维度是num_agents * act_dim。中间可以拼接后过几层全连接。
class Critic(nn.Module): def __init__(self, global_obs_dim, global_act_dim, hidden_dim=64): super().__init__() self.fc1 = nn.Linear(global_obs_dim + global_act_dim, hidden_dim) self.fc2 = nn.Linear(hidden_dim, hidden_dim) self.fc3 = nn.Linear(hidden_dim, 1) def forward(self, global_obs, global_act): x = torch.cat([global_obs, global_act], dim=-1) x = F.relu(self.fc1(x)) x = F.relu(self.fc2(x)) return self.fc3(x)这里global_act是所有智能体动作的拼接,维度是num_agents * act_dim。在训练时,对于第 i 个智能体的 Critic,输入的是所有智能体的动作,但更新 Actor 时只对第 i 个智能体的动作求梯度。
4.3 目标网络软更新与梯度裁剪
MADDPG 沿用 DDPG 的目标网络软更新方式:
def soft_update(target, source, tau=0.01): for target_param, param in zip(target.parameters(), source.parameters()): target_param.data.copy_(target_param.data * (1.0 - tau) + param.data * tau)tau=0.01是常见值,表示目标网络每次只跟随当前网络 1%。如果训练震荡,可以减小 tau;如果学习太慢,可以增大 tau。梯度裁剪也很重要,尤其是 Critic 的梯度,我一般会裁剪到 [-1, 1] 或 [-0.5, 0.5]。
torch.nn.utils.clip_grad_norm_(critic.parameters(), max_norm=1.0) torch.nn.utils.clip_grad_norm_(actor.parameters(), max_norm=1.0)4.4 训练循环的每一步
训练循环里,每个智能体有自己的 Actor、Critic、目标 Actor、目标 Critic 和优化器。采样一批经验后,先更新每个 Critic,再更新每个 Actor,最后软更新所有目标网络。
def train_step(agents, replay_buffer, batch_size, gamma=0.95): batch = replay_buffer.sample(batch_size) global_obs = torch.FloatTensor(batch['global_obs']) obs = torch.FloatTensor(batch['obs']) actions = torch.FloatTensor(batch['actions']) rewards = torch.FloatTensor(batch['rewards']) next_global_obs = torch.FloatTensor(batch['next_global_obs']) next_obs = torch.FloatTensor(batch['next_obs']) dones = torch.FloatTensor(batch['dones']) num_agents = len(agents) # 计算所有智能体的下一个动作 next_actions = [] for i, agent in enumerate(agents): next_actions.append(agent.target_actor(next_obs[:, i, :])) next_actions = torch.stack(next_actions, dim=1) # (batch, num_agents, act_dim) next_global_act = next_actions.view(batch_size, -1) # 更新每个 Critic for i, agent in enumerate(agents): with torch.no_grad(): target_q = agent.target_critic(next_global_obs, next_global_act) target_q = rewards[:, i:i+1] + gamma * (1 - dones) * target_q current_q = agent.critic(global_obs, actions.view(batch_size, -1)) critic_loss = F.mse_loss(current_q, target_q) agent.critic_optimizer.zero_grad() critic_loss.backward() torch.nn.utils.clip_grad_norm_(agent.critic.parameters(), 1.0) agent.critic_optimizer.step() # 更新每个 Actor for i, agent in enumerate(agents): # 当前智能体的动作由 Actor 给出,其他智能体的动作从经验里取 cur_actions = actions.clone() cur_actions[:, i, :] = agent.actor(obs[:, i, :]) cur_global_act = cur_actions.view(batch_size, -1) actor_loss = -agent.critic(global_obs, cur_global_act).mean() agent.actor_optimizer.zero_grad() actor_loss.backward() torch.nn.utils.clip_grad_norm_(agent.actor.parameters(), 1.0) agent.actor_optimizer.step() # 软更新所有目标网络 for agent in agents: soft_update(agent.target_actor, agent.actor) soft_update(agent.target_critic, agent.critic)这段代码里,next_actions是用目标 Actor 算出来的,用于计算目标 Q 值。更新 Actor 时,只把当前智能体的动作替换成 Actor 的输出,其他智能体的动作保持经验里的值。这是 MADDPG 的核心操作。gamma=0.95是折扣因子,对于这种短回合任务,0.95 比 0.99 更合适。
4.5 探索噪声的添加与衰减
DDPG 是确定性策略,必须加噪声才能探索。我一般用 Ornstein-Uhlenbeck 噪声或高斯噪声。高斯噪声简单,但 OU 噪声在连续控制里更平滑。下面是一个简单的高斯噪声衰减:
class GaussianNoise: def __init__(self, act_dim, sigma=0.2): self.act_dim = act_dim self.sigma = sigma self.t = 0 def sample(self): return np.random.normal(0, self.sigma, self.act_dim) def decay(self, rate=0.9995, min_sigma=0.05): self.sigma = max(min_sigma, self.sigma * rate)sigma=0.2是初始噪声标准差,min_sigma=0.05是下限。每回合结束后调用decay,噪声会慢慢减小,从探索转向利用。注意,噪声要加在动作上,然后裁剪到 [-1, 1]。
5. 避坑与排查:MADDPG 训练不收敛的五个血泪经验
5.1 奖励不升反降,智能体学会「躺平」
现象:训练初期奖励有一点上升,然后迅速下降,最后所有智能体都不动,奖励稳定在一个很低的值。
原因:碰撞惩罚太大,或者奖励设计有漏洞。智能体发现「不动」比「动」的惩罚更小,因为动可能撞到别人。这是典型的奖励黑客。
解决:调整奖励尺度,让靠近目标的奖励占主导。可以把碰撞惩罚从 -1.0 降到 -0.1,或者给「不动」加一个小的负奖励。另外,检查 done 条件是否太容易触发,导致回合太短,智能体来不及学习。
5.2 Critic 损失爆炸,出现 NaN
现象:训练几步后 Critic 损失变成 NaN,网络参数全乱。
原因:学习率太大,或者奖励没有缩放,TD 误差量级过大。另外,如果经验回放池里存了未归一化的观测,也会导致输入过大。
解决:把 Critic 学习率降到 1e-3 或 1e-4,Actor 学习率降到 1e-4。奖励乘以 0.1 或 0.01。观测归一化到 [-1, 1]。加梯度裁剪。如果还不行,检查经验回放池里是否有 NaN,可以在 push 时加断言。
5.3 智能体之间「串通」或「互相甩锅」
现象:在合作场景下,智能体学会了轮流去目标点,但整体效率很低;在竞争场景下,一个智能体碾压其他智能体。
原因:MADDPG 的 Critic 看到全局信息,如果奖励是共享的,智能体可能学会搭便车。如果奖励是独立的,Critic 可能无法正确评估每个智能体的贡献。
解决:对于合作场景,可以尝试给每个智能体独立的奖励,或者用反事实基线。对于竞争场景,确保每个智能体的 Critic 输入包含其他智能体的动作,这样它才能预测对手行为。另外,可以调整奖励函数,让每个智能体的贡献更明确。
5.4 训练速度慢,样本效率低
现象:跑了几万步,奖励还在原地踏步。
原因:经验回放池太小,或者 batch_size 太小。MADDPG 需要大量交互数据,因为每个 Critic 的输入维度是num_agents * (obs_dim + act_dim),参数量随智能体数量增长。
解决:把回放池容量加到 1e6,batch_size 加到 1024 或 2048。如果内存不够,可以用优先经验回放。另外,减少智能体数量或观测维度,先在小规模上验证算法正确性,再扩大规模。
5.5 执行时表现远差于训练时
现象:训练时奖励很高,但把 Actor 拿出来单独跑,表现一塌糊涂。
原因:训练时用了目标网络和噪声,执行时没用。或者训练时 Critic 过拟合了,Actor 依赖了 Critic 的错误评估。
解决:执行时只用 Actor 当前网络,不加噪声。如果训练时用了目标 Actor 算动作,执行时也要用当前 Actor。另外,可以在训练结束后用固定随机种子跑几个回合,看看方差大不大。如果方差大,说明策略不稳定,需要继续训练或调整超参数。
6. 进阶技巧:让 MADDPG 在复杂场景下更稳的两个改动
6.1 用注意力机制替换 Critic 的简单拼接
原始 MADDPG 的 Critic 把所有智能体的观测和动作直接拼接,当智能体数量增多时,输入维度线性增长,参数量爆炸。一个常见的改进是引入注意力机制:每个智能体的 Critic 只关注与自己相关的其他智能体。
具体做法是,把全局状态拆成每个智能体的局部观测,用一个注意力模块计算每个智能体对当前智能体的权重,然后加权求和作为 Critic 的输入。这样 Critic 的输入维度不随智能体数量增长,而且能学到「谁对我重要」。
class AttentionCritic(nn.Module): def __init__(self, obs_dim, act_dim, hidden_dim=64): super().__init__() self.obs_dim = obs_dim self.act_dim = act_dim # 每个智能体的嵌入 self.embed = nn.Linear(obs_dim + act_dim, hidden_dim) # 注意力权重 self.attn = nn.Linear(hidden_dim, 1) # Q 头 self.q_head = nn.Sequential( nn.Linear(hidden_dim * 2, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, 1) ) def forward(self, global_obs, global_act, num_agents): # global_obs: (batch, num_agents * obs_dim) # global_act: (batch, num_agents * act_dim) batch_size = global_obs.size(0) obs = global_obs.view(batch_size, num_agents, self.obs_dim) act = global_act.view(batch_size, num_agents, self.act_dim) # 拼接每个智能体的观测和动作 x = torch.cat([obs, act], dim=-1) # (batch, num_agents, obs_dim+act_dim) embed = F.relu(self.embed(x)) # (batch, num_agents, hidden_dim) # 计算注意力权重 attn_weights = F.softmax(self.attn(embed), dim=1) # (batch, num_agents, 1) # 加权求和 context = (embed * attn_weights).sum(dim=1) # (batch, hidden_dim) # 当前智能体的嵌入(这里假设第一个是当前智能体,实际要按索引取) self_embed = embed[:, 0, :] # (batch, hidden_dim) q_input = torch.cat([self_embed, context], dim=-1) return self.q_head(q_input)这个改动需要调整训练循环,因为每个智能体的 Critic 输入不再是全局拼接,而是经过注意力聚合的上下文。注意,注意力权重是每个智能体独立计算的,所以每个 Critic 有自己的注意力模块。
6.2 用优先经验回放加速收敛
优先经验回放(Prioritized Experience Replay)根据 TD 误差给经验排序,误差大的样本被采样的概率高。在 MADDPG 里,每个智能体的 Critic 有自己的 TD 误差,可以取平均或最大值作为优先级。
class PrioritizedReplayBuffer(ReplayBuffer): def __init__(self, capacity, num_agents, obs_dim, act_dim, alpha=0.6): super().__init__(capacity, num_agents, obs_dim, act_dim) self.alpha = alpha self.priorities = np.zeros(capacity) self.max_priority = 1.0 def push(self, *args): super().push(*args) self.priorities[self.ptr - 1] = self.max_priority def sample(self, batch_size, beta=0.4): probs = self.priorities[:self.size] ** self.alpha probs /= probs.sum() idxs = np.random.choice(self.size, batch_size, p=probs) batch = {k: v[idxs] for k, v in self.buffer.items()} # 重要性采样权重 weights = (self.size * probs[idxs]) ** (-beta) weights /= weights.max() batch['weights'] = weights batch['idxs'] = idxs return batch def update_priorities(self, idxs, priorities): self.priorities[idxs] = priorities self.max_priority = max(self.max_priority, priorities.max())在训练循环里,计算完 Critic 损失后,用 TD 误差更新优先级。注意,重要性采样权重要乘在 Critic 损失上,避免偏差。alpha=0.6控制优先级的程度,beta=0.4控制重要性采样的程度,通常 beta 会从 0.4 线性增加到 1.0。
这两个改动我都在实际项目里用过,注意力机制在智能体数量超过 5 个时效果明显,优先经验回放在奖励稀疏时能快 30% 左右。但要注意,改动越大,调试成本越高。我的习惯是先用原始 MADDPG 跑通一个简单环境,确认训练循环没问题,再逐步加改进。每次只改一个地方,改完跑 3 个随机种子,看平均曲线,不要被单次跑出来的「玄学」结果骗了。希望帮到你。
本文还有配套的精品资源,点击获取