推测解码（Speculative Decoding）2026 全景解析：从 Medusa 到 DFlash，LLM 推理加速的四大技术路线与 Python 实战

import numpy as np
import matplotlib
matplotlib.use('Agg')
import matplotlib.pyplot as plt

def calculate_speedup(accept_rate: float, k: int, method: str) -> float:
    """计算不同推测解码方法的理论加速比"""
    if method == "standard":
        # 标准自回归：无加速
        return 1.0
    elif method == "medusa":
        # Medusa: 加速比 = 1 / (1 / (accept_rate * k) + overhead)
        overhead = 0.1  # 额外解码头的开销
        return 1.0 / (1.0 / (accept_rate * k) + overhead)
    elif method == "eagle":
        # Eagle: 隐藏状态预测，接受率通常更高
        overhead = 0.08
        return 1.0 / (1.0 / (accept_rate * k) + overhead)
    elif method == "dflash":
        # DFlash: Block-level 并行，overhead 更低
        overhead = 0.05
        return 1.0 / (1.0 / (accept_rate * k) + overhead)
    elif method == "lookahead":
        # Lookahead: 无训练开销，但接受率较低
        overhead = 0.15
        return 1.0 / (1.0 / (accept_rate * k) + overhead)
    return 1.0

# 模拟不同接受率下的加速比
accept_rates = np.linspace(0.3, 0.95, 20)
k_values = [3, 5, 7]
methods = ["medusa", "eagle", "dflash", "lookahead"]

fig, axes = plt.subplots(1, len(k_values), figsize=(18, 5))

for idx, k in enumerate(k_values):
    ax = axes[idx]
    for method in methods:
        speedups = [calculate_speedup(ar, k, method) for ar in accept_rates]
        ax.plot(accept_rates, speedups, label=method.capitalize(), linewidth=2)
    
    ax.set_xlabel('Accept Rate (α)', fontsize=12)
    ax.set_ylabel('Speedup (×)', fontsize=12)
    ax.set_title(f'k = {k}', fontsize=14)
    ax.legend(fontsize=10)
    ax.grid(True, alpha=0.3)
    ax.set_ylim(0, max(5, max([calculate_speedup(0.95, k, m) for m in methods]) + 1))

plt.suptitle('Speculative Decoding Speedup Analysis (2026)', fontsize=16, fontweight='bold')
plt.tight_layout()
plt.savefig('/tmp/speculative_speedup.png', dpi=150, bbox_inches='tight')
print("✅ 加速比分析图已保存至 /tmp/speculative_speedup.png")

# 打印具体数值
print(f"\n{'方案':<12} | {'k=3, α=0.6':<12} | {'k=5, α=0.7':<12} | {'k=7, α=0.8':<12}")
print("-" * 56)
for method in methods:
    s1 = calculate_speedup(0.6, 3, method)
    s2 = calculate_speedup(0.7, 5, method)
    s3 = calculate_speedup(0.8, 7, method)
    print(f"{method.capitalize():<12} | {s1:>10.2f}×    | {s2:>10.2f}×    | {s3:>10.2f}×")

"""
完整的 Speculative Decoding 实现
包含：Draft Model、Target Model 验证、接受率统计
适用于教学和生产环境参考
"""
from typing import List, Tuple, Optional
import numpy as np
from collections import Counter

class SimpleTokenizer:
    """简化的字符级 Tokenizer，用于演示"""
    
    def __init__(self, vocab: List[str]):
        self.vocab = vocab
        self.char_to_id = {ch: i for i, ch in enumerate(vocab)}
        self.id_to_char = {i: ch for i, ch in enumerate(vocab)}
    
    def encode(self, text: str) -> List[int]:
        return [self.char_to_id.get(ch, 0) for ch in text]
    
    def decode(self, ids: List[int]) -> str:
        return ''.join(self.id_to_char.get(i, '') for i in ids)

class DraftModel:
    """
    小型草稿模型：使用 n-gram 统计进行快速预测
    实际场景中可以是小型 LLM（如 1B 参数的蒸馏模型）
    """
    
    def __init__(self, n: int = 3):
        self.n = n
        self.ngram_counts: dict = {}
        self.total_counts: dict = {}
    
    def train(self, texts: List[str]):
        """从语料中统计 n-gram 频率"""
        for text in texts:
            for i in range(len(text) - self.n):
                context = text[i:i + self.n - 1]
                next_char = text[i + self.n - 1]
                if context not in self.ngram_counts:
                    self.ngram_counts[context] = Counter()
                    self.total_counts[context] = 0
                self.ngram_counts[context][next_char] += 1
                self.total_counts[context] += 1
    
    def predict_next(self, context: str, top_k: int = 5) -> List[Tuple[str, float]]:
        """基于 n-gram 预测下一个 token"""
        ctx = context[-(self.n - 1):] if len(context) >= self.n - 1 else context
        if ctx in self.ngram_counts:
            counts = self.ngram_counts[ctx]
            total = self.total_counts[ctx]
            probs = [(ch, count / total) for ch, count in counts.most_common(top_k)]
            return probs
        # 回退：均匀分布
        return [(' ', 0.5), ('.', 0.2), (',', 0.15), ('\n', 0.1), ('a', 0.05)]
    
    def generate_draft(self, prefix: str, num_tokens: int) -> List[str]:
        """快速生成 num_tokens 个候选 token"""
        tokens = []
        context = prefix
        for _ in range(num_tokens):
            candidates = self.predict_next(context)
            chars, probs = zip(*candidates)
            chosen = np.random.choice(chars, p=probs)
            tokens.append(chosen)
            context += chosen
        return tokens


class TargetModel:
    """
    大型目标模型：使用完整的概率分布进行验证
    实际场景中可以是 70B+ 参数的 LLM
    """
    
    def __init__(self, n: int = 4):
        self.n = n
        self.ngram_counts: dict = {}
        self.total_counts: dict = {}
    
    def train(self, texts: List[str]):
        """训练更大的 n-gram 模型（模拟更大的参数规模）"""
        for text in texts:
            for i in range(len(text) - self.n):
                context = text[i:i + self.n - 1]
                next_char = text[i + self.n - 1]
                if context not in self.ngram_counts:
                    self.ngram_counts[context] = Counter()
                    self.total_counts[context] = 0
                self.ngram_counts[context][next_char] += 1
                self.total_counts[context] += 1
    
    def get_prob_distribution(self, context: str) -> dict:
        """获取完整的概率分布"""
        ctx = context[-(self.n - 1):] if len(context) >= self.n - 1 else context
        if ctx in self.ngram_counts:
            total = self.total_counts[ctx]
            return {ch: count / total for ch, count in self.ngram_counts[ctx].items()}
        return {}
    
    def verify_tokens(
        self, prefix: str, draft_tokens: List[str]
    ) -> Tuple[List[str], int]:
        """
        并行验证草稿模型生成的 token 序列
        
        Returns:
            (accepted_tokens, accept_count): 接受的 token 列表和接受数量
        """
        accepted = []
        context = prefix
        
        for i, draft_token in enumerate(draft_tokens):
            # 目标模型验证：检查 draft_token 是否在目标模型的高概率 token 中
            probs = self.get_prob_distribution(context)
            
            if draft_token in probs and probs[draft_token] > 0.1:
                accepted.append(draft_token)
                context += draft_token
            else:
                break
        
        return accepted, len(accepted)


class SpeculativeDecoder:
    """推测解码器：整合草稿模型和目标模型"""
    
    def __init__(self, draft: DraftModel, target: TargetModel, k: int = 5):
        self.draft = draft
        self.target = target
        self.k = k  # 每次推测的 token 数量
        self.stats = {
            'total_steps': 0,
            'total_accepted': 0,
            'total_draft_tokens': 0,
        }
    
    def generate(self, prompt: str, max_tokens: int) -> str:
        """使用推测解码生成文本"""
        result = prompt
        tokens_generated = 0
        
        while tokens_generated < max_tokens:
            remaining = max_tokens - tokens_generated
            k = min(self.k, remaining)
            
            # Step 1: 草稿模型快速生成 k 个 token
            draft_tokens = self.draft.generate_draft(result, k)
            self.stats['total_draft_tokens'] += k
            
            # Step 2: 目标模型并行验证
            accepted, accept_count = self.target.verify_tokens(result, draft_tokens)
            
            # Step 3: 接受匹配的 token
            if accepted:
                result += ''.join(accepted)
                tokens_generated += len(accepted)
                self.stats['total_accepted'] += len(accepted)
            
            # Step 4: 如果不匹配，用目标模型重新生成
            if accept_count < k:
                # 从断点处用目标模型重新生成 1 个 token
                probs = self.target.get_prob_distribution(result)
                if probs:
                    next_token = max(probs, key=probs.get)
                    result += next_token
                    tokens_generated += 1
            
            self.stats['total_steps'] += 1
            
            # 安全限制，防止无限循环
            if self.stats['total_steps'] > max_tokens * 3:
                break
        
        return result
    
    def get_accept_rate(self) -> float:
        """计算总体接受率"""
        if self.stats['total_draft_tokens'] == 0:
            return 0.0
        return self.stats['total_accepted'] / self.stats['total_draft_tokens']
    
    def get_speedup(self) -> float:
        """计算相对于标准自回归的加速比"""
        if self.stats['total_steps'] == 0:
            return 1.0
        # 标准方法需要 total_tokens 步，推测方法用了 total_steps 步
        total_tokens = self.stats['total_accepted'] + (
            self.stats['total_steps'] - self.stats['total_accepted'] // max(1, self.k)
        )
        return max(1.0, total_tokens / max(1, self.stats['total_steps']))


# ========== 使用示例 ==========
if __name__ == "__main__":
    # 训练语料（这里用英文示例，实际应用中需要大量文本）
    corpus = [
        "the quick brown fox jumps over the lazy dog. "
        "the quick brown fox is a fast animal. "
        "the lazy dog sleeps all day long. "
        "the fox and the dog are in the garden. "
        "a quick brown animal moves through the forest. "
        "the dog barks at the fox. the fox runs away. "
    ] * 10  # 重复以增加统计量
    
    # 初始化模型
    draft = DraftModel(n=3)
    target = TargetModel(n=4)
    
    draft.train(corpus)
    target.train(corpus)
    
    # 创建推测解码器
    decoder = SpeculativeDecoder(draft, target, k=3)
    
    # 生成文本
    prompt = "the "
    generated = decoder.generate(prompt, max_tokens=50)
    
    print(f"Prompt: {prompt}")
    print(f"Generated: {generated}")
    print(f"\n📊 性能统计:")
    print(f"   接受率: {decoder.get_accept_rate():.1%}")
    print(f"   理论加速比: {decoder.get_speedup():.2f}×")
    print(f"   总步数: {decoder.stats['total_steps']}")
    print(f"   接受 token 数: {decoder.stats['total_accepted']}")
    
    # 对比：标准自回归（纯目标模型）
    print(f"\n🔬 对比实验：改变 k 值的影响")
    for k in [1, 3, 5, 7]:
        dec = SpeculativeDecoder(draft, target, k=k)
        _ = dec.generate(prompt, max_tokens=50)
        print(f"   k={k}: 接受率={dec.get_accept_rate():.1%}, 加速比={dec.get_speedup():.2f}×")