Speculative Decoding最新进展2026

概述

Speculative Decoding(推测解码)已成为加速大语言模型(LLM)推理的标准技术。它通过使用小型draft模型生成候选token,再由大型target模型并行验证,在保持输出分布不变的前提下实现加速。然而,标准方法仍存在多个瓶颈:验证阶段的联合概率难计算、draft与verification的串行依赖、大词表带来的线性投影开销等。

2025-2026年,研究者提出了多项创新技术来突破这些瓶颈:

方法核心贡献论文
Hierarchical SD分层分支重采样,无损提升接受率arXiv:2601.05724
Speculative² Decoding (Saguaro)并行化draft与verificationarXiv:2603.03251 (ICLR 2026)
NanoSpec动态极小词表,40倍压缩arXiv:2605.26444
Speed-of-Light Bounds理论最优下界EACL 2026

1. 分层Speculative Decoding (HSD)

1.1 问题背景:联合不可追溯性

现有序列级验证方法面临**联合不可追溯性(Joint Intractability)**问题:

  • Token-wise验证:逐token独立接受,
  • 序列级验证:需要完整的联合概率
  • 联合分布计算量随token数指数增长,实际不可行

1.2 核心思想:层级分支重采样

HSD提出分层分支重采样策略,将联合分布分解为多层级条件分布:

Level 0: p_T(x_1)
Level 1: p_T(x_2 | x_1)
Level 2: p_T(x_3 | x_1, x_2)
...

关键创新

  • 每个层级只恢复该分支内的部分目标分布
  • 在最后接受的token位置立即进行重采样
  • 平衡超额概率质量与不足概率质量

1.3 形式化定义

为最后接受的token位置,为当前轮次长度:

def hierarchical_verification(draft_tokens, target_probs, draft_probs):
    """
    HSD验证流程
    """
    # 从后向前扫描
    k = gamma
    while k > 0 and accept_token(draft_tokens[k], target_probs[k], draft_probs[k]):
        k -= 1
    
    # 在位置k+1进行重采样
    residual_dist = compute_residual(target_probs[k+1], draft_probs[k+1])
    bonus_token = sample_from(residual_dist)
    
    return accepted_tokens[:k+1], bonus_token

1.4 理论保证

定理(HSD无损性)

这确保HSD在期望意义上恢复完整的目标分布,无信息损失。

1.5 实验结果

模型数据集接受率提升速度提升
Llama-3-8BHumanEval+8.2%+10.5%
Qwen2.5-7BMT-Bench+6.5%+8.3%
Llama-3-70BGSM8K+7.1%+9.8%

关键发现

  • 与EAGLE-3集成后,速度提升超过12%
  • 在多种模型家族上表现一致
  • 兼容多draft设置

1.6 代码实现

import torch
import torch.nn.functional as F
 
def hsd_verification(
    draft_tokens: torch.Tensor,      # [batch, seq_len]
    target_logits: torch.Tensor,     # [batch, seq_len, vocab_size]
    draft_logits: torch.Tensor,      # [batch, seq_len, vocab_size]
    gamma: int = 16
) -> tuple[torch.Tensor, torch.Tensor]:
    """
    Hierarchical Speculative Decoding验证
    
    Args:
        draft_tokens: draft模型生成的token序列
        target_logits: target模型的logits
        draft_logits: draft模型的logits
        gamma: 推测长度
    
    Returns:
        accepted: 被接受的token序列
        bonus: 额外采样的bonus token
    """
    batch_size, seq_len = draft_tokens.shape
    
    # 计算概率分布
    target_probs = F.softmax(target_logits, dim=-1)
    draft_probs = F.softmax(draft_logits, dim=-1)
    
    # 从后向前扫描找接受位置
    acceptance_mask = torch.zeros_like(draft_tokens, dtype=torch.bool)
    for pos in range(seq_len - 1, -1, -1):
        t_prob = target_probs[:, pos, draft_tokens[:, pos]]
        d_prob = draft_probs[:, pos, draft_tokens[:, pos]]
        ratio = (t_prob / d_prob.clamp(min=1e-8)).clamp(max=1.0)
        accept = torch.rand(batch_size) < ratio
        acceptance_mask[:, pos] = accept
        if not accept.all():
            break
    
    # 找到最后接受位置
    last_accept = acceptance_mask.float().cumsum(dim=1).argmax(dim=1)
    
    # 计算残差分布进行bonus采样
    bonus_tokens = []
    for b in range(batch_size):
        pos = last_accept[b].item() + 1
        if pos < seq_len:
            residual = (target_probs[b, pos] - draft_probs[b, pos]).clamp(min=0)
            residual = residual / residual.sum().clamp(min=1e-8)
            bonus = torch.multinomial(residual, 1).item()
        else:
            bonus = torch.argmax(target_probs[b, seq_len-1], dim=-1).item()
        bonus_tokens.append(bonus)
    
    accepted = draft_tokens.clone()
    for b in range(batch_size):
        accepted[b, last_accept[b].item()+1:] = 0
    
    return accepted, torch.tensor(bonus_tokens, device=draft_tokens.device)

2. Speculative Speculative Decoding (Saguaro)

2.1 问题背景:串行依赖瓶颈

标准Speculative Decoding存在内在的串行依赖

时间线: [Draft 1] → [Verify 1] → [Draft 2] → [Verify 2] → ...
         ↑ verifier必须等待draft完成

即使draft模型与target模型在不同硬件上,verification必须等待speculation完成才能开始。

2.2 核心思想:并行推测

SSD引入**“猜测的猜测”**机制:

  • 在verification进行期间,draft模型预先生成可能的验证结果
  • 如果实际验证结果在预测集合中,立即返回预先生成的token
  • 完全消除draft开销
时间线(SSD): [Draft 1 || Verify 1] → [Draft 2 || Verify 2] → ...
                   ↑ 并行执行

2.3 三大挑战与解决方案

挑战1:预测验证结果

不仅需要预测接受多少token,还需预测bonus token。

解决方案(Saguaro Cache)

class SaguaroCache:
    """
    使用最可能的draft logits预测bonus token
    """
    def __init__(self, draft_model, target_model):
        self.cache = {}
        
    def predict_bonus(self, draft_logits, target_logits):
        # 用draft logits预测bonus token,准确率高达90%
        top_k = torch.topk(draft_logits, k=3, dim=-1)
        # 检查top-k是否包含实际bonus
        bonus = self.sample_bonus(target_logits)
        return {
            'candidates': top_k.indices[0].tolist(),
            'bonus': bonus
        }

挑战2:接受率与预测准确率的权衡

  • 高接受率 → 宽预测集合 → 难以准确预测bonus
  • 低接受率 → 窄预测集合 → 预测准确但加速少

解决方案(Saguaro Sampling)

这一采样分布平衡了预测准确性和token质量。

挑战3:Cache Miss处理

当预测失败时的fallback策略:

批量大小最优Fallback策略
1立即重新speculate
8-16批量重新speculate
>32使用partial结果

2.4 理论分析

定理(SSD无损性)

SSD在期望意义上保持与标准SD相同的输出分布。

加速比分析

其中是Saguaro Cache命中率。

2.5 实验结果

配置方法速度提升绝对速度
Llama-3.1-70B (TP=4)AR Decoding1.0x50 tok/s
Llama-3.1-70B (TP=4)Standard SD3.3x165 tok/s
Llama-3.1-70B (TP=4)Saguaro SSD5.0x250 tok/s
  • 平均比最强SD基线快30%
  • 比自回归解码快5倍

2.6 代码实现

import torch
from typing import List, Tuple
 
class SaguaroSSD:
    """
    Speculative Speculative Decoding (Saguaro)
    """
    def __init__(self, draft_model, target_model, draft_device='cuda:0'):
        self.draft = draft_model
        self.target = target_model
        self.draft_device = draft_device
        self.cache = {}  # Saguaro Cache
        
    def forward_round(
        self, 
        input_ids: torch.Tensor,
        max_lookahead: int = 8
    ) -> Tuple[torch.Tensor, List[int]]:
        """
        SSD一轮:并行执行draft和verify
        """
        # Phase 1: 启动并行执行
        with torch.cuda.stream(self.draft_stream):
            # Draft model在前一个验证期间预先生成
            draft_output = self.draft.generate(
                input_ids,
                max_length=max_lookahead,
                return_dict=True
            )
            draft_tokens = draft_output.sequences[:, -max_lookahead:]
        
        # Phase 2: Target model验证
        with torch.no_grad():
            target_logits = self.target(
                torch.cat([input_ids, draft_tokens[0:1]], dim=-1)
            ).logits[-max_lookahead:]
        
        # Phase 3: 检查Saguaro Cache
        cache_key = self._get_cache_key(input_ids)
        if cache_key in self.cache:
            cached_result = self.cache[cache_key]
            if self._verify_cache_hit(cached_result, target_logits):
                return cached_result['tokens'], cached_result['bonus']
        
        # Phase 4: 标准验证流程
        accepted, bonus = self._standard_verify(
            draft_tokens, target_logits
        )
        
        # Phase 5: 更新Cache
        self._update_cache(cache_key, accepted, bonus)
        
        return accepted, bonus
    
    def _standard_verify(
        self, 
        draft_tokens: torch.Tensor,
        target_logits: torch.Tensor
    ) -> Tuple[torch.Tensor, int]:
        """
        标准验证流程(带Saguaro Sampling)
        """
        draft_probs = torch.softmax(self.draft(input_ids).logits, dim=-1)
        target_probs = torch.softmax(target_logits, dim=-1)
        
        # Saguaro Sampling: 平衡接受率和预测准确性
        saguaro_probs = torch.sqrt(draft_probs * target_probs.clamp(min=1e-8))
        saguaro_probs = saguaro_probs / saguaro_probs.sum(dim=-1, keepdim=True)
        
        # 验证过程
        accepted_len = 0
        for i in range(len(draft_tokens[0])):
            ratio = (target_probs[0, i, draft_tokens[0, i]] / 
                    draft_probs[0, i, draft_tokens[0, i]].clamp(min=1e-8))
            if torch.rand(1) < min(ratio.item(), 1.0):
                accepted_len += 1
            else:
                break
        
        # Bonus采样
        if accepted_len == len(draft_tokens[0]):
            bonus = torch.argmax(target_logits[-1], dim=-1).item()
        else:
            residual = (target_probs[0, accepted_len] - 
                       draft_probs[0, accepted_len]).clamp(min=0)
            bonus = torch.multinomial(residual / residual.sum(), 1).item()
        
        return draft_tokens[:, :accepted_len], bonus

3. NanoSpec:极小词表策略

3.1 问题背景:大词表瓶颈

现代LLM词表通常超过100k tokens(如Llama-3有128k,Qwen-2.5有152k),导致线性投影层成为计算瓶颈

Draft模型推理时间分解:
├── Attention计算: 30%
├── 前馈网络(FFN): 10%
└── LM Head(线性投影): 60%  ← 瓶颈!

3.2 核心思想:上下文感知的动态词表

NanoSpec的核心理念:每个生成步骤只需要极小的充分词表

核心洞察:语言生成的时序局部性

下一个token极大可能:

  • 出现在近期上下文中(“The cat sat on the ___ → mat”)
  • 是近期高概率候选的近端扩展

3.3 方法:动态词表构建

def build_minimal_vocabulary(
    input_ids: torch.Tensor,
    draft_logits: torch.Tensor,
    base_vocab_size: int = 3000,
    coverage_threshold: float = 0.99
) -> torch.Tensor:
    """
    NanoSpec: 为每个生成步骤动态构建极小词表
    
    Args:
        input_ids: 当前输入token序列
        draft_logits: draft模型的logits
        base_vocab_size: 基础词表大小(~3k)
        coverage_threshold: 覆盖率阈值
    
    Returns:
        active_vocab: 活跃token索引
    """
    vocab_size = draft_logits.shape[-1]
    
    # 1. 从近期上下文提取token
    context_tokens = set(input_ids[-64:].tolist())  # 最近64个token
    
    # 2. 从draft logits提取top-k候选
    top_k_probs, top_k_indices = torch.topk(
        draft_logits[0], 
        k=base_vocab_size
    )
    
    # 3. 合并:上下文token + top-k候选
    active_vocab = list(context_tokens | set(top_k_indices.tolist()))
    
    # 4. 覆盖率验证
    cumsum_probs = top_k_probs.cumsum(dim=0) / top_k_probs.sum()
    optimal_k = (cumsum_probs > coverage_threshold).nonzero()[0].item() + 1
    active_vocab = top_k_indices[:max(optimal_k, len(context_tokens))].tolist()
    
    return torch.tensor(active_vocab, device=draft_logits.device)

3.4 系统-算法协同设计

高动态、极小词表带来的稀疏内存访问问题:

问题:随机gather LM Head权重 → GPU内存访问效率低

解决方案

  1. 异步Gathering:词表查询与计算重叠
  2. GPU驻留状态管理:热点权重保持在shared memory
class AsyncVocabGather:
    """
    异步词表聚集,减少稀疏访问开销
    """
    def __init__(self, lm_head_weight, vocab_indices):
        self.weight = lm_head_weight
        self.vocab_indices = vocab_indices
        
    def forward(self, hidden_states):
        # 异步启动权重gather
        future = torch.cuda.Event()
        gathered_weight = self.weight[self.vocab_indices]
        
        # 与下一层计算重叠
        output = torch.matmul(hidden_states, gathered_weight.t())
        
        return output

3.5 实验结果

方法活跃词表大小草稿时间缩减端到端加速
Full Vocab128k0%1.0x
FR-Spec32k15%1.1x
DynaSpec27k18%1.12x
NanoSpec<3k51.6%1.17-1.29x

关键发现

  • 词表压缩超过40倍(128k → 3k)
  • 无需任何辅助训练参数
  • 与EAGLE-2、EAGLE-3兼容

3.6 代码实现

import torch
import torch.nn.functional as F
from typing import List
 
class NanoSpec:
    """
    NanoSpec: Minimalist In-Context Vocabulary for Speculative Decoding
    """
    def __init__(
        self, 
        draft_model,
        target_model,
        base_vocab_size: int = 3000
    ):
        self.draft = draft_model
        self.target = target_model
        self.base_vocab_size = base_vocab_size
        
    def build_context_aware_vocab(
        self,
        input_ids: torch.Tensor,
        draft_logits: torch.Tensor
    ) -> torch.Tensor:
        """
        动态构建上下文感知的极小词表
        """
        vocab_size = draft_logits.shape[-1]
        
        # 1. 上下文token集合
        context_window = 64
        context_tokens = set(input_ids[-context_window:].tolist())
        
        # 2. Top-k候选
        top_k = min(self.base_vocab_size * 2, vocab_size)
        _, top_indices = torch.topk(draft_logits[0], k=top_k)
        
        # 3. 合并并去重
        candidate_tokens = set(top_indices.tolist())
        active_tokens = list(context_tokens | candidate_tokens)
        
        # 4. 截取到目标大小
        if len(active_tokens) > self.base_vocab_size:
            # 按概率排序,优先保留高概率token
            token_probs = {
                t: draft_logits[0, t].item() for t in active_tokens
            }
            active_tokens = sorted(
                active_tokens, 
                key=lambda t: token_probs[t], 
                reverse=True
            )[:self.base_vocab_size]
        
        return torch.tensor(active_tokens, device=input_ids.device)
    
    def draft_with_minimal_vocab(
        self,
        input_ids: torch.Tensor,
        max_new_tokens: int = 16
    ) -> torch.Tensor:
        """
        使用极小词表进行draft
        """
        generated = input_ids.clone()
        
        for _ in range(max_new_tokens):
            with torch.no_grad():
                logits = self.draft(generated).logits[:, -1, :]
            
            # 动态构建极小词表
            active_vocab = self.build_context_aware_vocab(generated, logits)
            
            # 只在活跃词表上计算
            restricted_logits = logits[:, active_vocab]
            next_token = restricted_logits.argmax(dim=-1)
            
            generated = torch.cat([generated, next_token], dim=-1)
            
            if next_token.item() == 2:  # EOS token
                break
        
        return generated[:, input_ids.shape[1]:]

4. 理论极限:Speed-of-Light Bounds

4.1 问题背景

标准Speculative Decoding的加速比有理论上限吗?

4.2 分支随机游走框架

研究者利用**分支随机游走(Branching Random Walk)**刻画解码加速的理论极限:

核心洞察

其中是与draft-target匹配度相关的参数。

4.3 主要结论

  1. 下界:任何SD方法无法超过的加速比,其中是token接受率
  2. 实际意义:当前方法与理论极限的差距可作为效率评估指标
  3. 设计指导:揭示了draft模型质量与加速比的非线性关系

5. 方法对比与选择指南

方法核心优势适用场景额外开销
Hierarchical SD无损、提升接受率需要高接受率轻微计算
Saguaro SSD消除串行依赖大批量推理需要分离硬件
NanoSpec40倍词表压缩大词表模型需GPU优化
Speed-of-Light理论极限评估与基准仅理论

选择建议

输入: 模型词表大小、推理批量、硬件配置
输出: 推荐方法

if 词表大小 > 100k:
    if 有GPU优化能力:
        → NanoSpec
    else:
        → FR-Spec

if 批量大小 > 8 and 有分离硬件:
    → Saguaro SSD

if 接受率 < 0.7:
    → Hierarchical SD

if 需要评估效率:
    → Speed-of-Light Bounds

6. 未来研究方向

  1. 自适应方法:根据推理上下文动态选择最优SD变体
  2. 多模态扩展:将SD技术扩展到图像、音频生成
  3. 硬件协同:针对新型AI加速器(Groq、TPU)的SD优化
  4. 分布式场景:多设备SD的通信优化

参考文献