简介基于Keras实现的多智能体深度确定性策略梯度MADDPG代码包面向强化学习研究者与Python开发者适用于机器人协同、多人博弈、交通控制等连续动作多智能体场景。压缩包共21个文件、大小7.86MB其中12个H5文件是不同智能体及其目标网络训练好的权重可直接加载到预测脚本中观察协同效果7个Python脚本分别负责环境建模、神经网络构建、经验回放缓冲区、训练主流程与预测可视化另有README和许可证文件目录结构简洁。已有813人学习下载适合需要快速搭建多智能体强化学习实验的开发者参考。通过源码可掌握MADDPG中集中式Critic、分布式Actor的设计逻辑以及目标网络软更新、Ornstein-Uhlenbeck噪声、经验回放等关键实现既能用于学术研究也可在此基础上扩展更多智能体或调整奖励函数完成定制任务。1. MADDPG不只是“多智能体版DDPG”这套Keras实现能解决什么问题多智能体强化学习里最头疼的问题不是某个智能体训练不出来而是所有智能体凑在一起之后环境对每个智能体来说都变成了一个“移动靶”——队友的策略一直在变状态转移的规律就永远学不稳。MADDPGMulti-Agent Deep Deterministic Policy Gradient的核心思路是“集中训练、分散执行”训练时让每个智能体的Critic看到全局的观测和动作执行时Actor仍然只依赖自己的局部观测。这套基于Python和Keras的MADDPG实现把网络定义、经验回放、目标网络软更新和训练主循环全部落地适合正在研究多智能体算法、或者被多智能体环境训练发散问题折磨的开发者。看完这篇文章你不仅能把它跑起来观察训练曲线还能替换成自己的环境做二次开发。2. 算法核心与架构设计集中训练怎么“集中”分散执行怎么“分散”2.1 从DDPG到MADDPG两者到底差在哪DDPG是单智能体连续控制任务里的经典算法一个Actor网络根据状态输出动作一个Critic网络根据状态加动作估计Q值再用经验回放池打破样本相关性配合目标网络软更新来稳定训练。MADDPG的改动并不只是“多复制几份网络”而是在训练信息上做了重新分配。每个智能体都有一套自己的Actor和Critic但Actor的输入只有自己的观测Critic的输入却是所有智能体的观测加所有智能体的动作。这样设计的原因是单个智能体的视角下其他智能体都是环境的一部分如果Critic看不到它们的动作Q值目标就会随着策略更新而漂移让Critic看到全局信息相当于把多智能体环境变成了一个稳定的监督学习问题。训练时每个智能体的Critic都会独立更新但共享同一份经验回放数据。执行时Actor只做局部观测到动作的映射不依赖其他智能体的信息所以这套算法的部署方式叫“分散执行”。这种训练和执行的分离是MADDPG最核心的设计哲学。对比项DDPGMADDPGActor输入自身状态自身观测Critic输入自身状态自身动作所有智能体观测所有智能体动作经验回放单智能体经验多智能体共享经验池目标网络数量2个2N个N个智能体各一套2.2 项目目录结构与模块划分从工程角度看这个项目把MADDPG拆成了几个职责清晰的模块目录结构大致如下maddpg-keras/ ├── train.py # 训练主入口负责环境交互与循环调度 ├── config.py # 超参数集中配置学习率、tau、gamma、噪声参数 ├── envs/ │ ├── simple_spread.py # 协作场景多个智能体分散覆盖目标点 │ └── simple_tag.py # 对抗场景追捕者与逃跑者的博弈 ├── maddpg/ │ ├── agent.py # 单个智能体的学习逻辑含actor/critic更新 │ ├── actor.py # Actor网络构建 │ ├── critic.py # Critic网络构建 │ ├── replay_buffer.py # 多智能体经验回放区 │ └── utils.py # 软更新、噪声策略等工具函数 └── requirements.txt我比较喜欢这个划分方式的原因在于agent.py把“网络定义”和“学习逻辑”解耦了actor.py和critic.py只负责网络结构的构建而agent.py负责梯度更新和目标网络同步。这样不管是想改网络层数、换激活函数还是修改更新公式都不需要动其他模块。2.3 技术选型为什么用Keras函数式API而不是SequentialMADDPG的网络结构天然适合用Keras函数式API来实现。Sequential模型只能描述“输入到输出的一条直线”但Critic网络需要接受两个输入全局观测和全局动作而且动作不是直接拼在最外层输入而是拼接在第一层隐藏层后面这种结构用Sequential根本写不了。另一个考虑是Keras的模型序列化能力。训练多智能体时要把在线网络权重周期性地复制给目标网络get_weights和set_weights这套接口在Keras里非常成熟两种网络结构相同直接按列表元素做加权平均即可。如果用框架自带的低层API手写每次更新目标网络要处理几十个变量代码量会明显变大。依赖方面这个项目基于TensorFlow 2.x的tf.keras接口实现环境部分依赖gym接口约定数据操作依赖numpy。Python版本建议3.6以上TensorFlow 2.x在Python 3.6到3.9之间兼容性表现最好。如果你用的是TensorFlow 1.x的旧Keras函数命名和优化器接口会有一批小差异。3. 核心代码拆解Actor、Critic与经验回放的具体实现3.1 Actor网络把连续状态映射到连续动作Actor网络的任务是接收智能体的局部观测输出一个连续动作向量。由于MADDPG处理的是连续动作空间问题输出层需要把动作范围限制在环境允许的边界内这里有一个比较通用的做法输出层使用tanh激活把值压缩到[-1, 1]再乘以动作边界这样环境动作范围哪怕是[0, 5]这种非对称区间也能通过缩放映射过去。from tensorflow.keras.layers import Input, Dense, Lambda from tensorflow.keras.initializers import RandomUniform from tensorflow.keras.models import Model def build_actor(state_dim, action_dim, action_bound1.0): state_input Input(shape(state_dim,)) x Dense(64, activationrelu)(state_input) x Dense(64, activationrelu)(x) # 输出层用tanh把动作限制在[-1,1]再用Lambda层乘以动作边界 raw_action Dense(action_dim, activationtanh, kernel_initializerRandomUniform(-3e-3, 3e-3))(x) action_output Lambda(lambda a: a * action_bound)(raw_action) return Model(inputsstate_input, outputsaction_output)这里输出层的权重初始化用的是RandomUniform(-3e-3, 3e-3)初始化的范围比隐藏层小很多。原因在于Actor网络刚开始训练时如果输出层的初始权重太大动作就会直接打到边界上探索阶段完全被破坏策略会一直在一个很小范围内打转。把初始权重压到千分位级别可以让策略在训练早期输出一个接近零的平稳动作再靠噪声逐步探索。Lambda(lambda a: a * action_bound)这层是对tanh输出做缩放。如果你的环境动作范围是[-2, 2]把action_bound设为2即可如果是非对称范围比如[-1, 3]最好在环境交互时再做一次clip不要靠乘一个标量解决所有情况。3.2 Critic网络全局信息下的Q值估计Critic网络是MADDPG和DDPG差别的核心。训练时Critic的输入是所有智能体的观测拼接到一起再加上所有智能体的动作拼接到一起输出一个标量Q值。from tensorflow.keras.layers import Input, Dense, Concatenate from tensorflow.keras.initializers import RandomUniform from tensorflow.keras.models import Model def build_critic(global_obs_dim, global_act_dim): obs_input Input(shape(global_obs_dim,)) act_input Input(shape(global_act_dim,)) x Dense(64, activationrelu)(obs_input) # 动作不直接拼接在输入层而是拼接在第一层隐层之后 x Concatenate()([x, act_input]) x Dense(64, activationrelu)(x) q_output Dense(1, kernel_initializerRandomUniform(-3e-3, 3e-3))(x) return Model(inputs[obs_input, act_input], outputsq_output)注意这里动作没有直接拼在输入层而是拼在第一层隐藏层之后。这是实践里一个常见的选择让观测先经过一层特征提取再把动作信息融进去Q值的梯度信号能更有效地反传到动作上。如果你把动作直接拼在输入层等价于让网络从一开始就去拟合观测和动作的联合分布学习难度会增大。在实际训练中global_obs_dim是N个智能体观测维度之和global_act_dim是所有动作维度之和。一个容易踩坑的地方是每个智能体都有自己的Critic它们看到的全局输入是相同的但输出Q值的语义不同——第i个智能体的Critic输出的是“在这个全局状态下所有智能体采取这个联合动作时智能体i的期望回报”。3.3 经验回放池多智能体样本怎么存、怎么取多智能体经验回放和单智能体有个关键差异一条经验必须同时保存所有智能体的观测、动作和奖励因为训练任何一个Critic都需要全局信息。from collections import deque import numpy as np import random class ReplayBuffer: def __init__(self, capacity): self.buffer deque(maxlencapacity) def add(self, s, a, r, s_next, done): # s和a都是长度为智能体数量的列表每个元素是单个智能体的观测/动作 self.buffer.append((s, a, r, s_next, done)) def sample(self, batch_size): samples random.sample(self.buffer, batch_size) num_agents len(samples[0][0]) # 按智能体维度拆开每个智能体的样本形状为(batch_size, obs_dim) s [np.stack([x[0][i] for x in samples]) for i in range(num_agents)] a [np.stack([x[1][i] for x in samples]) for i in range(num_agents)] r [np.array([x[2][i] for x in samples]) for i in range(num_agents)] s_next [np.stack([x[3][i] for x in samples]) for i in range(num_agents)] done [np.array([x[4][i] for x in samples]) for i in range(num_agents)] return s, a, r, s_next, done这个回放池设计的关键点在于sample方法的返回值结构。返回值中的s是一个列表列表的每个元素对应一个智能体形状是(batch_size, obs_dim_i)。这个结构设计是为了方便后续训练时拼接全局观测——只用一行np.concatenate(s, axis1)就能得到所有智能体的全局观测矩阵。缓冲区的容量capacity建议设置得比DDPG大一些。多智能体训练时同一个经验会被多个智能体反复使用如果容量太小样本多样性不足Critic对全局状态空间的覆盖会很差。我一般会把容量设到50万条以上训练时会先填充一部分数据再开始更新避免学习初期被空缓冲区的随机噪声干扰。4. 训练流程与调参从环境交互到梯度更新的完整链路4.1 训练主循环reset、step、learn三步轮转训练主循环的逻辑比较直观每一回合重置环境每个智能体根据自身观测输出动作并加噪声执行联合动作得到下一组观测和奖励存入经验回放池当缓冲区数据量达标后开始批量学习。# train.py 核心循环 for episode in range(episodes): s env.reset() episode_reward np.zeros(env.num_agents) for step in range(max_steps): a [] for i in range(env.num_agents): action agents[i].actor.predict(s[i].reshape(1, -1), verbose0)[0] action np.clip(action agents[i].noise(), agents[i].action_low, agents[i].action_high) a.append(action) s_next, r, done, _ env.step(a) buffer.add(s, a, r, s_next, done) episode_reward r s s_next if any(done): break if len(buffer.buffer) batch_size * 20: for i in range(env.num_agents): agents[i].learn(*buffer.sample(batch_size))注意两个细节。第一r必须是长度等于智能体数量的数组每个智能体独立拿到自己的即时奖励这样各自的Critic才能学到对应的价值函数。第二在把采样得到的数据传给每个智能体的learn方法时每个智能体都会重新从缓冲区中独立抽样一次batch样本相当于用同一批环境交互数据训练了所有智能体数据利用率比每个智能体各存一份要高。4.2 Critic更新与Actor更新互补的损失函数learn方法是整个MADDPG实现的核心。Critic更新和Actor更新是两段独立的梯度计算它们共享同一个全局观测和全局动作输入但目标完全不同。# maddpg/agent.py 中单个智能体的learn方法 def learn(self, s, a, r, s_next, done): obs_full np.concatenate(s, axis1) # (batch, sum_obs_dim) act_full np.concatenate(a, axis1) # (batch, sum_act_dim) obs_full_next np.concatenate(s_next, axis1) # 用目标Actor网络计算下一时刻的联合动作 target_next_actions [] for i in range(self.num_agents): target_next_actions.append(agents[i].target_actor.predict(s_next[i], verbose0)) target_next_acts_full np.concatenate(target_next_actions, axis1) # 目标Q值 即时奖励 折扣 * 目标Critic估计 target_q self.target_critic.predict([obs_full_next, target_next_acts_full], verbose0) y r[self.idx].reshape(-1, 1) self.gamma * target_q * (1 - done[self.idx].reshape(-1, 1)) # 更新Critic最小化预测Q值与目标Q值的均方误差 with tf.GradientTape() as tape: q_pred self.critic([obs_full, act_full]) critic_loss tf.reduce_mean(tf.square(q_pred - y)) critic_grads tape.gradient(critic_loss, self.critic.trainable_variables) self.critic_optimizer.apply_gradients(zip(critic_grads, self.critic.trainable_variables)) # 更新Actor让当前Critic对当前Actor输出的Q值最大化 with tf.GradientTape() as tape: new_self_action self.actor(s[self.idx]) fake_actions [] for i in range(self.num_agents): if i self.idx: fake_actions.append(new_self_action) else: fake_actions.append(tf.constant(a[i])) fake_acts_full tf.concat(fake_actions, axis1) actor_q self.critic([tf.constant(obs_full), fake_acts_full]) actor_loss -tf.reduce_mean(actor_q) actor_grads tape.gradient(actor_loss, self.actor.trainable_variables) self.actor_optimizer.apply_gradients(zip(actor_grads, self.actor.trainable_variables)) soft_update(self.target_actor, self.actor, self.tau) soft_update(self.target_critic, self.critic, self.tau)y的计算是Critic更新的关键。在MADDPG里目标Q值用的是所有智能体的目标网络输出而不是各自在线网络的输出这保证了目标值在一段时间内保持稳定。要注意done[idx]的维度处理如果不做reshape(-1, 1)y的形状和q_pred不匹配Keras会直接报广播错误。Actor更新的技巧在于fake_actions的构建。当前智能体的动作换成了新Actor的输出其他智能体的动作保持batch采样时的原始动作然后一起喂给Critic。这样计算出的Q值梯度只反传到当前智能体的Actor上其他智能体的动作被当成了常数。Critic在计算时把其他智能体的动作当作已知输入正好体现了“集中训练”的思想。4.3 目标网络软更新tau参数怎么定软更新是让训练稳定的关键机制。如果直接把在线网络的权重复制给目标网络目标Q值每一步都会跟着当前策略剧烈变化训练容易发散。软更新的做法是每次用一个小比例tau把在线网络权重融合进目标网络def soft_update(target_network, source_network, tau0.01): target_weights target_network.get_weights() source_weights source_network.get_weights() new_weights [ tau * s (1 - tau) * t for t, s in zip(target_weights, source_weights) ] target_network.set_weights(new_weights)tau的取值直接决定训练稳定性。我一般习惯设置为0.01也就是目标网络每次只向在线网络靠近1%。调大tau到0.1会让目标网络跟踪速度变快但Critic损失曲线会明显抖动调小到0.001会让训练过程变慢但在奖励稀疏的场景下反而更稳。经验是先保持0.01跑几百个episode观察Critic损失是否稳定下降再按一个量级调整。噪声策略部分这个项目用Ornstein-Uhlenbeck过程OU噪声生成时间相关的探索噪声和单纯的高斯噪声相比OU噪声带有均值回归特性能让探索更平滑地围绕均值波动。在粒子追逐这类需要连续小幅调整方向的场景里OU噪声的探索效率比独立高斯噪声高不少。class OUNoise: def __init__(self, action_dim, mu0.0, theta0.15, sigma0.2): self.mu mu self.theta theta self.sigma sigma self.action_dim action_dim self.state np.ones(action_dim) * mu def reset(self): self.state np.ones(self.action_dim) * self.mu def sample(self): dx self.theta * (self.mu - self.state) self.sigma * np.random.randn(self.action_dim) self.state dx return self.state实际调参时一个比较有效的做法是前几百个episode用较大的sigma0.3强制探索后面逐渐衰减到0.05左右。这个项目里OUNoise在每回合env.reset()后都需要调用reset()不然上一个回合的噪声状态会沿用到新回合的初始阶段导致每个回合开始时的探索方向产生系统性偏移。5. 避坑记录Keras实现MADDPG最常见的四个翻车点5.1 特征维度拼接方向错误现象训练刚开始np.concatenate(s, axis1)就报维度不匹配或者能运行但网络输出全是同一个值训练曲线完全不动。原因经验回放池返回的s是“智能体列表”每个元素是(batch_size, obs_dim)。有的实现会把多个智能体的观测直接np.stack成(batch_size, num_agents, obs_dim)再喂给CriticDense层根本没法处理这种三维输入还有人把多个智能体观测在axis0上拼起来变成(batch_size * num_agents, obs_dim)虽然不报错但语义全错了。解决先把concat轴搞清楚。np.concatenate(s, axis1)得到的形状是(batch_size, sum(obs_dim))这才是Dense期望的二维输入。我在调试时会打印每一步的张量形状和张量形状表放在一起对照确认没问题再往下走。5.2 在训练循环内部反复compile模型现象训练初期损失正常下降某一次调整学习率或换优化器之后损失突然跳变之前积累的训练进度全部丢失。原因有的实现为了修改超参数在learn方法内部反复调用model.compile()。在Keras中每次调用compile()都会重新创建优化器变量意味着Adam里维护的动量项和历史梯度平方项全部清零相当于每更新一次就重置一次优化器状态。解决把compile()放在模型初始化阶段只在往模型中传入新的学习率时调用而不要放在训练循环里。如果是想动态调整学习率应该用tf.keras.optimizers.schedules来做学习率衰减而不是重新编译整个模型。5.3 目标网络和在线网络共用同一个模型实例现象训练跑到几百步后Critic损失不但不降反而快速上升所有智能体的动作全部挤到边界值。原因有的初学者顺手把目标网络直接指向Actor本身self.target_actor self.actor以为“反正软更新是把在线权重复制过去”。实际上这样两个模型共享同一个权重对象你更新在线网络的那一刻目标网络也被同步更新了软更新失去了“目标值暂时冻结”的意义Q值目标追着Q值预测跑训练必然发散。解决目标网络必须通过build_actor和build_critic再创建一份独立实例。初始化时用target_network.set_weights(source_network.get_weights())把在线权重复制过去之后只通过soft_update更新目标网络。5.4 Critic损失出现NaN现象训练正常跑了几百步后critic_loss突然变成nan之后所有智能体的动作输出也变成nan整个训练进程废掉。原因一般来说是数值溢出。Critic的Q值目标默认没有做裁剪如果环境的即时奖励数值范围很大比如几百几千目标Q值会积累到很大经过多层梯度反传后激活值溢出。另外Actor输出层和Critic输出层如果用了较大的初始权重也可能让第一轮梯度就爆掉。解决常规做法是把奖励先缩放到[-1, 1]范围或者在环境里直接除以一个固定的奖励缩放系数。同时在优化器里加clipvalue1.0或clipnorm0.5做梯度裁剪给NaN上个保险。输出层用第3章里的RandomUniform(-3e-3, 3e-3)初始化可以在源头降低数值爆炸概率。6. 二次开发指引把内置场景换成你自己的多智能体任务6.1 环境替换遵循一个简单的接口约定要在这个项目上做二次开发最快的方式是替换环境。这个项目的所有环境都遵循同一个自定义接口协议只需要实现三个方法class YourEnv: def reset(self): # 返回长度 智能体数量的列表每个元素是对应智能体的观测数组 return [obs_0, obs_1, ...] def step(self, actions): # actions长度 智能体数量的列表每个元素是对应智能体的动作数组 # 返回next_obs, rewards, done, info # rewards必须是长度 智能体数量的数组 # done可以是布尔值或长度 智能体数量的数组 return next_obs, rewards, done, {} def render(self): passreset和step的返回值结构必须严格遵守。一个常见的适配错误是把reset返回成所有智能体观测拼接后的大数组导致每个Actor拿到的输入维度成倍增加。记住这个项目里Actor的输入维度是单个智能体的观测维度不是全局维度。替换环境后需要在config.py里改三个参数obs_dim_per_agent、action_dim_per_agent、action_bound。如果观测维度或动作维度变了Actor和Critic网络的输入维度会在构建时自动匹配不需要改网络代码。6.2 智能体数量变化时改哪些地方从3个智能体改成5个智能体主要改动集中在两处。第一是环境的reset和step返回值智能体数量增加后观测列表长度跟着变化经验回放池的存储结构天然支持任意数量不用改。第二是Critic网络的输入维度global_obs_dim和global_act_dim会变大当观测维度总数超过256时建议把第一层隐藏层从64加到128或者加一层不然特征信息会被压缩得比较厉害。还有一个容易被忽略的问题智能体数量增加后batch_size要相应调大。经验回放池里每个智能体的样本矩阵都是(batch_size, obs_dim)当智能体数量从3增加到5时如果batch_size还保持32单个智能体在每个批次里学到的数据量其实变少了。我一般会把batch_size设置成128或256保证每个Critic更新的样本覆盖度。6.3 训练效率优化的几个实操技巧系统学完这套实现后性能体会收获更多。常见做法是Actor在计算动作时逐个使用actor.predict调用。但如果智能体数量较多每个predict都有额外的开销因此可以将所有智能体的观测batch化一次性将网络的前向过程跑完。动作维度不同时不能强行拼接。另一个容易被忽视的问题target_actor.predict在每次learn时都会执行一轮全部智能体的前向推理意味着每个learn调用带都增加了几次模型调用。可以考虑将target_actor[i].predict(s_next[i])的调用次数降为每个智能体每轮只计算一次然后重复利用。或者在learn里干脆使用同一批batch数据把所有智能体的目标网络输出计算放在同一个循环里用列表收集后再concat尽量减少网络前向调用次数。多智能体项目调试起来比单智能体麻烦得多从头跑一遍成本高。我的习惯是先在3智能体的简单场景上调试通再逐步增加智能体数量每一步都盯着Critic损失曲线确认没爆炸才继续下一步。这套Keras版的MADDPG好处是结构清晰改起来方便适合做这种增量调试。希望这次的拆解能帮你在自己的多智能体场景里少走几步弯路。本文还有配套的精品资源点击获取