在标准 Transformer 模型中,每一层包含一个自注意力模块(Self-Attention)和一个前馈神经网络(FFN)。MoE 的核心思想是用多个专家(Expert)替代 FFN,每个专家本身也是一个 FFN,但通过动态路由机制(门控网络)选择对每个输入 token 最相关的少数专家进行计算。这种设计可以在不显著增加计算量的情况下,大幅提升模型容量。
标准 Transformer 层的计算流程如下:
在 MoE 化的 Transformer 中,部分 FFN 层被替换为 MoE 层(通常每隔几层替换一次)。MoE 层的计算步骤如下:
(1)专家定义
(2)门控机制(Gating Mechanism)
(3)稀疏路由(Sparse Routing)
(4)加权输出
负载均衡(Load Balancing)
训练稳定性
通信开销(分布式训练)
python展开代码class MoELayer(nn.Module):
    def __init__(self, num_experts, hidden_size, expert_size, k=1):
        super().__init__()
        self.experts = nn.ModuleList([FFN(hidden_size, expert_size) for _ in range(num_experts)])
        self.gate = nn.Linear(hidden_size, num_experts)  # 门控网络
        self.k = k
    def forward(self, x):
        # 计算亲和度
        logits = self.gate(x)  # [batch_size, seq_len, num_experts]
        scores = torch.softmax(logits, dim=-1)
        
        # 选择 Top-K 专家
        topk_scores, topk_indices = torch.topk(scores, self.k, dim=-1)  # [batch, seq_len, k]
        
        # 稀疏计算(仅用选中的专家)
        outputs = []
        for i in range(self.k):
            expert_idx = topk_indices[:, :, i]
            expert_output = self.experts[expert_idx](x)  # 伪代码,需具体实现
            outputs.append(expert_output * topk_scores[:, :, i].unsqueeze(-1))
        
        return sum(outputs) + x  # 加权求和 + 残差连接
下图是DeepSeekMoE 的示意图。子图(a)展示了采用传统 top-2 路由策略的 MoE 层。子图(b)说明了细粒度专家分割策略。随后,子图(c)展示了共享专家隔离策略的集成,构成了完整的 DeepSeekMoE 架构。值得注意的是,在这三种架构中,专家参数的数量和计算成本保持不变。

DeepSeekMoE 是在传统混合专家(MoE)架构上的创新改进,通过细粒度专家分割和共享专家隔离两大核心策略,显著提升模型的专业化能力与参数效率,同时保持计算成本不变。以下是其核心设计理念与技术实现:
python展开代码class DeepSeekMoELayer(nn.Module):
    def __init__(self, num_experts, m, K, Ks):
        super().__init__()
        # 细粒度分割:每个专家拆分为 m 个子专家
        self.sub_experts = nn.ModuleList([
            SmallFFN(hidden_dim // m) for _ in range(num_experts * m)
        ])
        self.shared_experts = nn.ModuleList([  # 共享专家
            SmallFFN(hidden_dim // m) for _ in range(Ks)
        ])
        self.gate = nn.Linear(hidden_dim, num_experts * m)  # 门控网络
        self.K = K
        self.Ks = Ks
    def forward(self, x):
        # 共享专家计算(强制所有 token 经过)
        shared_out = sum([expert(x) for expert in self.shared_experts])
        # 细粒度路由计算
        logits = self.gate(x)
        scores = torch.softmax(logits, dim=-1)
        topk_scores, topk_indices = torch.topk(scores, self.K - self.Ks, dim=-1)
        
        # 加权求和路由专家输出
        routed_out = 0
        for i in range(self.K - self.Ks):
            expert_idx = topk_indices[:, :, i]
            routed_out += self.sub_experts[expert_idx](x) * topk_scores[:, :, i]
        return shared_out + routed_out + x  # 总输出



本文作者:Dong
本文链接:
版权声明:本博客所有文章除特别声明外,均采用 CC BY-NC。本作品采用《知识共享署名-非商业性使用 4.0 国际许可协议》进行许可。您可以在非商业用途下自由转载和修改,但必须注明出处并提供原作者链接。 许可协议。转载请注明出处!