执行摘要

• 一句话：为EAGLE推测解码添加自适应步数调整，根据接受长度动态切换运行时状态。
• 推荐动作：建议技术管理者和工程师精读此PR，重点关注：
  1) 自适应策略的EMA设计和滞后阈值如何平衡响应速度与稳定性；
  2) 运行时状态切换机制如何实现零开销原子操作，避免CUDA图重捕获；
  3) CUDA图兼容性检查的风险及潜在解决方案。对于实施类似动态调整的系统具有较高参考价值。

功能与动机

PR body指出，静态speculative_num_steps无法适应变化的工作负载：步数过小导致草稿模型能力未充分利用，步数过大则生成过多候选令牌被浪费。自适应机制旨在根据实际接受率动态调整步数，以最大化吞吐量，并提供了基准数据展示静态步数下的吞吐量瓶颈。

实现拆解

1. 新增自适应策略模块（adaptive_spec_params.py）：定义AdaptiveSpeculativeParams类，使用EMA（默认alpha=0.2）跟踪批次平均接受长度，根据候选步数列表（默认[1,3,7]）和滞后阈值决策何时上下调整步数。关键方法update()在满足预热批次和更新间隔后触发_recompute_params()进行步数切换。
2. 新增运行时状态管理模块（adaptive_runtime_state.py）：引入SpecRuntimeState数据类封装各阶段（draft、verify、extend）的注意力后端和CUDA图运行器；定义AdaptiveSpecWorker协议要求worker实现build_adaptive_runtime_state()和apply_runtime_state()；AdaptiveController负责初始化时预构建所有候选步数的状态池，并在on_verify_complete()中根据策略决策原子切换状态。
3. 修改EAGLEWorker核心逻辑（eagle_worker.py）：集成AdaptiveController，在初始化时根据server_args.speculative_adaptive标志创建控制器并注册初始状态；实现协议方法以支持状态构建和应用，通过_override_worker_state()临时覆盖服务器参数捕获CUDA图。
4. 测试和基准配套：新增单元测试test_adaptive_spec_params.py验证策略逻辑（如预热、间隔、滞后阈值），新增端到端测试test_adaptive_speculative.py启动真实服务器测试状态切换和GSM8K准确性；新增基准脚本bench_adaptive_speculative.py对比自适应与静态服务器性能。
5. 文档和配置更新：新增文档adaptive_speculative_decoding.md说明使用方法和参数，更新server_args.py添加speculative_adaptive和speculative_adaptive_config命令行选项。

关键文件：

• python/sglang/srt/speculative/adaptive_spec_params.py（模块 推测解码；类别 source；类型 core-logic；符号 load_adaptive_config, AdaptiveSpeculativeParams, init, update）: 定义了自适应策略核心类AdaptiveSpeculativeParams，负责EMA跟踪接受长度和步数决策逻辑。
• python/sglang/srt/speculative/adaptive_runtime_state.py（模块 推测解码；类别 source；类型 core-logic；符号 SpecRuntimeState, AdaptiveSpecWorker, build_adaptive_runtime_state, apply_runtime_state）: 定义了运行时状态管理核心，包括SpecRuntimeState数据类、AdaptiveSpecWorker协议和AdaptiveController控制器，负责状态池和原子切换。
• python/sglang/srt/speculative/eagle_worker.py（模块 推测解码；类别 source；类型 core-logic；符号 apply_runtime_state, build_adaptive_runtime_state, _override_worker_state, init）: 修改EAGLEWorker以支持自适应协议，集成AdaptiveController并实现状态构建和应用方法，是功能落地的核心。
• test/registered/unit/spec/test_adaptive_spec_params.py（模块 单元测试；类别 test；类型 test-coverage；符号 TestAdaptiveSpeculativeParams, test_initial_steps_snap_to_nearest_candidate_preferring_larger_step, test_update_respects_warmup_and_interval, test_empty_batches_do_not_consume_warmup_or_shift_steps）: 新增单元测试，验证AdaptiveSpeculativeParams的策略逻辑，覆盖预热、间隔、滞后阈值等关键行为。
• benchmark/bench_adaptive_speculative.py（模块 基准测试；类别 source；类型 dependency-wiring；符号 build_phase_plan, send_request, run_phase, summarize_phases）: 新增基准测试脚本，对比自适应与静态服务器的吞吐量、延迟和接受长度，用于性能验证。

关键符号：AdaptiveSpeculativeParams.update, AdaptiveController.on_verify_complete, EAGLEWorker.build_adaptive_runtime_state, EAGLEWorker.apply_runtime_state

关键源码片段

python/sglang/srt/speculative/adaptive_runtime_state.py

定义了运行时状态管理核心，包括SpecRuntimeState数据类、AdaptiveSpecWorker协议和AdaptiveController控制器，负责状态池和原子切换。

@dataclass
class SpecRuntimeState:
    """A complete set of runtime resources bound to a specific speculative decoding configuration.
​
    Each decode round runs three stages — draft, verify, extend — and every
    stage has shape-dependent resources (attention backends and CUDA graphs)
    that must match the current configuration. Switching adaptive steps
    means swapping the entire state atomically.
    """
    speculative_num_steps: int # 当前步数配置
    speculative_num_draft_tokens: int # 对应草稿令牌数（steps+1）
    draft_attn_backend: "AttentionBackend | None" # 草稿阶段注意力后端
    cuda_graph_runner: "EAGLEDraftCudaGraphRunner | None" # 草稿阶段 CUDA 图运行器
    target_attn_backend: "AttentionBackend" # 验证阶段注意力后端
    target_graph_runner: "CudaGraphRunner | CPUGraphRunner | None" # 验证阶段图运行器
    draft_extend_attn_backend: "AttentionBackend | None" # 扩展阶段注意力后端
    cuda_graph_runner_for_draft_extend: "EAGLEDraftExtendCudaGraphRunner | None" # 扩展阶段 CUDA 图运行器
​
class AdaptiveController:
    """Facade that owns adaptive decision-making and runtime state switching.
​
    Works with any worker that implements ``AdaptiveSpecWorker`` protocol.
    """
    def __init__(self, worker: AdaptiveSpecWorker, config_path: str | None = None):
        self.worker = worker
        cfg = load_adaptive_config(config_path) # 加载配置文件
        self.params = AdaptiveSpeculativeParams(
            initial_steps=worker.speculative_num_steps, config=cfg
        )
        self._states: dict[int, SpecRuntimeState] = {} # 状态池，键为步数
​
    def init_states(self) -> None:
        """Build and register runtime states for all candidate steps."""
        for steps in self.params.candidate_steps:
            if steps in self._states:
                continue
            # 委托 worker 构建对应步数的运行时状态
            state = self.worker.build_adaptive_runtime_state(
                speculative_num_steps=steps,
                speculative_num_draft_tokens=steps + 1,
            )
            self._states[steps] = state
        self._activate(self.params.current_steps) # 激活初始状态
​
    def on_verify_complete(self, accept_lengths: list[int]) -> None:
        """Feed verify results; switch runtime state if EMA warrants it."""
        if self.params.update(accept_lengths): # 如果策略决策步数变化
            self._activate(self.params.current_steps) # 激活新步数对应状态

python/sglang/srt/speculative/eagle_worker.py

修改EAGLEWorker以支持自适应协议，集成AdaptiveController并实现状态构建和应用方法，是功能落地的核心。

def __init__(self, server_args: ServerArgs, gpu_id: int, tp_rank: int, dp_rank: Optional[int],
             moe_ep_rank: int, attn_cp_rank: int, moe_dp_rank: int, nccl_port: int,
             target_worker: TpModelWorker):
    # ... 原有初始化代码 ...
    self.speculative_num_steps = server_args.speculative_num_steps
    self.speculative_num_draft_tokens = server_args.speculative_num_draft_tokens
​
    # Adaptive speculative
    self.adaptive_controller: Optional[AdaptiveController] = None
    if server_args.speculative_adaptive: # 检查是否启用自适应
        self.adaptive_controller = AdaptiveController(
            self, config_path=server_args.speculative_adaptive_config
        )
    # ... 后续初始化 ...
    if self.adaptive_controller is not None:
        # 注册初始运行时状态，以便控制器管理
        self.adaptive_controller.register(
            SpecRuntimeState(
                speculative_num_steps=self.speculative_num_steps,
                speculative_num_draft_tokens=self.speculative_num_draft_tokens,
                draft_attn_backend=self.draft_attn_backend,
                cuda_graph_runner=self.cuda_graph_runner,
                target_attn_backend=self.target_worker.model_runner.attn_backend,
                target_graph_runner=self.target_worker.model_runner.graph_runner,
                draft_extend_attn_backend=self.draft_extend_attn_backend,
                cuda_graph_runner_for_draft_extend=self.cuda_graph_runner_for_draft_extend,
            )
        )
        self.adaptive_controller.init_states() # 预构建所有候选步数的状态
​
def apply_runtime_state(self, state: SpecRuntimeState) -> None:
    """Apply a pre-built runtime state to this worker."""
    if self.speculative_num_steps == state.speculative_num_steps:
        return # 步数未变，无需切换
    logger.info(f"Switch adaptive runtime state: steps {self.speculative_num_steps} -> {state.speculative_num_steps}")
    self.speculative_num_steps = state.speculative_num_steps
    self.speculative_num_draft_tokens = state.speculative_num_draft_tokens
    # 原子更新各阶段资源引用
    self.draft_attn_backend = state.draft_attn_backend
    self.cuda_graph_runner = state.cuda_graph_runner
    self.target_worker.model_runner.attn_backend = state.target_attn_backend
    self.target_worker.model_runner.graph_runner = state.target_graph_runner
    self.draft_extend_attn_backend = state.draft_extend_attn_backend
    self.cuda_graph_runner_for_draft_extend = state.cuda_graph_runner_for_draft_extend

评论区精华

• 设计简化建议：gemini-code-assist建议简化AdaptiveSpeculativeParams.__init__中的current_steps初始化逻辑，作者可能已采纳以提高可读性。
• CUDA图兼容性风险：chatgpt-codex-connector警告在cuda_graph_runner.py和eagle_draft_extend_cuda_graph_runner.py中新增的令牌数检查可能导致DP注意力模式下各rank执行路径分歧，引发同步问题或性能回归；此问题在review中未明确解决。
• 测试覆盖要求：Qiaolin-Yu要求添加单元测试和端到端测试，作者后续提交中添加了test_adaptive_spec_params.py和test_adaptive_speculative.py，满足要求并获得批准。
• 类型注解改进：Qiaolin-Yu建议在adaptive_runtime_state.py中使用具体类型而非object，作者回复“done”表示已修复。

• 功能扩展讨论：issue评论中liquanfeng询问是否支持EAGLE3，作者回复已添加支持。

  • 简化自适应参数初始化逻辑 (design): 作者可能已采纳，但review中未显式确认修改。
  • CUDA图令牌数检查引入同步风险 (correctness): 此风险在review中未明确解决，可能需后续验证或修复。
  • 要求添加单元测试和端到端测试 (testing): 作者后续提交中添加了test_adaptive_spec_params.py和test_adaptive_speculative.py，满足要求并获得批准。

风险与影响

• 风险：
  • CUDA图同步风险：cuda_graph_runner.py中新增的is_num_tokens_supported检查在DP注意力模式下可能因各rank本地令牌数不同导致CUDA图与eager执行路径分歧，引发集体操作不同步或崩溃（chatgpt-codex-connector指出）。
  • 自适应策略振荡：AdaptiveSpeculativeParams的EMA参数和滞后阈值配置不当可能导致步数频繁切换，增加开销并降低吞吐量；默认参数虽经过调优，但需用户理解配置。
  • 兼容性限制：目前仅支持EAGLE topk=1，PR body提到EAGLEWorkerV2支持为后续TODO，且未覆盖其他推测算法（如N-gram），限制了使用范围。
  • 新增复杂性：运行时状态池管理和原子切换增加了代码复杂度，可能引入隐蔽bug，尤其是在多线程或分布式环境下。
  • 配置错误静默失败：chatgpt-codex-connector指出speculative_adaptive_config标志未自动启用自适应行为，若用户仅提供配置路径而遗漏--speculative-adaptive，系统将静默回退到静态解码。
• 影响：
  • 用户影响：为用户提供了动态优化推测解码吞吐量的能力，尤其是在接受率变化的工作负载下可提升性能；新增命令行选项和配置文件增加了使用灵活性，但需学习新参数。
  • 系统影响：在核心推测解码路径中引入状态切换逻辑，可能轻微增加运行时开销，但设计为零开销引用交换；预构建多个CUDA图可能增加内存占用，但避免了在线重捕获。
  • 团队影响：为推测解码模块增加了重要功能，需要维护新组件和测试；与近期历史PR（如PR 22908、22832）同属推测解码改进线，展现了该领域的持续投入。
  • 风险标记：CUDA图同步风险, 自适应策略振荡, topk=1限制, 配置静默失败

关联脉络

• PR #22908 [AMD] Resolve Qwen3.5 MTP (speculative decoding) radix cache conflict.: 同属推测解码功能线，涉及设备感知和缓存冲突修复，本PR的自适应机制可能与此类底层优化交互。
• PR #22832 [sgl] fix incorrect behavior in cuda graph draft extend: 修复CUDA图推测解码扩展逻辑，本PR新增自适应状态切换同样依赖CUDA图运行器，需关注兼容性。
• PR #22088 [sgl] add support for weight update function in spedec: 扩展EAGLE推测解码工作者功能，本PR在此基础上新增自适应参数调整，属于同一模块的持续演进。