一、引言

在当今人工智能领域，大型语言模型（LLM）正不断打破能力与规模的记录，一些模型的参数量已达数千亿。然而，近期一种趋势让这些巨型模型在保持高性能的同时，还能兼顾效率，那就是**Mixture-of-Experts（MoE，混合专家系统）**层。像DeepSeek、Mistral Mixtral和阿里巴巴的Qwen 3等新模型，都借助这一技术，在降低计算成本的基础上实现了高水准的性能表现。本文将深入剖析MoE技术的原理、优势及其具体实现方式，并通过实例代码展示如何构建基于MoE的模型。

二、Mixture-of-Experts（MoE）是什么？

简单来说，Mixture-of-Experts模型就像是一个由多个专家组成的小组，以及一个能够决定每个输入由哪些专家处理的智能调度员。与传统的密集型模型（每个参数都参与每个输入的处理）不同，MoE是一种稀疏型模型，它只激活一部分参数（即“专家”）来处理特定的输入。

这一概念最早可以追溯到20世纪90年代初。1991年的一篇题为《Adaptive Mixture of Local Experts》（本地专家自适应混合）的论文中，研究人员提出分别训练不同的神经网络处理数据的不同部分，并通过一个门控网络将每个输入分配给合适的“专家”。这种早期的MoE系统在训练周期上比常规网络减少了50%，这表明让专家专注于特定任务可以带来效率上的提升。

如今，MoE在大型AI模型中迎来了复兴。其核心思想仍然是将一个庞大的模型分解为多个小型组件，让每个组件在特定的模式上变得非常擅长，并利用一个门控机制将输入引导至最相关的组件。在实际应用中，比如在一个Transformer模型的MoE层中，可能会有8个专家前馈网络，当一个标记（单词）通过该层时，模型的路由器将选择最相关的1或2个专家来处理这个标记，而忽略其他专家。这种方式使得模型的总参数量可以非常庞大（各个专家的参数量相加），但在处理每个输入时只激活少量参数，从而节省了计算和内存资源。

三、Mixture-of-Experts（MoE）如何工作？

以下是Mixture-of-Experts（MoE）工作原理的详细阐述：

（一）核心概念

MoE模型主要由以下几个关键组件构成：

专家（Experts） ：这些是专门化子网络，通常是比较简单的前馈网络或其他层，每个输入都可能被路由到它们。每个专家可能会学习处理数据的某些特定特征。
路由器 / 门控网络（Router/Gating Network） ：这就好比一个“交通警察”，它会查看输入（或标记），并决定应该由哪个专家（或专家组合）来处理它。路由器会输出一组权重或概率，为专家分配权重，例如，“专家3应该主要处理这个输入，专家1稍微参与一些，其他专家则完全不参与”。
稀疏激活（Sparse Activation） ：对于每个输入，只激活权重最高的专家，通常是前k个专家。其他专家在这个输入的处理过程中保持不活跃状态。这种稀疏性正是MoE高效的原因 —— 它避免了在每个示例上运行整个网络。

以一个具体的例子来说明：假设计算机有多个顾问（专家），但对于每个项目（输入），只召集最相关的几个顾问来工作，其余的则闲置。这样一来，公司可以拥有广泛的专长（高容量），而无需在每个项目上部署所有顾问。

（二）工作流程

在MoE层内部，其工作流程如下：

想象我们有一个包含多个专家网络的MoE层。当一个输入进入时：

路由器计算专家评分 ：输入被送入路由器（门控网络），该网络为每个专家生成一个评分或权重。这些评分通常通过SoftMax转换为概率，使得它们的总和为1。例如，路由器可能输出权重为[0.75, 0.05, 0.20]，表示三个专家中，专家0被认为与这个输入最相关。
选择顶级专家 ：基于路由器的评分，模型选择前k个专家（通常k=1或k=2）来实际使用。在上述示例中，这将是专家0（以及可能的专家2）。其他专家将被跳过，不参与此输入的处理。
专家生成输出 ：每个选中的专家独立处理输入（或标记），并生成其输出（例如，其预测或对数据的转换）。
合并专家输出 ：通常通过加权求和的方式，使用路由器的概率来合并活跃专家的输出。例如，如果专家0的权重为0.75，专家2的权重为0.20，那么最终输出 = 0.75×（专家0的输出） + 0.20×（专家2的输出）（而专家1的0.05的输出则被有效忽略）。在许多实现中，前k个专家的输出被平均或相加（加权后），形成该层的输出，并继续传递到模型的下一部分。

这种设计意味着模型的容量（总参数量）可以极高，而每个输入的计算量却相对较低。例如，Mistral的Mixtral模型在每一层都有8个专家，总参数量约为467亿，但每个标记仅使用约129亿参数的专家。这就好比模型是一个由8个专家组成的委员会，但任何一个标记只咨询其中的2个专家，因此在推理时的工作量相当于一个约130亿参数的模型。这就是DeepSeek、Mixtral、Qwen3等模型能够在拥有数千亿参数的情况下，而不会导致推理速度成比例下降的原因。

（三）训练挑战

当然，要使MoE正常工作并非易事 —— 路由器必须学会将正确的输入发送到正确的专家。如果它将所有内容都发送到一个专家，那么这个专家就会成为瓶颈，而其他专家则被闲置。如果路由器随机分配输入，专家们就无法实现专业化。诸如负载均衡损失、路由中的噪声（例如，Noisy Top – k 门控）或自适应路由（如DeepSeek所使用的）等技术被用来确保所有专家都能得到训练和使用。在这里我们暂不深入探讨这些细节，但需要明白，为了有效地训练MoE模型并避免专家塌缩，大量的研究工作正在进行。

四、Mixture-of-Experts（MoE）实战：基于PyTorch的代码示例

为了更直观地理解MoE的实际应用，我们将通过一个文本分类任务的案例来展示如何构建基于MoE的模型。我们将使用PyTorch构建一个用于情感分析的IMDB数据集分类器，并在模型中集成MoE结构。

（一）数据准备与环境搭建

安装和导入库 ：安装并导入Hugging Face Transformers和Datasets库，并导入PyTorch。同时检查是否有GPU可用，并加载必要的模块（分词器、模型等）。

!pip install transformers datasets -q
import torch
from torch import nn
from torch.utils.data import DataLoader
from transformers import AutoTokenizer, AutoModel
from datasets import load_dataset
import matplotlib.pyplot as plt
from tqdm import tqdm

加载数据集 ：使用datasets.load_dataset加载IMDB数据集，该数据集包含预定义的训练集和测试集。示例中将测试集用作验证集。

# 检查设备
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print("Using device:", device)
# 加载IMDB数据集
dataset = load_dataset("imdb")
# 使用测试集作为验证集
train_data = dataset["train"]
val_data = dataset["test"]

分词处理 ：由于我们使用基于Transformer的模型（DistilRoBERTa）作为基础，因此需要对文本进行分词。下载预训练的DistilRoBERTa分词器，并将其应用于文本，将每个评论截断或填充到固定长度（256个标记）。

# 分词处理训练和验证数据
model_name = "distilroberta-base" # 使用较小的模型以提高效率
tokenizer = AutoTokenizer.from_pretrained(model_name)
def tokenize_function(examples):
    return tokenizer(examples["text"], padding="max_length", truncation=True, max_length=256)
# 应用分词函数
train_data = train_data.map(tokenize_function, batched=True)
val_data = val_data.map(tokenize_function, batched=True)

数据加载器准备 ：将分词后的数据集转换为PyTorch张量，并加载到DataLoader对象中，以便进行批量处理。训练集被批量处理（批次大小为128）并打乱顺序，而验证集被批量处理（批次大小为256）不打乱顺序。

# 将数据集转换为PyTorch张量并创建DataLoader
train_data.set_format(type="torch", columns=["input_ids", "attention_mask", "label"])
val_data.set_format(type="torch", columns=["input_ids", "attention_mask", "label"])
train_dataloader = DataLoader(train_data, batch_size=128, shuffle=True, pin_memory=True)
val_dataloader = DataLoader(val_data, batch_size=256, shuffle=False, pin_memory=True)

（二）定义MoE模型架构

接下来，定义一个自定义模型类MoEClassifier，在Transformer编码器之上实现Mixture-of-Experts结构。

# 定义Mixture of Experts模型
class MoEClassifier(nn.Module):
    def __init__(self, base_model_name="distilroberta-base", num_labels=2, num_experts=3):
        super(MoEClassifier, self).__init__()
        # 基础Transformer模型（不带分类头）
        self.base_model = AutoModel.from_pretrained(base_model_name)
        self.hidden_size = self.base_model.config.hidden_size
        self.num_labels = num_labels
        self.num_experts = num_experts
        # 专家头（每个都是简单的线性分类器）
        self.experts = nn.ModuleList([nn.Linear(self.hidden_size, self.num_labels) for _ in range(num_experts)])
        # 门控网络，用于生成每个专家的权重
        self.gate = nn.Linear(self.hidden_size, self.num_experts)    
    def forward(self, input_ids, attention_mask, labels=None):
        # Transformer前向传播
        outputs = self.base_model(input_ids=input_ids, attention_mask=attention_mask)
        # 使用[CLS]标记表示（第一个标记）作为池化输出
        last_hidden_state = outputs.last_hidden_state  # (batch_size, seq_len, hidden_size)
        pooled_output = last_hidden_state[:, 0, :]  # (batch_size, hidden_size)
        # 计算门控权重（对专家进行SoftMax）
        gating_logits = self.gate(pooled_output)  # (batch_size, num_experts)
        gating_weights = torch.softmax(gating_logits, dim=1)  # (batch_size, num_experts)
        # 计算每个专家的 logits
        expert_logits = torch.stack([expert(pooled_output) for expert in self.experts], dim=1)  # (batch_size, num_experts, num_labels)
        # 按门控概率加权专家输出
        weighted_logits = expert_logits * gating_weights.unsqueeze(2)  # (batch_size, num_experts, num_labels)
        final_logits = weighted_logits.sum(dim=1)  # (batch_size, num_labels)
        loss = None
        if labels is not None:
            loss_fn = nn.CrossEntropyLoss()
            loss = loss_fn(final_logits, labels)
        # 返回类似HuggingFace模型输出的字典
        return {"loss": loss, "logits": final_logits} if loss is not None else {"logits": final_logits}

基础Transformer模型 ：该模型使用DistilRoBERTa作为特征提取器。代码中通过AutoModel.from_pretrained(base_model_name)加载不带分类头的Transformer模型，该模型将输出文本中每个标记的隐藏状态。
专家层 ：在Transformer之上，MoEClassifier定义了多个专家“头”。在本实现中，每个专家是一个简单的线性层，接收Transformer的隐藏状态并输出类别 logits。如果num_experts=3（如代码中设置），则创建3个形状为[hidden_size -> num_labels]的线性层，并将它们存储在名为self.experts的nn.ModuleList中。在IMDB情感分析任务中，num_labels=2（正面或负面情感），因此每个专家输出2个数值（类别 logits）。
门控网络 ：关键部分是self.gate，另一个线性层，形状为[hidden_size -> num_experts]。这是路由器。给定Transformer的隐藏表示，该层将为每个专家生成一组分数（logits）。如果有3个专家，门控网络将输出3个数字，表示每个专家的权重。
前向传播逻辑 ：当调用模型处理输入时，内部执行以下步骤：
- 输入文本通过DistilRoBERTa基础模型。假设输出隐藏状态的大小为(batch_size, seq_len, hidden_size)，我们取第一个标记（通常是[CLS]标记）的嵌入作为整个序列的池化表示。这将得到形状为(batch_size, hidden_size)的pooled_output张量 —— 每个输入示例一个向量。
- 门控网络（self.gate）处理pooled_output，生成gating_logits，形状为(batch_size, num_experts)。对于批次中的每个示例，我们现在为每个专家生成一个分数。对这些logits应用SoftMax，得到专家的门控权重（概率）。例如，输出可能是[0.7, 0.2, 0.1]，分别对应专家0、1、2，这意味着专家0被认为与该输入最相关。
- 每个专家头（线性层）也被应用于pooled_output。代码通过列表推导式在一行中实现这一点：计算每个专家对输入的logits。结果expert_logits可以视为形状为(batch_size, num_experts, num_labels)的张量，包含每个专家的预测分数。例如，专家0可能输出[-1.2, 2.3]（更倾向于“正面”），专家1输出[0.5, 0.1]（更倾向于“负面”），等等，针对给定输入。
- 混合专家 ：现在使用门控权重将这些专家输出进行组合。代码将每个专家的logits乘以其权重，并将它们相加：final_logits = sum(weight_i * expert_i_logits)。这生成形状为(batch_size, num_labels)的最终预测logits，用于每个输入，整合所选专家的贡献。值得注意的是，即使计算了所有专家的logits，权重会使得权重非常低的专家贡献几乎可以忽略不计。在实践中，可以通过仅计算顶级专家来优化，但为了简化，此实现计算所有专家的logits，并依赖权重来抑制不相关的专家。
- 如果在训练期间提供标签，模型可以计算交叉熵损失，并返回最终logits（和损失）。在推理期间（无标签），它将仅输出logits。

（三）训练MoE模型

在定义模型后，示例代码继续训练模型（尽管为了演示仅训练1个周期）。训练循环包括一些标准元素和几个有趣的细节：

# 初始化模型、优化器和混合精度缩放器
num_experts = 3
model = MoEClassifier(base_model_name=model_name, num_experts=num_experts).to(device)
optimizer = torch.optim.AdamW(model.parameters(), lr=2e-5)
scaler = torch.cuda.amp.GradScaler()# 初始化用于绘制损失的列表
train_losses = []
val_losses = []# 带有实时损失绘制的训练循环（训练和验证）
epochs = 1
for epoch in range(epochs):
    model.train()
    total_train_loss = 0.0
    progress_bar = tqdm(train_dataloader, desc=f"Epoch {epoch+1}/{epochs}", ncols=100, position=0, leave=True)
    for batch in progress_bar:
        # 将数据移动到设备
        input_ids = batch["input_ids"].to(device, non_blocking=True)
        attention_mask = batch["attention_mask"].to(device, non_blocking=True)
        labels = batch["label"].to(device, non_blocking=True)
        optimizer.zero_grad() # 带有混合精度的前向传播
        with torch.cuda.amp.autocast():
            outputs = model(input_ids=input_ids, attention_mask=attention_mask, labels=labels)
            loss = outputs["loss"] # 反向传播和优化
        scaler.scale(loss).backward()
        scaler.step(optimizer)
        scaler.update()
        total_train_loss += loss.item() # 更新进度条的当前损失
        progress_bar.set_postfix(loss=total_train_loss / (progress_bar.n + 1)) # 跟踪训练损失以实时绘制
        train_losses.append(total_train_loss / (progress_bar.n + 1)) # 计算平均训练损失
    avg_train_loss = total_train_loss / len(train_dataloader) # 现在在验证集上评估
    model.eval()
    total_val_loss = 0.0
    with torch.no_grad():
        for batch in val_dataloader:
            input_ids = batch["input_ids"].to(device, non_blocking=True)
            attention_mask = batch["attention_mask"].to_device, non_blocking=True)
            labels = batch["label"].to(device, non_blocking=True) # 带有混合精度的前向传播
            with torch.cuda.amp.autocast():
                outputs = model(input_ids=input_ids, attention_mask=attention_mask, labels=labels)
                loss = outputs["loss"]
            total_val_loss += loss.item() # 计算平均验证损失
    avg_val_loss = total_val_loss / len(val_dataloader) # 跟踪验证损失以实时绘制
    val_losses.append(avg_val_loss)
    print(f"Epoch {epoch+1}/{epochs} - Train Loss: {avg_train_loss:.4f} - Validation Loss: {avg_val_loss:.4f}")

模型初始化 ：创建MoEClassifier的实例，指定num_experts=3（并使用DistilRoBERTa作为基础）。模型被移动到可用设备（如果有GPU）。使用AdamW优化器（学习率为2e-5）对模型的所有参数进行优化。
混合精度训练 ：代码使用PyTorch的自动混合精度（torch.cuda.amp.autocast和GradScaler）来加速GPU上的训练，使用float16。这是一项性能优化措施 —— 并不改变MoE逻辑，但有助于高效处理大型Transformer模型。
周期循环 ：迭代指定的周期数（在本例中为epochs = 1）。对于每个周期，将模型设置为训练模式，并循环遍历train_dataloader中的批次。对于每个批次：
- 输入（input_ids和attention_mask）和标签被移动到设备。
- 在自动铸造上下文中调用模型，以获取输出和损失。在幕后，模型将根据批次计算门控权重、专家输出，并组合logits以获取最终预测，然后通过比较final_logits和真实标签计算交叉熵损失。
- 对损失进行缩放并反向传播（scaler.scale(loss).backward()），优化器更新模型权重。这会更新Transformer权重以及专家和门控网络的参数。随着时间的推移，门控网络将学会路由输入，专家将专门化以降低损失。
- 累积训练损失以监控进度。
验证：在训练周期后，模型在验证集（IMDB测试集）上进行评估，以计算平均验证损失。这在无梯度模式下进行（仅前向传播）。

即使只进行简短的训练，模型也会开始拟合数据。关键是训练MoE模型与训练常规模型没有根本区别 —— 主要区别在于门控网络和专家的参数与模型其他部分同时学习。在像DeepSeek这样的复杂MoE模型中，还有一些额外的技巧来稳定训练（谷歌的MoE研究提到了一些用于平衡专家的额外损失等），但概念上是相同的：计算所选专家和路由器的梯度，并更新它们。

（四）推理：使用MoE模型（及解释）

训练完成后，示例代码展示了如何在新文本上使用模型进行推理。它定义了一个自定义的infer()函数，该函数不仅获取预测结果，还打印出MoE内部的工作情况，以实现透明性。让我们看一个例子：

import torch
import torch.nn.functional as F

def infer(model, input_data):
    model.eval()  # 将模型设置为评估模式
    device = next(model.parameters()).device  # 获取模型的设备（CPU或GPU）
    input_data = input_data.to(device)        # 将输入数据移动到与模型相同的设备上
    # 将输入数据转换为与模型权重相同的类型（通常是float32）
    input_data = input_data.to(dtype=torch.long)
    # 如果可用，使用混合精度以提高效率
    with torch.no_grad():
        with torch.cuda.amp.autocast():
            # 第1步：将输入通过基础Transformer模型进行前向传播
            outputs = model.base_model(input_ids=input_data)  # 获取Transformer输出
            pooled_output = outputs.last_hidden_state[:, 0, :]  # [CLS]标记表示
            # 第2步：门控网络为每个专家生成权重
            gating_logits = model.gate(pooled_output)  # 门控网络的原始输出（logits）
            gating_weights = F.softmax(gating_logits, dim=-1)  # 将其转换为总和为1的概率
            # 第3步：每个专家为输入生成自己的logits
            expert_outputs = [expert(pooled_output) for expert in model.experts]  # 专家输出列表（每个专家一个张量）
            # 第4步：使用门控权重合并专家输出
            combined_logits = torch.zeros_like(expert_outputs[0])
            for w, logits in zip(gating_weights[0], expert_outputs):
                combined_logits += w * logits  # 对专家的logits进行加权求和
    # 将张量移动到CPU以便打印（如果在半精度下转换为float以便清晰）
    gating_weights_cpu = gating_weights[0].detach().cpu().float()   # 1D张量，长度为num_experts
    expert_logits_cpu = [logit.detach().cpu().float() for logit in expert_outputs]  # 专家logits列表
    combined_logits_cpu = combined_logits.detach().cpu().float()    # 形状为（num_classes，）的张量
    # 确定最终预测标签（假设为二元分类，“正面”或“负面”）
    pred_index = int(torch.argmax(combined_logits_cpu))  # 取logit值最高的索引
    prediction_label = "positive" if pred_index == 1 else "negative"
    # 打印详细的MoE推理解释
    # 解释门控权重
    gw_list = gating_weights_cpu.tolist()
    print(f"Gating weights (for each expert): {gw_list}")
    print(" - 门控网络根据输入的相关性为专家分配这些权重。")
    for i, w in enumerate(gw_list):
        print(f"   Expert {i} weight: {w:.4f}（来自Expert {i}的贡献比例）")
    # 解释每个专家的logits和单独预测
    for i, logits in enumerate(expert_logits_cpu):
        logits_list = logits.tolist()
        print(f"Expert {i} logits: {logits_list}")
        if len(logits_list) == 2:
            # 如果是二元分类，确定这个专家更倾向的类别
            neg_logit, pos_logit = logits_list[0], logits_list[1]
            expert_pred = "positive" if pos_logit > neg_logit else "negative"
            print(f" - Expert {i} would predict '{expert_pred}' ("
                  f"{'pos' if expert_pred == 'positive' else 'neg'} logit is higher).")
    # 解释组合logits的计算
    comb_list = combined_logits_cpu.tolist()
    print(f"Combined logits (weighted sum of experts): {comb_list}")
    print(" - 每个组合logit是通过将专家的logits乘以它们的门控权重并相加得到的。")
    if len(comb_list) == 2:
        print(f"   (例如，combined_neg_logit = "
              f"{' + '.join([f'{w:.4f}*{expert_logits_cpu[i][0].item():.4f}' for i, w in enumerate(gw_list)])})")
        print(f"   (and combined_pos_logit = "
              f"{' + '.join([f'{w:.4f}*{expert_logits_cpu[i][1].item():.4f}' for i, w in enumerate(gw_list)])})")
    # 解释最终预测
    print(f"Final Prediction: {prediction_label.upper()}")
    print(f" - 模型预测'{prediction_label}'情感，因为该类别的组合logit最高。")
    # 准备结果字典
    result = {
        "expert_logits": [logit.tolist() for logit in expert_logits_cpu],
        "gating_weights": gw_list,
        "combined_logits": comb_list,
        "prediction_label": prediction_label
    }
    return result

# 示例输入文本
# sample_text = "It was illogical  movie but it was fun"
sample_text = "I loved the movie! It was so exciting and fun!"# 对输入文本进行分词
inputs = tokenizer(sample_text, return_tensors="pt", padding=True, truncation=True, max_length=256)# 使用infer函数进行推理
result = infer(model, inputs["input_ids"])# 显示结果（由infer函数返回的字典）
print(result)

假设我们输入文本：“I loved the movie! It was so exciting and fun!”（我爱这部电影！它太刺激了，太有趣了！），这是一个正面情感的评论。infer函数将对该文本进行分词，并将其输入到模型中。以下是该函数的输出示例：

Gating weights (for each expert): [0.17755267024040222, 0.47388124465942383, 0.34856608510017395]
 - The gating network assigns these weights to the experts based on input relevance.
   Expert 0 weight: 0.1776 (proportion of contribution from Expert 0)
   Expert 1 weight: 0.4739 (proportion of contribution from Expert 1)
   Expert 2 weight: 0.3486 (proportion of contribution from Expert 2)
Expert 0 logits: [[-2.15625, 2.126953125]]
Expert 1 logits: [[-2.73828125, 2.421875]]
Expert 2 logits: [[-2.86328125, 2.265625]]
Combined logits (weighted sum of experts): [[-2.6796875, 2.31640625]]
 - Each combined logit is computed by summing the experts' logits multiplied by their gating weights.
Final Prediction: POSITIVE
 - The model predicts 'positive' sentiment because that class has the highest combined logit.
{'expert_logits': [[[-2.15625, 2.126953125]], [[-2.73828125, 2.421875]], [[-2.86328125, 2.265625]]], 'gating_weights': [0.17755267024040222, 0.47388124465942383, 0.34856608510017395], 'combined_logits': [[-2.6796875, 2.31640625]], 'prediction_label': 'positive'}

从这个例子中我们可以看到：

门控权重 ：函数打印出该输入对应的每个专家的权重。例如，在一次运行中，结果可能显示：Gating weights (for each expert): [0.7612, 0.0801, 0.1587]。这意味着专家0被分配了约76%的责任，专家1约8%，专家2约16%。这表明路由器认为专家0对于这个特定的评论是最相关的，专家2有些相关，而专家1不太相关。
专家输出 ：接下来，它展示了每个专家的原始logits（分数）用于负面和正面类别。例如：
- 专家0 logits：[-2.5215, 2.2305]
- 专家1 logits：[-0.5479, 1.1855]
- 专家2 logits：[-1.2412, 1.9307]

在二元分类中，更高的第二个数字表示“正面”类别的投票。可以看出，专家0强烈投票支持正面（因为2.23远大于-2.52），专家1也倾向于正面（1.185大于-0.548），专家2同样倾向于正面（1.931大于-1.241）。所有三个专家实际上都表示“这可能是一个正面的评论”，尽管专家0是最有信心的。

组合logits ：函数然后打印出组合logits，这些logits是在对专家的logits进行加权求和后得到的。例如，它可能计算为：Combined logits: [[-2.1602, 2.0996]]。这是如何计算的呢？实际上是0.7612×Expert0_logits + 0.0801×Expert1_logits + 0.1587×Expert2_logits。如果我们计算正面logit：0.7612×2.2305 + 0.0801×1.1855 + 0.1587×1.9307≈2.10，这与组合后的正面logit相符。函数甚至打印出组合公式的分解，以便清晰展示。
最终预测 ：最后，它输出模型的预测标签：POSITIVE。这很有道理，因为组合后的logits显示正面类别的分数更高（2.0996 > -2.1602）。函数还解释道：“模型预测‘正面’情感，因为该类别的组合logit最高。” 我们已经确认MoE模型正确地将该示例分类为正面。

很酷的是，我们可以窥探MoE的决策过程。我们看到专家0在该输入中承担了大部分权重，而专家0确实对正面情感信号非常强烈，这推动了最终预测。如果出现不同类型的输入，例如一个非常微妙或负面的评论，专家2可能会获得更高的权重并接管任务。这正是MoE提供鲁棒性和效率的方式：如果一个专家对某个输入很自信且合适，路由器会高度依赖它。如果一个不同的输入传来，路由器可能会选择不同的专家。

五、结论

Mixture-of-Experts（MoE）是一种令人兴奋的架构模式，它允许AI模型通过添加更多的“专家”来水平扩展，而无需使每个推理步骤变慢。通过仅激活每个输入的少数专家，MoE模型（如DeepSeek、Mixtral和Qwen3）实现了高容量和专业化的最佳组合，同时在效率上可与小得多的模型相媲美。

在本文中，我们介绍了MoE的概念，并通过一个简单的示例展示了基于MoE的模型如何运作。我们看到了路由器网络如何学习将任务分配给专家子网络，以及如何将输出组合起来生成最终答案。这种技术正在推动最新的突破 —— DeepSeek R1的6710亿参数模型使用MoE，每个标记仅使用约370亿参数，而Qwen – 3的MoE模型通过将子任务委托给不同的专家来处理复杂的推理任务。

随着研究的继续，我们可能会看到MoE以更具创意的方式被应用（以及在训练稳定性和专家平衡方面得到改进）。对于任何构建或探索AI模型的人来说，MoE是一个有用的技术，当需要在不超出计算预算的情况下扩展模型容量时。毕竟，有时使用专家团队比使用单一的“万事通”模型更聪明（也更快）！