Speculative Decoding最新进展2026
概述
Speculative Decoding(推测解码)已成为加速大语言模型(LLM)推理的标准技术。它通过使用小型draft模型生成候选token,再由大型target模型并行验证,在保持输出分布不变的前提下实现加速。然而,标准方法仍存在多个瓶颈:验证阶段的联合概率难计算、draft与verification的串行依赖、大词表带来的线性投影开销等。
2025-2026年,研究者提出了多项创新技术来突破这些瓶颈:
| 方法 | 核心贡献 | 论文 |
|---|---|---|
| Hierarchical SD | 分层分支重采样,无损提升接受率 | arXiv:2601.05724 |
| Speculative² Decoding (Saguaro) | 并行化draft与verification | arXiv: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_token1.4 理论保证
定理(HSD无损性):
这确保HSD在期望意义上恢复完整的目标分布,无信息损失。
1.5 实验结果
| 模型 | 数据集 | 接受率提升 | 速度提升 |
|---|---|---|---|
| Llama-3-8B | HumanEval | +8.2% | +10.5% |
| Qwen2.5-7B | MT-Bench | +6.5% | +8.3% |
| Llama-3-70B | GSM8K | +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 Decoding | 1.0x | 50 tok/s |
| Llama-3.1-70B (TP=4) | Standard SD | 3.3x | 165 tok/s |
| Llama-3.1-70B (TP=4) | Saguaro SSD | 5.0x | 250 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], bonus3. 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内存访问效率低
解决方案:
- 异步Gathering:词表查询与计算重叠
- 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 output3.5 实验结果
| 方法 | 活跃词表大小 | 草稿时间缩减 | 端到端加速 |
|---|---|---|---|
| Full Vocab | 128k | 0% | 1.0x |
| FR-Spec | 32k | 15% | 1.1x |
| DynaSpec | 27k | 18% | 1.12x |
| NanoSpec | <3k | 51.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 主要结论
- 下界:任何SD方法无法超过的加速比,其中是token接受率
- 实际意义:当前方法与理论极限的差距可作为效率评估指标
- 设计指导:揭示了draft模型质量与加速比的非线性关系
5. 方法对比与选择指南
| 方法 | 核心优势 | 适用场景 | 额外开销 |
|---|---|---|---|
| Hierarchical SD | 无损、提升接受率 | 需要高接受率 | 轻微计算 |
| Saguaro SSD | 消除串行依赖 | 大批量推理 | 需要分离硬件 |
| NanoSpec | 40倍词表压缩 | 大词表模型 | 需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. 未来研究方向
- 自适应方法:根据推理上下文动态选择最优SD变体
- 多模态扩展:将SD技术扩展到图像、音频生成
- 硬件协同:针对新型AI加速器(Groq、TPU)的SD优化
- 分布式场景:多设备SD的通信优化