1. 从单智能体到多智能体为什么我要啃MADDPG这块硬骨头最开始接触强化学习的时候我和大多数人一样从Q-Learning、DQN这些经典算法入手跑个CartPole、打砖块什么的感觉还挺有成就感。但很快我就发现一个问题这些算法解决的都是单智能体在静态环境中的决策问题。一旦场景里出现多个需要同时决策的智能体整个问题的复杂度就完全不一样了。举个生活化的例子。单智能体强化学习就像你一个人开车路况虽然复杂但至少其他车不会根据你的驾驶策略来调整自己的行为。而多智能体强化学习就像你在一个没有交通规则的十字路口周围每辆车都在实时判断其他车的意图并做出反应。这时候环境是非平稳的——你的策略一变其他智能体的最优策略也跟着变传统的Q-Learning直接歇菜。MADDPGMulti-Agent Deep Deterministic Policy Gradient就是专门为解决这类问题设计的。它基于Actor-Critic框架采用集中式训练、分布式执行的思路让每个智能体在训练时能获取全局信息执行时只依赖局部观测。这个思路在理论上很优雅但真正落地的时候坑不少。我选择用百度飞桨的PaddlePaddle和PARL框架来实现原因有几个一是PARL对多智能体场景有比较成熟的抽象不用自己从头搭训练循环二是飞桨的静态图动态图混合编程模式在调试的时候很方便三是中文文档相对友好遇到问题查资料的成本低。这篇文章我会把整个实现过程拆开揉碎讲清楚包括环境搭建、算法原理、代码实现、调参经验、踩坑记录适合有一定强化学习基础、想往多智能体方向进阶的朋友参考。2. MADDPG算法核心原理拆解2.1 为什么传统Actor-Critic在多智能体场景下会失效要理解MADDPG的设计动机得先搞清楚传统方法在多智能体环境里到底遇到了什么问题。假设有N个智能体每个智能体的策略参数为θ_i那么智能体i的期望回报J(θ_i)依赖于所有智能体的策略。当其他智能体的策略发生变化时智能体i的最优策略也会改变。这就导致经验回放池中的数据变得过时——你之前存进去的(s, a, r, s)四元组是在其他智能体旧策略下产生的用这些数据来训练当前策略梯度方向可能完全是错的。更具体地说在单智能体场景下环境转移概率P(s|s,a)是固定的所以Q函数的贝尔曼更新是稳定的。但在多智能体场景下从智能体i的视角看环境转移变成了P(s|s, a_1, a_2, ..., a_N)其中a_j (j≠i)是由其他智能体的策略决定的。这些策略在训练过程中不断变化导致从智能体i的角度看环境是非平稳的。我打个比方你在跟一群人玩石头剪刀布但规则是每个人可以随时改变自己的出拳策略。你之前总结的对手喜欢出石头这个经验可能下一秒就失效了。这就是非平稳性带来的核心困难。2.2 集中式训练分布式执行的核心思想MADDPG的核心创新点在于Critic网络能看到全局信息。具体来说每个智能体i有一个Actor网络μ_i(o_i; θ_i)只接收自己的局部观测o_i输出动作a_i每个智能体i有一个Critic网络Q_i(x, a_1, ..., a_N; φ_i)接收全局状态x和所有智能体的联合动作输出Q值训练时Critic用全局信息来评估当前联合动作的好坏指导Actor更新执行时每个Actor只需要自己的局部观测就能输出动作不需要其他智能体的信息这个设计的精妙之处在于训练时利用全局信息解决了非平稳性问题执行时又保持了分布式决策的灵活性。Critic知道其他智能体的策略所以能准确评估当前动作的价值不会因为其他智能体策略变化而导致Q值估计失准。用前面的十字路口类比训练阶段相当于每个司机都有一个上帝视角的教练能看到所有车的位置和意图告诉司机你刚才那个决策在全局来看是好是坏。但真正上路开车时司机只需要看自己周围的状况就能做出判断不需要知道所有车的完整信息。2.3 算法流程与关键公式MADDPG的完整训练流程可以拆成以下几个步骤对于每个智能体i从回放池中采样一批经验(x, a_1, ..., a_N, r_1, ..., r_N, x)然后Critic更新最小化以下损失函数L(φ_i) E[(Q_i(x, a_1, ..., a_N) - y_i)^2]其中目标值 y_i r_i γ * Q_i(x, a_1, ..., a_N)|_{a_jμ_j(o_j)}注意这里的目标动作a_j来自目标Actor网络μ_j而不是当前Actor网络。这是DDPG中target network的标准做法用来稳定训练。Actor更新最大化以下目标函数J(θ_i) E[Q_i(x, a_1, ..., a_{i-1}, μ_i(o_i), a_{i1}, ..., a_N)]梯度为 ∇θ_i J E[∇θ_i μ_i(o_i) * ∇a_i Q_i(x, a_1, ..., a_N)|_{a_iμ_i(o_i)}]目标网络软更新θ_i ← τθ_i (1-τ)θ_i φ_i ← τφ_i (1-τ)φ_i这里的τ通常取0.01表示每次只更新一点点保持目标网络的稳定性。注意MADDPG的Critic输入维度会随着智能体数量线性增长。如果有N个智能体每个动作维度为d那么Critic的输入维度就是状态维度 N*d。当智能体数量较多时Critic网络会变得很大训练效率会明显下降。这是MADDPG的一个固有局限。3. 基于PARL框架的代码实现全流程3.1 环境准备与依赖安装先把基础环境搭好。我用的配置是Python 3.8 PaddlePaddle 2.4 PARL 2.1这个组合实测下来比较稳定。pip install paddlepaddle2.4.0 pip install parl2.1.0 pip install gym0.21.0 pip install numpy1.21.0这里有个坑要提醒gym的版本不要装太新的0.26之后的版本API改动比较大PARL里的一些示例代码可能跑不通。我一开始装了gym 0.26结果环境reset的返回值从obs变成了(obs, info)导致一堆代码报错。后来降到0.21就顺利了。另外如果你有GPU的话建议装paddlepaddle-gpu版本多智能体训练的计算量比单智能体大很多用CPU跑会非常慢。我实测下来同样的MADDPG场景GPU版本比CPU版本快大概8到10倍。3.2 多智能体环境封装PARL对多智能体环境的封装思路是每个智能体有自己的观测空间和动作空间环境负责协调所有智能体的交互。我以一个简化的多智能体追逃场景为例来说明。import gym import numpy as np from gym import spaces class MultiAgentEnv(gym.Env): def __init__(self, n_agents3, n_opponents2): self.n_agents n_agents self.n_opponents n_opponents self.n n_agents n_opponents # 每个智能体的观测自身位置速度其他智能体相对位置 self.obs_dim 4 2 * (self.n - 1) # 动作二维连续力 self.act_dim 2 self.observation_space spaces.Box( low-np.inf, highnp.inf, shape(self.obs_dim,), dtypenp.float32 ) self.action_space spaces.Box( low-1.0, high1.0, shape(self.act_dim,), dtypenp.float32 ) def reset(self): # 随机初始化所有智能体位置 self.agent_pos np.random.uniform(-1, 1, (self.n, 2)) self.agent_vel np.zeros((self.n, 2)) return self._get_obs() def _get_obs(self): obs [] for i in range(self.n_agents): # 自身状态 self_state np.concatenate([ self.agent_pos[i], self.agent_vel[i] ]) # 其他智能体相对位置 others [] for j in range(self.n): if j ! i: rel_pos self.agent_pos[j] - self.agent_pos[i] others.append(rel_pos) obs.append(np.concatenate([self_state] others)) return np.array(obs, dtypenp.float32) def step(self, actions): # actions shape: (n_agents, act_dim) for i in range(self.n_agents): self.agent_vel[i] actions[i] * 0.1 self.agent_vel[i] np.clip(self.agent_vel[i], -1, 1) self.agent_pos[i] self.agent_vel[i] * 0.1 obs self._get_obs() rewards self._compute_reward() done False info {} return obs, rewards, done, info def _compute_reward(self): # 简化的奖励函数追逃距离 rewards np.zeros(self.n_agents) for i in range(self.n_agents): for j in range(self.n_agents, self.n): dist np.linalg.norm( self.agent_pos[i] - self.agent_pos[j] ) rewards[i] - dist return rewards这个环境封装有几个关键点需要注意观测设计每个智能体的观测只包含自己的状态和其他智能体的相对位置不包含全局绝对坐标。这样做是为了模拟部分可观测的场景也更符合实际应用中的情况。动作空间用连续的二维力向量范围限制在[-1, 1]。连续动作空间是MADDPG的适用场景如果是离散动作应该考虑用MADDPG的离散版本或者其他算法。奖励设计这里用了最简单的距离负奖励。实际项目中奖励函数的设计往往是最耗时间的部分需要反复调整才能让智能体学到期望的行为。3.3 Actor和Critic网络定义在PARL框架下我们需要分别定义Actor和Critic的Model类。import paddle import paddle.nn as nn import parl from parl import layers class ActorModel(parl.Model): def __init__(self, obs_dim, act_dim, hidden_dim64): super().__init__() 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 paddle.nn.functional.relu(self.fc1(obs)) x paddle.nn.functional.relu(self.fc2(x)) # tanh将输出限制在[-1, 1] return paddle.tanh(self.fc3(x)) class CriticModel(parl.Model): def __init__(self, global_obs_dim, global_act_dim, hidden_dim64): super().__init__() self.fc1 nn.Linear( global_obs_dim global_act_dim, hidden_dim ) self.fc2 nn.Linear(hidden_dim, hidden_dim) self.fc3 nn.Linear(hidden_dim, 1) def forward(self, global_obs, global_act): x paddle.concat([global_obs, global_act], axis-1) x paddle.nn.functional.relu(self.fc1(x)) x paddle.nn.functional.relu(self.fc2(x)) return self.fc3(x)这里有个细节值得展开说Actor的输出层用tanh激活函数把动作限制在[-1, 1]范围内。这是因为我们的动作空间定义就是[-1, 1]。如果你的动作空间范围不同需要相应地调整缩放。比如动作范围是[-2, 2]那就在tanh输出后乘以2。Critic的输入是全局观测和全局动作的拼接。全局观测维度 n_agents * obs_dim全局动作维度 n_agents * act_dim。当智能体数量增加时这个输入维度会快速增长。我试过5个智能体的场景Critic的输入维度就到了512 52 70维网络参数量明显增加训练时间也相应变长。3.4 MADDPG Agent的实现接下来是核心的Agent类它封装了每个智能体的Actor、Critic以及目标网络。class MADDPGAgent(parl.Agent): def __init__(self, algorithm, obs_dim, act_dim, global_obs_dim, global_act_dim): super().__init__(algorithm) self.obs_dim obs_dim self.act_dim act_dim self.global_obs_dim global_obs_dim self.global_act_dim global_act_dim def build_model(self): self.model ActorModel(self.obs_dim, self.act_dim) self.target_model ActorModel(self.obs_dim, self.act_dim) self.target_model.set_state_dict(self.model.state_dict()) def predict(self, obs): obs paddle.to_tensor(obs, dtypefloat32) if len(obs.shape) 1: obs obs.unsqueeze(0) action self.model(obs) return action.numpy() def learn(self, obs, global_obs, global_act, rewards, next_obs, next_global_obs, terminals, other_agents_actions): # 调用algorithm的learn方法 return self.algorithm.learn( obs, global_obs, global_act, rewards, next_obs, next_global_obs, terminals, other_agents_actions )PARL的Agent抽象层帮我们处理了模型构建、参数同步这些琐碎的事情。但多智能体场景下有个特殊之处每个智能体的Critic需要知道所有智能体的动作。所以在learn的时候需要把其他智能体的动作也传进来。3.5 训练循环与经验回放训练循环是整个实现中最容易出问题的部分。多智能体场景下经验回放池的管理比单智能体复杂得多。class ReplayBuffer: def __init__(self, capacity, n_agents, obs_dim, act_dim): self.capacity capacity self.n_agents n_agents self.obs_dim obs_dim self.act_dim act_dim self.ptr 0 self.size 0 self.obs np.zeros((capacity, n_agents, obs_dim)) self.acts np.zeros((capacity, n_agents, act_dim)) self.rewards np.zeros((capacity, n_agents)) self.next_obs np.zeros((capacity, n_agents, obs_dim)) self.terminals np.zeros((capacity, n_agents)) def push(self, obs, acts, rewards, next_obs, terminals): self.obs[self.ptr] obs self.acts[self.ptr] acts self.rewards[self.ptr] rewards self.next_obs[self.ptr] next_obs self.terminals[self.ptr] terminals self.ptr (self.ptr 1) % self.capacity self.size min(self.size 1, self.capacity) def sample(self, batch_size): idxs np.random.randint(0, self.size, sizebatch_size) return ( self.obs[idxs], self.acts[idxs], self.rewards[idxs], self.next_obs[idxs], self.terminals[idxs] )训练主循环的逻辑def train(): env MultiAgentEnv(n_agents3, n_opponents2) n_agents 3 obs_dim env.obs_dim act_dim env.act_dim global_obs_dim obs_dim * (n_agents 2) global_act_dim act_dim * (n_agents 2) # 为每个智能体创建Agent agents [] for i in range(n_agents): model ActorModel(obs_dim, act_dim) critic CriticModel(global_obs_dim, global_act_dim) algorithm MADDPG(model, critic, ...) agent MADDPGAgent(algorithm, obs_dim, act_dim, global_obs_dim, global_act_dim) agents.append(agent) buffer ReplayBuffer(100000, n_agents, obs_dim, act_dim) for episode in range(5000): obs env.reset() episode_reward np.zeros(n_agents) for step in range(100): # 每个智能体根据自己的观测选择动作 actions [] for i, agent in enumerate(agents): action agent.predict(obs[i]) # 添加探索噪声 noise np.random.normal(0, 0.1, act_dim) action np.clip(action noise, -1, 1) actions.append(action[0]) actions np.array(actions) next_obs, rewards, done, _ env.step(actions) buffer.push(obs, actions, rewards, next_obs, np.zeros(n_agents)) obs next_obs episode_reward rewards # 每步都更新 if buffer.size 1000: batch buffer.sample(256) for i, agent in enumerate(agents): agent.learn(*prepare_batch(batch, i)) if episode % 100 0: print(fEpisode {episode}, fReward: {episode_reward.mean():.2f})注意探索噪声的方差需要随着训练逐渐衰减。我一开始用固定的0.1结果训练后期智能体一直在抖动收敛不了。后来改成从0.3线性衰减到0.01效果明显改善。4. 调参实战那些文档里不会写的经验4.1 学习率与网络结构的配合MADDPG对学习率非常敏感。我试过几组不同的配置总结下来配置Actor学习率Critic学习率效果A1e-41e-3收敛慢但稳定B1e-31e-3震荡严重不收敛C1e-41e-4收敛极慢D3e-41e-3收敛快且稳定最终我选了配置DActor用3e-4Critic用1e-3。Critic的学习率通常要比Actor大因为Critic需要先准确评估当前策略的价值Actor才能根据准确的梯度方向更新。如果Critic还没学好Actor就急着更新很容易跑偏。网络结构方面我一开始用了128-128的两层隐藏层结果参数量太大训练很慢。后来改成64-64效果差不多但速度快了近一倍。对于中小规模的多智能体场景64维的隐藏层基本够用了。4.2 软更新系数τ的选择τ控制目标网络的更新速度。τ越大目标网络跟得越紧但训练越不稳定τ越小目标网络越稳定但学习速度越慢。我实测下来τ0.01是一个比较平衡的值。试过τ0.1训练曲线抖动明显τ0.001收敛速度慢得让人着急。如果你发现训练过程中Q值一直在震荡可以试着把τ调小一点。4.3 经验回放池大小的权衡回放池太小数据多样性不够容易过拟合回放池太大旧数据太多与当前策略分布不匹配。我的经验是回放池大小至少是batch_size的100倍。比如batch_size256回放池至少25600。我一般设100000对于大多数中小规模场景够用了。另外多智能体场景下有个特殊问题不同智能体的经验是耦合的。你不能只存单个智能体的(s, a, r, s)必须存所有智能体的联合经验。因为Critic需要全局信息来评估。这也是为什么上面的ReplayBuffer要存(n_agents, obs_dim)这样的结构。4.4 奖励尺度的影响奖励的尺度对训练影响很大。如果奖励值范围在[-100, 100]梯度会很大容易导致训练不稳定如果范围在[-0.01, 0.01]梯度太小学习速度极慢。我一般会把奖励归一化到[-1, 1]或者[-10, 10]这个量级。具体做法是先跑几个episode统计奖励的均值和标准差然后用( r - mean ) / std来归一化。这个操作看起来简单但对训练稳定性的提升非常明显。5. 常见问题与排查技巧实录5.1 训练不收敛的排查思路多智能体训练不收敛是最常见的问题。我总结了一个排查清单症状可能原因排查方法奖励一直不上升学习率太大降低学习率10倍试试奖励震荡剧烈探索噪声太大衰减噪声方差Q值爆炸奖励尺度太大归一化奖励智能体行为单一探索不足增大初始噪声训练后期性能下降过拟合增大回放池减小网络我遇到过一次很诡异的情况训练前1000个episode奖励稳步上升然后突然崩掉奖励直接掉到初始水平。排查了很久才发现是目标网络更新出了问题。当时我用的τ0.1目标网络跟得太快导致目标Q值不稳定最终引发了训练崩溃。把τ降到0.01后就再也没出现过这个问题。5.2 智能体之间串通或摆烂的处理多智能体训练中经常出现两种极端要么所有智能体学出一样的策略串通要么某些智能体完全放弃摆烂。串通问题通常是因为Critic的全局信息太强导致每个智能体都能准确预测其他智能体的行为最终收敛到同一个策略。解决办法是给Critic的输入加一点噪声或者限制Critic能看到的信息范围。摆烂问题通常是因为奖励分配不均。某些智能体的奖励一直很低梯度更新方向不明确慢慢就放弃了。解决办法是调整奖励函数给每个智能体设置独立的、可达成的子目标。5.3 计算资源不足时的优化策略多智能体训练对计算资源的需求是单智能体的N倍N为智能体数量。如果资源有限可以考虑以下优化共享Critic网络如果所有智能体的观测和动作空间相同可以共享一个Critic网络参数量减少N倍降低更新频率不必每步都更新可以每10步更新一次减小batch_size从256降到64显存占用减少4倍使用混合精度训练飞桨支持AMP可以显著减少显存占用我试过共享Critic的方案在智能体同质的场景下效果几乎不损失但训练速度提升了近3倍。不过如果智能体异质比如追逃场景中追者和逃者共享Critic就不太合适了。5.4 从仿真到实际部署的注意事项仿真环境训练好的策略直接部署到实际系统中往往会性能骤降。原因主要有两个一是仿真与现实的观测分布不一致二是实际系统中的延迟和噪声。我的经验是在仿真训练时就要加入噪声和延迟。具体做法是在观测上叠加高斯噪声在动作执行上加入随机延迟。这样训练出来的策略鲁棒性会好很多。另外部署前一定要做域随机化把仿真环境中的物理参数如质量、摩擦系数随机化让策略适应参数变化。6. 多智能体强化学习的扩展方向MADDPG虽然经典但也有一些固有局限。比如Critic输入维度随智能体数量线性增长当智能体数量超过10个时训练会变得非常困难。针对这个问题后续有很多改进工作比如用注意力机制来压缩全局信息或者用图神经网络来建模智能体之间的交互关系。另外MADDPG假设所有智能体都能获取全局状态信息来训练Critic。但在很多实际场景中全局状态是不可观测的。这时候可以考虑用通信学习的方法让智能体之间学习交换什么信息来辅助训练。我在实际项目中还发现一个有趣的现象智能体数量增加时每个智能体的策略反而更容易收敛。这听起来反直觉但仔细想想也合理——智能体多了环境中的交互更丰富每个智能体收到的梯度信号更多样反而有助于跳出局部最优。当然这个现象只在智能体数量适中3到8个时成立再多就会因为维度灾难而适得其反。最后分享一个我在调试MADDPG时的小技巧先把智能体数量设为2把算法跑通再逐步增加。两个智能体的场景最简单容易定位问题。等两个智能体能稳定收敛了再扩展到3个、5个。这样排查问题的成本会低很多。我一开始就设了5个智能体结果调了两周都没收敛后来降到2个一天就跑通了然后再逐步加回去整个过程顺畅多了。