简介：基于Keras实现的多智能体深度确定性策略梯度（MADDPG）代码包，面向强化学习研究者与Python开发者，适用于机器人协同、多人博弈、交通控制等连续动作多智能体场景。压缩包共21个文件、大小7.86MB，其中12个H5文件是不同智能体及其目标网络训练好的权重，可直接加载到预测脚本中观察协同效果；7个Python脚本分别负责环境建模、神经网络构建、经验回放缓冲区、训练主流程与预测可视化，另有README和许可证文件，目录结构简洁。已有813人学习下载，适合需要快速搭建多智能体强化学习实验的开发者参考。通过源码可掌握MADDPG中集中式Critic、分布式Actor的设计逻辑，以及目标网络软更新、Ornstein-Uhlenbeck噪声、经验回放等关键实现，既能用于学术研究，也可在此基础上扩展更多智能体或调整奖励函数完成定制任务。
1. MADDPG不只是“多智能体版DDPG”：这套Keras实现能解决什么问题
多智能体强化学习里最头疼的问题不是某个智能体训练不出来，而是所有智能体凑在一起之后，环境对每个智能体来说都变成了一个“移动靶”——队友的策略一直在变，状态转移的规律就永远学不稳。MADDPG（Multi-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最核心的设计哲学。
| 对比项 | DDPG | MADDPG |
|---|---|---|
| Actor输入 | 自身状态 | 自身观测 |
| 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而不是Sequential
MADDPG的网络结构天然适合用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_bound=1.0): state_input = Input(shape=(state_dim,)) x = Dense(64, activation='relu')(state_input) x = Dense(64, activation='relu')(x) # 输出层用tanh把动作限制在[-1,1]，再用Lambda层乘以动作边界 raw_action = Dense(action_dim, activation='tanh', kernel_initializer=RandomUniform(-3e-3, 3e-3))(x) action_output = Lambda(lambda a: a * action_bound)(raw_action) return Model(inputs=state_input, outputs=action_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, activation='relu')(obs_input) # 动作不直接拼接在输入层，而是拼接在第一层隐层之后 x = Concatenate()([x, act_input]) x = Dense(64, activation='relu')(x) q_output = Dense(1, kernel_initializer=RandomUniform(-3e-3, 3e-3))(x) return Model(inputs=[obs_input, act_input], outputs=q_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(maxlen=capacity) 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, axis=1)就能得到所有智能体的全局观测矩阵。
缓冲区的容量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), verbose=0)[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, axis=1) # (batch, sum_obs_dim) act_full = np.concatenate(a, axis=1) # (batch, sum_act_dim) obs_full_next = np.concatenate(s_next, axis=1) # 用目标Actor网络计算下一时刻的联合动作 target_next_actions = [] for i in range(self.num_agents): target_next_actions.append(agents[i].target_actor.predict(s_next[i], verbose=0)) target_next_acts_full = np.concatenate(target_next_actions, axis=1) # 目标Q值 = 即时奖励 + 折扣 * 目标Critic估计 target_q = self.target_critic.predict([obs_full_next, target_next_acts_full], verbose=0) 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, axis=1) 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, tau=0.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, mu=0.0, theta=0.15, sigma=0.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用较大的sigma=0.3强制探索，后面逐渐衰减到0.05左右。这个项目里OUNoise在每回合env.reset()后都需要调用reset()，不然上一个回合的噪声状态会沿用到新回合的初始阶段，导致每个回合开始时的探索方向产生系统性偏移。
5. 避坑记录：Keras实现MADDPG最常见的四个翻车点
5.1 特征维度拼接方向错误
现象：训练刚开始，np.concatenate(s, axis=1)就报维度不匹配，或者能运行但网络输出全是同一个值，训练曲线完全不动。
原因：经验回放池返回的s是“智能体列表”，每个元素是(batch_size, obs_dim)。有的实现会把多个智能体的观测直接np.stack成(batch_size, num_agents, obs_dim)再喂给Critic，Dense层根本没法处理这种三维输入；还有人把多个智能体观测在axis=0上拼起来，变成(batch_size * num_agents, obs_dim)，虽然不报错，但语义全错了。
解决：先把concat轴搞清楚。np.concatenate(s, axis=1)得到的形状是(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]范围，或者在环境里直接除以一个固定的奖励缩放系数。同时在优化器里加clipvalue=1.0或clipnorm=0.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好处是结构清晰，改起来方便，适合做这种增量调试。希望这次的拆解能帮你在自己的多智能体场景里少走几步弯路。
本文还有配套的精品资源，点击获取
