多智能体协同

一句话概述

多智能体强化学习（MARL）研究多个智能体在共享环境中协同或竞争的学习问题。核心挑战包括非平稳性（每个智能体都在学习，环境从他人视角看是动态的）、信用分配（团队奖励如何分到个体）和维度爆炸（联合动作空间随智能体数量指数增长）。CTDE（集中训练、分布执行）范式是当前最主流的多智能体学习框架。

教学与演示

什么是多智能体强化学习

传统强化学习假设一个智能体在一个（从它的视角看）固定的环境中学习。但现实中很多问题涉及多个智能体——比如自动驾驶汽车之间互相避让、足球机器人配合进球、无人机编队飞行。

在多智能体系统中，每个智能体的行为会影响其他智能体的观测和奖励。当我方智能体在更新策略时，对方或同伴的策略也在变化，这违反了RL中「环境是平稳的」这一基本假设。

大白话 单人RL就像一个人在空房间里学走路——地板不会动。多智能体RL就像一群人同时在房间里跳舞——你每走一步，别人也在动，你永远不知道下一秒地板会变成什么样。

马尔可夫博弈

多智能体的数学框架是马尔可夫博弈（Markov Game），也叫随机博弈（Stochastic Game）。它是MDP向多智能体的自然推广。

一个N个智能体的马尔可夫博弈定义为：

其中每个智能体i有自己的动作空间A^i和奖励函数R^i。状态转移P依赖于所有智能体的联合动作。

根据奖励函数的关系，马尔可夫博弈分为三类：

undefined

大白话 马尔可夫博弈就是MDP的「多人联机版」——状态转移不再只依赖你一个人，而是所有人一起决定。奖励也变成了每人一份，有的共享（队友拿奖励你也拿），有的对冲（对手拿奖励你就亏）。

非平稳性问题

MARL的核心挑战是非平稳性（Non-stationarity）。当一个智能体更新策略时，它对其他智能体来说就像环境变了。

举个例子：你在和另一个RL智能体玩剪刀石头布。你的对手学到了你总出石头，于是它开始出布。这让你收到的奖励突然下降——不是因为你的策略变差了，而是因为对手的策略变好了。

从智能体i的视角看，转移概率 P(s'|s, a^i) 不是固定的——因为它依赖于其他智能体的策略π^(-i)，而这些策略在训练中不断变化。

主要解决方案：

undefined

大白话 非平稳性就是说——你以为你在玩超级玛丽，结果游戏规则时不时悄悄变化。你的策略刚学会跳过一个坑，结果那个坑的位置变了。这种感觉就是多智能体学习中的常态。

信用分配问题

在合作任务中，团队获得一个全局奖励。问题来了：这个奖励是谁的功劳？

如果一场足球赛赢了，每个队员都得到「赢球」这个奖励。但后卫防守很辛苦、前锋跑位很积极、守门员扑出关键球——他们的贡献值完全不同，却收到同样的奖励信号。这就是信用分配问题（Credit Assignment Problem）。

主要方法：

undefined

大白话 团队项目得了A+，每个人都沾光。但老师想知道谁最努力、谁划水——这就是信用分配。在RL里，如果所有人都收到相同的奖励信号，划水的人也会觉得自己做对了，根本学不到真正的技能。

集中训练、分布执行（CTDE）

CTDE（Centralized Training with Decentralized Execution）是当前最流行的多智能体训练范式。

核心思想很简单：训练时「开挂」——Critic能看到所有智能体的全局信息（所有观测、所有动作）；执行时回归现实——Actor只用自己的局部观测做决策。

这解决了两个矛盾：

undefined

大白话 训练时你是「上帝视角」，能看到所有人的位置和动作，用这个全局信息来评价每个人的决策好不好。真正上场时，你只能看到自己眼前的画面，但你的决策已经被上帝视角的训练「熏陶」过了。

什么用

多智能体协同的核心价值：

undefined

大白话 一只蚂蚁什么也做不了，一万只蚂蚁能搭桥过河。多智能体系统的魅力在于：局部简单规则 + 交互 = 全局智能行为。

哪些坑

核心代码演示

下面实现多智能体协同的三个核心组件：独立Q-Learning、联合动作空间和价值分解。

"""
独立Q-Learning (Independent Q-Learning) - 最简单的MARL方法
每个智能体独立学习，把其他智能体当作环境的一部分
"""
import numpy as np

# ===== 简单的多智能体网格世界 =====
class SimpleMARLEnv:
    """
    两个智能体在3x3网格中协作
    目标：两人同时到达目标位置才能获得奖励
    """
    def __init__(self):
        self.size = 3
        self.n_agents = 2
        self.n_actions = 4  # 上、下、左、右
        self.action_names = ['上', '下', '左', '右']
        self.goal = np.array([2, 2])  # 目标位置在右下角
        self.reset()
    
    def reset(self):
        # 智能体位置：智能体0在左上角，智能体1在左下角
        self.positions = np.array([[0, 0], [2, 0]])
        return self.positions.copy()
    
    def step(self, actions):
        """
        执行所有智能体的动作
        actions: list of int，每个智能体的动作 [a1, a2]
        """
        action_deltas = np.array([[-1, 0], [1, 0], [0, -1], [0, 1]])
        rewards = np.zeros(self.n_agents)
        
        for i in range(self.n_agents):
            delta = action_deltas[actions[i]]
            new_pos = self.positions[i] + delta
            # 边界检查
            new_pos = np.clip(new_pos, 0, self.size - 1)
            
            # 碰撞检查：两个智能体不能在同一位置
            other_agent = 1 - i  # 另一个智能体的索引
            if np.array_equal(new_pos, self.positions[other_agent]):
                new_pos = self.positions[i].copy()  # 被阻挡
            
            self.positions[i] = new_pos
        
        # 检查是否都到达目标
        both_at_goal = all(
            np.array_equal(self.positions[i], self.goal) for i in range(self.n_agents)
        )
        if both_at_goal:
            rewards = np.ones(2) * 10.0  # 团队奖励
            done = True
        else:
            rewards = np.ones(2) * -0.1  # 每步小惩罚
            done = False
        
        return self.positions.copy(), rewards, done
    
    def get_state_index(self):
        """将两个智能体的位置编码为状态索引"""
        return self.positions[0, 0] * 27 + self.positions[0, 1] * 9 + \
               self.positions[1, 0] * 3 + self.positions[1, 1]

# ===== 独立Q-Learning训练 =====
def train_independent_ql(env, n_episodes=500):
    """每个智能体维护自己的Q表，独立学习"""
    n_states = env.size ** 4  # 状态数 = 位置组合总数
    n_actions = env.n_actions
    
    # 每个智能体独立的Q表 [n_states, n_actions]
    Q_tables = [np.zeros((n_states, n_actions)) for _ in range(env.n_agents)]
    
    alpha = 0.1    # 学习率
    gamma = 0.95   # 折扣因子
    epsilon = 0.3  # 探索率
    rewards_history = []
    
    for ep in range(n_episodes):
        env.reset()
        state_idx = env.get_state_index()
        total_reward = 0
        done = False
        step = 0
        
        while not done and step < 50:
            actions = []
            for i in range(env.n_agents):
                # ε-greedy选择动作（每个智能体独立选择）
                if np.random.random() < epsilon:
                    actions.append(np.random.randint(n_actions))
                else:
                    actions.append(np.argmax(Q_tables[i][state_idx]))
            
            next_pos, rewards, done = env.step(actions)
            next_state_idx = env.get_state_index()
            total_reward += rewards[0]
            
            # 每个智能体独立更新自己的Q表
            for i in range(env.n_agents):
                best_next = np.max(Q_tables[i][next_state_idx])
                Q_tables[i][state_idx, actions[i]] += alpha * (
                    rewards[i] + gamma * best_next * (1 - done) -
                    Q_tables[i][state_idx, actions[i]]
                )
            
            state_idx = next_state_idx
            step += 1
        
        rewards_history.append(total_reward)
        
        if ep % 100 == 0:
            avg = np.mean(rewards_history[-100:])
            print(f"Episode {ep:3d} | 平均奖励: {avg:.2f}")
    
    return rewards_history

env = SimpleMARLEnv()
history = train_independent_ql(env, n_episodes=500)
print(f"最终平均奖励: {np.mean(history[-50:]):.2f}")
print("独立Q-Learning: 简单但有非平稳性问题")

"""
联合动作空间与维度爆炸演示
展示多智能体动作组合随智能体数量指数增长的问题
"""
import numpy as np

def demonstrate_joint_action_space(n_agents, n_actions_per_agent):
    """
    演示联合动作空间的指数增长
    
    如果N个智能体各有M个可选动作：
    联合动作数 = M^N
    """
    joint_actions = n_actions_per_agent ** n_agents
    print(f"智能体数量: {n_agents}")
    print(f"每智能体动作数: {n_actions_per_agent}")
    print(f"联合动作数: {joint_actions:,}")
    
    # 估算存储一个Q表所需内存
    memory_per_entry = 4  # float32 = 4 bytes
    total_memory = joint_actions * memory_per_entry
    if total_memory > 1e9:
        print(f"Q表内存: {total_memory / 1e9:.1f} GB — 太大了！")
    else:
        print(f"Q表内存: {total_memory / 1e6:.1f} MB")
    return joint_actions

print("=== 联合动作空间随智能体数量增长 ===\n")
n_actions = 5  # 移动方向: 上下左右+停留
for n in [2, 3, 4, 5, 6, 8, 10]:
    demonstrate_joint_action_space(n, n_actions)
    print()

print("结论：4个智能体（625种联合动作）已经很大，")
print("10个智能体（~9.7M种组合）直接存储Q表不可行。")
print("这就是需要值分解（VDN/QMIX）和集中Critic的原因！")

# ===== 解决方案：值分解 =====
print("\n=== 值分解（Value Decomposition）演示 ===")

def vdn_decompose(global_q, individual_qs):
    """
    VDN：全局Q等于个体Q之和
    Q_tot(s, a) = Σ_i Q_i(s_i, a_i)
    
    假设误差反向传播梯度时：∂Q_tot/∂Q_i = 1
    意味着每个智能体的Q贡献梯度权重相同
    """
    return np.sum(individual_qs)

def qmix_decompose(global_q, individual_qs, mixing_weights):
    """
    QMIX：带权重的单调分解
    Q_tot(s, a) = f(Q_1, Q_2, ..., Q_N | s) 
    
    其中f是一个单调递增函数（权重≥0），由超网络从全局状态生成
    这比VDN更灵活，但仍保证 ∂Q_tot/∂Q_i ≥ 0（单调性约束）
    """
    # 简化演示：带权重的和（实际QMIX用超网络产生权重）
    weighted_sum = np.sum(individual_qs * mixing_weights)
    # 加一个全局偏置
    bias = 0.0  # 实际QMIX中由超网络生成
    return weighted_sum + bias

# 模拟5个智能体的个体Q值
n_agents = 5
individual_qs = np.array([2.0, 1.5, 0.5, 3.0, 1.0])  # 各智能体的Q值
print(f"个体Q值: {individual_qs}")

# VDN求和
global_q_vdn = vdn_decompose(None, individual_qs)
print(f"VDN全局Q (和): {global_q_vdn:.1f}")

# QMIX加权和
mixing_weights = np.array([1.2, 0.8, 1.0, 1.5, 0.5])  # 超网络输出
global_q_qmix = qmix_decompose(None, individual_qs, mixing_weights)
print(f"QMIX全局Q (加权和): {global_q_qmix:.1f}")

print("\nVDN/QMIX思想：将全局Q分解为个体Q的组合")
print("优势：只需学习N×M个Q值，而非M^N个！")

"""
CTDE (Centralized Training Decentralized Execution) 框架
训练时Critic用全局信息，执行时Actor只用局部观测
"""
import numpy as np

class CTDEAgent:
    """
    CTDE智能体：独立的Actor（分散执行）和Critic（集中训练）
    """
    def __init__(self, agent_id, obs_dim, n_actions, global_dim):
        self.id = agent_id
        self.obs_dim = obs_dim
        self.n_actions = n_actions
        self.global_dim = global_dim
        
        # Actor：只用局部观测 → 输出动作概率
        # 简化：用线性模型表示
        self.actor_weights = np.random.randn(obs_dim, n_actions) * 0.1
        
        # Critic：用全局信息（所有智能体的观测+动作）→ 估计Q值
        self.critic_weights = np.random.randn(global_dim) * 0.1
        
    def get_action(self, obs, epsilon=0.1):
        """分散执行：Actor只用自己的局部观测"""
        logits = obs @ self.actor_weights
        probs = np.exp(logits - np.max(logits))
        probs = probs / np.sum(probs)
        
        if np.random.random() < epsilon:
            return np.random.randint(self.n_actions)
        return np.argmax(probs)
    
    def get_q_value(self, global_state):
        """集中训练：Critic用全局信息评估动作"""
        return np.dot(global_state, self.critic_weights)

# ===== 模拟CTDE训练流程 =====
n_agents = 3
obs_dim = 5      # 每个智能体的局部观测维度
n_actions = 4    # 每智能体的动作数
global_dim = obs_dim * n_agents + n_agents  # 全局状态 = 所有观测 + 所有动作

# 创建智能体
agents = [CTDEAgent(i, obs_dim, n_actions, global_dim) 
          for i in range(n_agents)]

# 训练阶段（集中Critic）
print("=== 训练阶段（集中）===")
print("1. 收集所有智能体的局部观测和动作")
print("2. 拼接为全局状态 [obs1, obs2, obs3, a1, a2, a3]")
print("3. 用全局Critic评估Q值")
print("4. 反向传播更新Actor和Critic")

# 执行阶段（分散Actor）
print("\n=== 执行阶段（分散）===")
local_obs = np.random.randn(obs_dim)  # 智能体0的局部观测
action = agents[0].get_action(local_obs, epsilon=0)
print(f"智能体0只用自己的观测 {local_obs[:2]}... 选择动作 {action}")
print("无需全局信息，无需通信！")

print("\nCTDE精髓：训练时开上帝视角，执行时回归现实")

概念关系图谱

重点答疑

大白话 多智能体强化学习最难的地方不是「怎么让一个好智能体做得更好」，而是「怎么让一群同时学习的智能体别互相踩脚」。CTDE就像安排了一个总教练——训练时能看到全场，执教每个人；比赛时队员们各自用自己的判断力。

💡 核心要点：多智能体系统的三个核心挑战——非平稳性（环境在变）、信用分配（功劳归谁）、维度爆炸（组合太多）——CTDE通过集中Critic+分散Actor一揽子解决了这三个问题。

undefined

章节单词汇总