尧图网站建设 
简介一套基于Python的多智能体强化学习实现方案完整覆盖VDN、QMIX、QTRAN、QPLEX四种经典价值分解算法并配有对应模型文件。资源面向具备一定强化学习基础、正在完成课程设计或期末大作业的本科生与研究生也适合希望快速上手多智能体算法的研究者。工程实现了MultiAgentController智能体控制器、SMAC星际争霸环境封装、ReplayBuffer经验回放模块同时区分on-policy与off-policy数据使用方式便于理解不同算法的训练范式。压缩包共131个文件包含36个Python源码、29个npy与25个pkl模型权重、18张训练过程png图、4个PDF说明文档及TensorBoard事件文件整体约9.05MB结构清晰可直接运行或二次开发。目前已有342人学习下载借助预训练模型和可视化日志可快速对比四种算法在多智能体协作任务中的收敛速度与最终表现是开展实验和撰写报告的实用参考。1. 多智能体强化学习模板速览VDN到QPLEX的四代思路演进多智能体强化学习MARL在课程设计和实际项目里最容易卡住的地方不是算法原理看不懂而是没有一个能直接跑通、能改参、能出图的工程底子。这份源码给出了一个比较标准的MARL训练框架核心覆盖VDN、QMIX、QTRAN、QPLEX四种值分解算法配套了对应的模型文件和SMAC环境接口适合用来做期末大作业、课程设计或者是作为自己研究路线的起点。你可以把它理解成一个「算法全家桶」同一个环境、同一套buffer、同一套训练管线换一个mixing模块就换一种算法对比实验的对照组天然整齐。仓库里的核心模块拆得比较清楚MultiAgentController负责agent网络和动作生成SMAC负责环境交互ReplayBuffer负责数据存取三者解耦之后替换算法或者替换环境都不需要动主流程。下面按实际动手的顺序从环境封装讲到最后如何判断模型是否真正收敛。2. 环境接口与经验回放SMAC的API封装与ReplayBuffer的存储语义2.1 MultiAgentController统一的agent网络抽象在动手跑训练之前最好先把MultiAgentController这个类的作用弄清楚。它包含的是多智能体共享或独立的agent网络对值分解类算法来说这个网络输出的是每个智能体独立的Q值不包含mixing net也不包含critic net。采样的动作由它生成计算individual Q值也由它负责所以它本质上是一个「每个智能体一份Q网络」的管理器。值得注意的一个设计点是agent网络通常用DRQN结构带一个GRU层因为SMAC这类部分可观测环境里每个智能体只能看到自己的局部观测需要利用历史信息来推断真实状态。常见的做法是输入维度是obs_shape经过一个全连接层得到embedding再过一个GRU最后接一个全连接层输出动作空间维度的Q值。代码写出来大致是这个形状class DRQN(tf.keras.Model): def __init__(self, obs_shape, n_actions, hidden_dim64): super().__init__() self.fc1 tf.keras.layers.Dense(hidden_dim, activationrelu) self.gru tf.keras.layers.GRU(hidden_dim, return_sequencesTrue, return_stateTrue) self.fc2 tf.keras.layers.Dense(n_actions) def call(self, obs, hidden_state): x self.fc1(obs) x, new_hidden self.gru(x, initial_statehidden_state) q_values self.fc2(x) return q_values, new_hidden这里return_sequencesTrue是为了在训练时支持episode级别的反传return_stateTrue是为了在采样时拿到更新后的隐藏状态。实战中经常有人只把最后一帧的Q值拿来算loss这样梯度无法沿时间维反传效果会明显变差。MultiAgentController内部通常会维护一份hidden_state的字典按agent编号存储避免不同智能体之间的状态互相串扰。2.2 SMAC接口的三个核心方法SMAC是星际争霸多智能体环境也是这个领域最常用的benchmark之一。它的核心API实际上是围绕reset()、get_obs()、get_avail_actions()、step()这几个方法展开的。即便你打算在作业里换成自己的环境也建议按照SMAC提供的那套语义来做一层薄封装因为后续所有算法代码都会依赖这层接口。封装时最重要的是把每个step返回的数据整理成统一的字典格式。一般包含obs形状是(n_agents, obs_shape)state形状是(state_shape,)这是全局状态供mixing net和central critic使用actions形状是(n_agents,)reward是一个标量done布尔值以及avail_actions形状是(n_agents, n_actions)用于屏蔽非法动作。很多初学者会在avail_actions上踩坑——SMAC里有些动作是无效的比如攻击不存在的敌人如果不做mask网络会学到输出永远指向非法动作训练出来的策略根本没法部署。2.3 ReplayBuffer的两种存储语义transition与episodeReplayBuffer在MARL里的设计比单智能体要微妙一些。单智能体DQN只需要存transition也就是(s, a, r, s)四元组多智能体场景下VDN、QMIX、QTRAN、QPLEX都是off-policy算法按理说存transition就够了但因为agent网络带GRU如果要学时序特征sample的时候必须能恢复出完整的上文这时候就需要按episode粒度存储。两种方式各有适用场景。存transition的好处是sample简单、均匀随机即可坏处是GRU的初始hidden state不好处理通常会截断一段固定长度的子序列来近似这会牺牲一点信息。存episode的好处是样本是完整的轨迹训练时可以直接从头开始展开GRU得到每个时间步的hidden state坏处是buffer里存的是变长序列batch处理时需要做padding或者在sample时只取固定长度的episode。常见的做法是环境本身强制episode长度一致SMAC的episode长度在大部分地图下是固定的所以按episode存既不浪费也不复杂。class ReplayBuffer: def __init__(self, max_episodes, episode_limit): self.buffer [] self.max_episodes max_episodes self.episode_limit episode_limit def add_episode(self, episode): if len(self.buffer) self.max_episodes: self.buffer.pop(0) self.buffer.append(episode) def sample(self, batch_size): episodes random.sample(self.buffer, batch_size) obs_batch [] actions_batch [] rewards_batch [] dones_batch [] for ep in episodes: obs_batch.append(ep[obs]) actions_batch.append(ep[actions]) rewards_batch.append(ep[rewards]) dones_batch.append(ep[dones]) return (tf.stack(obs_batch), tf.stack(actions_batch), tf.stack(rewards_batch), tf.stack(dones_batch))这里add_episode是整个episode存进去而不是逐帧存。sample时按batch_size取若干个完整episode形状是(batch_size, episode_limit, n_agents, obs_shape)。要注意的是这个实现里episode_limit必须是一致的如果你的环境允许变长episode就需要用tf.ragged.stack或者在sample之后做动态padding。这是MARL里一个很典型的工程取舍算法论文里默认episode等长实际环境里往往不是需要自己在buffer层处理。on-policy和off-policy的区别在这里也体现得很直接。QMIX这类off-policy算法可以从一个旧的buffer里反复采样来训练只要行为策略的覆盖范围足够广数据可以重复利用而如果换成PPO、MAPPO这类on-policy算法每轮训练完必须清空buffer因为当前策略已经更新了旧数据不再符合重要性采样的前提。源码里这个buffer没有做清空动作说明它默认面向的是值分解算法这条路线。3. VDN到QMIX单调性约束与混合网络的实现3.1 VDN的加性分解逻辑VDNValue-Decomposition Networks是最早把「中心化训练、去中心化执行」落到值分解框架里的方法之一。它的核心假设非常朴素联合Q值可以拆成每个智能体Q值之和也就是Q_tot sum(Q_i)。这个假设没有引入任何额外参数实现上就是一个简单的sum操作所以VDN通常作为baseline的第一个对照算法。但VDN的问题也在于过于简单。它假设智能体之间对联合Q值的贡献是线性可加的这在实际战斗中往往不成立。比如两个智能体配合夹击一个敌人任何一个单独行动都拿不到击杀奖励但合在一起能拿到这种「超加性」的合作效应VDN的线性分解表达不了。这正好是后面QMIX要解决的问题。VDN的训练loss本身不复杂核心是把TD error算在Q_tot上再用梯度反传到每个agent的Q网络上def train_step_vdn(batch, q_net, optimizer, gamma0.99): obs batch[obs] # (batch, episode_limit, n_agents, obs_shape) actions batch[actions] rewards batch[rewards] # (batch, episode_limit, 1) dones batch[dones] with tf.GradientTape() as tape: q_values, _ q_net(obs) # (batch, episode_limit, n_agents, n_actions) batch_idx tf.range(tf.shape(obs)[0]) time_idx tf.range(tf.shape(obs)[1]) selected_q tf.gather_nd(q_values, tf.stack([ batch_idx[None, :, None] * tf.ones_like(actions), time_idx[None, :, None] * tf.ones_like(actions), actions ], axis-1)) # 按动作索引取Q值 q_tot tf.reduce_sum(selected_q, axis-1) # (batch, episode_limit) next_q_values, _ q_net(obs) next_q_tot tf.reduce_sum(tf.reduce_max(next_q_values, axis-1), axis-1) targets rewards[..., 0] gamma * (1 - dones[..., 0]) * next_q_tot loss tf.reduce_mean((tf.stop_gradient(targets) - q_tot) ** 2) grads tape.gradient(loss, q_net.trainable_variables) optimizer.apply_gradients(zip(grads, q_net.trainable_variables)) return loss这段代码里tf.gather_nd的索引构造比较绕它的作用是从q_values里按每个时间步每个智能体的实际动作取出对应的Q值。如果不做这一步梯度没办法沿着「被选中动作」这条路反传。next_q_tot用的是目标Q值的max——也就是假设每个智能体下一步都选自己Q最大的动作这个操作叫double sampling的简易版实际效果还行但如果想更稳可以再加一个target network来算next Q。3.2 QMIX的单调性约束如何解决非线性合作QMIX的动机是在保持VDN「每个智能体独立决策」的前提下把联合Q值的表达能力提上去。它的做法是用一个mixing network来拟合Q_tot输入是每个agent的Q值输出是联合Q值同时用一组hypernetwork来生成mixing network的权重条件输入是全局状态state。这看起来只是把线性加和换成了神经网络但真正的关键约束是单调性约束∂Q_tot / ∂Q_i 0。这个约束保证了对每个智能体来说如果它的Q_i变大Q_tot不会变小于是「每个智能体都选自己Q最大的动作」这个贪心策略在全局层面也是最优的。实现这个约束的标准做法是限制mixing network第一层和第二层的权重非负激活函数用绝对值。class QMixNet(tf.keras.Model): def __init__(self, n_agents, state_shape, hidden_dim32): super().__init__() self.n_agents n_agents self.hyper_w1 tf.keras.layers.Dense(hidden_dim, activationrelu) self.hyper_w2 tf.keras.layers.Dense(n_agents, activationrelu) self.b1 tf.keras.layers.Dense(hidden_dim, activationrelu) self.b2 tf.keras.layers.Dense(1) def call(self, q_vals, states): # q_vals: (batch, n_agents) w1 tf.abs(self.hyper_w1(states)) # (batch, hidden_dim) w1 tf.reshape(w1, (-1, self.n_agents, 1)) b1 self.b1(states) hidden tf.nn.elu(tf.matmul(q_vals[:, None, :], w1)[..., 0] b1) w2 tf.abs(self.hyper_w2(states)) w2 tf.reshape(w2, (-1, 1, self.n_agents)) b2 self.b2(states) q_tot tf.matmul(hidden[:, None, :], w2)[..., 0] b2 return q_tottf.abs在这里不是随便加的它是单调性约束的工程落点。去掉绝对值的话QMIX就退化成普通的mixing network不再有理论保证。训练时q_vals来自agent网络的Q值输出states是全局状态loss是(r gamma * max Q_tot_next - Q_tot)^2。这个loss直接从Q_tot层面算梯度通过mixing network反传到agent网络。3.3 两种算法在主循环里的切换方式因为VDN和QMIX的agent网络完全一样只是mixing部分不同切换算法只需要换一个mixing模块。源码里MultiAgentController只负责生成agent的Q值训练时把Q值传给不同的mixing层即可。这样可以避免为每个算法重复写一套环境采样和buffer逻辑。实际跑实验对比时我建议把两种mixing net都保留训练脚本里加一个--alg vdn|qmix参数就好。4. QTRAN与QPLEX从可证明分解到优势函数分解4.1 QTRAN的可行性条件与实现代价QMIX加了单调性约束换来的是实现简单和训练稳定但代价是这个约束本身在理论上限制了它能表达的函数类。QTRAN想解决的是「任意可分解的联合Q值函数都能被近似」这个更一般的问题。它不再约束Q_tot对Q_i的单调性而是把联合Q函数分解为Q_tot(s, a) sum(Q_i(s, a_i)) V(s)其中V(s)是状态相关的修正项用于捕捉多智能体之间的非线性协作。QTRAN的训练目标是让上述等式在最优动作上成立同时用不等式约束保证其它动作的Q值不越界。听起来很完备实际操作里QTRAN有两个明显的工程痛点。第一它需要额外的V(s)网络这个网络同样由全局状态输入但输出的修正值要同时平衡两类loss约束调参敏感度比QMIX高很多第二它的loss里有一个对所有动作的联合Q值约束项即Q_joint(s, a) sum(Q_i) V(s)这个约束在动作空间比较大的环境中计算量会爆炸。实践里多数实现会做一次采样近似也就是只用batch里出现过的动作来估计牺牲一部分理论保证换取可训练性。4.2 QPLEX的优势函数分解QPLEX可以看成QTRAN的实用化改造。它保留了「分解可行性」的目标但换了一个更巧妙的路径不是直接分解Q函数而是分解优势函数。定义A_i(s, a_i) Q_i(s, a_i) - V_i(s)QPLEX用Inequality Constraint把联合优势函数分解为A_tot sum(A_i) lambda * A_tot再通过mask机制确保在最优联合动作上分解误差为零。从代码实现角度看QPLEX的核心多了一个duplex dueling network结构。它要对每个agent计算对应的优势值并且用一个可学习的mask来约束联合动作的选择空间。这个mask的维度是(batch, n_agents, n_actions)理论上它充当了QTRAN中那个全局不等式约束的替代品。所以QPLEX在训练时的参数量明显比QMIX大相应的在SMAC的多数地图上它的收敛上限也会更高。4.3 实验配置一张地图上的四算法对照如何设计如果这是你的期末作业建议不要直接拿四套算法分别跑十个地图那样工作量太大而且很难看出对比规律。更合理的实验设计思路是选一张相对简单的对称地图比如3m和一张需要配合的地图比如2s3z或3s_vs_5z先跑通VDN和QMIX确认两个算法在3m上都能学到接近最优的策略再在这个基础上叠加QTRAN和QPLEX。超参数方面这份源码的惯例和大多数MARL开源项目一致lr通常取1e-4到5e-4之间batch_size取32或64buffer容量按episode数算一般不小于5000条episode。特别提醒一下QMIX和QPLEX的mixing网络对state的scale比较敏感如果训练曲线loss不降先看一眼state的数值范围必要时做一个running normalization。# 一个可参考的训练运行方式 python main.py --map 3m --alg qmix --batch_size 64 --lr 3e-4 \ --buffer_size 5000 --episode_limit 40 --n_steps 20000这里--n_steps对应的是一次训练过程中采样的总步数不是训练轮数。在SMAC环境里每一步是所有智能体同时执行一个动作所以n_steps20000意味着大概20000步环境交互。--episode_limit要和地图的实际步数上限对齐3m是40步2s3z是150步设小了会截断episode设大了buffer里会出现大量doneTrue的帧影响TD target的计算。5. 训练收敛性验证模型加载、评估指标与三个典型坑5.1 加载模型文件跑一次evaluation仓库里提供了对应的模型文件也就是那些.tfevents之外的checkpoint。加载模型做evaluation的常见做法是先重建agent网络和mixing网络再恢复tf.train.Checkpoint里的变量。注意如果训练时用了tf.function装饰训练step那么保存的checkpoint可能包含_tf_function相关的缓存变量恢复时需要先执行一次推理让变量格式对齐。# 恢复checkpoint并评测 checkpoint tf.train.Checkpoint(agentq_net, mixingmixing_net) checkpoint.restore(tf.train.latest_checkpoint(./models)) n_test_episodes 100 win_count 0 for _ in range(n_test_episodes): obs env.reset() hidden_state None done False while not done: q_vals, hidden_state q_net(obs[None, ...], hidden_state) actions tf.argmax(q_vals, axis-1).numpy()[0] obs, reward, done, info env.step(actions) if info[battle_won]: win_count 1 print(fWin rate: {win_count / n_test_episodes:.2f})这段代码里hidden_state要初始化为None让GRU从零状态开始。测试时动作选择直接argmax不做epsilon-greedy因为评估的是当前策略的确定性表现。info[battle_won]是SMAC自身返回的战斗结果标志如果换了自己的环境需要定义对应的胜负判断指标。5.2 训练曲线到底该看图里的哪一条很多人拿到这份源码第一件事是看TensorBoard里那些events.out.tfevents.*文件。这些文件是训练过程的原始事件日志可以用tensorboard --logdir runs直接打开。但要注意的是loss曲线并不能完整地反映求解质量。在MARL环境里reward的绝对数值和地图属性强相关3m这张图的理论最大reward可能是20而2s3z可能是50所以你没法横向对比不同地图的loss值。更实用的指标是TD-error的绝对值以及TD-error在训练后半段是否已经进入一个低方差区间。TD-error能反映当前Q函数和真实回报之间的拟合程度。如果TD-error稳定在一个很小的值附近震荡说明Q值的估计已经收敛这时再看胜率是否达标。如果胜率始终上不去但TD-error很小多半是探索策略出了问题典型表现是整个训练过程中每个episode都在重复同一套低效动作导致buffer里缺乏高质量的正样本。5.3 三个反复出现的坑第一个坑是avail_actions没有正确传入算法。VDN和QMIX在计算max Q_next时如果不对非法动作做mask网络会倾向于选择非法动作导致目标Q值被系统性高估。处理方式是在计算max Q_tot之前把非法动作位置的Q值设为一个很大的负数比如-1e9。第二个坑是episode长度不一致导致的batch维度爆炸。如果你的环境是变长的用tf.stack构造batch时会直接报错。可以先用tf.RaggedTensor存储在每个训练step里动态pad到batch内的最大长度同时在计算loss时用mask把padding位置的值屏蔽掉。第三个坑是on-policy和off-policy的数据混用。这份源码的buffer是off-policy语义如果你的作业里想对比MAPPO这类on-policy算法不能直接复用buffer否则策略更新之后旧数据已经失效训练会得到一组杂乱无章的reward曲线。常见做法是on-policy算法单独写一个rollout_buffer每个iteration采集一批数据用完即弃。5.4 一个实用调试技巧单agent单步冒烟测试在正式训练之前先做一个很小的冒烟测试把环境换成最简单的单步任务比如两个agent各选择一个动作给定一个固定的reward表然后跑10个batch的训练。这种方式可以快速验证agent网络、mixing网络和梯度反传链路是否完整。如果这个最简任务都学不到正确的Q值排序那大概率是前向传播的维度拼错了而不是算法本身的收敛问题。看训练loss之前先确保你的代码能在10秒钟之内把loss降下来多做这一步能省掉很多Debug时间。本文还有配套的精品资源点击获取 多智能体强化学习算法实现：从VDN到QPLEX的值分解演进
简介一套基于Python的多智能体强化学习实现方案完整覆盖VDN、QMIX、QTRAN、QPLEX四种经典价值分解算法并配有对应模型文件。资源面向具备一定强化学习基础、正在完成课程设计或期末大作业的本科生与研究生也适合希望快速上手多智能体算法的研究者。工程实现了MultiAgentController智能体控制器、SMAC星际争霸环境封装、ReplayBuffer经验回放模块同时区分on-policy与off-policy数据使用方式便于理解不同算法的训练范式。压缩包共131个文件包含36个Python源码、29个npy与25个pkl模型权重、18张训练过程png图、4个PDF说明文档及TensorBoard事件文件整体约9.05MB结构清晰可直接运行或二次开发。目前已有342人学习下载借助预训练模型和可视化日志可快速对比四种算法在多智能体协作任务中的收敛速度与最终表现是开展实验和撰写报告的实用参考。1. 多智能体强化学习模板速览VDN到QPLEX的四代思路演进多智能体强化学习MARL在课程设计和实际项目里最容易卡住的地方不是算法原理看不懂而是没有一个能直接跑通、能改参、能出图的工程底子。这份源码给出了一个比较标准的MARL训练框架核心覆盖VDN、QMIX、QTRAN、QPLEX四种值分解算法配套了对应的模型文件和SMAC环境接口适合用来做期末大作业、课程设计或者是作为自己研究路线的起点。你可以把它理解成一个「算法全家桶」同一个环境、同一套buffer、同一套训练管线换一个mixing模块就换一种算法对比实验的对照组天然整齐。仓库里的核心模块拆得比较清楚MultiAgentController负责agent网络和动作生成SMAC负责环境交互ReplayBuffer负责数据存取三者解耦之后替换算法或者替换环境都不需要动主流程。下面按实际动手的顺序从环境封装讲到最后如何判断模型是否真正收敛。2. 环境接口与经验回放SMAC的API封装与ReplayBuffer的存储语义2.1 MultiAgentController统一的agent网络抽象在动手跑训练之前最好先把MultiAgentController这个类的作用弄清楚。它包含的是多智能体共享或独立的agent网络对值分解类算法来说这个网络输出的是每个智能体独立的Q值不包含mixing net也不包含critic net。采样的动作由它生成计算individual Q值也由它负责所以它本质上是一个「每个智能体一份Q网络」的管理器。值得注意的一个设计点是agent网络通常用DRQN结构带一个GRU层因为SMAC这类部分可观测环境里每个智能体只能看到自己的局部观测需要利用历史信息来推断真实状态。常见的做法是输入维度是obs_shape经过一个全连接层得到embedding再过一个GRU最后接一个全连接层输出动作空间维度的Q值。代码写出来大致是这个形状class DRQN(tf.keras.Model): def __init__(self, obs_shape, n_actions, hidden_dim64): super().__init__() self.fc1 tf.keras.layers.Dense(hidden_dim, activationrelu) self.gru tf.keras.layers.GRU(hidden_dim, return_sequencesTrue, return_stateTrue) self.fc2 tf.keras.layers.Dense(n_actions) def call(self, obs, hidden_state): x self.fc1(obs) x, new_hidden self.gru(x, initial_statehidden_state) q_values self.fc2(x) return q_values, new_hidden这里return_sequencesTrue是为了在训练时支持episode级别的反传return_stateTrue是为了在采样时拿到更新后的隐藏状态。实战中经常有人只把最后一帧的Q值拿来算loss这样梯度无法沿时间维反传效果会明显变差。MultiAgentController内部通常会维护一份hidden_state的字典按agent编号存储避免不同智能体之间的状态互相串扰。2.2 SMAC接口的三个核心方法SMAC是星际争霸多智能体环境也是这个领域最常用的benchmark之一。它的核心API实际上是围绕reset()、get_obs()、get_avail_actions()、step()这几个方法展开的。即便你打算在作业里换成自己的环境也建议按照SMAC提供的那套语义来做一层薄封装因为后续所有算法代码都会依赖这层接口。封装时最重要的是把每个step返回的数据整理成统一的字典格式。一般包含obs形状是(n_agents, obs_shape)state形状是(state_shape,)这是全局状态供mixing net和central critic使用actions形状是(n_agents,)reward是一个标量done布尔值以及avail_actions形状是(n_agents, n_actions)用于屏蔽非法动作。很多初学者会在avail_actions上踩坑——SMAC里有些动作是无效的比如攻击不存在的敌人如果不做mask网络会学到输出永远指向非法动作训练出来的策略根本没法部署。2.3 ReplayBuffer的两种存储语义transition与episodeReplayBuffer在MARL里的设计比单智能体要微妙一些。单智能体DQN只需要存transition也就是(s, a, r, s)四元组多智能体场景下VDN、QMIX、QTRAN、QPLEX都是off-policy算法按理说存transition就够了但因为agent网络带GRU如果要学时序特征sample的时候必须能恢复出完整的上文这时候就需要按episode粒度存储。两种方式各有适用场景。存transition的好处是sample简单、均匀随机即可坏处是GRU的初始hidden state不好处理通常会截断一段固定长度的子序列来近似这会牺牲一点信息。存episode的好处是样本是完整的轨迹训练时可以直接从头开始展开GRU得到每个时间步的hidden state坏处是buffer里存的是变长序列batch处理时需要做padding或者在sample时只取固定长度的episode。常见的做法是环境本身强制episode长度一致SMAC的episode长度在大部分地图下是固定的所以按episode存既不浪费也不复杂。class ReplayBuffer: def __init__(self, max_episodes, episode_limit): self.buffer [] self.max_episodes max_episodes self.episode_limit episode_limit def add_episode(self, episode): if len(self.buffer) self.max_episodes: self.buffer.pop(0) self.buffer.append(episode) def sample(self, batch_size): episodes random.sample(self.buffer, batch_size) obs_batch [] actions_batch [] rewards_batch [] dones_batch [] for ep in episodes: obs_batch.append(ep[obs]) actions_batch.append(ep[actions]) rewards_batch.append(ep[rewards]) dones_batch.append(ep[dones]) return (tf.stack(obs_batch), tf.stack(actions_batch), tf.stack(rewards_batch), tf.stack(dones_batch))这里add_episode是整个episode存进去而不是逐帧存。sample时按batch_size取若干个完整episode形状是(batch_size, episode_limit, n_agents, obs_shape)。要注意的是这个实现里episode_limit必须是一致的如果你的环境允许变长episode就需要用tf.ragged.stack或者在sample之后做动态padding。这是MARL里一个很典型的工程取舍算法论文里默认episode等长实际环境里往往不是需要自己在buffer层处理。on-policy和off-policy的区别在这里也体现得很直接。QMIX这类off-policy算法可以从一个旧的buffer里反复采样来训练只要行为策略的覆盖范围足够广数据可以重复利用而如果换成PPO、MAPPO这类on-policy算法每轮训练完必须清空buffer因为当前策略已经更新了旧数据不再符合重要性采样的前提。源码里这个buffer没有做清空动作说明它默认面向的是值分解算法这条路线。3. VDN到QMIX单调性约束与混合网络的实现3.1 VDN的加性分解逻辑VDNValue-Decomposition Networks是最早把「中心化训练、去中心化执行」落到值分解框架里的方法之一。它的核心假设非常朴素联合Q值可以拆成每个智能体Q值之和也就是Q_tot sum(Q_i)。这个假设没有引入任何额外参数实现上就是一个简单的sum操作所以VDN通常作为baseline的第一个对照算法。但VDN的问题也在于过于简单。它假设智能体之间对联合Q值的贡献是线性可加的这在实际战斗中往往不成立。比如两个智能体配合夹击一个敌人任何一个单独行动都拿不到击杀奖励但合在一起能拿到这种「超加性」的合作效应VDN的线性分解表达不了。这正好是后面QMIX要解决的问题。VDN的训练loss本身不复杂核心是把TD error算在Q_tot上再用梯度反传到每个agent的Q网络上def train_step_vdn(batch, q_net, optimizer, gamma0.99): obs batch[obs] # (batch, episode_limit, n_agents, obs_shape) actions batch[actions] rewards batch[rewards] # (batch, episode_limit, 1) dones batch[dones] with tf.GradientTape() as tape: q_values, _ q_net(obs) # (batch, episode_limit, n_agents, n_actions) batch_idx tf.range(tf.shape(obs)[0]) time_idx tf.range(tf.shape(obs)[1]) selected_q tf.gather_nd(q_values, tf.stack([ batch_idx[None, :, None] * tf.ones_like(actions), time_idx[None, :, None] * tf.ones_like(actions), actions ], axis-1)) # 按动作索引取Q值 q_tot tf.reduce_sum(selected_q, axis-1) # (batch, episode_limit) next_q_values, _ q_net(obs) next_q_tot tf.reduce_sum(tf.reduce_max(next_q_values, axis-1), axis-1) targets rewards[..., 0] gamma * (1 - dones[..., 0]) * next_q_tot loss tf.reduce_mean((tf.stop_gradient(targets) - q_tot) ** 2) grads tape.gradient(loss, q_net.trainable_variables) optimizer.apply_gradients(zip(grads, q_net.trainable_variables)) return loss这段代码里tf.gather_nd的索引构造比较绕它的作用是从q_values里按每个时间步每个智能体的实际动作取出对应的Q值。如果不做这一步梯度没办法沿着「被选中动作」这条路反传。next_q_tot用的是目标Q值的max——也就是假设每个智能体下一步都选自己Q最大的动作这个操作叫double sampling的简易版实际效果还行但如果想更稳可以再加一个target network来算next Q。3.2 QMIX的单调性约束如何解决非线性合作QMIX的动机是在保持VDN「每个智能体独立决策」的前提下把联合Q值的表达能力提上去。它的做法是用一个mixing network来拟合Q_tot输入是每个agent的Q值输出是联合Q值同时用一组hypernetwork来生成mixing network的权重条件输入是全局状态state。这看起来只是把线性加和换成了神经网络但真正的关键约束是单调性约束∂Q_tot / ∂Q_i 0。这个约束保证了对每个智能体来说如果它的Q_i变大Q_tot不会变小于是「每个智能体都选自己Q最大的动作」这个贪心策略在全局层面也是最优的。实现这个约束的标准做法是限制mixing network第一层和第二层的权重非负激活函数用绝对值。class QMixNet(tf.keras.Model): def __init__(self, n_agents, state_shape, hidden_dim32): super().__init__() self.n_agents n_agents self.hyper_w1 tf.keras.layers.Dense(hidden_dim, activationrelu) self.hyper_w2 tf.keras.layers.Dense(n_agents, activationrelu) self.b1 tf.keras.layers.Dense(hidden_dim, activationrelu) self.b2 tf.keras.layers.Dense(1) def call(self, q_vals, states): # q_vals: (batch, n_agents) w1 tf.abs(self.hyper_w1(states)) # (batch, hidden_dim) w1 tf.reshape(w1, (-1, self.n_agents, 1)) b1 self.b1(states) hidden tf.nn.elu(tf.matmul(q_vals[:, None, :], w1)[..., 0] b1) w2 tf.abs(self.hyper_w2(states)) w2 tf.reshape(w2, (-1, 1, self.n_agents)) b2 self.b2(states) q_tot tf.matmul(hidden[:, None, :], w2)[..., 0] b2 return q_tottf.abs在这里不是随便加的它是单调性约束的工程落点。去掉绝对值的话QMIX就退化成普通的mixing network不再有理论保证。训练时q_vals来自agent网络的Q值输出states是全局状态loss是(r gamma * max Q_tot_next - Q_tot)^2。这个loss直接从Q_tot层面算梯度通过mixing network反传到agent网络。3.3 两种算法在主循环里的切换方式因为VDN和QMIX的agent网络完全一样只是mixing部分不同切换算法只需要换一个mixing模块。源码里MultiAgentController只负责生成agent的Q值训练时把Q值传给不同的mixing层即可。这样可以避免为每个算法重复写一套环境采样和buffer逻辑。实际跑实验对比时我建议把两种mixing net都保留训练脚本里加一个--alg vdn|qmix参数就好。4. QTRAN与QPLEX从可证明分解到优势函数分解4.1 QTRAN的可行性条件与实现代价QMIX加了单调性约束换来的是实现简单和训练稳定但代价是这个约束本身在理论上限制了它能表达的函数类。QTRAN想解决的是「任意可分解的联合Q值函数都能被近似」这个更一般的问题。它不再约束Q_tot对Q_i的单调性而是把联合Q函数分解为Q_tot(s, a) sum(Q_i(s, a_i)) V(s)其中V(s)是状态相关的修正项用于捕捉多智能体之间的非线性协作。QTRAN的训练目标是让上述等式在最优动作上成立同时用不等式约束保证其它动作的Q值不越界。听起来很完备实际操作里QTRAN有两个明显的工程痛点。第一它需要额外的V(s)网络这个网络同样由全局状态输入但输出的修正值要同时平衡两类loss约束调参敏感度比QMIX高很多第二它的loss里有一个对所有动作的联合Q值约束项即Q_joint(s, a) sum(Q_i) V(s)这个约束在动作空间比较大的环境中计算量会爆炸。实践里多数实现会做一次采样近似也就是只用batch里出现过的动作来估计牺牲一部分理论保证换取可训练性。4.2 QPLEX的优势函数分解QPLEX可以看成QTRAN的实用化改造。它保留了「分解可行性」的目标但换了一个更巧妙的路径不是直接分解Q函数而是分解优势函数。定义A_i(s, a_i) Q_i(s, a_i) - V_i(s)QPLEX用Inequality Constraint把联合优势函数分解为A_tot sum(A_i) lambda * A_tot再通过mask机制确保在最优联合动作上分解误差为零。从代码实现角度看QPLEX的核心多了一个duplex dueling network结构。它要对每个agent计算对应的优势值并且用一个可学习的mask来约束联合动作的选择空间。这个mask的维度是(batch, n_agents, n_actions)理论上它充当了QTRAN中那个全局不等式约束的替代品。所以QPLEX在训练时的参数量明显比QMIX大相应的在SMAC的多数地图上它的收敛上限也会更高。4.3 实验配置一张地图上的四算法对照如何设计如果这是你的期末作业建议不要直接拿四套算法分别跑十个地图那样工作量太大而且很难看出对比规律。更合理的实验设计思路是选一张相对简单的对称地图比如3m和一张需要配合的地图比如2s3z或3s_vs_5z先跑通VDN和QMIX确认两个算法在3m上都能学到接近最优的策略再在这个基础上叠加QTRAN和QPLEX。超参数方面这份源码的惯例和大多数MARL开源项目一致lr通常取1e-4到5e-4之间batch_size取32或64buffer容量按episode数算一般不小于5000条episode。特别提醒一下QMIX和QPLEX的mixing网络对state的scale比较敏感如果训练曲线loss不降先看一眼state的数值范围必要时做一个running normalization。# 一个可参考的训练运行方式 python main.py --map 3m --alg qmix --batch_size 64 --lr 3e-4 \ --buffer_size 5000 --episode_limit 40 --n_steps 20000这里--n_steps对应的是一次训练过程中采样的总步数不是训练轮数。在SMAC环境里每一步是所有智能体同时执行一个动作所以n_steps20000意味着大概20000步环境交互。--episode_limit要和地图的实际步数上限对齐3m是40步2s3z是150步设小了会截断episode设大了buffer里会出现大量doneTrue的帧影响TD target的计算。5. 训练收敛性验证模型加载、评估指标与三个典型坑5.1 加载模型文件跑一次evaluation仓库里提供了对应的模型文件也就是那些.tfevents之外的checkpoint。加载模型做evaluation的常见做法是先重建agent网络和mixing网络再恢复tf.train.Checkpoint里的变量。注意如果训练时用了tf.function装饰训练step那么保存的checkpoint可能包含_tf_function相关的缓存变量恢复时需要先执行一次推理让变量格式对齐。# 恢复checkpoint并评测 checkpoint tf.train.Checkpoint(agentq_net, mixingmixing_net) checkpoint.restore(tf.train.latest_checkpoint(./models)) n_test_episodes 100 win_count 0 for _ in range(n_test_episodes): obs env.reset() hidden_state None done False while not done: q_vals, hidden_state q_net(obs[None, ...], hidden_state) actions tf.argmax(q_vals, axis-1).numpy()[0] obs, reward, done, info env.step(actions) if info[battle_won]: win_count 1 print(fWin rate: {win_count / n_test_episodes:.2f})这段代码里hidden_state要初始化为None让GRU从零状态开始。测试时动作选择直接argmax不做epsilon-greedy因为评估的是当前策略的确定性表现。info[battle_won]是SMAC自身返回的战斗结果标志如果换了自己的环境需要定义对应的胜负判断指标。5.2 训练曲线到底该看图里的哪一条很多人拿到这份源码第一件事是看TensorBoard里那些events.out.tfevents.*文件。这些文件是训练过程的原始事件日志可以用tensorboard --logdir runs直接打开。但要注意的是loss曲线并不能完整地反映求解质量。在MARL环境里reward的绝对数值和地图属性强相关3m这张图的理论最大reward可能是20而2s3z可能是50所以你没法横向对比不同地图的loss值。更实用的指标是TD-error的绝对值以及TD-error在训练后半段是否已经进入一个低方差区间。TD-error能反映当前Q函数和真实回报之间的拟合程度。如果TD-error稳定在一个很小的值附近震荡说明Q值的估计已经收敛这时再看胜率是否达标。如果胜率始终上不去但TD-error很小多半是探索策略出了问题典型表现是整个训练过程中每个episode都在重复同一套低效动作导致buffer里缺乏高质量的正样本。5.3 三个反复出现的坑第一个坑是avail_actions没有正确传入算法。VDN和QMIX在计算max Q_next时如果不对非法动作做mask网络会倾向于选择非法动作导致目标Q值被系统性高估。处理方式是在计算max Q_tot之前把非法动作位置的Q值设为一个很大的负数比如-1e9。第二个坑是episode长度不一致导致的batch维度爆炸。如果你的环境是变长的用tf.stack构造batch时会直接报错。可以先用tf.RaggedTensor存储在每个训练step里动态pad到batch内的最大长度同时在计算loss时用mask把padding位置的值屏蔽掉。第三个坑是on-policy和off-policy的数据混用。这份源码的buffer是off-policy语义如果你的作业里想对比MAPPO这类on-policy算法不能直接复用buffer否则策略更新之后旧数据已经失效训练会得到一组杂乱无章的reward曲线。常见做法是on-policy算法单独写一个rollout_buffer每个iteration采集一批数据用完即弃。5.4 一个实用调试技巧单agent单步冒烟测试在正式训练之前先做一个很小的冒烟测试把环境换成最简单的单步任务比如两个agent各选择一个动作给定一个固定的reward表然后跑10个batch的训练。这种方式可以快速验证agent网络、mixing网络和梯度反传链路是否完整。如果这个最简任务都学不到正确的Q值排序那大概率是前向传播的维度拼错了而不是算法本身的收敛问题。看训练loss之前先确保你的代码能在10秒钟之内把loss降下来多做这一步能省掉很多Debug时间。本文还有配套的精品资源点击获取