简介这份资源是基于MADDPG的多智能体博弈对抗算法Python实现项目源码面向计算机相关专业正在做毕业设计、课程设计或期末大作业的学生以及需要多智能体强化学习实战练习的学习者。项目经导师指导并认可通过评审分98分可作为高分毕设参考方案。压缩包共14个文件约1.6MB以10个Python源码文件为核心涵盖MADDPG主算法、DDPG基础实现、网络结构、经验回放缓冲区、强化学习工具函数及多智能体环境测试脚本另含cfg配置、txt说明与gitignore等辅助文件目录结构清晰便于按模块阅读与二次开发。已有189人学习关注。读者可从中获得完整的多智能体博弈对抗实现思路、算法模块拆分方式、训练与测试脚本组织方法以及环境配置参考适合用来理解MADDPG的代码落地流程并迁移到自己的课题中。1. MADDPG 多智能体博弈对抗从标题到能跑起来的距离多智能体博弈对抗这件事真正上手才会发现难点从来不是把 MADDPG 的公式抄对而是让几个智能体在同一个环境里打起来还能收敛。MADDPGMulti-Agent Deep Deterministic Policy Gradient解决的核心问题是每个智能体的策略在训练中不断变化导致从单个智能体视角看环境是非平稳的。它的做法是把所有智能体的观测和动作拼进 Critic 的输入让 Critic 在训练时开天眼而 Actor 执行时只看自己的局部观测。这个设计直接决定了你在写代码时Critic 网络和 Actor 网络的输入维度是不一样的这是后面所有坑的源头。这篇面向的是想用 Python 从零搭一套多智能体对抗训练框架的从业者。你可能已经看过 MADDPG 的论文知道它基于 DDPG 扩展而来但打开代码不知道 replay buffer 里该存什么、Critic 的输入怎么拼、多个智能体怎么共享参数。下面按环境怎么搭 → 网络怎么写 → 训练怎么跑 → 坑在哪的顺序把一套可复现的实现路径讲清楚。环境用最经典的 MPEMulti-Agent Particle Environment里的 simple_tag 或 simple_adversary 场景Python 3.8 以上PyTorch 1.10 以上不需要 GPU 也能跑通小规模对抗。2. 博弈对抗环境与 MADDPG 的接口设计2.1 为什么对抗场景比协作场景更难调协作场景里所有智能体共享一个奖励大家朝一个方向优化即使策略有波动整体趋势是明确的。对抗场景不一样追捕方得分意味着逃跑方失分两个策略在互相拆台。MADDPG 的 Critic 虽然能看到全局信息但 Actor 只能根据局部观测做决策这就导致一个典型现象——训练初期追捕方很快学会围堵逃跑方却因为探索不足一直往墙角跑梯度信号被追捕方主导逃跑方的 Actor 几乎不更新。常见做法是在对抗场景里给每个智能体单独维护一套 Actor 和 Critic不共享参数。如果智能体数量多、观测维度一致可以考虑共享底层特征提取层但输出层必须独立。我一般会在环境封装层就把每个智能体的观测、动作、奖励分开存避免后面 replay buffer 里数据错位。2.2 环境封装把 MPE 的返回值整理成训练能用的格式MPE 环境原生的step()返回的是所有智能体的观测列表、奖励列表、done 标志和额外信息。直接拿这个去喂网络维度对不上是小事关键是每个智能体的观测维度可能不同动作空间也可能不同。下面这段封装代码把环境输出整理成按智能体索引的字典结构。import numpy as np import gym from gym import spaces class MultiAgentEnvWrapper: 把 MPE 环境封装成按 agent_id 索引的字典接口 def __init__(self, env_namesimple_adversary, max_steps100): self.env gym.make(env_name) self.max_steps max_steps self.n_agents self.env.n_agents self.step_count 0 # 按智能体分别记录观测和动作空间 self.obs_dims [] self.act_dims [] self.act_highs [] for i in range(self.n_agents): obs_space self.env.observation_space[i] act_space self.env.action_space[i] self.obs_dims.append(obs_space.shape[0]) self.act_dims.append(act_space.shape[0]) self.act_highs.append(act_space.high[0]) def reset(self): self.step_count 0 obs_list self.env.reset() return {i: obs_list[i] for i in range(self.n_agents)} def step(self, action_dict): # action_dict: {agent_id: np.array} actions [action_dict[i] for i in range(self.n_agents)] obs_list, reward_list, done_list, info self.env.step(actions) self.step_count 1 obs {i: obs_list[i] for i in range(self.n_agents)} rewards {i: reward_list[i] for i in range(self.n_agents)} # MPE 的 done 是列表取第一个即可所有智能体同步结束 done done_list[0] if isinstance(done_list, (list, tuple)) else done_list if self.step_count self.max_steps: done True return obs, rewards, done, info这段封装的关键点有三个。第一obs_dims和act_dims按智能体分别记录因为 simple_adversary 里追捕方和逃跑方的观测维度确实不同。第二step()接收的是字典内部转成列表再调环境这样上层训练循环不用关心 MPE 的接口细节。第三done的处理要小心MPE 不同版本返回的 done 可能是列表也可能是布尔值这里做了兼容。参数max_steps控制单局最大步数对抗场景一般设 50 到 100太短学不到策略太长训练慢。2.3 Replay Buffer 里到底存什么MADDPG 的 replay buffer 和单智能体 DDPG 最大的区别是存的是全局状态。每个 transition 需要包含所有智能体的观测、所有智能体的动作、每个智能体各自的奖励、下一时刻所有智能体的观测、done 标志。Critic 训练时从 buffer 里采样把全局观测和全局动作拼起来作为输入。import random from collections import deque class MultiAgentReplayBuffer: def __init__(self, capacity100000): self.buffer deque(maxlencapacity) def push(self, obs_dict, act_dict, rew_dict, next_obs_dict, done): obs_dict: {i: np.array}, act_dict: {i: np.array} self.buffer.append((obs_dict, act_dict, rew_dict, next_obs_dict, done)) def sample(self, batch_size): batch random.sample(self.buffer, batch_size) obs_batch, act_batch, rew_batch, next_obs_batch, done_batch zip(*batch) return obs_batch, act_batch, rew_batch, next_obs_batch, done_batch def __len__(self): return len(self.buffer)这里没有把数据转成 tensor因为不同智能体的观测维度不同转 tensor 的时机放在训练循环里按智能体分别处理更灵活。capacity设 10 万是经验值对抗场景状态空间不大10 万足够覆盖多种博弈局面。如果显存或内存紧张降到 5 万也能跑但采样多样性会下降。3. Actor-Critic 网络实现与参数共享策略3.1 Critic 为什么要吃全局信息MADDPG 的核心设计就在 Critic 的输入上。每个智能体 i 的 Critic 接收的是所有智能体的观测拼接 所有智能体的动作拼接。这样 Critic 在评估某个动作好不好时知道其他智能体当前在做什么从而缓解非平稳问题。Actor 则只接收自己的局部观测输出自己的动作。import torch import torch.nn as nn import torch.nn.functional as F class Actor(nn.Module): def __init__(self, obs_dim, act_dim, act_high, hidden_dim64): super().__init__() self.act_high act_high 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)) # tanh 输出 [-1,1]再缩放到动作空间范围 return torch.tanh(self.fc3(x)) * self.act_high class Critic(nn.Module): def __init__(self, total_obs_dim, total_act_dim, hidden_dim64): super().__init__() self.fc1 nn.Linear(total_obs_dim total_act_dim, hidden_dim) self.fc2 nn.Linear(hidden_dim, hidden_dim) self.fc3 nn.Linear(hidden_dim, 1) def forward(self, all_obs, all_act): x torch.cat([all_obs, all_act], dim-1) x F.relu(self.fc1(x)) x F.relu(self.fc2(x)) return self.fc3(x)Actor 的输出层用 tanh 再乘act_high这是 DDPG 系列的标准做法保证输出在合法动作范围内。Critic 的输入维度是total_obs_dim total_act_dim即所有智能体观测维度之和加上所有智能体动作维度之和。隐藏层 64 维对 MPE 场景够用如果换成更复杂的对抗环境可以加到 128 或 256但要注意训练时间会明显增加。3.2 每个智能体独立网络还是共享参数这是 MADDPG 实现里最容易纠结的地方。我的经验是对抗场景一律独立网络。原因很直接——追捕方和逃跑方的策略目标相反共享参数会让梯度互相抵消。协作场景可以考虑共享但即使共享Critic 也必须独立因为每个智能体的奖励函数不同。class MADDPGAgent: 单个智能体的 Actor-Critic 封装 def __init__(self, agent_id, obs_dim, act_dim, act_high, total_obs_dim, total_act_dim, lr1e-3, gamma0.95): self.agent_id agent_id self.gamma gamma self.actor Actor(obs_dim, act_dim, act_high) self.critic Critic(total_obs_dim, total_act_dim) self.target_actor Actor(obs_dim, act_dim, act_high) self.target_critic Critic(total_obs_dim, total_act_dim) # 初始化目标网络参数 self.target_actor.load_state_dict(self.actor.state_dict()) self.target_critic.load_state_dict(self.critic.state_dict()) self.actor_optim torch.optim.Adam(self.actor.parameters(), lrlr) self.critic_optim torch.optim.Adam(self.critic.parameters(), lrlr) def select_action(self, obs, noise_scale0.1): obs_t torch.FloatTensor(obs).unsqueeze(0) with torch.no_grad(): action self.actor(obs_t).squeeze(0).numpy() # 训练时加高斯噪声做探索 action np.random.normal(0, noise_scale, sizeaction.shape) return np.clip(action, -self.actor.act_high, self.actor.act_high)gamma设 0.95 是 MPE 场景的常用值因为单局步数不多折扣因子不需要太大。noise_scale控制探索强度训练初期可以设 0.2后期降到 0.05。注意select_action里先no_grad再转 numpy避免把计算图带进 replay buffer。3.3 目标网络的软更新DDPG 系列用软更新soft update而不是硬拷贝让目标网络缓慢跟随在线网络训练更稳定。更新系数tau一般设 0.01。def soft_update(target_net, online_net, tau0.01): for target_param, online_param in zip(target_net.parameters(), online_net.parameters()): target_param.data.copy_( tau * online_param.data (1 - tau) * target_param.data )tau越小目标网络越稳定但跟随越慢。0.01 是经验值如果发现训练震荡严重可以降到 0.005如果收敛太慢可以升到 0.02。这个参数没有理论最优靠观察 loss 曲线调。4. 训练循环从采样到梯度更新的完整链路4.1 一轮训练的完整流程训练循环的结构是每个 episode 开始时 reset 环境然后循环 step每个 step 里所有智能体各自选动作、执行、存 buffer当 buffer 积累到一定量后开始采样更新。下面是一个 episode 的核心逻辑。def run_episode(env, agents, buffer, batch_size1024, warmup5000): obs_dict env.reset() episode_rewards {i: 0 for i in range(env.n_agents)} done False while not done: act_dict {} for i, agent in enumerate(agents): act_dict[i] agent.select_action(obs_dict[i], noise_scale0.1) next_obs_dict, rew_dict, done, _ env.step(act_dict) buffer.push(obs_dict, act_dict, rew_dict, next_obs_dict, done) for i in range(env.n_agents): episode_rewards[i] rew_dict[i] obs_dict next_obs_dict # buffer 足够大才开始训练 if len(buffer) warmup: update_all_agents(agents, buffer, batch_size) return episode_rewardswarmup设 5000 步是让 buffer 先积累足够的随机探索数据避免一开始就用少量偏差大的数据训练导致策略崩溃。这个值可以按环境复杂度调MPE 场景 5000 够用更复杂的环境可能需要 1 万到 2 万。4.2 Critic 的更新用全局信息算 TD 误差Critic 的更新逻辑是用目标 Actor 算出下一时刻所有智能体的动作拼上下一时刻的全局观测喂给目标 Critic 得到目标 Q 值然后和当前 Critic 的 Q 值算均方误差。def update_critic(agent, agents, batch, gamma0.95): obs_batch, act_batch, rew_batch, next_obs_batch, done_batch batch # 拼接所有智能体的观测和动作 all_obs torch.cat([torch.FloatTensor([o[i] for o in obs_batch]) for i in range(len(agents))], dim-1) all_act torch.cat([torch.FloatTensor([a[i] for a in act_batch]) for i in range(len(agents))], dim-1) all_next_obs torch.cat([torch.FloatTensor([o[i] for o in next_obs_batch]) for i in range(len(agents))], dim-1) # 目标动作每个智能体的目标 Actor 根据 next_obs 输出 target_acts [] for i, a in enumerate(agents): next_obs_i torch.FloatTensor([o[i] for o in next_obs_batch]) with torch.no_grad(): target_acts.append(a.target_actor(next_obs_i)) all_target_act torch.cat(target_acts, dim-1) with torch.no_grad(): target_q agent.target_critic(all_next_obs, all_target_act) reward torch.FloatTensor([r[agent.agent_id] for r in rew_batch]).unsqueeze(1) done_mask torch.FloatTensor([1 - float(d) for d in done_batch]).unsqueeze(1) target_q reward gamma * done_mask * target_q current_q agent.critic(all_obs, all_act) critic_loss F.mse_loss(current_q, target_q) agent.critic_optim.zero_grad() critic_loss.backward() agent.critic_optim.step() return critic_loss.item()这段代码里done_mask的处理很关键。如果 episode 结束目标 Q 值不应该再加下一时刻的折扣回报所以用1 - done做掩码。all_obs的拼接顺序必须和网络定义时的顺序一致否则 Critic 学到的关联是错的。我一般按 agent_id 从小到大拼训练和推理保持一致。4.3 Actor 的更新只用自己的局部观测Actor 的更新目标是最大化 Critic 给出的 Q 值。注意这里 Actor 只用自己的局部观测但 Critic 需要全局信息。所以更新 Actor 时其他智能体的动作从 replay buffer 里取真实动作只有当前智能体的动作由 Actor 重新生成。def update_actor(agent, agents, batch): obs_batch, act_batch, _, _, _ batch # 当前智能体的动作由 Actor 重新生成 obs_i torch.FloatTensor([o[agent.agent_id] for o in obs_batch]) new_act_i agent.actor(obs_i) # 其他智能体的动作从 buffer 取 all_acts [] for i, a in enumerate(agents): if i agent.agent_id: all_acts.append(new_act_i) else: all_acts.append(torch.FloatTensor([act[i] for act in act_batch])) all_obs torch.cat([torch.FloatTensor([o[i] for o in obs_batch]) for i in range(len(agents))], dim-1) all_act torch.cat(all_acts, dim-1) actor_loss -agent.critic(all_obs, all_act).mean() agent.actor_optim.zero_grad() actor_loss.backward() agent.actor_optim.step() return actor_loss.item()Actor loss 取负号是因为要最大化 Q 值。这里有个容易翻车的地方new_act_i是有梯度的而其他智能体的动作是torch.FloatTensor没有梯度拼接后 Critic 的梯度只会回传到当前 Actor不会影响其他智能体。这正是 MADDPG 的设计意图——每个智能体只优化自己的 Actor。4.4 训练超参设置参考参数推荐值说明lr_actor1e-3Actor 学习率太大容易震荡lr_critic1e-3Critic 学习率可与 Actor 相同gamma0.95折扣因子MPE 场景常用tau0.01软更新系数batch_size1024采样批量buffer_capacity100000replay buffer 容量warmup5000预热步数noise_scale0.1~0.2探索噪声标准差max_steps50~100单局最大步数这些值不是金科玉律但按这套配置在 simple_adversary 上跑 2000 到 3000 个 episode 能看到明显的策略分化。如果 loss 一直不降先检查 Critic 输入拼接顺序和 replay buffer 里的数据是否对齐。5. 避坑与排查多智能体对抗训练里最容易翻车的五件事5.1 现象训练几百轮后所有智能体动作趋同原因通常是 Actor 输出层没有做动作范围缩放或者缩放系数用错了。MPE 里不同智能体的动作范围可能不同如果统一用act_high1.0实际环境要求的是别的范围动作被裁剪后信息丢失策略退化成输出极值。解决在环境封装时逐个读取action_space[i].high传给对应的 Actor。如果动作空间是多维的act_high要取向量而不是标量。5.2 现象Critic loss 剧烈震荡Q 值越来越大原因一般是目标网络更新太快或者gamma设得太大。对抗场景里奖励有正有负如果gamma接近 1目标 Q 值会累积很大的方差。解决先把tau降到 0.005观察 loss 是否平滑。如果还震荡把gamma降到 0.9。另外检查done_mask是否正确应用episode 结束时没有截断目标 Q 值会导致 Q 值无限增长。5.3 现象某个智能体的 Actor loss 始终为 0 或不变原因通常是这个智能体的 Critic 没有学到有效梯度或者 replay buffer 里这个智能体的数据比例失衡。对抗场景里如果一方很快碾压另一方弱势方的 transition 奖励几乎恒定Critic 输出常数Actor 梯度消失。解决检查 replay buffer 里各智能体的数据是否均匀。可以在采样时按智能体分层采样保证每个智能体的 transition 都被采到。另外适当增大noise_scale让弱势方有更多探索。5.4 现象训练初期 reward 上升很快之后突然崩掉这是典型的过拟合到某个博弈模式。追捕方学会了一种围堵方式逃跑方也学会了对应的逃跑路线双方陷入局部纳什均衡一旦环境初始状态稍有变化就崩溃。解决在环境 reset 时增加随机性比如随机化智能体初始位置。另外可以定期降低noise_scale而不是线性降到 0保留少量探索。经验做法是每 500 个 episode 把noise_scale乘以 0.9最低不低于 0.02。5.5 现象GPU 上训练比 CPU 还慢MPE 场景的网络很小数据在 CPU 和 GPU 之间来回拷贝的开销远大于计算本身。另外 replay buffer 存在 CPU 内存里每次采样都要转 tensor 再搬到 GPU这个搬运是瓶颈。解决小规模对抗场景直接用 CPU 训练。如果一定要用 GPU把整个 batch 一次性转成 tensor 再搬不要逐个 transition 搬。另外torch.set_num_threads(4)可以限制 CPU 线程数避免多线程竞争。6. 让对抗策略真正分化的两个进阶技巧第一个技巧是奖励重塑。MPE 原生奖励在对抗场景里往往太稀疏追捕方要围住逃跑方才有奖励前期几乎学不到东西。我一般会在环境封装层加一个距离惩罚项追捕方离逃跑方越远每步扣一个小额负奖励。这个惩罚系数从 0.01 开始试太大会让追捕方不敢动太小没效果。加上之后simple_adversary 上追捕方的胜率从 30% 左右能提到 70% 以上。第二个技巧是优先经验回放。对抗场景里大部分 transition 是平淡的双方都在远离彼此或者原地打转真正决定胜负的关键 transition 很少。给 replay buffer 里每条数据加一个 TD 误差权重采样时按权重抽能让 Critic 更快学到关键局面。实现上不用上完整的 Prioritized Experience Replay简单做法是每次采样后把 TD 误差大的 transition 复制一份放回 buffer成本低但有效。验证策略是否真的分化不要只看总 reward。我的习惯是每个 episode 记录追捕方和逃跑方的平均距离、追捕方的包围角度、逃跑方的存活步数。如果总 reward 在涨但平均距离没变说明策略在钻奖励函数的空子不是真的学会了对抗。这套指标比 loss 曲线更能说明问题。我自己踩过最深的坑是早期为了省事让所有智能体共享一个 Critic结果追捕方和逃跑方的梯度互相抵消训练了三天 loss 都没动。后来改成每个智能体独立 Critic同样的代码半天就出效果。多智能体对抗这件事网络结构上的偷懒最后都会在训练曲线上还回来。希望帮到你。本文还有配套的精品资源点击获取