MoE 起源：1991 年 Adaptive Mixtures of Local Experts 论文解读

对于公式 (2) 求偏导：

$ \frac{\partial E^{c}}{\partial o_{i}^{c}}=-2 p_{i}^{c}\left(d^{c}-o_{i}^{c}\right) $

在这个新的误差函数中，一个局部专家的目标不会直接受到其他局部专家权重的影响。

仍然存在一些间接的影响，因为如果其他某个专家改变其权重，这可能会导致门控网络改变分配给各个专家的权重，但至少这些权重变化不会改变某个局部专家在给定训练样本上所感知到的误差。

如果门控网络和局部专家都通过梯度下降法针对这个新的误差函数进行训练，系统往往会为每个训练样本分配一个单独的专家。

只要某个专家给出的误差小于所有专家误差的加权平均值（利用门控网络的输出来决定如何对每个专家的误差进行加权），它对该样本的权重会增加，否则权重会降低。

尽管如此，这种方法也有局限：

• 冷启动问题：初期所有专家表现差，门控网络难以做出好选择；
• 局部最优：某个专家偶然表现好，就垄断资源，其他专家无法成长；
• 方差大：因为输出是随机的，训练可能不稳定；
• 非确定性输出：同一输入多次前向传播可能得到不同结果。

软竞争

问题设定

• 输入：样本特征 $x$
• 真实目标：连续值输出 $d^c \in \mathbb{R}^D$，表示第 $c$ 个任务或模态下的目标向量（例如在多任务学习或条件生成中）
• 模型结构：Mixture of Experts（MoE），包含多个'专家'网络和一个门控（gating）网络
• 专家输出：第 $i$ 个专家对第 $c$ 个任务的预测输出为 $o_i^c \in \mathbb{R}^D$
• 门控权重：门控网络输出的归一化权重 $p_i^c = \text{Gate}(x)_i$，满足 $\sum_i p_i^c = 1$ 且 $p_i^c \in (0,1)$，表示第 $i$ 个专家在任务 $c$ 下的置信度或责任分配

模型预测分布建模

在 MoE 架构中，最终的预测不是单一专家的输出，而是多个专家输出的加权组合。但不同于简单的加权平均，这里通过概率视角建模整体输出的似然。

假设模型对目标 $d^c$ 的预测服从一个由专家输出构成的高斯混合分布（Gaussian Mixture Model, GMM）：

$ P(d^c | x) = \sum_{i} p_i^c \cdot \mathcal{N}(d^c \mid o_i^c, I) $

其中：

• $ \mathcal{N}(d^c \mid o_i^c, I) \propto \exp\left(-\frac{1}{2} |d^c - o_i^c|^2\right) $ 是以第 $i$ 个专家输出为中心的标准高斯核
• $ p_i^c $ 是门控网络提供的先验权重（即第 $i$ 个专家被选中的概率）

因此，整体似然为各专家生成目标 $d^c$ 的加权概率密度之和。

对数似然函数

取该混合分布的对数似然：

$ \log P(d^c | x) = \log \left( \sum_{i} p_i^c \cdot \frac{1}{(2\pi)^{D/2}} \exp\left(-\frac{1}{2} |d^c - o_i^c|^2\right) \right) $

忽略常数项（如 $(2\pi)^{D/2}$，因为它们不依赖于模型参数），可得有效对数似然为：

$ \log P(d^c | x) \propto \log \left( \sum_{i} p_i^c \exp\left(-\frac{1}{2} |d^c - o_i^c|^2\right) \right) $

我们希望最大化这个对数似然，使真实目标 $d^c$ 在模型预测的混合分布下具有高概率。

损失函数（Negative Log-Likelihood）

等价地，定义单个任务/样本的损失函数为负对数似然：

$ E^{c} = - \log \left( \sum_{i} p_{i}^{c} \exp\left(-\frac{1}{2} | d^{c} - o_{i}^{c} |^2 \right) \right) \quad \tag{3} $

这就是 MoE 架构中常用的 混合专家负对数似然损失（MoE NLL Loss）。

• $ \exp\left(-\frac{1}{2} |d^c - o_i^c|^2\right) $：衡量第 $i$ 个专家输出 $o_i^c$ 与真实目标 $d^c$ 的相似性（越接近，值越大）
• $ p_i^c $：门控网络赋予该专家的权重，反映其'被信任程度'
• 整个求和项是加权相似性总和，代表整体模型解释数据的能力
• 取负对数后，变成最小化目标：当某个 $p_i^c$ 较大且对应 $o_i^c$ 接近 $d^c$ 时，损失较小；否则损失较大

✅ 该损失自动鼓励两种机制：专家准确性：每个专家应尽量准确预测其负责的数据；门控合理性：门控网络应将高权重分配给那些能更好拟合当前输入的专家

整个训练集的损失

对于 $N$ 个样本（或 $N$ 个任务实例），总损失为所有任务/样本上的平均损失：

$ J = \frac{1}{N} \sum_{c=1}^N E^c = -\frac{1}{N} \sum_{c=1}^N \log \left( \sum_{i} p_{i}^{c} \exp\left(-\frac{1}{2} | d^{c} - o_{i}^{c} |^2 \right) \right) $

该损失可用于端到端训练 MoE 模型中的专家网络和门控网络。

为什么 (3) 比 (2) 更好？

| 特性 | Eq (2): $ \sum p_i |d - o_i|^2 $ | Eq (3): $ -\log \sum p_i e^{-\frac{1}{2}|d - o_i|^2} $ | | — | — | — | | 是否基于概率模型 | 否 | ✅ 是 | | 是否抑制大误差影响 | 否 | ✅ 是（指数衰减） | | 是否包含正则化 | 否 | ✅ 是（KL 项） | | 是否接近'选择最佳专家' | 线性加权 | ✅ 非线性聚合，类似 soft-min | | 训练稳定性 | 一般 | ✅ 更好 |

实验复现

实验代码

使用多层感知机作为专家网络，简单的前馈网络作为 Gating Network 复现了 MoE，并且根据公式 1、2、3 定义了损失函数，在数据集上合并了 MNIST 和 Fashion MNIST 两个数据集，验证模型在处理 MNIST 和 Fashion MNIST 两种看起来差异较大的任务时，看 Gating Network 是否会自动分配不同的专家。

import torch
import torch.nn as nn
import torch.optim as optim
import torchvision
import torchvision.transforms as transforms
import matplotlib.pyplot as plt
import numpy as np
import torch.nn.functional as F
from torch.utils.data import DataLoader, ConcatDataset
from tqdm import tqdm

# 定义单个专家的神经网络
class SingleExpert(nn.Module):
    def __init__(self, input_dim=28*28, output_dim=20):
        super(SingleExpert, self).__init__()
        self.fc1 = nn.Linear(input_dim, 256)
        self.fc2 = nn.Linear(256, 128)
        self.fc3 = nn.Linear(128, output_dim)

    def forward(self, x):
        x = x.view(-1, 28*28)
        x = torch.relu(self.fc1(x))
        x = torch.relu(self.fc2(x))
        x = self.fc3(x)
        return x

# 定义 MoE 模型
class MoE(nn.Module):
    def __init__(self, input_dim=28*28, output_dim=20, num_experts=4):
        super(MoE, self).__init__()
        self.num_experts = num_experts
        # 定义专家
        self.experts = nn.ModuleList([SingleExpert(input_dim, output_dim) for _ in range(num_experts)])
        # 定义门网络（决定使用哪个专家）
        self.gating_network = nn.Linear(input_dim, num_experts)

    def forward(self, x):
        x_flat = x.view(-1, 28*28)
        # 获取门网络的概率
        gate_outputs = torch.softmax(self.gating_network(x_flat), dim=1)
        # Shape: [batch_size, num_experts]
        # 获取每个专家的输出
        expert_outputs = torch.stack([expert(x) for expert in self.experts], dim=1)
        # Shape: [batch_size, num_experts, output_dim]
        # 随机选择一个专家基于门网络的概率
        expert_indices = torch.multinomial(gate_outputs, num_samples=1).squeeze()
        # Shape: [batch_size]
        # 收集对应于采样索引的专家输出
        final_output = expert_outputs[torch.arange(x.size(0)), expert_indices]
        # Shape: [batch_size, output_dim]
        return final_output, expert_outputs, gate_outputs, expert_indices

# 定义 MoE 模型的损失函数
def moe_loss(targets, expert_outputs, gate_outputs):
    """
    计算 MoE 模型的损失。
    Arguments:
    - targets: 真实输出向量，形状 [batch_size, output_dim]
    - expert_outputs: 每个专家的输出向量，形状 [batch_size, num_experts, output_dim]
    - gate_outputs: 门网络的输出（每个专家的概率），形状 [batch_size, num_experts]
    Returns:
    - loss: 计算的损失值
    """
    # 计算目标输出和每个专家输出之间的平方误差
    errors = torch.sum((expert_outputs - targets.unsqueeze(1))**2, dim=2)
    # Shape: [batch_size, num_experts]
    # 计算负半误差的指数（高斯似然）
    weighted_errors = torch.exp(-0.5 * errors)
    # Shape: [batch_size, num_experts]
    # 通过门网络的输出（概率）加权误差
    weighted_errors = gate_outputs * weighted_errors
    # Shape: [batch_size, num_experts]
    # 对专家求和并取对数以获得负对数似然
    loss = -torch.log(torch.sum(weighted_errors, dim=1) + 1e-8)
    # Shape: [batch_size]
    # 返回批次的平均损失
    return loss.mean()

# 定义 MoE 模型的均方误差硬损失
def mse_hard_loss(targets, expert_outputs, gate_outputs):
    """
    计算 MoE 模型的均方误差硬损失。
    Arguments:
    - targets: 真实输出向量，形状 [batch_size, output_dim]
    - expert_outputs: 每个专家的输出向量，形状 [batch_size, num_experts, output_dim]
    - gate_outputs: 门网络的输出（每个专家的概率），形状 [batch_size, num_experts]
    Returns:
    - loss: 计算的损失值
    """
    # Step 1: 使用门控权重对专家输出进行加权求和（融合输出）
    fused_output = torch.sum(gate_outputs.unsqueeze(-1) * expert_outputs, dim=1)
    # [batch_size, output_dim]
    # Step 2: 计算融合输出与目标之间的 L2 距离平方
    reconstruction_error = torch.sum((fused_output - targets)**2, dim=1)
    # [batch_size]
    # Step 3: 取 batch 平均作为最终损失
    loss = torch.mean(reconstruction_error)
    # 返回批次的平均损失
    return loss

# 定义 MoE 模型的均方误差软损失
def mse_soft_loss(targets, expert_outputs, gate_outputs):
    """
    计算 MoE 模型的均方误差损失。
    Arguments:
    - targets: 真实输出向量，形状 [batch_size, output_dim]
    - expert_outputs: 每个专家的输出向量，形状 [batch_size, num_experts, output_dim]
    - gate_outputs: 门网络的输出（每个专家的概率），形状 [batch_size, num_experts]
    Returns:
    - loss: 计算的损失值
    """
    errors = torch.sum((expert_outputs - targets.unsqueeze(1))**2, dim=2)
    # Shape: [batch_size, output_dim]
    weighted_errors = gate_outputs * errors
    # Shape: [batch_size, output_dim]
    # 对专家求和并取平均以获得均方误差
    loss = torch.mean(weighted_errors)
    # 返回批次的平均损失
    return loss

def one_hot_encoding(labels, num_classes=10):
    # 确保标签矩阵在与标签相同的设备上创建
    return torch.eye(num_classes, device=labels.device)[labels]

# 训练循环，跟踪专家选择
def train(model, dataloader, optimizer, num_epochs=10, loss_name='moe'):
    model.train()
    # 初始化专家选择计数器
    expert_selection_count = torch.zeros(model.num_experts, device=device)
    for epoch in range(num_epochs):
        running_loss = 0.0
        correct = 0
        total = 0
        # 使用 tqdm 显示进度
        for inputs, labels in tqdm(dataloader):
            inputs, labels = inputs.to(device), labels.to(device)
            optimizer.zero_grad()
            # 前向传播
            final_output, expert_outputs, gate_outputs, expert_indices = model(inputs)
            # 统计专家选择次数
            for idx in expert_indices:
                expert_selection_count[idx] += 1
            # 将标签转换为独热编码
            one_hot_labels = one_hot_encoding(labels, num_classes=len(combined_classes))
            # 计算 MoE 损失
            if loss_name == 'moe':
                loss = moe_loss(one_hot_labels, expert_outputs, gate_outputs)
            elif loss_name == 'mse_hard':
                loss = mse_hard_loss(one_hot_labels, expert_outputs, gate_outputs)
            elif loss_name == 'mse_soft':
                loss = mse_soft_loss(one_hot_labels, expert_outputs, gate_outputs)
            loss.backward()
            optimizer.step()
            running_loss += loss.item()
            _, predicted = torch.max(final_output, 1)
            total += labels.size(0)
            correct += (predicted == labels).sum().item()
        print(f'Epoch [{epoch + 1}/{num_epochs}], Loss: {running_loss / len(dataloader):.4f}, Accuracy: {100* correct / total:.2f}%')
    # 输出专家选择次数
    print("\n专家选择次数统计（训练集）：")
    for i, count in enumerate(expert_selection_count):
        print(f'专家 {i}: 被选择 {count.item()} 次')
    return expert_selection_count

# 测试循环，跟踪专家选择（用于数据集层面和类别层面）
def test_with_expert_statistics(model, dataloader, dataset_name="", num_classes=20):
    model.eval()
    correct = 0
    total = 0
    # 初始化专家选择计数器（数据集层面）
    expert_selection_count = torch.zeros(model.num_experts, device=device)
    with torch.no_grad():
        for inputs, labels in dataloader:
            inputs, labels = inputs.to(device), labels.to(device)
            # 前向传播
            final_output, _, _, expert_indices = model(inputs)
            # 统计专家选择次数（数据集层面）
            for idx in expert_indices:
                expert_selection_count[idx] += 1
            # 获取预测
            _, predicted = torch.max(final_output, 1)
            total += labels.size(0)
            correct += (predicted == labels).sum().item()
        accuracy = 100 * correct / total
        print(f'\n{dataset_name} 测试集准确率：{accuracy:.2f}%')
        # 输出专家选择次数统计（数据集层面）
        print(f"\n{dataset_name} 专家选择次数统计（数据集层面）：")
        for i, count in enumerate(expert_selection_count):
            print(f'专家 {i}: 被选择 {count.item()} 次，占比 {100* count.item()/ total:.2f}%')
    return accuracy, expert_selection_count

if __name__ == "__main__":
    # 主实验
    batch_size = 1024
    num_experts = 4
    loss_name = ['moe', 'mse_hard', 'mse_soft'][1]
    device = torch.device('cpu') if not torch.cuda.is_available() else torch.device('cuda:0')
    
    # 定义数据预处理
    transform = transforms.Compose([
        transforms.ToTensor(),
        transforms.Normalize((0.5,), (0.5,))
    ])
    
    # 加载 MNIST 和 Fashion-MNIST 数据集
    mnist_train = torchvision.datasets.MNIST(root='./data', train=True, download=True, transform=transform)
    mnist_test = torchvision.datasets.MNIST(root='./data', train=False, download=True, transform=transform)
    fashion_mnist_train = torchvision.datasets.FashionMNIST(root='./data', train=True, download=True, transform=transform)
    fashion_mnist_test = torchvision.datasets.FashionMNIST(root='./data', train=False, download=True, transform=transform)
    
    # 修改 Fashion-MNIST 的标签，使其与 MNIST 标签区分开来（加上 10）
    fashion_mnist_train.targets = fashion_mnist_train.targets + 10
    fashion_mnist_test.targets = fashion_mnist_test.targets + 10
    
    # 合并训练集和测试集
    combined_train_data = ConcatDataset([mnist_train, fashion_mnist_train])
    combined_test_data = ConcatDataset([mnist_test, fashion_mnist_test])
    
    # 打印合并后的数据集信息
    print(f"训练集样本数：{len(combined_train_data)}")
    print(f"测试集样本数：{len(combined_test_data)}")
    
    # 定义 MNIST 和 Fashion-MNIST 的类别名称
    mnist_classes = [str(i) for i in range(10)]  # MNIST 类别是 '0' 到 '9'
    fashion_mnist_classes = ['T-shirt/top', 'Trouser', 'Pullover', 'Dress', 'Coat', 'Sandal', 'Shirt', 'Sneaker', 'Bag', 'Ankle boot']
    
    # 合并 MNIST 和 Fashion-MNIST 的类别名称
    combined_classes = mnist_classes + fashion_mnist_classes
    
    # 创建数据加载器
    train_loader = DataLoader(combined_train_data, batch_size=batch_size, shuffle=True)
    test_loader = DataLoader(combined_test_data, batch_size=batch_size, shuffle=False)
    
    # MoE 模型
    moe_model = MoE(output_dim=len(combined_classes), num_experts=num_experts).to(device)
    optimizer_moe = optim.Adam(moe_model.parameters(), lr=0.001)
    
    print("\n训练 MoE 模型...")
    train_expert_selection = train(moe_model, train_loader, optimizer_moe, num_epochs=10, loss_name=loss_name)
    print("测试 MoE 模型在 MNIST 和 Fashion-MNIST 上...")
    
    # 测试在单个数据集上
    print("测试在 MNIST 数据集上...")
    mnist_test_loader = DataLoader(mnist_test, batch_size=batch_size, shuffle=False)
    test_accuracy_mnist, mnist_expert_selection = test_with_expert_statistics(
        moe_model, mnist_test_loader, dataset_name="MNIST", num_classes=len(combined_classes))
    print("测试在 Fashion-MNIST 数据集上...")
    fashion_mnist_test_loader = DataLoader(fashion_mnist_test, batch_size=batch_size, shuffle=False)
    test_accuracy_fashion_mnist, fashion_mnist_expert_selection = test_with_expert_statistics(
        moe_model, fashion_mnist_test_loader, dataset_name="Fashion-MNIST", num_classes=len(combined_classes))

训练参数固定如下：

• batch=1024
• epoch=10
• expert=4

对比三种损失函数：

• 公式 1 mse-hard
• 公式 2 mse-soft
• 公式 3 moe

实验分析

公式 1 mse-hard

训练日志如下

Epoch [1/10], Loss: 0.3535, Accuracy: 64.07% Epoch [2/10], Loss: 0.1851, Accuracy: 80.30% Epoch [3/10], Loss: 0.1515, Accuracy: 84.36% Epoch [4/10], Loss: 0.1331, Accuracy: 86.52% Epoch [5/10], Loss: 0.1210, Accuracy: 88.03% Epoch [6/10], Loss: 0.1123, Accuracy: 89.14% Epoch [7/10], Loss: 0.1066, Accuracy: 89.77% Epoch [8/10], Loss: 0.0995, Accuracy: 90.36% Epoch [9/10], Loss: 0.0943, Accuracy: 90.98% Epoch [10/10], Loss: 0.0908, Accuracy: 91.47% 专家选择次数统计（训练集）： 专家 0: 被选择 537500.0 次 专家 1: 被选择 167879.0 次 专家 2: 被选择 194856.0 次 专家 3: 被选择 299765.0 次 测试 MoE 模型在 MNIST 和 Fashion-MNIST 上... 测试在 MNIST 数据集上... MNIST 测试集准确率：93.73% MNIST 专家选择次数统计（数据集层面）： 专家 0: 被选择 4584.0 次，占比 45.84% 专家 1: 被选择 1927.0 次，占比 19.27% 专家 2: 被选择 1194.0 次，占比 11.94% 专家 3: 被选择 2295.0 次，占比 22.95% 测试在 Fashion-MNIST 数据集上... Fashion-MNIST 测试集准确率：85.36% Fashion-MNIST 专家选择次数统计（数据集层面）： 专家 0: 被选择 4290.0 次，占比 42.90% 专家 1: 被选择 1249.0 次，占比 12.49% 专家 2: 被选择 1622.0 次，占比 16.22% 专家 3: 被选择 2839.0 次，占比 28.39% 

公式 2 mse-soft

训练日志如下

Epoch [1/10], Loss: 0.0924, Accuracy: 79.95% Epoch [2/10], Loss: 0.0470, Accuracy: 89.93% Epoch [3/10], Loss: 0.0381, Accuracy: 91.64% Epoch [4/10], Loss: 0.0341, Accuracy: 92.55% Epoch [5/10], Loss: 0.0311, Accuracy: 93.20% Epoch [6/10], Loss: 0.0286, Accuracy: 93.82% Epoch [7/10], Loss: 0.0271, Accuracy: 94.16% Epoch [8/10], Loss: 0.0255, Accuracy: 94.49% Epoch [9/10], Loss: 0.0241, Accuracy: 94.82% Epoch [10/10], Loss: 0.0230, Accuracy: 95.05% 专家选择次数统计（训练集）： 专家 0: 被选择 884.0 次 专家 1: 被选择 1861.0 次 专家 2: 被选择 377796.0 次 专家 3: 被选择 819459.0 次 测试 MoE 模型在 MNIST 和 Fashion-MNIST 上... 测试在 MNIST 数据集上... MNIST 测试集准确率：97.57% MNIST 专家选择次数统计（数据集层面）： 专家 0: 被选择 0.0 次，占比 0.00% 专家 1: 被选择 0.0 次，占比 0.00% 专家 2: 被选择 14.0 次，占比 0.14% 专家 3: 被选择 9986.0 次，占比 99.86% 测试在 Fashion-MNIST 数据集上... Fashion-MNIST 测试集准确率：88.30% Fashion-MNIST 专家选择次数统计（数据集层面）： 专家 0: 被选择 1.0 次，占比 0.01% 专家 1: 被选择 2.0 次，占比 0.02% 专家 2: 被选择 6474.0 次，占比 64.74% 专家 3: 被选择 3523.0 次，占比 35.23% 

公式 3 moe

训练日志如下

Epoch [1/10], Loss: 0.1878, Accuracy: 79.58% Epoch [2/10], Loss: 0.0981, Accuracy: 89.75% Epoch [3/10], Loss: 0.0807, Accuracy: 91.44% Epoch [4/10], Loss: 0.0717, Accuracy: 92.37% Epoch [5/10], Loss: 0.0658, Accuracy: 92.96% Epoch [6/10], Loss: 0.0612, Accuracy: 93.35% Epoch [7/10], Loss: 0.0576, Accuracy: 93.79% Epoch [8/10], Loss: 0.0549, Accuracy: 94.12% Epoch [9/10], Loss: 0.0520, Accuracy: 94.49% Epoch [10/10], Loss: 0.0492, Accuracy: 94.74% 专家选择次数统计（训练集）： 专家 0: 被选择 2878.0 次 专家 1: 被选择 132251.0 次 专家 2: 被选择 1063670.0 次 专家 3: 被选择 1201.0 次 测试 MoE 模型在 MNIST 和 Fashion-MNIST 上... 测试在 MNIST 数据集上... MNIST 测试集准确率：97.67% MNIST 专家选择次数统计（数据集层面）： 专家 0: 被选择 0.0 次，占比 0.00% 专家 1: 被选择 0.0 次，占比 0.00% 专家 2: 被选择 10000.0 次，占比 100.00% 专家 3: 被选择 0.0 次，占比 0.00% 测试在 Fashion-MNIST 数据集上... Fashion-MNIST 测试集准确率：87.77% Fashion-MNIST 专家选择次数统计（数据集层面）： 专家 0: 被选择 0.0 次，占比 0.00% 专家 1: 被选择 2576.0 次，占比 25.76% 专家 2: 被选择 7424.0 次，占比 74.24% 专家 3: 被选择 0.0 次，占比 0.00% 

分析

总体性能对比

专家分布对比总结