自回归生成是大模型推理的"阿喀琉斯之踵":每生成一个token需要完整前向传播,无法有效利用GPU并行性。多Token预测(Multi-Token Prediction, MTP)通过在单次前向传播中预测多个未来token,突破这一瓶颈。到2026年,MTP已成为推理加速的标准技术之一。

1. 自回归生成的瓶颈

1.1 推理延迟分析

标准自回归生成的推理延迟:

$$\text{Latency} = T \times (t_{prefill} + t_{decode})$$

其中 $T$ 为生成长度,$t_{prefill}$ 为预填充延迟,$t_{decode}$ 为逐token生成延迟。由于KV Cache复用,$t_{decode}$ 主导整个生成过程。

以70B模型生成100个token为例:

  • $t_{decode} \approx 30\text{ms/token}$(A100 80GB)
  • 总延迟 $\approx 3\text{秒}$

1.2 硬件利用率

自回归生成的硬件利用率极低:

阶段FLOPs利用率瓶颈
Prefill60-80%计算吞吐
Decode5-15%内存带宽

Decode阶段大部分时间花在从显存加载模型权重上,这是Memory-Bound而非Compute-Bound。

2. 多Token预测原理

2.1 核心思想

MTP的核心思想:在预测下一个token的同时,预测未来多个token

训练目标从单步预测扩展到多步预测:

$$\mathcal{L}{MTP} = \sum{k=1}^{K} \lambda_k \cdot \mathcal{L}(\hat{y}{t+k} | y{<t+k})$$

其中 $K$ 为预测步数,$\lambda_k$ 为第 $k$ 步的权重。

2.2 架构设计

主流MTP架构有三种方案:

方案1:独立输出头

class IndependentHeadsMTP(nn.Module):
    def __init__(self, base_model, num_heads=4):
        super().__init__()
        self.base = base_model
        self.output_heads = nn.ModuleList([
            nn.Linear(base_model.d_model, base_model.vocab_size)
            for _ in range(num_heads)
        ])
    
    def forward(self, x):
        hidden = self.base(x)  # [batch, seq_len, d_model]
        outputs = []
        for head in self.output_heads:
            outputs.append(head(hidden))  # 共享hidden,独立head
        return outputs  # [num_heads, batch, seq_len, vocab_size]

方案2:级联MTP(Meta的MTP方案)

每个预测头基于前一个预测头的结果进行预测:

class CascadedMTP(nn.Module):
    def forward(self, x):
        hidden = self.base(x)
        
        outputs = []
        current_hidden = hidden
        
        for k in range(self.num_heads):
            # 预测第k+1个token
            logits = self.output_heads[k](current_hidden)
            outputs.append(logits)
            
            # 将预测token的embedding加入hidden,用于预测下一个
            pred_token = logits.argmax(dim=-1)
            token_emb = self.embed(pred_token)
            current_hidden = torch.cat([current_hidden, token_emb], dim=1)
            current_hidden = self.transform_layers[k](current_hidden)
        
        return outputs

方案3:共享Transformer层

DeepSeek-V3采用的方案:使用共享底层,顶层独立:

class SharedLayerMTP(nn.Module):
    def __init__(self, d_model, num_layers, num_heads):
        super().__init__()
        # 共享底层
        self.shared_layers = nn.ModuleList([
            TransformerLayer(d_model) for _ in range(num_layers - 2)
        ])
        # 每个预测头有独立的顶层
        self.head_top_layers = nn.ModuleList([
            TransformerLayer(d_model) for _ in range(num_heads)
        ])
        self.output_heads = nn.ModuleList([
            nn.Linear(d_model, vocab_size) for _ in range(num_heads)
        ])
    
    def forward(self, x):
        # 共享层
        hidden = x
        for layer in self.shared_layers:
            hidden = layer(hidden)
        
        # 独立顶层
        outputs = []
        for i in range(self.num_heads):
            head_hidden = self.head_top_layers[i](hidden)
            outputs.append(self.output_heads[i](head_hidden))
        
        return outputs

3. 训练策略

3.1 损失权重设计

不同预测步的难度不同,需要设计合理的损失权重:

def compute_mtp_loss(logits_list, targets, loss_weights=None):
    """计算MTP损失"""
    if loss_weights is None:
        # 指数衰减权重
        loss_weights = [0.5 ** k for k in range(len(logits_list))]
    
    total_loss = 0
    for k, (logits, weight) in enumerate(zip(logits_list, loss_weights)):
        # 第k个输出预测第k+1个token
        target = targets[:, k+1:]
        loss = F.cross_entropy(
            logits[:, :-k-1, :].reshape(-1, logits.size(-1)),
            target.reshape(-1)
        )
        total_loss += weight * loss
    
    return total_loss

实验表明,合理的权重策略是:

  • 第1步(标准next token): 权重1.0
  • 第2步: 权重0.5
  • 第3步: 权重0.25
  • 第4步: 权重0.125

3.2 课程学习

从易到难的训练策略:先训练单步预测,再逐步增加预测步数:

def curriculum_training(model, data, epoch):
    if epoch < 10:
        # 前10轮:仅训练第1步
        model.num_active_heads = 1
    elif epoch < 20:
        # 10-20轮:训练前2步
        model.num_active_heads = 2
    else:
        # 20轮后:训练全部
        model.num_active_heads = model.num_heads

4. 推理加速

4.1 贪心MTP解码

最简单的方式:取所有预测token,只要置信度足够高就接受:

def greedy_mtp_decode(model, prompt, max_new_tokens, confidence_threshold=0.8):
    generated = prompt
    while len(generated) < max_new_tokens:
        # 模型输出K个预测
        logits_list = model(generated)
        
        # 贪心选择:取每个预测的最高概率token
        tokens = []
        confidences = []
        for logits in logits_list:
            probs = F.softmax(logits[-1], dim=-1)
            token = probs.argmax()
            confidence = probs.max()
            tokens.append(token)
            confidences.append(confidence)
        
        # 接受所有置信度高于阈值的token
        accepted = []
        for token, conf in zip(tokens, confidences):
            if conf > confidence_threshold:
                accepted.append(token)
            else:
                break  # 一旦遇到低置信度,停止接受
        
        if not accepted:
            # 保守策略:仅接受第1个token
            accepted = [tokens[0]]
        
        generated = generated + accepted
    
    return generated

4.2 MTP + Speculative Decoding

MTP可以作为Speculative Decoding的draft model:

def mtp_speculative_decode(target_model, draft_model, prompt, max_tokens):
    generated = prompt
    while len(generated) < max_tokens:
        # draft model生成候选序列(使用MTP)
        candidates = draft_model.generate_mtp(generated, k=4)
        
        # target model验证
        with torch.no_grad():
            target_logits = target_model(generated + candidates)
        
        # 接受策略:nucleus sampling
        accepted_len = 0
        for i, candidate in enumerate(candidates):
            # 计算target model对candidate的概率
            p_target = F.softmax(target_logits[i], dim=-1)[candidate]
            p_draft = F.softmax(draft_logits[i], dim=-1)[candidate]
            
            # 接受概率
            p_accept = min(1, p_target / p_draft)
            if random.random() < p_accept:
                accepted_len = i + 1
            else:
                break
        
        # 接受前accepted_len个token
        generated = generated + candidates[:accepted_len]
        
        # 如果拒绝了某个token,用target model重新采样
        if accepted_len < len(candidates):
            new_token = sample_from_adjusted_distribution(
                target_logits[accepted_len]
            )
            generated.append(new_token)
    
    return generated

5. 效果评估

5.1 加速比

模型规模MTP步数加速比质量损失
7B21.6x<0.5%
7B42.1x~1%
70B21.7x<0.3%
70B42.3x~0.8%
400B21.8x<0.2%

5.2 预测准确率

MTP不同步的准确率呈指数衰减:

预测步数准确率典型场景
1 (标准)70-85%
240-55%
320-30%
410-15%很低

关键:即使后续步准确率不高,只要部分token能被正确预测,推理加速效果依然显著。

6. 2026年实践建议

6.1 实施路径

  1. 训练阶段:从标准训练开始,后期加入MTP损失微调
  2. 推理阶段:使用MTP + Speculative Decoding组合
  3. 部署阶段:根据实际场景调整接受阈值

6.2 适用场景

MTP特别适合:代码生成、数学推理等具有可预测模式的任务。对于创意写作等多样性高的任务,MTP的加速效果有限。

多Token预测代表了推理加速的重要方向。随着模型能力的提升,未来MTP可能从当前的2-4步扩展到10步以上,实现数量级的推理加速。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞,认知差异创造无限可能。