
简介基于Python与MADDPG的多智能体博弈对抗算法工程面向强化学习入门及进阶学习者也适合作为毕业设计、课程设计、大作业或工程实训项目。资源完整包含从DDPG到MADDPG的模块化实现覆盖智能体网络结构、经验回放缓冲、策略更新逻辑、训练主流程与测试环境以10个Python脚本为核心配套说明文档、项目配置、依赖清单和Git管理文件共14个文件压缩包仅19KB体量小且目录层次清晰便于快速阅读和二次开发同时也方便按模块替换网络结构或修改奖励设置。目前已有267人学习下载可作为理解多智能体博弈中集中训练与分散执行机制的直观样例也能在对抗策略研究、算法对比实验中直接作为基线代码参考通过自行运行训练并观察agent演化可进一步掌握多智能体强化学习的关键调试思路与工程组织方式。1. 多智能体博弈对抗为什么用MADDPG而不是独立训练DDPG做多智能体博弈对抗项目最常见的开局误区是给每个智能体各训练一个DDPG。表面看每个智能体都有自己的Actor-Critic独立更新好像没什么问题。但跑起来就会发现训练曲线从第二个回合开始就再也不稳定——因为对某个智能体而言其他智能体的策略也在不断变化环境本质上是非平稳的单个智能体的Q值估计一直在追移动靶。MADDPG解决这个问题的方式是集中式训练、分布式执行训练阶段Critic能看见所有智能体的观测和动作从全局视角评估策略执行阶段Actor只依赖局部观测决策。这套方案既能处理连续动作、又能处理对抗博弈Python也可以直接落地。接下来按“架构→实现→训练→踩坑→验收”的顺序把一份完整可参考的MADDPG博弈对抗方案拆开讲清楚。2. 总体架构设计从环境到训练器的六个模块拆解2.1 模块清单每个文件在工程里的职责先给一份模块地图方便建立整体认知文件名模块角色核心职责mag.py博弈环境定义对抗场景、状态转移、奖励结算buffer.py经验回放存储并采样所有智能体的联合经验network.py网络定义Actor与Critic的PyTorch实现DDPG.py单智能体组件封装单个智能体的Actor-Critic与目标网络MADDPG.py多智能体协调集中式更新逻辑调用DDPG组件完成训练main.py训练入口环境-采样-存储-更新的主循环test.py / ceshi.py评估入口加载模型、关闭噪声、统计对抗结果这个结构遵循一个原则环境、算法、训练流程三个层面分离。mag.py只暴露reset和step两个接口不管算法细节MADDPG.py只关心如何从经验中更新参数不关心环境怎么实现main.py把两者粘合在一起。这样做的直接好处是想换一个博弈场景比如把追逃博弈改成猎鹿博弈或多智能体导航只需要重写mag.py的reset和step算法层完全不用动。对毕设和课设来说这也意味着你可以在环境上做文章而不必每次改动都牵动整个训练框架。我还会在模块清单后面补一张数据维度表防止写代码时把维度搞混。假设环境是2个智能体每个智能体观测维度为8位置、速度、相对位移等动作维度为2加速度的x、y分量那么缓冲区单条经验的结构就是2个8维观测、2个2维动作、2个1维奖励、2个8维下一观测、2个布尔结束标志。采样批次维度为(agent_num, batch_size, obs_dim)。这个约定在后面写代码时会被反复用到项目里所有模块都统一遵守它能少踩一半的维度坑。2.2 训练链路与测试链路的数据流训练链路的数据流是这样的mag.py的step返回每个智能体的观测、动作、奖励、结束标志main.py把这些组合成一条“联合经验”存入buffer.py当缓冲区样本足够后MADDPG.py从buffer.py采样一个批次拆成每个智能体独立的数据送入各自的Actor和Critic进行更新。关键点在于一条经验里必须同时包含所有智能体的观测和动作因为集中式Critic需要完整信息做全局评估缺一个智能体的数据Q值估计就不成立。测试链路则完全相反关闭探索噪声加载训练好的Actor参数让每个智能体只用自己的Actor做确定性决策跑若干个完整回合统计捕获成功率和平均回合长度。两条链路分开的意义在于训练链路中Critic依赖全局信息测试链路中只有Actor参与推理正好对应CTDE的核心思想。这个设计也方便调试——训练出问题时可以先跑测试链路定位是“策略本身没学好”还是“训练节奏有问题”而不是一头扎进loss曲线里猜。2.3 为什么这个结构适合毕设、课设和项目交付从交付角度看这个结构有几个明显优势。第一模块数量适中六个主要文件就可以把算法讲清楚答辩时每个文件对应一个知识点不容易出现“代码能跑但说不清”的尴尬。第二环境与算法解耦使得算法评估更客观你可以在同一套环境下把MADDPG和独立DDPG做对比用同一份环境代码跑对比实验证明MADDPG在处理非平稳环境上的收益。第三可替换性强所有超参数都集中在main.py顶部的配置字典里方便做消融实验。我一般不建议在这个项目里引入太多额外依赖。PyTorch加NumPy已经覆盖全部需求如果加上可视化只需要在测试脚本里用matplotlib画轨迹不需要上TensorBoard之类的重型工具。MADDPG的变体很多但作为毕设或课设先把标准版本跑通再谈改进才是稳的节奏。项目名“基于PythonMADDPG的多智能体博弈对抗算法”核心交付物就是这份能跑的训练框架和一份能说明白原理的报告上面的模块划分已经足够撑起整篇报告的结构了。3. 核心代码实现缓冲区、网络结构与集中式Critic更新3.1 buffer.py多智能体经验回放怎么写多智能体经验回放和单智能体的最大区别是每次存储必须把“所有智能体的观测、动作、奖励、下一观测、结束标志”打包成一条记录而不是每个智能体单独存一条。单独存会导致采样时拿不到同一时刻的全局信息集中式Critic就失去了意义。# buffer.py - 多智能体联合经验回放 import numpy as np from collections import deque class MultiAgentReplayBuffer: def __init__(self, capacity, agent_num): self.capacity capacity self.agent_num agent_num self.buffer deque(maxlencapacity) def store(self, obs_n, act_n, rew_n, next_obs_n, done_n): # obs_n: list长度智能体数每个元素是(obs_dim,)的数组 self.buffer.append((obs_n, act_n, rew_n, next_obs_n, done_n)) def sample(self, batch_size): idx np.random.choice(len(self.buffer), batch_size, replaceFalse) obs_batch [[] for _ in range(self.agent_num)] act_batch [[] for _ in range(self.agent_num)] rew_batch [[] for _ in range(self.agent_num)] next_obs_batch [[] for _ in range(self.agent_num)] done_batch [[] for _ in range(self.agent_num)] for i in idx: obs_n, act_n, rew_n, next_obs_n, done_n self.buffer[i] for a in range(self.agent_num): obs_batch[a].append(obs_n[a]) act_batch[a].append(act_n[a]) rew_batch[a].append(rew_n[a]) next_obs_batch[a].append(next_obs_n[a]) done_batch[a].append(done_n[a]) return (np.array(obs_batch, dtypenp.float32), np.array(act_batch, dtypenp.float32), np.array(rew_batch, dtypenp.float32), np.array(next_obs_batch, dtypenp.float32), np.array(done_batch, dtypenp.float32))这段代码的关键设计有两点。一是联合经验store时把2个智能体的数据打包成一条记录存入deque二是按智能体重新组织sample返回的每个变量形状都是(agent_num, batch_size, obs_dim)而不是把所有智能体混在一起。MADDPG更新时每个智能体更新自己的网络但Critic要看到所有智能体的信息后续代码里会用torch.cat把不同智能体的数据在最后一维拼接起来。缓冲区容量一般设置在50000到100000之间。太小会让训练样本缺乏多样性太大则会储存过多旧策略下的经验在对抗博弈这种策略快速变化的场景里会拖慢收敛。batch_size常用256训练开始前先攒够一定数量经验再进入更新阶段。注意全项目统一约定“第一维为agent索引第二维为batch第三维为特征”。这个约定在buffer和MADDPG之间必须严格一致否则拼接出来的张量含义就错了。3.2 network.pyActor和Critic的结构设计网络结构不复杂但有两个地方和单智能体DDPG不同Actor输入只包含当前智能体的局部观测输出连续动作Critic输入则是所有智能体的观测加所有智能体的动作输出一个全局Q值。# network.py - 策略网络与价值网络 import torch import torch.nn as nn import torch.nn.functional as F class Actor(nn.Module): def __init__(self, obs_dim, action_dim, hidden64): super().__init__() self.fc1 nn.Linear(obs_dim, hidden) self.fc2 nn.Linear(hidden, hidden) self.mu nn.Linear(hidden, action_dim) def forward(self, obs): x F.relu(self.fc1(obs)) x F.relu(self.fc2(x)) return torch.tanh(self.mu(x)) class Critic(nn.Module): def __init__(self, obs_dim_all, act_dim_all, hidden64): super().__init__() self.fc1 nn.Linear(obs_dim_all act_dim_all, hidden) self.fc2 nn.Linear(hidden, hidden) self.q nn.Linear(hidden, 1) def forward(self, obs_all, act_all): x torch.cat([obs_all, act_all], dim-1) x F.relu(self.fc1(x)) x F.relu(self.fc2(x)) return self.q(x)Actor输出的tanh会把动作限制在[-1, 1]环境内部通过线性缩放映射到实际物理范围。Critic输入维度是“所有智能体观测维度之和加所有智能体动作维度之和”这决定了Critic在更新时必须拿到完整的全局信息。这种结构有代价智能体数量增加时Critic输入维度线性膨胀训练成本上升所以这套方案最适合2到4个智能体的中小规模对抗场景。hidden取64对低维观测8到16维够用如果观测维度超过32建议把hidden加到128同时加深一层。训练前检查一下最后一层权重初始化过大的初始值会导致早期Q值输出巨大、梯度爆炸。通常的做法是让最后一层输出初始范围控制在[-0.1, 0.1]尤其是Critic的q层。3.3 MADDPG.py集中式Critic的更新逻辑DDPG.py的作用是把每个智能体内部的Actor、Critic、Actor_target、Critic_target组装成独立组件并提供各自的优化器。MADDPG.py则负责跨智能体的协调更新核心是集中式Critic的loss计算。# MADDPG.py - 集中式训练、分布式执行 import torch import torch.nn.functional as F import numpy as np from DDPG import DDPGAgent class MADDPG: def __init__(self, agent_num, obs_dim, act_dim, cfg): self.agent_num agent_num self.agents [DDPGAgent(obs_dim, act_dim, agent_num, cfg) for _ in range(agent_num)] def select_action(self, obs_n, noise_scale0.0): actions [] with torch.no_grad(): for i, agent in enumerate(self.agents): a agent.actor(torch.FloatTensor(obs_n[i])).numpy() if noise_scale 0: a np.random.randn(*a.shape) * noise_scale actions.append(np.clip(a, -1, 1)) return actions def update(self, sample, agent_idx): obs_batch, act_batch, rew_batch, next_obs_batch, done_batch sample agent self.agents[agent_idx] # 拼接所有智能体的观测与动作形状为(batch_size, sum_dim) obs_all torch.cat([torch.FloatTensor(obs_batch[i]) for i in range(self.agent_num)], dim-1) act_all torch.cat([torch.FloatTensor(act_batch[i]) for i in range(self.agent_num)], dim-1) # 目标Q用目标Actor生成下一时刻所有智能体的动作 with torch.no_grad(): next_act_all torch.cat([ self.agents[i].actor_target(torch.FloatTensor(next_obs_batch[i])) for i in range(self.agent_num) ], dim-1) next_obs_all torch.cat([ torch.FloatTensor(next_obs_batch[i]) for i in range(self.agent_num) ], dim-1) target_q agent.critic_target(next_obs_all, next_act_all) target_q (torch.FloatTensor(rew_batch[agent_idx]).unsqueeze(1) cfg[gamma] * (1 - torch.FloatTensor( done_batch[agent_idx]).unsqueeze(1)) * target_q) current_q agent.critic(obs_all, act_all) critic_loss F.mse_loss(current_q, target_q) agent.critic_optimizer.zero_grad() critic_loss.backward() agent.critic_optimizer.step() # Actor更新将自己的动作替换为新策略输出最大化Critic的Q值 new_act agent.actor(torch.FloatTensor(obs_batch[agent_idx])) new_act_all [] for i in range(self.agent_num): if i agent_idx: new_act_all.append(new_act) else: new_act_all.append(torch.FloatTensor(act_batch[i])) new_act_all torch.cat(new_act_all, dim-1) actor_loss -agent.critic(obs_all, new_act_all).mean() agent.actor_optimizer.zero_grad() actor_loss.backward() agent.actor_optimizer.step() # 软更新目标网络tau通常取0.01 for target_param, param in zip(agent.actor_target.parameters(), agent.actor.parameters()): target_param.data.copy_(cfg[tau] * param.data (1 - cfg[tau]) * target_param.data) for target_param, param in zip(agent.critic_target.parameters(), agent.critic.parameters()): target_param.data.copy_(cfg[tau] * param.data (1 - cfg[tau]) * target_param.data)更新逻辑有三个容易写错的点。第一目标Q计算时不要梯度所以包在torch.no_grad()里并且用的是目标网络actor_target和critic_target不是在线网络。第二Actor更新的关键技巧把当前智能体动作替换成新策略输出其他智能体动作保持经验里的原始值这样Critic的梯度才能正确传回当前Actor。第三软更新系数tau决定目标网络跟随速度一般取0.01每步更新一次取太大会让目标网络几乎等于在线网络训练会变得更不稳定。4. 训练与测试闭环从main.py到test.py的完整流程4.1 mag.py环境博弈对抗场景怎么建模mag.py里定义了一个简化的追逃博弈。两个智能体在二维平面上运动0号是追捕者1号是逃跑者。每个智能体的观测是自己和对方的横纵坐标、速度分量组成8维向量动作是x、y方向的加速度指令范围[-1, 1]环境内部做简单的物理积分。奖励设计是博弈的关键。追捕者的奖励是“离对方的负距离”距离越小奖励越大若距离小于捕获半径则额外加10分并结束回合。逃跑者的奖励对称正距离越大奖励越大被抓到额外扣10分。这样两个智能体的目标在数学上相互对抗训练出的策略才有博弈价值。环境reset时随机初始化双方位置避免每次都从同一个点出发导致策略过拟合。环境里还要做一步动作缩放把网络输出的[-1, 1]映射到实际加速度范围[-3, 3]。tanh输出的边界是饱和区梯度很小如果直接把[-1, 1]当加速度用智能体学到的力偏小运动明显迟钝。缩放后动作空间利用率更高收敛速度也会快一些。4.2 main.py训练主循环超参数与探索噪声# main.py - 训练主循环 import torch import numpy as np from buffer import MultiAgentReplayBuffer from MADDPG import MADDPG from mag import MultiAgentGame cfg { actor_lr: 1e-4, critic_lr: 1e-3, gamma: 0.95, tau: 0.01, batch_size: 256, buffer_size: 100000, max_episodes: 2000, noise_init: 0.5, noise_decay: 0.995, update_freq: 100, } env MultiAgentGame() agent_num 2 obs_dim env.obs_dim act_dim env.act_dim maddpg MADDPG(agent_num, obs_dim, act_dim, cfg) buffer MultiAgentReplayBuffer(cfg[buffer_size], agent_num) noise cfg[noise_init] for episode in range(cfg[max_episodes]): obs_n env.reset() ep_reward [0.0, 0.0] step 0 done False while not done: actions maddpg.select_action(obs_n, noise_scalenoise) next_obs_n, reward_n, done env.step(actions) buffer.store(obs_n, actions, reward_n, next_obs_n, done) obs_n next_obs_n ep_reward[0] reward_n[0] ep_reward[1] reward_n[1] step 1 if len(buffer.buffer) cfg[batch_size] and step % cfg[update_freq] 0: sample buffer.sample(cfg[batch_size]) for i in range(agent_num): maddpg.update(sample, i) if step 500: done True noise max(0.01, noise * cfg[noise_decay]) if episode % 50 0: print(fep {episode}, reward {ep_reward}, noise {noise:.3f})节奏设计上update_freq100表示每个回合内每收集100步经验才做一次更新而不是每步都更新。这样做的原因有两个MADDPG的Critic输入维度大每步更新计算开销高等步更新能让缓冲区在训练前期积累更多不同的经验采样更稳定。噪声从0.5开始按0.995指数衰减到0.01前期保证充分探索后期逐渐收敛到确定性策略。如果发现训练后期不收敛可以把noise_init调大或者把noise_decay改成0.998让探索持续得更久。超参数之间是强耦合的不建议只调一个。比如gamma从0.95改成0.99会明显改变Q值的尺度这时也需要把critic_lr调低否则容易出现Q值异常放大。我一般先默认参数跑200个回合观察奖励是否在一个量级上稳定再决定是否需要系统性调参。调参这事儿多少有点玄学但核心方向是噪声负责探索空间学习率负责稳定更新gamma负责权衡长期回报三者要一起动。4.3 test.py评估胜率、平均奖励与可视化# test.py - 关闭噪声加载模型评估 import torch import numpy as np from MADDPG import MADDPG from mag import MultiAgentGame env MultiAgentGame() maddpg MADDPG(2, env.obs_dim, env.act_dim, cfg) for i, agent in enumerate(maddpg.agents): agent.actor.load_state_dict(torch.load(factor_{i}.pth)) win_count 0 round_len [] for episode in range(100): obs_n env.reset() done False step 0 while not done: actions maddpg.select_action(obs_n, noise_scale0.0) # 确定性推理 obs_n, reward_n, done env.step(actions) step 1 if step 500: break if env.below_capture_radius(): win_count 1 round_len.append(step) print(f捕获成功率: {win_count / 100:.2f}) print(f平均回合长度: {np.mean(round_len):.1f})评估时最重要的一件事是关闭噪声noise_scale0.0表示完全确定性推理这既符合分布式执行的设定也消除了随机性对胜率统计的干扰。100个回合的统计数据已经能看出策略是否稳定。如果只想快速看一眼策略效果可以用ceshi.py跑几个回合并打印轨迹坐标把数据落到csv再用matplotlib画图。我一般还会记录训练过程中每50个回合的测试胜率做成一条曲线用来观察是否出现过拟合某个初始位置的问题。如果胜率很高但换随机种子后大幅下降说明模型记住了特定起始位置而不是学到通用对抗策略——这时候需要在环境reset里强化位置随机性扩大初始化范围。5. 常见踩坑与排查收敛震荡、维度错位与测试退化5.1 训练震荡不收敛现象奖励曲线一路上窜下跳完全没有上升趋势甚至越训练越差Actor输出的动作长期停在边界值比如全-1或全1。原因最常见的是探索噪声过大且没有衰减。噪声一直维持在0.5附近时策略总是在被随机扰动支配Critic学到的Q值对应的是大量随机动作下的价值测试时噪声一关策略就失效了。另一个原因是critic_lr和actor_lr比例不当Critic学得快、Actor学得慢策略更新跟不上价值估计的变化。解决先检查噪声衰减曲线确保训练后期噪声低于0.05再把critic_lr和actor_lr的倍数压到5倍以内比如actor_lr1e-4、critic_lr3e-4。如果仍然震荡把batch_size从256降到128减少单次更新的方差。做排查时先把这两步固定下来再谈其他调整。5.2 维度错位导致Loss异常现象训练loss一开始就出现NaN或者Critic的Q值和reward尺度差几个数量级报错信息常是“shape mismatch”或者RuntimeError里提示mat1和mat2维度不一致。原因MADDPG中拼接所有智能体数据时极易排错维度。常见错误有两个一是把(agent_num, batch_size, obs_dim)误当成(batch_size, agent_num * obs_dim)直接用忘了先按智能体拆开再拼接二是拼接顺序在buffer的sample和MADDPG的update里不一致比如buffer按agent 0、1排列update里却把agent 1的数据拼到了前面。解决写update之前先写30行左右的维度检查代码逐个打印sample返回的shape和每次torch.cat后的shape确认无误后再删掉。维度约定要写进注释明确标明“第一维是batch第二维才是agent”。另外torch.cat之前用.reshape(batch_size, -1)预先整理维度能提前暴露很多隐性错误。5.3 测试效果远差于训练现象训练时奖励曲线不错跑到test.py里一评估捕获成功率比训练时低20%以上甚至策略表现得像随机游走。原因一个常见原因是训练时存在噪声策略在噪声加持下“侥幸”获得高奖励测试时去掉噪声实际动作进入Actor输出的欠拟合区域效果反而变差。另一个原因是环境重置逻辑在训练和测试中不一致比如训练时随机位置范围大测试时缩到某个角落导致策略泛化失败。解决先把训练和测试的reset逻辑统一用同一个随机范围再把噪声衰减阶段拉长确保策略在低噪声下也得到充分优化。如果问题来自动作饱和检查Actor最后一层输出分布若超过80%的动作集中在±0.95以上就缩小动作缩放系数给策略留出更多“中间态”。5.4 梯度爆炸与Q值发散现象训练几千步后Critic loss突然从0.1跳到1e6随后所有输出变成NaN模型彻底废掉。原因MADDPG的目标Q值里用奖励乘以gamma累加如果奖励尺度过大比如几百上千而网络初始化范围是±0.05Q值目标远超网络表达能力梯度回传后权重被推得过大。另一个原因是软更新tau太接近1目标网络几乎同步在线网络TD误差被不断放大。解决先把奖励做缩放把最大奖励控制在10以内检查tau值不要超过0.05在optimizer更新前加上梯度裁剪用torch.nn.utils.clip_grad_norm_把Actor和Critic的梯度都限制在0.5以内。做完这三步大部分发散问题都能被按住。5.5 缓冲区永不更新导致模型不动现象训练日志里loss完全不变Actor输出也几乎不变调大学习率依然没有反应整个训练像是卡住了。原因最常见的是“热身条件”没满足——缓冲区经验数量没达到batch_size之前不触发更新。如果更新代码写在了内层循环的外面或者update_freq的判断逻辑在某个回合里一直不成立更新就静默跳过了。另一个原因是sample时replaceFalse但batch_size大于缓冲区有效样本数导致np.random.choice抛错后训练中断表面上看是报错实际上是更新逻辑本身没进循环。解决在训练循环里加一个计数器打印第一次“update triggered”是否出现同时在sample函数的入口打印缓冲区真实长度。这种问题往往不是算法问题而是代码控制流问题先在main.py的缩进和更新条件上排雷比调任何超参数都有效。6. 验收与答辩用对比实验和场景问答证明算法价值6.1 能直接展示的三类成果第一类是收敛曲线与对比图同一套环境、同一份超参数跑MADDPG和独立DDPG各2000回合画出每个回合的平均奖励曲线。MADDPG的曲线应该更早进入稳定区间且波动更小。注意对比实验的随机种子要固定环境reset里统一调用同一个numpy seed否则实验不可复现答辩时会被问住。第二类是策略可视化把测试轨迹画到二维图上追捕者应呈现出“堵截”而不是单纯“追尾”的路径——这是区分“学懂了”和“背答案”的重要证据。演示时可以专门挑两个初始位置一个正对追捕者一个侧向逃跑展示策略的泛化能力。第三类是消融实验关闭集中式Critic、改成独立DDPG或者去掉经验回放对比结果。消融实验不一定要每次重新完整训练可以用最后500个回合的checkpoint各跑20次评估用均值加标准差来说明问题。6.2 答辩时高概率被问的三个问题第一个问题“为什么不能用独立DDPG”答案要点是环境非平稳性其他智能体策略变化让单个智能体的Q函数失真MADDPG的集中式Critic在训练时用全局信息稳定了目标Q。第二个问题“怎么证明你的算法实现是对的”答案要点是先用小规模简单场景验证组件正确再接入对抗场景也可以对比连续动作随机策略作为下界。第三个问题“扩展到3个智能体还行吗”答案要点是扩展性代价主要在Critic输入维度3到4个智能体直接可用更多智能体需要引入注意力机制或分组训练。答辩时坦白承认局限反而加分比如直接说明“这套方案未处理智能体数量动态变化环境重写需要改step接口”。我吃过一次亏当年演示时一直强调效果却被问“换一个初始分布还能收敛吗”现场答不上来。从那以后我每次验收都强制走一遍“随机换种子、换初始位置分布、关噪声重测”三连训练代码每改一次就重跑一次评估不让“看着收敛”蒙混过关。多做这一步心里就有底了希望这些经验能帮到你。本文还有配套的精品资源点击获取