简介这份资源面向强化学习方向的学生、研究人员与算法工程师提供多智能体深度确定性策略梯度MADDPG算法的完整可运行程序代码用于理解并复现混合协作与竞争环境下的多智能体决策方法。MADDPG 由 2017 年论文《Multi-Agent Actor-Critic for Mixed Cooperative-Competitive Environments》提出在 DDPG 基础上扩展至多智能体场景是入门多智能体强化学习MADRL的经典算法。压缩包共 21 个文件约 24KB以 6 个 Python 源码文件为核心涵盖主训练入口、网络结构、经验回放缓冲区与多智能体环境等模块另含 5 个 XML 工程配置、5 个 pyc 编译缓存及若干 txt 说明文件目录结构清晰便于按模块阅读与调试。目前已有 1312 人学习下载适合希望从代码层面掌握 MADDPG 训练流程、网络设计与多智能体交互机制的读者参考实践。1. 从一场「各自为战」的翻车说起MADDPG 到底解决了什么如果你跑过单智能体强化学习再直接把它套到多智能体环境里大概率会经历一次血泪翻车每个智能体各自维护一个 Q 网络训练时环境对每个个体来说都在不停变化——因为其他智能体也在学、也在改策略。这种「移动靶」问题在学术上叫非平稳性non-stationarity直观后果就是回报曲线剧烈震荡今天能收敛明天直接崩掉。MADDPGMulti-Agent Deep Deterministic Policy Gradient就是冲着这个痛点来的它把每个智能体的 Critic 做成「全局视角」输入所有智能体的观测和动作而 Actor 仍然只用局部观测做决策。这样一来训练时 Critic 看到的是一个相对稳定的联合状态-动作空间执行时又不需要全局信息天然适配去中心化部署。这份资源围绕算法实现、训练流程和调参细节展开适合已经写过 DDPG、想跨到多智能体协作或对抗场景的 Python 从业者。2. MADDPG 的核心机制集中训练、分散执行怎么落地2.1 为什么 Critic 要「开天眼」Actor 却只能看局部先把最容易被忽略的一点讲透MADDPG 不是简单地把 DDPG 复制 N 份。它的关键设计是 CTDECentralized Training with Decentralized Execution——训练阶段允许 Critic 拿到全局信息执行阶段每个 Actor 只依赖自己的局部观测。具体到网络结构假设有 N 个智能体第 i 个智能体的 Critic 输入是所有智能体的观测拼接concat(o_1, o_2, ..., o_N)所有智能体的动作拼接concat(a_1, a_2, ..., a_N)输出是一个标量 Q 值。而第 i 个 Actor 的输入只有o_i输出是a_i。这个不对称设计是刻意的Critic 只在训练时用执行时被丢掉所以「开天眼」不会带来部署负担。为什么这样能缓解非平稳性因为对第 i 个 Critic 来说它评估的是「在所有人当前联合策略下我采取动作 a_i 的价值」。当其他智能体策略更新时Critic 的输入里包含了它们的新动作Q 值的估计目标会随之调整而不是像独立 Q-learning 那样把别人的变化当成环境噪声。常见做法是给每个智能体配一个独立的 Critic而不是共享一个——共享 Critic 在智能体异构观测维度、动作空间不同时会直接报维度错误。2.2 网络结构与经验回放缓冲区的实现下面这段是 MADDPG 里最核心的数据结构经验回放缓冲区。它和单智能体版本最大的区别是存的是所有智能体的联合转移而不是单个(s, a, r, s)。import numpy as np import random from collections import deque class MultiAgentReplayBuffer: def __init__(self, capacity, num_agents, obs_dims, act_dims): # capacity: 缓冲区最大容量经验条数 # num_agents: 智能体数量 N # obs_dims: 每个智能体的观测维度列表长度 N # act_dims: 每个智能体的动作维度列表长度 N self.capacity capacity self.num_agents num_agents self.obs_dims obs_dims self.act_dims act_dims self.buffer deque(maxlencapacity) def store(self, obs, actions, rewards, next_obs, dones): # obs: 列表每个元素是第 i 个智能体的观测形状 (obs_dims[i],) # actions: 列表每个元素是第 i 个智能体的动作 # rewards: 列表每个元素是第 i 个智能体的即时奖励 # next_obs: 同 obs 结构 # dones: 列表每个元素是该智能体是否终止 self.buffer.append((obs, actions, rewards, next_obs, dones)) def sample(self, batch_size): batch random.sample(self.buffer, batch_size) # 转置后按智能体维度拆分方便逐个计算 TD 目标 obs_batch [np.array([b[0][i] for b in batch]) for i in range(self.num_agents)] act_batch [np.array([b[1][i] for b in batch]) for i in range(self.num_agents)] rew_batch [np.array([b[2][i] for b in batch]) for i in range(self.num_agents)] next_obs_batch [np.array([b[3][i] for b in batch]) for i in range(self.num_agents)] done_batch [np.array([b[4][i] for b in batch]) for i in range(self.num_agents)] return obs_batch, act_batch, rew_batch, next_obs_batch, done_batch逻辑说明store每次存一整条联合经验sample采样后按智能体维度转置得到每个智能体各自的 batch。参数上capacity一般取 1e5 到 1e6太小会导致早期经验被快速覆盖太大则采样效率下降、内存吃紧。batch_size常见 256 或 512多智能体场景下建议比单智能体略大因为联合状态空间更稀疏。提示deque(maxlencapacity)会自动丢弃最旧的经验不需要手动维护索引。但如果你要做优先经验回放PER就得换成带优先级的数组结构deque不够用。2.3 训练循环从采样到参数更新的完整链路训练主循环的骨架如下重点是 Critic 的 TD 目标怎么算——它用的是「所有智能体目标 Actor 输出的动作」而不是当前 Actor 的动作。import torch import torch.nn as nn import torch.optim as optim def train_step(agents, target_agents, buffer, batch_size, gamma0.95): obs_b, act_b, rew_b, next_obs_b, done_b buffer.sample(batch_size) obs_b [torch.tensor(o, dtypetorch.float32) for o in obs_b] act_b [torch.tensor(a, dtypetorch.float32) for a in act_b] rew_b [torch.tensor(r, dtypetorch.float32).unsqueeze(1) for r in rew_b] next_obs_b [torch.tensor(o, dtypetorch.float32) for o in next_obs_b] done_b [torch.tensor(d, dtypetorch.float32).unsqueeze(1) for d in done_b] # 1. 用目标 Actor 计算下一步所有智能体的动作 next_actions [target_agents[i].actor(next_obs_b[i]) for i in range(len(agents))] for i, agent in enumerate(agents): # 2. 目标 Critic 输入所有智能体的 next_obs 和 next_actions 拼接 target_critic_input torch.cat(next_obs_b next_actions, dim1) target_q target_agents[i].critic(target_critic_input) y rew_b[i] gamma * (1 - done_b[i]) * target_q.detach() # 3. 当前 Critic 输入当前 obs 和真实动作拼接 critic_input torch.cat(obs_b act_b, dim1) q agent.critic(critic_input) critic_loss nn.MSELoss()(q, y) agent.critic_optim.zero_grad() critic_loss.backward() # 梯度裁剪多智能体场景下梯度爆炸很常见 nn.utils.clip_grad_norm_(agent.critic.parameters(), 0.5) agent.critic_optim.step() # 4. Actor 更新只用自己的局部观测梯度通过 Critic 回传 pred_actions [agents[j].actor(obs_b[j]) for j in range(len(agents))] pred_actions[i] agent.actor(obs_b[i]) # 替换成当前智能体的新动作 actor_critic_input torch.cat(obs_b pred_actions, dim1) actor_loss -agent.critic(actor_critic_input).mean() agent.actor_optim.zero_grad() actor_loss.backward() nn.utils.clip_grad_norm_(agent.actor.parameters(), 0.5) agent.actor_optim.step() # 5. 软更新目标网络 soft_update(agent.critic, target_agents[i].critic, tau0.01) soft_update(agent.actor, target_agents[i].actor, tau0.01) def soft_update(net, target_net, tau): for p, tp in zip(net.parameters(), target_net.parameters()): tp.data.copy_(tau * p.data (1 - tau) * tp.data)逻辑说明第 1 步用目标 Actor 生成下一步动作这是 DDPG 系算法的标准做法避免用当前策略估计导致的自举偏差。第 2 步的target_critic_input把next_obs_b和next_actions两个列表直接拼接顺序必须和 Critic 网络定义时的输入顺序一致否则维度对得上但语义全乱。第 4 步 Actor 更新时只替换第 i 个智能体的动作为新策略输出其他智能体保持当前策略这样 Critic 的梯度才能正确反映「我单独改变动作」的影响。参数上gamma取 0.95 是协作场景的常见值如果任务回合很长、奖励稀疏可以调到 0.99tau取 0.01 是软更新系数太大目标网络漂移快、训练不稳太小则学习滞后。梯度裁剪阈值 0.5 是我踩过坑之后固定下来的多智能体 Critic 输入维度高不裁剪很容易出现 loss 变 NaN。3. 环境适配与参数调优让训练真正跑起来3.1 观测与动作空间的维度对齐多智能体项目里最常见的翻车不是算法写错而是维度对不上。假设你用某个协作环境N3 个智能体每个观测维度 8动作维度 2。那么组件输入维度输出维度说明Actor_i82只用局部观测Critic_i3×8 3×2 301全局观测全局动作联合经验存储obs: 3×8, act: 3×2—每条经验含全部智能体如果智能体异构比如观测维度分别是 8、10、6那 Critic 输入就是8106 222 30但每个 Actor 的输入维度各不相同。这时候千万别用一个共享的 Actor 网络必须每个智能体独立建网络。我一般会在初始化时打印一遍每个网络的输入输出维度确认无误再开训。动作空间如果是连续的Actor 最后一层用tanh把输出压到[-1, 1]再按环境的动作范围缩放。如果是离散动作MADDPG 就不适用了得换 MADDPG 的离散变体或者 QMIX 那类值分解方法。3.2 探索噪声与奖励设计DDPG 系算法靠外部噪声做探索MADDPG 也一样。常见做法是 OU 噪声Ornstein-Uhlenbeck或高斯噪声。OU 噪声有时间相关性适合惯性系统高斯噪声实现简单大多数场景够用。class GaussianNoise: def __init__(self, action_dim, sigma0.2): # sigma: 噪声标准差随训练衰减 self.action_dim action_dim self.sigma sigma def sample(self): return np.random.normal(0, self.sigma, sizeself.action_dim) def decay(self, rate0.9995, min_sigma0.02): # 每回合调用一次让探索逐渐收敛 self.sigma max(min_sigma, self.sigma * rate)参数说明sigma初始 0.2 是常用起点动作范围大的任务可以到 0.3rate取 0.9995 意味着大约 2000 回合后噪声降到初始的一半。min_sigma别设成 0否则后期完全没有探索遇到局部最优就出不来。奖励设计上协作任务常用全局共享奖励所有智能体拿同一个 reward对抗任务则各自拿自己的。共享奖励会让信用分配变难——某个智能体摆烂也能蹭到团队奖励。如果发现训练后期有智能体「躺平」可以加个体奖励项做混合比如r_i 0.7 * r_global 0.3 * r_individual。3.3 训练不收敛时的排查顺序训练曲线不涨或者震荡按这个顺序查先看 Critic loss 是否爆炸或变 NaN。是的话降学习率、加梯度裁剪、检查输入拼接顺序。再看动作输出是否饱和在 ±1。饱和说明 Actor 输出被 tanh 压死降 Actor 学习率或减小噪声。然后看经验回放里 reward 的分布。如果全是 0 或全负说明任务太难或奖励太稀疏先简化环境。最后看目标网络更新频率。tau太大是震荡的常见元凶。4. 避坑指南多智能体训练里最容易翻车的五件事4.1 现象训练初期回报直接崩到负无穷原因Critic 输入拼接顺序和网络定义不一致或者某个智能体的观测没归一化导致 Q 值量级失控。解决固定拼接顺序为「先所有 obs再所有 act」并在拼接前对观测做 running mean/std 归一化。我一般会在第一次 forward 后打印 Q 值的均值和方差正常应该在个位数到几十之间。4.2 现象某个智能体学会了「摆烂」团队任务失败原因共享奖励下的信用分配问题摆烂智能体的 Actor 梯度被其他智能体的贡献稀释。解决引入个体奖励项或者用反事实基线counterfactual baseline给每个智能体算优势函数。简单做法是给每个智能体的奖励减去团队平均奖励突出个体差异。4.3 现象训练到一半 loss 突然变 NaN原因多智能体 Critic 输入维度高梯度累积后爆炸或者经验回放里混入了异常值比如环境返回了 inf。解决梯度裁剪阈值设 0.5并在store时检查np.isfinite异常经验直接丢弃。另外学习率别超过 1e-3Critic 常用 1e-3Actor 常用 1e-4。4.4 现象评估时表现远差于训练时原因训练时 Critic 用了全局信息但评估时如果误把全局信息喂给 Actor或者忘了切到 eval 模式BatchNorm/Dropout 行为不一致。解决评估时只给 Actor 局部观测并调用actor.eval()。如果网络里没有 BatchNorm也要养成切模式的习惯避免以后加层时踩坑。4.5 现象换了个随机种子结果完全不可复现原因多智能体环境本身有随机性加上 PyTorch、NumPy、Python 内置 random 三处种子没统一。解决在训练脚本开头固定三处种子并设置torch.backends.cudnn.deterministic True。即便如此多线程环境下的浮点累加顺序仍可能带来微小差异评估时多跑几个 episode 取平均。5. 进阶技巧用参数共享和课程学习把训练效率拉满5.1 同构智能体的参数共享如果所有智能体的观测和动作维度相同可以让它们共享一套 Actor 和 Critic 参数只是输入各自的观测。这样做的好处是样本效率翻倍——每个智能体的经验都在更新同一套网络。实现上很简单只建一个 Actor 和一个 Critic训练时把 N 个智能体的数据拼成一个 batch 一起 forward。但要注意参数共享要求智能体「角色对称」。如果环境里智能体分工不同比如一个负责侦察、一个负责搬运共享参数会让网络学出一个「四不像」策略。判断标准是把智能体编号打乱后任务最优策略是否不变。不变就可以共享。5.2 课程学习从简单场景逐步加难度多智能体任务直接上完整难度探索空间太大经常几万回合都不见奖励。我一般会设计一个难度递增的课程阶段智能体数量任务难度训练回合12单目标、无碰撞200022多目标、有障碍300033多目标、有障碍500043完整任务8000每个阶段结束后保存模型下一阶段加载上一阶段的参数继续训。这样比从头训完整任务快 2 到 3 倍。切换阶段时经验回放缓冲区要清空否则旧难度的经验会干扰新策略。5.3 验证训练是否真的有效的三个检查点训练跑起来不代表学对了。我习惯在三个时间点做检查第一训练 1000 回合后看 Critic 的 Q 值是否随回合数上升。如果 Q 值一直不动说明 Critic 没学到东西先别管 Actor。第二训练 5000 回合后固定随机种子跑 20 个 episode看成功率。如果成功率波动超过 30%说明策略还不稳定继续训或者降学习率。第三训练结束后把每个智能体的动作序列打出来人工看是否符合任务逻辑。比如协作搬运任务里两个智能体的动作应该呈现互补模式而不是同向乱动。从那以后我每次开新多智能体项目都强制先跑一个 2 智能体的最小验证确认 CTDE 链路通了再上规模。希望帮到你。本文还有配套的精品资源点击获取