训练分布

一句话概述

CTDE（集中训练分布执行）是多智能体强化学习的核心范式。本讲深入剖析CTDE的具体实现方案：值分解方法（VDN/QMIX）将全局Q函数分解为个体Q的组合，策略梯度方法（MADDPG/COMA）使用集中Critic为分散Actor提供训练信号，反事实基线实现精确的个体信用分配。

教学与演示

CTDE范式回顾

CTDE解决了多智能体RL中一个根本矛盾：训练需要全局信息，执行只能用局部信息。

undefined

CTDE不是单一算法，而是一类算法的设计范式。根据Critic的工作方式，可以分为两大流派：

流派一：值分解（Value Decomposition）

undefined

流派二：集中Critic（Centralized Critic）

undefined

大白话 CTDE的两大流派就像两种不同的「教练」：值分解派教练说「每个人做好自己就行，加起来就是团队分数」；集中Critic派教练说「我要看全局动态来告诉你该怎么改进」。

VDN：加性值分解

VDN（Value Decomposition Networks）是最简的值分解方法，由DeepMind在2017年提出。它的核心假设极其简单：

其中每个智能体i的个体Q_i只依赖于自己的局部观测s_i和动作a_i，全局Q_tot是所有个体Q的简单算术和。

训练方式：用标准的DQN损失训练，但梯度会自动分配到各个智能体：

因为∂Q_tot/∂Q_i = 1，所以全局TD误差的梯度「平均」分配给每个智能体。这意味着每个智能体得到的梯度信号完全相同——VDN没有区分「谁贡献大、谁贡献小」。

执行方式：每个智能体独立做greedy选择：

由于Q_tot是个体Q的简单求和，各自最大化等价于联合最大化——这是VDN的「IGM原则」（Individual-Global-Max）的自然满足。

大白话 VDN就像团队按人头平分奖金——不管谁功劳大，每个人拿一样多。虽然不公平，但在简单任务中效果还不错，因为大致正确的信号总比没信号好。

QMIX：单调值分解

QMIX（2018）发现VDN的加性约束太强了——不是所有问题中全局Q都等于个体Q的简单相加。例如，只有在两个智能体都做出正确动作时（「与」逻辑），全局Q才高；任何一人出错就全盘皆输。加性分解无法刻画这种「协同效应」。

QMIX引入了一个混合网络（Mixing Network） 来实现更灵活的分解：

其中f_mix是一个超网络（Hypernetwork），输入全局状态s，输出混合网络的权重和偏置。关键约束：混合网络的权重必须始终保持非负，从而保证：

单调性确保：提高任何个体Q_i绝对不会降低全局Q_tot。这意味着分散执行时的argmax操作仍然有效——每个智能体选择最大化自身Q_i的动作，不会损害全局利益。

大白话 QMIX把VDN的「固定总和」升级为「加权组合」，权重由全局状态动态决定——就像教练根据场上局势动态调整「谁承担更多责任」。但QMIX不能表示「非单调」的信用分配（比如一个人出错反而需要另一人加倍努力补偿）。这也是为什么后来有了QPLEX等非单调扩展。

MADDPG：多智能体DDPG

MADDPG（Multi-Agent Deep Deterministic Policy Gradient）采取完全不同的路线：为每个智能体配备一个集中Critic。

这个Critic接收所有智能体的观测和动作作为输入，输出智能体i的Q值。换句话说，每个智能体都有一个「全知」的Critic来告诉自己该动作有多好。

MADDPG的Actor更新使用DDPG风格的策略梯度：

关键技巧：推断其他智能体策略。MADDPG还维护了对其他智能体策略的估计（通过学习预测他们的动作），这有助于在训练时模拟他们的行为，减轻非平稳性。

大白话 MADDPG给每个智能体配了一个「上帝视角教练」：这个教练知道所有人的位置、动作和策略，然后告诉智能体「在当前局面下你该怎么做」。训练时的「开挂」让执行时即使没有全局信息，策略也已经很优秀了。

COMA：反事实信用分配

COMA（Counterfactual Multi-Agent Policy Gradients）解决了VDN/QMIX的一个盲点：它们不显式区分个体贡献，梯度只是隐式分配的。

COMA的核心是一个反事实优势函数：

这个公式的含义是：智能体i的「优势」= 实际Q值 - 所有可能动作Q值的期望（保持其他智能体的动作不变）。如果A^i > 0，说明智能体i选择的动作比「平均」好——贡献为正；如果A^i < 0，说明比平均差——拖后腿了。

这提供了精确的「个体贡献」信号。

大白话 COMA就像比赛结束后教练逐帧回放：假如你还是你，但队友都保持不变，你选择另一个动作会怎样？如果实际动作的得分更高，说明你做对了；反之说明你做错了。这种「反事实思维」让你清楚地知道自己贡献了多少。

什么用

CTDE方法和值分解的实际应用：

undefined

大白话 CTDE在多智能体领域的地位就像ResNet在图像分类——不太可能是最终答案，但它是当前最可靠、最通用的基线框架。

哪些坑

核心代码演示

"""
VDN和QMIX的值分解实现对比
展示从加性分解到超网络单调分解的演变
"""
import numpy as np

# ===== 1. VDN：加性值分解 =====
class VDN:
    """
    VDN：Q_tot = Σ_i Q_i(s_i, a_i)
    最简值分解，适用于回报是线性可加的任务
    """
    def __init__(self, n_agents):
        self.n_agents = n_agents
    
    def compute_q_tot(self, individual_qs):
        """
        计算全局Q值
        
        参数：
        - individual_qs: 每个智能体的个体Q值 [n_agents]
        """
        return np.sum(individual_qs)
    
    def compute_gradient(self, td_error):
        """
        全局TD误差的梯度分配
        ∂L/∂Q_i = ∂L/∂Q_tot * ∂Q_tot/∂Q_i = TD_error * 1
        每个智能体收到相同的梯度！
        """
        return np.ones(self.n_agents) * td_error

# ===== 2. QMIX：超网络单调分解 =====
class QMIX:
    """
    QMIX：Q_tot = f(Q_1, ..., Q_N | s)，f是非负权重的单调函数
    混合网络的权重和偏置由超网络从全局状态生成
    """
    def __init__(self, n_agents, state_dim, mixing_dim=32):
        self.n_agents = n_agents
        self.state_dim = state_dim
        self.mixing_dim = mixing_dim
        
        # 超网络：全局状态 → 混合网络权重
        # W1: [mixing_dim, n_agents], b1: [mixing_dim], W2: [1, mixing_dim], b2: [1]
        self.hyper_w1 = np.random.randn(state_dim, mixing_dim * n_agents) * 0.01
        self.hyper_b1 = np.random.randn(state_dim, mixing_dim) * 0.01
        self.hyper_w2 = np.random.randn(state_dim, mixing_dim) * 0.01
        self.hyper_b2 = np.random.randn(state_dim, 1) * 0.01
    
    def _abs(self, x):
        """绝对值操作，确保权重非负（单调性约束）"""
        return np.abs(x)
    
    def compute_q_tot(self, individual_qs, global_state):
        """
        QMIX的前向传播
        1. 超网络从全局状态生成非负权重
        2. 混合网络对个体Q做加权组合
        
        参数：
        - individual_qs: 个体Q值 [n_agents]
        - global_state: 全局状态 [state_dim]
        """
        n_agents = self.n_agents
        
        # Layer 1: 超网络生成第一层权重和偏置
        w1_full = global_state @ self.hyper_w1  # [mixing_dim * n_agents]
        w1 = self._abs(w1_full.reshape(self.mixing_dim, n_agents))  # 非负约束
        b1 = global_state @ self.hyper_b1  # [mixing_dim]
        
        # 混合网络第一层：h = ELU(W1 * Q_individual + b1)
        # 使用ELU保持非线性（在负数区域平滑）
        h = w1 @ individual_qs + b1  # [mixing_dim]
        h = np.maximum(h, 0) + np.minimum(0, np.exp(h) - 1)  # ELU
        
        # Layer 2: 超网络生成第二层权重和偏置
        w2 = self._abs(global_state @ self.hyper_w2)  # [mixing_dim] -> 非负
        b2 = global_state @ self.hyper_b2  # [1]
        
        # 混合网络第二层：Q_tot = W2 * h + b2
        q_tot = np.dot(w2, h) + b2  # 标量
        
        return q_tot.item()
    
    def check_monotonicity(self, individual_qs, global_state):
        """
        验证单调性：∂Q_tot/∂Q_i ≥ 0 for all i
        即：提高任何个体Q不应该降低全局Q
        """
        eps = 1e-6
        q_base = self.compute_q_tot(individual_qs, global_state)
        derivatives = np.zeros(self.n_agents)
        
        for i in range(self.n_agents):
            qs_perturbed = individual_qs.copy()
            qs_perturbed[i] += eps
            q_new = self.compute_q_tot(qs_perturbed, global_state)
            derivatives[i] = (q_new - q_base) / eps
        
        return derivatives

# ===== 演示 =====
np.random.seed(42)
n_agents, state_dim = 5, 10

vdn = VDN(n_agents)
qmix = QMIX(n_agents, state_dim)

# 模拟个体Q值和全局状态
individual_qs = np.array([3.0, 1.5, 2.5, 0.8, 4.0])
global_state = np.random.randn(state_dim)

# VDN
q_tot_vdn = vdn.compute_q_tot(individual_qs)
print(f"VDN全局Q: {q_tot_vdn:.2f} (等于个体Q之和: {np.sum(individual_qs):.2f})")

# QMIX
q_tot_qmix = qmix.compute_q_tot(individual_qs, global_state)
print(f"QMIX全局Q: {q_tot_qmix:.2f} (非简单求和，由全局状态动态加权)")

# 验证单调性
derivatives = qmix.check_monotonicity(individual_qs, global_state)
print(f"\n∂Q_tot/∂Q_i (单调性验证): {derivatives}")
print(f"所有导数 >= 0: {np.all(derivatives >= 0)}")
print("单调性确保了分散执行时各自贪心不会损害全局利益！")

"""
MADDPG的核心：集中Critic + 分散Actor
每个智能体维护对其他智能体策略的估计
"""
import numpy as np

class MADDPGAgent:
    """MADDPG智能体：集中Critic评估，分散Actor执行"""
    def __init__(self, agent_id, obs_dim, n_actions, 
                 global_obs_dim, global_act_dim):
        self.id = agent_id
        self.obs_dim = obs_dim
        self.n_actions = n_actions
        
        # Actor: 只用局部观测 → 输出动作
        self.actor = np.random.randn(obs_dim, n_actions) * 0.1
        
        # 集中Critic: 全局观测+全局动作 → Q值
        critic_input_dim = global_obs_dim + global_act_dim
        self.critic = np.random.randn(critic_input_dim, 1) * 0.1
        
        # 对其他智能体策略的估计模型（减轻非平稳性）
        # 用其他智能体的观测预测他们的动作
        self.policy_estimators = {}  # agent_id -> 预测模型
    
    def act(self, obs):
        """分散执行：只用局部观测"""
        logits = obs @ self.actor
        probs = np.exp(logits - np.max(logits))
        return probs / np.sum(probs)
    
    def q_value(self, global_input):
        """集中Critic评估"""
        return np.dot(global_input, self.critic).item()
    
    def estimate_others_actions(self, others_obs, other_ids):
        """推断其他智能体的策略（用于训练时模拟他们的行为）"""
        estimated = {}
        for oid in other_ids:
            if oid in self.policy_estimators:
                estimator = self.policy_estimators[oid]
                estimated[oid] = np.argmax(others_obs @ estimator)
            else:
                estimated[oid] = np.random.randint(self.n_actions)
        return estimated

# ===== MADDPG训练流程演示 =====
n_agents, obs_dim, n_actions = 3, 5, 4
global_obs_dim = obs_dim * n_agents
global_act_dim = n_actions * n_agents

# 创建智能体
agents = []
for i in range(n_agents):
    agent = MADDPGAgent(i, obs_dim, n_actions, 
                         global_obs_dim, global_act_dim)
    agents.append(agent)

# 模拟一轮训练
print("=== MADDPG训练流程 ===\n")
# 每个智能体产生观测和动作
observations = [np.random.randn(obs_dim) for _ in range(n_agents)]
actions = [np.argmax(agents[i].act(observations[i])) for i in range(n_agents)]

# 构建全局输入（集中Critic）
global_obs = np.concatenate(observations)
global_act_onehot = np.eye(n_actions)[actions].flatten()
global_input = np.concatenate([global_obs, global_act_onehot])

# 每个智能体的集中Critic评估
for i in range(n_agents):
    q_val = agents[i].q_value(global_input)
    print(f"智能体{i}的集中Critic Q值: {q_val:.4f}")

print(f"\n关键洞察：")
print(f"1. 每个智能体的Critic都看到了完整的全局状态和全局动作")
print(f"2. Actor只用局部观测，执行时不需要全局信息")
print(f"3. 策略估计器帮助模拟其他智能体的行为变化")

"""
COMA反事实基线计算
精确评估每个智能体的个体贡献
"""
import numpy as np

def compute_coma_advantage(agent_id, actions_taken, all_q_values,
                            policy_probs, n_actions):
    """
    计算COMA反事实优势
    
    A^i(s, a) = Q(s, a) - Σ π^i(a'|o_i) * Q(s, (a', a^{-i}))
    
    参数：
    - agent_id: 当前智能体ID
    - actions_taken: 所有智能体实际执行的动作 [n_agents]
    - all_q_values: Q(s, 所有可能的联合动作组合)
    - policy_probs: 智能体i的策略概率 π^i(·|o_i) [n_actions]
    - n_actions: 动作空间大小
    
    返回：
    - advantage: 反事实优势值
    """
    # 实际执行的联合动作的Q值
    actual_q = all_q_values[tuple(actions_taken)]
    
    # 反事实基线：保持其他智能体动作不变，对智能体i的所有动作求期望
    counterfactual_q = 0.0
    other_actions = list(actions_taken)
    for a in range(n_actions):
        other_actions[agent_id] = a  # 替换智能体i的动作
        q_for_a = all_q_values[tuple(other_actions)]
        counterfactual_q += policy_probs[a] * q_for_a
    
    # 反事实优势 = 实际Q - 反事实基线
    advantage = actual_q - counterfactual_q
    return advantage

# ===== 演示COMA在一个简单场景 =====
np.random.seed(42)
n_agents, n_actions = 2, 3  # 2个智能体，各3个动作

# 智能体0的策略：均匀分布
policy_0 = np.array([1/3, 1/3, 1/3])

# 模拟Q值表（仅用于演示，实际中由神经网络估计）
# Q[s, a0, a1] 三维数组
q_values = np.zeros((n_actions, n_actions))
q_values[0, 0] = 5.0  # 两人都选0 → 高Q值
q_values[0, 1] = 2.0
q_values[0, 2] = 1.0
q_values[1, 0] = 3.0
q_values[1, 1] = 4.0
q_values[1, 2] = 2.0
q_values[2, 0] = 1.0
q_values[2, 1] = 2.0
q_values[2, 2] = 3.0

# 场景：智能体0选了动作0，智能体1选了动作0
actions_taken = [0, 0]
actual_q = q_values[0, 0]  # 实际Q = 5.0

# COMA计算智能体0的优势
adv_0 = compute_coma_advantage(
    0, actions_taken, q_values, policy_0, n_actions
)

# 验证：反事实基线 = (5.0 + 3.0 + 1.0) / 3 = 3.0
# COMA优势 = 5.0 - 3.0 = 2.0（正，说明智能体0贡献好）
print(f"实际Q值: {actual_q}")
print(f"反事实基线期望: {(5.0+3.0+1.0)/3:.1f}")
print(f"COMA优势（智能体0）: {adv_0:.1f}")
print(f"解读：智能体0的动作(0)比平均水平好 {adv_0:.1f}")

# 如果智能体0选了动作2（差动作）
actions_taken_bad = [2, 0]
adv_0_bad = compute_coma_advantage(
    0, actions_taken_bad, q_values, policy_0, n_actions
)
print(f"\n如果智能体0选动作2（差动作）:")
print(f"COMA优势: {adv_0_bad:.1f}（负值表示拖后腿了）")
print("\nCOMA提供了精确的个体贡献信号，远超简单的平均分配！")

概念关系图谱

重点答疑

大白话 CTDE就像一个开挂的游戏教练：训练时能看小地图、知道每个人的操作和策略，然后针对性地指导你；上场时你只能凭自己的屏幕做判断，但训练时的指导已经内化成你的本能了。

💡 核心要点：值分解（VDN/QMIX）和集中Critic（MADDPG/COMA）是CTDE的两条腿。值分解靠「分解全局Q为个体Q之和」隐式分配信用，集中Critic靠「全局信息直接评价动作」给予精确反馈。QMIX的单调性约束虽有限制，但在大多数实际任务中已足够。

undefined

章节单词汇总