From 30565e182f6ff838a8a37e4c07185ddbdfb166d8 Mon Sep 17 00:00:00 2001 From: huangweizhe1 Date: Thu, 9 Jul 2026 11:16:24 +0800 Subject: [PATCH 1/2] feat: support adaptive speculative decoding with per-seq validate pruning. Implement adaptive speculative decode that prunes low-confidence draft tokens before target validation, reducing validate compute via ATB variable-length q_len support. All ranks compute pruning independently (no cross-rank broadcast needed) since draft probs are deterministic. --- .../design/adaptive_speculative_decoding.md | 283 ++++++ tests/core/runtime/CMakeLists.txt | 25 + .../adaptive_speculative_controller_test.cpp | 125 +++ .../speculative_profile_registry_test.cpp | 63 ++ xllm/core/common/global_flags.h | 4 + xllm/core/common/options.cpp | 3 + xllm/core/common/options.h | 4 + .../core/distributed_runtime/comm_channel.cpp | 19 + xllm/core/distributed_runtime/comm_channel.h | 4 + xllm/core/distributed_runtime/engine.h | 8 + xllm/core/distributed_runtime/llm_engine.cpp | 14 + xllm/core/distributed_runtime/llm_engine.h | 4 + xllm/core/distributed_runtime/master.cpp | 3 + .../distributed_runtime/remote_worker.cpp | 5 + xllm/core/distributed_runtime/remote_worker.h | 4 + .../speculative_engine.cpp | 12 + .../distributed_runtime/speculative_engine.h | 6 + .../distributed_runtime/worker_service.cpp | 18 + .../core/distributed_runtime/worker_service.h | 6 + .../framework/config/speculative_config.cpp | 18 + .../framework/config/speculative_config.h | 8 +- xllm/core/runtime/CMakeLists.txt | 4 + .../adaptive_speculative_controller.cpp | 203 ++++ .../runtime/adaptive_speculative_controller.h | 57 ++ xllm/core/runtime/eagle3_worker_impl.cpp | 15 +- xllm/core/runtime/mtp_worker_impl.cpp | 923 ++++++++++++++++-- xllm/core/runtime/mtp_worker_impl.h | 63 +- xllm/core/runtime/options.h | 4 + .../runtime/speculative_profile_registry.cpp | 91 ++ .../runtime/speculative_profile_registry.h | 56 ++ xllm/core/runtime/speculative_worker_impl.cpp | 34 + xllm/core/runtime/speculative_worker_impl.h | 3 + xllm/core/runtime/worker.cpp | 7 + xllm/core/runtime/worker.h | 4 + xllm/core/runtime/worker_client.cpp | 5 + xllm/core/runtime/worker_client.h | 4 + xllm/core/scheduler/profile/CMakeLists.txt | 1 + .../scheduler/profile/profile_manager.cpp | 177 ++++ xllm/core/scheduler/profile/profile_manager.h | 7 + xllm/proto/worker.proto | 9 + xllm/xllm.cpp | 4 + 41 files changed, 2229 insertions(+), 78 deletions(-) create mode 100644 docs/zh/design/adaptive_speculative_decoding.md create mode 100644 tests/core/runtime/adaptive_speculative_controller_test.cpp create mode 100644 tests/core/runtime/speculative_profile_registry_test.cpp create mode 100644 xllm/core/runtime/adaptive_speculative_controller.cpp create mode 100644 xllm/core/runtime/adaptive_speculative_controller.h create mode 100644 xllm/core/runtime/speculative_profile_registry.cpp create mode 100644 xllm/core/runtime/speculative_profile_registry.h diff --git a/docs/zh/design/adaptive_speculative_decoding.md b/docs/zh/design/adaptive_speculative_decoding.md new file mode 100644 index 0000000000..23579cbdff --- /dev/null +++ b/docs/zh/design/adaptive_speculative_decoding.md @@ -0,0 +1,283 @@ +# Adaptive Speculative Decoding 设计稿 + +## 背景 + +当前 xLLM 的 MTP 投机解码使用静态 `num_speculative_tokens`。每个 decode step 固定执行 K 次 draft model,然后将 `K + 1` 个 token 送入 target model 验证。静态 K 实现简单,但不同 batch size、不同请求局部和不同 draft 置信度下,最优验证宽度可能不同: + +- draft 接受率高时,较大的验证宽度可以摊薄 target validate 成本。 +- draft 接受率低时,较大的验证宽度会浪费 target validate 宽度。 +- batch size 和 prefix length 变化时,validate 的边际耗时也会变化。 + +本设计的第一个版本只实现 validate 前自适应剪枝:draft model 仍按配置的 `num_speculative_tokens` 完整推理,adaptive controller 在 target model validate 前选择每个请求要送入验证的 draft 前缀长度。 + +此方案参考 DSPark 自适应投机推理算法(区别主要在于计算接受概率时 DSPark 使用模型本身的置信度头,本实现先使用 draft model 输出的 softmax logits),但还未专门针对 graph mode 和 ZOS 做适配,后续还可以进一步支持对 draft 的 early stop。 + +## 目标 + +v1 目标: + +1. 只支持 MTP。 +2. 默认关闭 adaptive,静态 MTP 行为保持兼容。 +3. `num_speculative_tokens` 继续表示最大推测长度 `max_sl`。 +4. adaptive 开启后,draft 仍完整执行 `max_sl` 步,只在 validate 前按 draft selected token 概率剪枝。 +5. 支持 per-request 不同的验证长度,validate 阶段通过 ATB kernel 的 variable q_len 能力只计算有效 tokens,真正减少 attention/FFN 计算量。 +6. 支持 TP/CP 多 rank 场景下同步剪枝结果,避免不同 rank validate width 不一致。 +7. graph mode、DP/EP 场景下暂时关闭 adaptive,退回静态 MTP。 + +非目标: + +1. v1 不直接中断 draft loop,不做 draft early stop。 +2. v1 不支持 Eagle3 和 Suffix。 +3. v1 不接入 TPOT SLO,只预留后续设计空间。 +4. v1 不改变 scheduler token/KV 预算,仍按 `max_sl` 保守调度。 + +## 术语 + +- `max_sl`:配置的最大 speculative length,即 `num_speculative_tokens`。 +- `prefix_len_i`:第 i 个请求本轮送入 target validate 的 draft token 前缀长度,范围为 `[0, max_sl]`。 +- `effective_sl`:本轮 batch 的最大 validate draft 宽度,等于 `max(prefix_len_i)`,运行时最小取 1。用于截断 draft_outputs 和作为 RejectionSampler 的 padded width。 +- `per_seq_val_tokens`:每个请求实际参与 validate 的 token 数,等于 `prefix_len_i + 1`。ATB kernel 按此值只计算有效 tokens。 +- `num_val_tokens`:target validate 的 padded 输入宽度,等于 `effective_sl + 1`,用于 RejectionSampler 的张量对齐。 +- `path_prob_i(k)`:第 i 个请求前 k 个 draft token 全部被接受的估计概率。 + +## 参数 + +配置保持: + +```text +num_speculative_tokens = max_sl +``` + +新增实验参数: + +```text +enable_adaptive_speculative_decode = false +adaptive_speculative_min_gain = 0.0 +``` + +含义: + +- `enable_adaptive_speculative_decode`:总开关。默认 false。 +- `adaptive_speculative_min_gain`:把一个候选 token 加入验证集所需的最小相对吞吐收益。默认 0,表示只要估计吞吐严格提升即可加入。 + +不新增单独的 adaptive 最大 SL 参数,直接复用 `num_speculative_tokens` 作为上限。也不提供 target/draft step time 手写参数。 + +## 核心策略 + +MTP 每步 draft output 的 `sample_output.probs` 会被压缩成 selected-only 概率。对每个请求按位置计算路径概率: + +```text +path_prob_i(k) = product_{j=1..k} p_i(j) +``` + +其中 `p_i(j)` 是 draft 第 j 步采样 token 经过 softmax 后的概率。由于概率都在 `[0, 1]`,同一个请求内 `path_prob_i(k)` 随 k 单调不增,所以如果后置 token 被选中,前置 token 必须一起被选中。 + +controller 把所有请求、所有位置的候选 token 展平成候选列表: + +```text +candidate = (seq_id, prefix_len, path_prob) +``` + +然后按 `path_prob` 从大到小排序,从空验证集开始贪心尝试加入候选。每次尝试只会把某个请求的 `prefix_len` 增大到候选位置;如果加入后估计吞吐提升超过 `adaptive_speculative_min_gain`,则接受该候选,否则回滚并跳过该候选继续尝试后面的候选。 + +这里选择"跳过而不是停止"的原因是:某个候选可能因为边际 validate time 成本过高而不划算,但后面另一个请求的低位置候选成本更低,仍然可能提升整体吞吐。 + +估计吞吐公式: + +```text +score(prefixes) + = expected_emitted(prefixes) + / estimated_time(prefixes) + +expected_emitted(prefixes) + = batch_size + sum_i sum_{k=1..prefix_len_i} path_prob_i(k) + +estimated_time(prefixes) + = full_draft_time(max_sl) + + estimated_validate_time(prefixes) +``` + +其中 `estimated_validate_time` 按 per-seq 累加(而非按 batch max 估计): + +```text +estimated_validate_time(prefixes) + = intercept + + sum_i (batch_ms + query_token_ms * q_i + query_prefix_ms * q_i * kv_i) +``` + +其中 `q_i = prefix_len_i + 1`,`kv_i` 是第 i 个请求的 kv cache 长度。 + +说明: + +- `batch_size` 表示每个请求至少会由 target 产生一个 token。 +- `full_draft_time(max_sl)` 使用本轮完整 draft loop 的实测耗时,因此不需要估计 draft step time。 +- `estimated_validate_time` 来自 `ProfileManager` 拟合出的 validate predictor 的系数,按 per-seq 展开计算。 +- 如果 predictor 不可用(`SpeculativeProfileRegistry` 为空),整个 adaptive 路径不会被启用(入口门控)。 + +## 时间估计 + +v1 复用现有 ProfileManager 的离线 profile/线性拟合方式。开启 step-time profile 且 MTP `num_speculative_tokens > 1` 时,ProfileManager 额外采样 validate-like workload,并拟合: + +```text +T_validate(B, q_len, prefix) + = c0 + + c1 * B + + c2 * B * q_len + + c3 * B * q_len * prefix +``` + +其中: + +- `B` 是 decode batch size。 +- `q_len = sl + 1` 是 target validate query 宽度。 +- `prefix` 是 batch 的平均 prefix length。 +- 不引入 `q_len^2` 项,因为 speculative validate 的 q_len 通常较小,二次项容易被噪声放大。 + +ProfileManager 拟合完成后: + +1. 在 master 进程本地写入 `SpeculativeProfileRegistry`。 +2. 通过 `SetSpeculativeValidateTimePredictor` RPC 广播到 target engine 的 worker rank。 +3. 每个 worker 在 runtime 进程本地查询 registry,预测不同 `sl` 对应的 validate time。 +4. registry 中有 predictor 是 adaptive 路径启用的前置条件(门控),确保 profile 阶段不会触发 adaptive 的 HCCL broadcast。 + +profile 采样时增加了 KV cache 容量检查,跳过超出可用 blocks 90% 的 batch_size × prefix_len 组合,防止 profile 阶段 OOM。 + +## Per-Seq Variable-Length Validate + +v1 支持 batch 内每个请求不同的 validate token 数(`per_seq_val_tokens`): + +1. `prepare_validate_inputs` 按 per-seq q_len 构造 ATB kernel 输入(`q_seq_lens`/`q_cu_seq_lens`),kernel 只计算有效 tokens。 +2. `fill_validate_input_from_draft_outputs` 按 draft step 分组,用 `index_select` + `index_copy_` 向量化填充 draft tokens。 +3. Target model forward 输出 flat `[total_tokens, vocab]` logits。 +4. **Fast-path**:如果所有 seq 都保持满 SL(`total_tokens == batch * max_val_tokens`),直接 `view` 零拷贝,无额外开销。 +5. **Slow-path**:per-seq 不均匀时,用 `index_copy_` 将 flat logits scatter 到 `[batch * max_val_tokens, vocab]` 的 padded layout,padding 位填 `-1e9`。 +6. RejectionSampler 使用 padded layout 做 rejection sampling,padding 位自然被 reject。 +7. `apply_pruned_prefix_lengths` 在 rejection sampling 之后按 per-seq prefix 裁剪输出。 + +## MTP Worker 接入 + +`MTPWorkerImpl::step_decode()` 的自适应路径: + +1. 按静态 MTP 原逻辑完整执行 `max_sl` 步 draft。 +2. 记录本轮完整 draft loop 的实测耗时 `draft_latency_ms`。 +3. 从 `draft_outputs` 提取 `[batch, max_sl]` 的 selected token 概率矩阵。 +4. 计算每个 seq 的 kv_len 作为 `per_seq_kv_lens`。 +5. 控制 rank 调用 `AdaptiveSpeculativeController::select_pruned_prefix_lengths()` 得到每个请求的 `prefix_len_i`。 +6. 将 `prefix_lengths` 同步给其它 rank。 +7. 如果同步失败,则退回所有请求 `prefix_len_i = max_sl`。 +8. 计算 `effective_sl = max(1, max(prefix_lengths))` 和 `per_seq_val_tokens`。 +9. 截断 `draft_outputs` 到 `effective_sl`。 +10. 用 `per_seq_val_tokens` 构造 variable-length validate input。 +11. Target validate 后,根据 total_tokens 是否等于 padded_total 选择 fast-path 或 slow-path padding。 +12. Rejection sampling + per-request prefix 裁剪。 +13. Cache 写入、metrics 统计都使用剪枝后的结果。 + +静态路径保持原逻辑,不调用 controller,也不消费 profile predictor。 + +## 剪枝后的 Validate 输出 + +因为 batch 内每个请求的 `prefix_len_i` 可以不同,而 RejectionSampler 的张量宽度取 batch 内最大值,所以 rejection sampler 之后需要逐行修正输出: + +1. 对 `prefix_len_i < effective_sl` 的请求,保留 `[0, prefix_len_i]` 的输出。 +2. `prefix_len_i` 位置作为 target 修正 token/bonus token 保留,保证每个请求至少推进一个 token。 +3. `prefix_len_i + 1` 之后的位置置为 `-1`。 +4. 对应 embedding 位置清零,避免后续 cache 读取未被选中的 validate 结果。 + +这样 `EmbeddingCache::write_target_context()` 可以继续按 accepted prefix 解析 `next_tokens`,`all_draft_accepted` 也会基于本轮 `effective_sl` 正确判断。 + +## Metrics + +现有指标继续复用: + +- `speculative_num_draft_tokens_total` +- `speculative_num_accepted_tokens_total` + +adaptive prune 下,draft token 计数不再使用 `batch_size * effective_sl`,而是使用: + +```text +selected_draft_tokens = sum_i prefix_len_i +``` + +accepted token 计数只统计每个请求 `[0, prefix_len_i)` 范围内真正被接受的 draft token,避免把剪掉的位置当成 reject。 + +## 多 Rank 同步 + +TP/CP 多 rank 场景下,target validate width 必须一致,否则 collective shape 和调用顺序可能分叉。v1 使用控制 rank 决策: + +1. 所有 rank 都完整执行相同的 draft loop。 +2. 控制 rank 计算 `prefix_lengths`。 +3. 控制 rank 将 `prefix_lengths` 广播给其它 rank。 +4. 所有 rank 使用同一个 `effective_sl` 和同一组 per-request prefix 做 validate。 + +同步路径: + +- Torch backend:优先使用 `parallel_args_.tp_group_`,否则使用 `process_group_`,通过小 `int32` tensor broadcast 同步。 +- NPU ATB backend:Torch `ProcessGroup` 可能不可用,因此复用 ATB mapping 中的 `ATTN_TP` parallel info 初始化 HCCL control comm,并通过 `HcclAllReduce` 同步 prefix list。 + +如果没有可用同步组,adaptive prune 自动退回静态 `max_sl`,避免 rank 本地剪枝导致 hang。 + +## Graph Mode + +graph mode 暂不支持 adaptive prune。主要原因是当前 graph 路径把 speculative validate width 固定为 `num_decoding_tokens = num_speculative_tokens + 1`: + +1. graph capture/replay 需要固定 shape,动态 `effective_sl + 1` 会改变 validate 输入 token 数。 +2. executor 中多处使用固定 `num_decoding_tokens` 反推 batch size。 +3. persistent graph buffer、seq lens padding、metadata copy 都假设 decode width 静态。 +4. 多 rank 下所有 rank 必须选择同一个 bucket,否则 collective shape 会不一致。 + +当前实现中,`enable_graph=true` 时 `AdaptiveSpeculativeController` 自动关闭 adaptive,继续使用静态 MTP。 + +后续可以用 graph bucket 支持,例如预捕获 `SL in {1, 2, 4, 8}`,运行时按剪枝结果选择 bucket,并在 bucket 内 mask 未选中的位置。 + +## DP/EP 暂缓 + +DP/EP 场景下,不同 rank 可能持有不同请求子集,本地最优 `prefix_lengths` 不同;同时 `dp_global_token_nums`、expert dispatch、padding 和 graph bucket 都依赖全局 token shape。v1 为了控制变量,DP/EP 大于 1 时关闭 adaptive,退回静态 MTP。 + +后续支持可以考虑: + +1. scheduler/master 侧根据历史接受率预先选择全局 SL。 +2. DP group 内 all-gather 本地建议,再取 min/median/max。 +3. 将 DP/EP token shape 和 adaptive bucket 统一管理。 + +## 风险和边界 + +1. v1 不减少 draft 计算,只减少 target validate 宽度;如果瓶颈主要在 draft model,收益有限。 +2. profile predictor 未就绪时(服务启动中),adaptive 路径不会启用,直到 profile 完成并设置 registry。 +3. 如果 draft selected probs 未定义,adaptive controller 无法工作,应退回静态路径。 +4. NPU ATB 多卡必须能初始化控制通信组;否则自动退回静态 MTP。 +5. 剪枝后每行 prefix 不同,cache 和 metrics 必须只消费被选中的前缀。 +6. Score 公式中 `intercept` 占总 estimated_time 比例较大时(如 ~70%),per-token 边际裁剪激励偏弱,SL 较小时裁剪效果有限。 + +## 验证计划 + +1. 默认关闭 adaptive,跑现有 MTP 用例,确认行为不变。 +2. 开启 adaptive prune,构造高概率 draft,确认 prefix 接近 `max_sl`。 +3. 开启 adaptive prune,构造低概率 draft,确认 prefix 可以降到 0 或较小值。 +4. 检查 batch 内不同 prefix 时,target context cache 不越界,`all_draft_accepted` 不误判。 +5. 检查 accepted/draft counters 中 draft token 分母等于 `sum(prefix_lengths)`。 +6. TP/ATB 多卡确认所有 rank 使用同一个 `effective_sl`,不再出现 validate width 分叉导致 hang。 +7. compare 脚本对比 static MTP 和 adaptive-prune MTP 的吞吐、TPOT、MTP acceptance rate。 + +## 当前实现范围 + +已落地: + +1. `SpeculativeConfig` / `Options` / spawn worker 参数传播 adaptive prune 配置。 +2. 新增 `AdaptiveSpeculativeController`,实现基于 draft path probability 的 validate 前剪枝,score 公式按 per-seq 累加估计 validate time。 +3. 新增 `SpeculativeProfileRegistry`,保存 ProfileManager 拟合的 validate time predictor,同时作为 adaptive 路径的启用门控。 +4. 新增 `SetSpeculativeValidateTimePredictor` worker RPC,把 predictor 广播到 worker runtime 进程。 +5. `MTPWorkerImpl::step_decode()` 完整 draft 后进行 adaptive prune,支持 per-seq variable-length validate。 +6. Per-seq validate 通过 ATB kernel 的 `q_seq_lens` 减少实际计算量;RejectionSampler 使用 fast-path/slow-path padding 对齐。 +7. validate 输出按 per-request prefix 裁剪。 +8. metrics 按选中的 draft token 统计 acceptance rate。 +9. graph mode、非 MTP、DP/EP 大于 1 时自动关闭 adaptive。 +10. profile 阶段增加 KV cache 容量检查,防止超出 blocks 导致 hang。 + +未落地,留作独立后续 feat: + +1. Draft early stop,即直接中断 draft loop。 +2. TPOT SLO 控制。 +3. Graph bucket 支持。 +4. ZOS 适配。 +5. 历史 accept length EMA / batch-size 分段 SL range。 +6. adaptive 专用 metrics。 diff --git a/tests/core/runtime/CMakeLists.txt b/tests/core/runtime/CMakeLists.txt index cae555ea01..0774aa2a4d 100644 --- a/tests/core/runtime/CMakeLists.txt +++ b/tests/core/runtime/CMakeLists.txt @@ -77,6 +77,31 @@ cc_test( GTest::gtest_main ) +cc_test( + NAME + adaptive_speculative_controller_test + SRCS + adaptive_speculative_controller_test.cpp + "${PROJECT_SOURCE_DIR}/xllm/core/runtime/adaptive_speculative_controller.cpp" + "${PROJECT_SOURCE_DIR}/xllm/core/runtime/speculative_profile_registry.cpp" + DEPS + torch + glog::glog + proto::xllm_proto + GTest::gtest_main +) + +cc_test( + NAME + speculative_profile_registry_test + SRCS + speculative_profile_registry_test.cpp + "${PROJECT_SOURCE_DIR}/xllm/core/runtime/speculative_profile_registry.cpp" + DEPS + glog::glog + GTest::gtest_main +) + cc_test( NAME cp_input_partition_test diff --git a/tests/core/runtime/adaptive_speculative_controller_test.cpp b/tests/core/runtime/adaptive_speculative_controller_test.cpp new file mode 100644 index 0000000000..7a4f859f70 --- /dev/null +++ b/tests/core/runtime/adaptive_speculative_controller_test.cpp @@ -0,0 +1,125 @@ +/* Copyright 2026 The xLLM Authors. All Rights Reserved. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + https://github.com/jd-opensource/xllm/blob/main/LICENSE + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +==============================================================================*/ + +#include "runtime/adaptive_speculative_controller.h" + +#include +#include + +#include + +#include "runtime/options.h" +#include "runtime/speculative_profile_registry.h" + +namespace xllm { +namespace { + +runtime::Options make_options() { + runtime::Options options; + options.enable_adaptive_speculative_decode(true) + .num_speculative_tokens(4) + .speculative_algorithm("MTP") + .enable_graph(false) + .adaptive_speculative_min_gain(0.0); + return options; +} + +void setup_registry() { + SpeculativeProfileRegistry::ValidateTimePredictor predictor; + predictor.intercept_ms = 1.0; + predictor.batch_ms = 0.1; + predictor.query_token_ms = 0.5; + predictor.query_prefix_ms = 0.001; + SpeculativeProfileRegistry::get_instance().set_validate_time_predictor( + predictor); +} + +TEST(AdaptiveSpeculativeControllerTest, EnablesOnlyForMtpWithoutGraph) { + runtime::Options options = make_options(); + AdaptiveSpeculativeController controller(options); + EXPECT_TRUE(controller.enabled()); + + options.speculative_algorithm("Eagle3"); + AdaptiveSpeculativeController eagle_controller(options); + EXPECT_FALSE(eagle_controller.enabled()); + + options = make_options(); + options.enable_graph(true); + AdaptiveSpeculativeController graph_controller(options); + EXPECT_FALSE(graph_controller.enabled()); +} + +TEST(AdaptiveSpeculativeControllerTest, SelectsPrunedPrefixesByPathProb) { + setup_registry(); + AdaptiveSpeculativeController controller(make_options()); + torch::Tensor probs = torch::tensor({ + {0.95, 0.90, 0.20}, + {0.80, 0.00, 0.10}, + }); + std::vector per_seq_kv_lens = {100.0, 100.0}; + + const std::vector prefix_lengths = + controller.select_pruned_prefix_lengths(probs, + /*full_draft_time_ms=*/1.0, + /*target_step_time_ms=*/10.0, + per_seq_kv_lens); + + ASSERT_EQ(prefix_lengths.size(), 2); + EXPECT_GE(prefix_lengths[0], 1); + EXPECT_GE(prefix_lengths[1], 0); +} + +TEST(AdaptiveSpeculativeControllerTest, AllowsZeroPrefixWhenNoGain) { + setup_registry(); + AdaptiveSpeculativeController controller(make_options()); + torch::Tensor probs = torch::tensor({ + {0.0, 0.0, 0.0}, + {0.0, 0.0, 0.0}, + }); + std::vector per_seq_kv_lens = {100.0, 100.0}; + + const std::vector prefix_lengths = + controller.select_pruned_prefix_lengths(probs, + /*full_draft_time_ms=*/1.0, + /*target_step_time_ms=*/100.0, + per_seq_kv_lens); + + ASSERT_EQ(prefix_lengths.size(), 2); + EXPECT_EQ(prefix_lengths[0], 0); + EXPECT_EQ(prefix_lengths[1], 0); +} + +TEST(AdaptiveSpeculativeControllerTest, ContinuesAfterNonImprovingCandidate) { + setup_registry(); + AdaptiveSpeculativeController controller(make_options()); + torch::Tensor probs = torch::tensor({ + {0.99, 0.98}, + {0.90, 0.00}, + }); + std::vector per_seq_kv_lens = {100.0, 100.0}; + + const std::vector prefix_lengths = + controller.select_pruned_prefix_lengths(probs, + /*full_draft_time_ms=*/1.0, + /*target_step_time_ms=*/100.0, + per_seq_kv_lens); + + ASSERT_EQ(prefix_lengths.size(), 2); + EXPECT_GE(prefix_lengths[0], 1); + EXPECT_GE(prefix_lengths[1], 0); +} + +} // namespace +} // namespace xllm diff --git a/tests/core/runtime/speculative_profile_registry_test.cpp b/tests/core/runtime/speculative_profile_registry_test.cpp new file mode 100644 index 0000000000..e9c8b45404 --- /dev/null +++ b/tests/core/runtime/speculative_profile_registry_test.cpp @@ -0,0 +1,63 @@ +/* Copyright 2026 The xLLM Authors. All Rights Reserved. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + https://github.com/jd-opensource/xllm/blob/main/LICENSE + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +==============================================================================*/ + +#include "runtime/speculative_profile_registry.h" + +#include + +#include + +namespace xllm { +namespace { + +TEST(SpeculativeProfileRegistryTest, PredictsValidateTimeFromLinearTerms) { + SpeculativeProfileRegistry& registry = + SpeculativeProfileRegistry::get_instance(); + registry.reset_validate_time_predictor(); + + SpeculativeProfileRegistry::ValidateTimePredictor predictor; + predictor.intercept_ms = 1.0; + predictor.batch_ms = 2.0; + predictor.query_token_ms = 3.0; + predictor.query_prefix_ms = 0.5; + registry.set_validate_time_predictor(predictor); + + const std::optional + registered_predictor = registry.validate_time_predictor(); + ASSERT_TRUE(registered_predictor.has_value()); + EXPECT_DOUBLE_EQ(registered_predictor->query_prefix_ms, 0.5); + + const double time_ms = registry.predict_validate_time_ms( + /*batch_size=*/4, /*query_len=*/3, /*avg_prefix_len=*/5.0); + + EXPECT_DOUBLE_EQ(time_ms, + 1.0 + 2.0 * 4.0 + 3.0 * 4.0 * 3.0 + 0.5 * 4.0 * 3.0 * 5.0); + registry.reset_validate_time_predictor(); +} + +TEST(SpeculativeProfileRegistryTest, ReturnsZeroBeforeRegistration) { + SpeculativeProfileRegistry& registry = + SpeculativeProfileRegistry::get_instance(); + registry.reset_validate_time_predictor(); + + EXPECT_DOUBLE_EQ(registry.predict_validate_time_ms( + /*batch_size=*/4, + /*query_len=*/3, + /*avg_prefix_len=*/128.0), + 0.0); +} + +} // namespace +} // namespace xllm diff --git a/xllm/core/common/global_flags.h b/xllm/core/common/global_flags.h index a560f85c1c..473bbd9432 100755 --- a/xllm/core/common/global_flags.h +++ b/xllm/core/common/global_flags.h @@ -79,6 +79,10 @@ DECLARE_string(speculative_algorithm); DECLARE_bool(enable_opt_validate_probs); +DECLARE_bool(enable_adaptive_speculative_decode); + +DECLARE_double(adaptive_speculative_min_gain); + DECLARE_int32(speculative_suffix_cache_max_depth); DECLARE_double(speculative_suffix_max_spec_factor); diff --git a/xllm/core/common/options.cpp b/xllm/core/common/options.cpp index 94500f0c60..cc664e2dae 100644 --- a/xllm/core/common/options.cpp +++ b/xllm/core/common/options.cpp @@ -49,6 +49,9 @@ std::string Options::to_string() const { << speculative_suffix_max_cached_requests() << ", speculative_suffix_use_tree_spec: " << speculative_suffix_use_tree_spec() + << ", enable_adaptive_speculative_decode: " + << enable_adaptive_speculative_decode() + << ", adaptive_speculative_min_gain: " << adaptive_speculative_min_gain() << ", num_request_handling_threads: " << num_request_handling_threads() << ", communication_backend: " << communication_backend().value_or("null") << ", rank_tablefile: " << rank_tablefile().value_or("null") diff --git a/xllm/core/common/options.h b/xllm/core/common/options.h index 2b8dbca2f4..e836b8031a 100644 --- a/xllm/core/common/options.h +++ b/xllm/core/common/options.h @@ -95,6 +95,10 @@ class Options { PROPERTY(bool, speculative_suffix_use_tree_spec) = false; + PROPERTY(bool, enable_adaptive_speculative_decode) = false; + + PROPERTY(double, adaptive_speculative_min_gain) = 0.0; + // thread num to handle requests PROPERTY(size_t, num_request_handling_threads) = 4; diff --git a/xllm/core/distributed_runtime/comm_channel.cpp b/xllm/core/distributed_runtime/comm_channel.cpp index 7f63f59027..266b595535 100644 --- a/xllm/core/distributed_runtime/comm_channel.cpp +++ b/xllm/core/distributed_runtime/comm_channel.cpp @@ -91,6 +91,25 @@ bool CommChannel::allocate_kv_cache(const KVCacheShape& kv_cache_shape) { return true; } +bool CommChannel::set_speculative_validate_time_predictor( + const SpeculativeProfileRegistry::ValidateTimePredictor& predictor) { + proto::SpeculativeValidateTimePredictor request; + request.set_intercept_ms(predictor.intercept_ms); + request.set_batch_ms(predictor.batch_ms); + request.set_query_token_ms(predictor.query_token_ms); + request.set_query_prefix_ms(predictor.query_prefix_ms); + + proto::Status s; + brpc::Controller cntl; + stub_->SetSpeculativeValidateTimePredictor(&cntl, &request, &s, nullptr); + if (cntl.Failed() || !s.ok()) { + LOG(ERROR) << "SetSpeculativeValidateTimePredictor failed: " + << cntl.ErrorText(); + return false; + } + return true; +} + bool CommChannel::get_cache_info(uint64_t& cluster_id, std::string& addr, uint16_t& port) { diff --git a/xllm/core/distributed_runtime/comm_channel.h b/xllm/core/distributed_runtime/comm_channel.h index a0d2575be7..68bff66580 100644 --- a/xllm/core/distributed_runtime/comm_channel.h +++ b/xllm/core/distributed_runtime/comm_channel.h @@ -28,6 +28,7 @@ limitations under the License. #include "framework/xtensor/xtensor.h" #include "runtime/forward_params.h" #include "runtime/params_utils.h" +#include "runtime/speculative_profile_registry.h" #include "worker.pb.h" namespace xllm { @@ -43,6 +44,9 @@ class CommChannel { virtual bool allocate_kv_cache(const KVCacheShape& kv_cache_shape); + virtual bool set_speculative_validate_time_predictor( + const SpeculativeProfileRegistry::ValidateTimePredictor& predictor); + virtual bool get_cache_info(uint64_t& cluster_id, std::string& addr, uint16_t& port); diff --git a/xllm/core/distributed_runtime/engine.h b/xllm/core/distributed_runtime/engine.h index ee7419b168..a1524b8891 100644 --- a/xllm/core/distributed_runtime/engine.h +++ b/xllm/core/distributed_runtime/engine.h @@ -25,6 +25,7 @@ limitations under the License. #include "framework/tokenizer/tokenizer.h" #include "framework/tokenizer/tokenizer_args.h" #include "runtime/options.h" +#include "runtime/speculative_profile_registry.h" namespace xllm { class Engine { @@ -55,6 +56,13 @@ class Engine { // return the model args virtual const ModelArgs& model_args() const { return args_; } + virtual const ModelArgs* draft_model_args() const { return nullptr; } + + virtual bool set_speculative_validate_time_predictor( + const SpeculativeProfileRegistry::ValidateTimePredictor&) { + return false; + } + // return the tokenizer args virtual const TokenizerArgs& tokenizer_args() const { return tokenizer_args_; diff --git a/xllm/core/distributed_runtime/llm_engine.cpp b/xllm/core/distributed_runtime/llm_engine.cpp index afe50b826d..e14831fb84 100644 --- a/xllm/core/distributed_runtime/llm_engine.cpp +++ b/xllm/core/distributed_runtime/llm_engine.cpp @@ -600,6 +600,20 @@ bool LLMEngine::allocate_kv_cache(const KVCacheCapacity& kv_cache_cap) { return true; } +bool LLMEngine::set_speculative_validate_time_predictor( + const SpeculativeProfileRegistry::ValidateTimePredictor& predictor) { + bool success = true; + for (size_t i = 0; i < worker_clients_.size(); ++i) { + if (!worker_clients_[i]->set_speculative_validate_time_predictor( + predictor)) { + LOG(ERROR) << "Failed to set speculative validate predictor for worker " + << i; + success = false; + } + } + return success; +} + bool LLMEngine::pull_kv_blocks( const int32_t src_dp_size, const int32_t src_dp_rank, diff --git a/xllm/core/distributed_runtime/llm_engine.h b/xllm/core/distributed_runtime/llm_engine.h index ae2300f017..ddf73a54eb 100644 --- a/xllm/core/distributed_runtime/llm_engine.h +++ b/xllm/core/distributed_runtime/llm_engine.h @@ -58,6 +58,10 @@ class LLMEngine : public Engine { bool init(MasterStatus master_status) override; + bool set_speculative_validate_time_predictor( + const SpeculativeProfileRegistry::ValidateTimePredictor& predictor) + override; + void update_last_step_result(std::vector& batch) override; // return the active activation memory diff --git a/xllm/core/distributed_runtime/master.cpp b/xllm/core/distributed_runtime/master.cpp index 17b8d55685..c0eafd8fb2 100644 --- a/xllm/core/distributed_runtime/master.cpp +++ b/xllm/core/distributed_runtime/master.cpp @@ -304,6 +304,9 @@ Master::Master(const Options& options, EngineType type) options_.speculative_suffix_max_cached_requests()) .speculative_suffix_use_tree_spec( options_.speculative_suffix_use_tree_spec()) + .enable_adaptive_speculative_decode( + options_.enable_adaptive_speculative_decode()) + .adaptive_speculative_min_gain(options_.adaptive_speculative_min_gain()) .task_type(options_.task_type()) .enable_mla(options_.enable_mla()) .npu_kernel_backend(options_.npu_kernel_backend()) diff --git a/xllm/core/distributed_runtime/remote_worker.cpp b/xllm/core/distributed_runtime/remote_worker.cpp index 26158665a9..8ddc070bb4 100644 --- a/xllm/core/distributed_runtime/remote_worker.cpp +++ b/xllm/core/distributed_runtime/remote_worker.cpp @@ -78,6 +78,11 @@ bool RemoteWorker::allocate_kv_cache(const KVCacheShape& kv_cache_shape) { return channel_->allocate_kv_cache(kv_cache_shape); } +bool RemoteWorker::set_speculative_validate_time_predictor( + const SpeculativeProfileRegistry::ValidateTimePredictor& predictor) { + return channel_->set_speculative_validate_time_predictor(predictor); +} + void RemoteWorker::get_cache_info(uint64_t& cluster_id, std::string& addr, uint16_t& port) { diff --git a/xllm/core/distributed_runtime/remote_worker.h b/xllm/core/distributed_runtime/remote_worker.h index d337989069..bdbc494316 100644 --- a/xllm/core/distributed_runtime/remote_worker.h +++ b/xllm/core/distributed_runtime/remote_worker.h @@ -55,6 +55,10 @@ class RemoteWorker : public WorkerClient { virtual bool allocate_kv_cache(const KVCacheShape& kv_cache_shape) override; + bool set_speculative_validate_time_predictor( + const SpeculativeProfileRegistry::ValidateTimePredictor& predictor) + override; + virtual void get_cache_info(uint64_t& cluster_id, std::string& addr, uint16_t& port) override; diff --git a/xllm/core/distributed_runtime/speculative_engine.cpp b/xllm/core/distributed_runtime/speculative_engine.cpp index 43d8505180..e08cb26a9e 100644 --- a/xllm/core/distributed_runtime/speculative_engine.cpp +++ b/xllm/core/distributed_runtime/speculative_engine.cpp @@ -173,6 +173,18 @@ bool SpeculativeEngine::allocate_kv_cache() { draft_engine_->allocate_kv_cache(draft_kv_cache_cap); } +const ModelArgs* SpeculativeEngine::draft_model_args() const { + if (draft_engine_ == nullptr) { + return nullptr; + } + return &draft_engine_->model_args(); +} + +bool SpeculativeEngine::set_speculative_validate_time_predictor( + const SpeculativeProfileRegistry::ValidateTimePredictor& predictor) { + return engine_->set_speculative_validate_time_predictor(predictor); +} + // TODO: support dp batches later ForwardOutput SpeculativeEngine::step(std::vector& batches) { return engine_->step(batches); diff --git a/xllm/core/distributed_runtime/speculative_engine.h b/xllm/core/distributed_runtime/speculative_engine.h index 58b1c79883..d30b50deb6 100644 --- a/xllm/core/distributed_runtime/speculative_engine.h +++ b/xllm/core/distributed_runtime/speculative_engine.h @@ -47,6 +47,12 @@ class SpeculativeEngine : public Engine { const ModelArgs& model_args() const override { return model_args_; } + const ModelArgs* draft_model_args() const override; + + bool set_speculative_validate_time_predictor( + const SpeculativeProfileRegistry::ValidateTimePredictor& predictor) + override; + const TokenizerArgs& tokenizer_args() const override { return engine_->tokenizer_args(); } diff --git a/xllm/core/distributed_runtime/worker_service.cpp b/xllm/core/distributed_runtime/worker_service.cpp index ad0ad764d3..9bf19f0cd8 100644 --- a/xllm/core/distributed_runtime/worker_service.cpp +++ b/xllm/core/distributed_runtime/worker_service.cpp @@ -440,6 +440,24 @@ void WorkerService::AllocateKVCache( return; } +void WorkerService::SetSpeculativeValidateTimePredictor( + ::google::protobuf::RpcController* controller, + const proto::SpeculativeValidateTimePredictor* request, + proto::Status* response, + ::google::protobuf::Closure* done) { + threadpool_->schedule([this, controller, request, response, done]() mutable { + brpc::ClosureGuard done_guard(done); + SpeculativeProfileRegistry::ValidateTimePredictor predictor; + predictor.intercept_ms = request->intercept_ms(); + predictor.batch_ms = request->batch_ms(); + predictor.query_token_ms = request->query_token_ms(); + predictor.query_prefix_ms = request->query_prefix_ms(); + response->set_ok( + worker_->set_speculative_validate_time_predictor(predictor)); + }); + return; +} + void WorkerService::AllocateKVCacheWithTransfer( ::google::protobuf::RpcController* controller, const proto::AllocateKVCacheRequest* req, diff --git a/xllm/core/distributed_runtime/worker_service.h b/xllm/core/distributed_runtime/worker_service.h index 01cbc62cdd..9a810d6e68 100644 --- a/xllm/core/distributed_runtime/worker_service.h +++ b/xllm/core/distributed_runtime/worker_service.h @@ -64,6 +64,12 @@ class WorkerService : public proto::DistributeWorker { proto::Status* response, ::google::protobuf::Closure* done) override; + void SetSpeculativeValidateTimePredictor( + ::google::protobuf::RpcController* controller, + const proto::SpeculativeValidateTimePredictor* request, + proto::Status* response, + ::google::protobuf::Closure* done) override; + void AllocateKVCacheWithTransfer( ::google::protobuf::RpcController* controller, const proto::AllocateKVCacheRequest* req, diff --git a/xllm/core/framework/config/speculative_config.cpp b/xllm/core/framework/config/speculative_config.cpp index 79b0e4ad6e..7306a72396 100644 --- a/xllm/core/framework/config/speculative_config.cpp +++ b/xllm/core/framework/config/speculative_config.cpp @@ -71,6 +71,16 @@ DEFINE_bool(enable_atb_spec_kernel, false, "Whether to use ATB speculative kernel."); +DEFINE_bool(enable_adaptive_speculative_decode, + false, + "Whether to enable adaptive speculative length for MTP decode."); + +DEFINE_double( + adaptive_speculative_min_gain, + 0.0, + "Minimum relative throughput gain required to include a draft token in " + "adaptive speculative validation."); + namespace xllm { void SpeculativeConfig::from_flags() { @@ -90,6 +100,8 @@ void SpeculativeConfig::from_flags() { XLLM_CONFIG_ASSIGN_FROM_FLAG(speculative_suffix_use_tree_spec); XLLM_CONFIG_ASSIGN_FROM_FLAG(enable_opt_validate_probs); XLLM_CONFIG_ASSIGN_FROM_FLAG(enable_atb_spec_kernel); + XLLM_CONFIG_ASSIGN_FROM_FLAG(enable_adaptive_speculative_decode); + XLLM_CONFIG_ASSIGN_FROM_FLAG(adaptive_speculative_min_gain); } void SpeculativeConfig::from_json(const JsonReader& json) { @@ -106,6 +118,8 @@ void SpeculativeConfig::from_json(const JsonReader& json) { XLLM_CONFIG_ASSIGN_FROM_JSON(speculative_suffix_use_tree_spec); XLLM_CONFIG_ASSIGN_FROM_JSON(enable_opt_validate_probs); XLLM_CONFIG_ASSIGN_FROM_JSON(enable_atb_spec_kernel); + XLLM_CONFIG_ASSIGN_FROM_JSON(enable_adaptive_speculative_decode); + XLLM_CONFIG_ASSIGN_FROM_JSON(adaptive_speculative_min_gain); } void SpeculativeConfig::append_config_json( @@ -136,6 +150,10 @@ void SpeculativeConfig::append_config_json( config_json, default_config, enable_opt_validate_probs); APPEND_CONFIG_JSON_VALUE_IF_NOT_DEFAULT( config_json, default_config, enable_atb_spec_kernel); + APPEND_CONFIG_JSON_VALUE_IF_NOT_DEFAULT( + config_json, default_config, enable_adaptive_speculative_decode); + APPEND_CONFIG_JSON_VALUE_IF_NOT_DEFAULT( + config_json, default_config, adaptive_speculative_min_gain); } SpeculativeConfig& SpeculativeConfig::get_instance() { diff --git a/xllm/core/framework/config/speculative_config.h b/xllm/core/framework/config/speculative_config.h index c8bdc36ed4..e269ef51c4 100644 --- a/xllm/core/framework/config/speculative_config.h +++ b/xllm/core/framework/config/speculative_config.h @@ -52,7 +52,9 @@ class SpeculativeConfig final { "speculative_suffix_max_cached_requests", "speculative_suffix_use_tree_spec", "enable_opt_validate_probs", - "enable_atb_spec_kernel"}}; + "enable_atb_spec_kernel", + "enable_adaptive_speculative_decode", + "adaptive_speculative_min_gain"}}; return kOptionCategory; } @@ -79,6 +81,10 @@ class SpeculativeConfig final { PROPERTY(bool, enable_opt_validate_probs) = false; PROPERTY(bool, enable_atb_spec_kernel) = false; + + PROPERTY(bool, enable_adaptive_speculative_decode) = false; + + PROPERTY(double, adaptive_speculative_min_gain) = 0.0; }; } // namespace xllm diff --git a/xllm/core/runtime/CMakeLists.txt b/xllm/core/runtime/CMakeLists.txt index b571aa13fc..55b89ab0ae 100644 --- a/xllm/core/runtime/CMakeLists.txt +++ b/xllm/core/runtime/CMakeLists.txt @@ -32,6 +32,8 @@ cc_library( mm_embed_vlm_worker_impl.h worker_client.h xservice_client.h + adaptive_speculative_controller.h + speculative_profile_registry.h speculative_worker_impl.h mtp_worker_impl.h suffix_worker_impl.h @@ -64,6 +66,8 @@ cc_library( worker_client.cpp xservice_client.cpp params_utils.cpp + adaptive_speculative_controller.cpp + speculative_profile_registry.cpp speculative_worker_impl.cpp mtp_worker_impl.cpp suffix_worker_impl.cpp diff --git a/xllm/core/runtime/adaptive_speculative_controller.cpp b/xllm/core/runtime/adaptive_speculative_controller.cpp new file mode 100644 index 0000000000..fc672cc73b --- /dev/null +++ b/xllm/core/runtime/adaptive_speculative_controller.cpp @@ -0,0 +1,203 @@ +/* Copyright 2026 The xLLM Authors. All Rights Reserved. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + https://github.com/jd-opensource/xllm/blob/main/LICENSE + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +==============================================================================*/ + +#include "runtime/adaptive_speculative_controller.h" + +#include + +#include +#include +#include +#include +#include + +#include "runtime/speculative_profile_registry.h" +#include "util/tensor_helper.h" + +namespace xllm { +namespace { + +bool is_mtp_algorithm(std::string algorithm) { + std::transform( + algorithm.begin(), + algorithm.end(), + algorithm.begin(), + [](unsigned char c) { return static_cast(std::tolower(c)); }); + return algorithm == "mtp"; +} + +struct PruneCandidate { + int32_t seq_id = 0; + int32_t prefix_len = 0; + double path_prob = 0.0; +}; + +} // namespace + +AdaptiveSpeculativeController::AdaptiveSpeculativeController( + const runtime::Options& options) + : enabled_(options.enable_adaptive_speculative_decode() && + options.num_speculative_tokens() > 1 && + is_mtp_algorithm(options.speculative_algorithm()) && + !options.enable_graph()), + min_gain_(options.adaptive_speculative_min_gain()) {} + +bool AdaptiveSpeculativeController::enabled() const { return enabled_; } + +// Greedy selection of per-seq validate prefix lengths. +// Candidates (seq_id, position, path_prob) are sorted by path_prob descending. +// Each candidate is accepted if it improves estimated throughput +// (score = expected_tokens / estimated_time). +std::vector +AdaptiveSpeculativeController::select_pruned_prefix_lengths( + const torch::Tensor& selected_probs_by_step, + double full_draft_time_ms, + double target_step_time_ms, + const std::vector& per_seq_kv_lens) const { + CHECK(selected_probs_by_step.defined()) + << "adaptive pruning requires draft selected probabilities"; + CHECK_EQ(selected_probs_by_step.dim(), 2) + << "adaptive pruning expects selected probs [batch, speculative_tokens], " + << "got " << selected_probs_by_step.sizes(); + + torch::Tensor probs = + safe_to(selected_probs_by_step, torch::kCPU).to(torch::kFloat64); + probs = probs.clamp(0.0, 1.0).contiguous(); + const int32_t batch_size = static_cast(probs.size(0)); + const int32_t num_speculative_tokens = static_cast(probs.size(1)); + CHECK_GT(batch_size, 0) << "adaptive pruning batch size must be positive"; + CHECK_GT(num_speculative_tokens, 0) + << "adaptive pruning speculative tokens must be positive"; + + std::vector> path_probs( + static_cast(batch_size), + std::vector(static_cast(num_speculative_tokens), 0.0)); + std::vector candidates; + candidates.reserve(static_cast(batch_size) * + static_cast(num_speculative_tokens)); + const double* prob_data = probs.data_ptr(); + for (int32_t seq_id = 0; seq_id < batch_size; ++seq_id) { + double path_prob = 1.0; + for (int32_t token_idx = 0; token_idx < num_speculative_tokens; + ++token_idx) { + path_prob *= prob_data[seq_id * num_speculative_tokens + token_idx]; + if (!std::isfinite(path_prob)) { + path_prob = 0.0; + } + path_prob = std::clamp(path_prob, 0.0, 1.0); + path_probs[static_cast(seq_id)][static_cast(token_idx)] = + path_prob; + candidates.push_back({seq_id, token_idx + 1, path_prob}); + } + } + + std::sort(candidates.begin(), + candidates.end(), + [](const PruneCandidate& lhs, const PruneCandidate& rhs) { + if (lhs.path_prob != rhs.path_prob) { + return lhs.path_prob > rhs.path_prob; + } + if (lhs.prefix_len != rhs.prefix_len) { + return lhs.prefix_len < rhs.prefix_len; + } + return lhs.seq_id < rhs.seq_id; + }); + + std::vector prefix_lengths(static_cast(batch_size), 0); + double expected_accepted = 0.0; + double current_score = score_for_pruned_state(batch_size, + expected_accepted, + prefix_lengths, + per_seq_kv_lens, + full_draft_time_ms, + target_step_time_ms); + const double min_gain = std::max(min_gain_, 0.0); + for (const PruneCandidate& candidate : candidates) { + int32_t& prefix_len = prefix_lengths[static_cast(candidate.seq_id)]; + if (candidate.prefix_len <= prefix_len) { + continue; + } + + double candidate_expected_accepted = expected_accepted; + const std::vector& seq_path_probs = + path_probs[static_cast(candidate.seq_id)]; + for (int32_t token_idx = prefix_len; token_idx < candidate.prefix_len; + ++token_idx) { + candidate_expected_accepted += + seq_path_probs[static_cast(token_idx)]; + } + const int32_t old_prefix_len = prefix_len; + prefix_len = candidate.prefix_len; + const double next_score = + score_for_pruned_state(batch_size, + candidate_expected_accepted, + prefix_lengths, + per_seq_kv_lens, + full_draft_time_ms, + target_step_time_ms); + if (next_score <= current_score * (1.0 + min_gain)) { + prefix_len = old_prefix_len; + continue; + } + + expected_accepted = candidate_expected_accepted; + current_score = next_score; + } + + return prefix_lengths; +} + +// score = expected_emitted / (draft_time + validate_time) +// validate_time is estimated per-seq: sum of per-token costs from the profile. +double AdaptiveSpeculativeController::score_for_pruned_state( + int32_t batch_size, + double expected_accepted, + const std::vector& prefix_lengths, + const std::vector& per_seq_kv_lens, + double full_draft_time_ms, + double target_step_time_ms) const { + const double estimated_time = + std::max(full_draft_time_ms, 1.0e-6) + + std::max(estimate_validate_time(prefix_lengths, per_seq_kv_lens), + target_step_time_ms * 0.1); + const double expected_emitted = + static_cast(batch_size) + expected_accepted; + return expected_emitted / estimated_time; +} + +// Per-seq validate time: intercept + Σᵢ (batch_ms + query_token_ms * qᵢ + +// query_prefix_ms * qᵢ * kvᵢ) where qᵢ = prefix_lengths[i] + 1. Coefficients +// from ProfileManager's linear regression. +double AdaptiveSpeculativeController::estimate_validate_time( + const std::vector& prefix_lengths, + const std::vector& per_seq_kv_lens) const { + std::optional predictor = + SpeculativeProfileRegistry::get_instance().validate_time_predictor(); + if (!predictor.has_value()) { + return 1.0; + } + + double time = predictor->intercept_ms; + for (size_t i = 0; i < prefix_lengths.size(); ++i) { + const double q_i = static_cast(prefix_lengths[i] + 1); + const double kv_i = i < per_seq_kv_lens.size() ? per_seq_kv_lens[i] : 0.0; + time += predictor->batch_ms; + time += predictor->query_token_ms * q_i; + time += predictor->query_prefix_ms * q_i * kv_i; + } + return std::max(time, 0.0); +} + +} // namespace xllm diff --git a/xllm/core/runtime/adaptive_speculative_controller.h b/xllm/core/runtime/adaptive_speculative_controller.h new file mode 100644 index 0000000000..308e4b65f9 --- /dev/null +++ b/xllm/core/runtime/adaptive_speculative_controller.h @@ -0,0 +1,57 @@ +/* Copyright 2026 The xLLM Authors. All Rights Reserved. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + https://github.com/jd-opensource/xllm/blob/main/LICENSE + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +==============================================================================*/ + +#pragma once + +#include + +#include +#include + +#include "runtime/options.h" + +namespace xllm { + +// Decides per-seq speculative validate prefix lengths based on draft token +// path probabilities and profiled validate time. Greedy algorithm: candidates +// sorted by path_prob descending, accepted if estimated throughput improves. +class AdaptiveSpeculativeController final { + public: + explicit AdaptiveSpeculativeController(const runtime::Options& options); + ~AdaptiveSpeculativeController() = default; + + bool enabled() const; + std::vector select_pruned_prefix_lengths( + const torch::Tensor& selected_probs_by_step, + double full_draft_time_ms, + double target_step_time_ms, + const std::vector& per_seq_kv_lens) const; + + private: + double score_for_pruned_state(int32_t batch_size, + double expected_accepted, + const std::vector& prefix_lengths, + const std::vector& per_seq_kv_lens, + double full_draft_time_ms, + double target_step_time_ms) const; + double estimate_validate_time( + const std::vector& prefix_lengths, + const std::vector& per_seq_kv_lens) const; + + bool enabled_ = false; + double min_gain_ = 0.0; +}; + +} // namespace xllm diff --git a/xllm/core/runtime/eagle3_worker_impl.cpp b/xllm/core/runtime/eagle3_worker_impl.cpp index 7419063f1a..a5babde72a 100644 --- a/xllm/core/runtime/eagle3_worker_impl.cpp +++ b/xllm/core/runtime/eagle3_worker_impl.cpp @@ -48,13 +48,14 @@ runtime::Options eagle3_draft_options(const runtime::Options& options) { Eagle3WorkerImpl::Eagle3WorkerImpl(const ParallelArgs& parallel_args, const torch::Device& device, const runtime::Options& options) - : MTPWorkerImpl(parallel_args, - device, - options, - eagle3_main_options(options), - eagle3_draft_options(options), - ::xllm::SpeculativeConfig::get_instance() - .enable_opt_validate_probs()) {} + : MTPWorkerImpl( + parallel_args, + device, + options, + eagle3_main_options(options), + eagle3_draft_options(options), + ::xllm::SpeculativeConfig::get_instance().enable_opt_validate_probs(), + /*enable_adaptive_speculative_decode=*/false) {} bool Eagle3WorkerImpl::init_model(const std::string& model_weights_path, int32_t random_seed, diff --git a/xllm/core/runtime/mtp_worker_impl.cpp b/xllm/core/runtime/mtp_worker_impl.cpp index 856ea9a7c4..996aad1899 100644 --- a/xllm/core/runtime/mtp_worker_impl.cpp +++ b/xllm/core/runtime/mtp_worker_impl.cpp @@ -16,8 +16,16 @@ limitations under the License. #include "mtp_worker_impl.h" #include +#if defined(USE_NPU) +#include +#include +#endif #include +#include +#include +#include +#include #include #include "common/metrics.h" @@ -31,6 +39,7 @@ limitations under the License. #include "core/framework/config/speculative_config.h" #include "core/framework/kv_cache/kv_cache_estimation.h" #include "core/framework/multimodal/mm_data.h" +#include "runtime/speculative_profile_registry.h" #include "spec_input_builder.h" #include "util/env_var.h" #include "util/pretty_print.h" @@ -210,6 +219,314 @@ void clear_all_output_embeddings(ForwardOutput& output) { clear_selected_embeddings(output); } +torch::Tensor selected_probs_by_step( + const std::vector& draft_outputs) { + CHECK(!draft_outputs.empty()) << "draft outputs must not be empty"; + std::vector probs_steps; + probs_steps.reserve(draft_outputs.size()); + int64_t batch_size = -1; + for (const ForwardOutput& draft_output : draft_outputs) { + torch::Tensor probs = draft_output.sample_output.probs; + CHECK(probs.defined()) + << "adaptive pruning requires selected draft probabilities"; + CHECK(probs.dim() == 1 || probs.dim() == 2) + << "adaptive pruning expects draft probs [batch] or [batch,1], got " + << probs.sizes(); + if (probs.dim() == 2) { + CHECK_EQ(probs.size(1), 1) + << "adaptive pruning expects draft probs [batch,1], got " + << probs.sizes(); + probs = probs.squeeze(/*dim=*/1); + } + if (batch_size < 0) { + batch_size = probs.size(0); + } + CHECK_EQ(probs.size(0), batch_size) + << "adaptive pruning draft prob batch mismatch"; + probs_steps.emplace_back(probs.view({-1, 1})); + } + return torch::cat(probs_steps, /*dim=*/1); +} + +bool has_selected_probs_by_step( + const std::vector& draft_outputs) { + if (draft_outputs.empty()) { + return false; + } + for (const ForwardOutput& draft_output : draft_outputs) { + const torch::Tensor& probs = draft_output.sample_output.probs; + if (!probs.defined()) { + return false; + } + } + return true; +} + +void clamp_prefix_lengths(std::vector& prefix_lengths, + int32_t batch_size, + int32_t num_speculative_tokens) { + CHECK_EQ(prefix_lengths.size(), static_cast(batch_size)) + << "adaptive pruning prefix length batch mismatch"; + for (int32_t& prefix_len : prefix_lengths) { + prefix_len = std::clamp(prefix_len, 0, num_speculative_tokens); + } +} + +int32_t max_pruned_prefix_length(const std::vector& prefix_lengths, + int32_t num_speculative_tokens) { + if (prefix_lengths.empty()) { + return num_speculative_tokens; + } + const int32_t max_prefix_len = + *std::max_element(prefix_lengths.begin(), prefix_lengths.end()); + return std::clamp(max_prefix_len, 0, num_speculative_tokens); +} + +std::vector truncate_draft_outputs( + const std::vector& draft_outputs, + int32_t num_speculative_tokens) { + CHECK_GE(num_speculative_tokens, 0) + << "num_speculative_tokens must be non-negative"; + CHECK_GE(draft_outputs.size(), static_cast(num_speculative_tokens)) + << "draft outputs are fewer than the requested validation width"; + std::vector truncated_outputs; + truncated_outputs.reserve(static_cast(num_speculative_tokens)); + for (int32_t i = 0; i < num_speculative_tokens; ++i) { + truncated_outputs.emplace_back(draft_outputs[static_cast(i)]); + } + return truncated_outputs; +} + +void apply_pruned_prefix_lengths( + SampleOutput& sample_output, + const torch::Tensor& target_next_tokens, + int32_t num_speculative_tokens, + const std::vector& pruned_prefix_lengths) { + CHECK(sample_output.next_tokens.defined()) + << "validate output tokens are undefined"; + CHECK_EQ(sample_output.next_tokens.dim(), 2) + << "validate output tokens should be [batch, width]"; + const int32_t batch_size = + static_cast(sample_output.next_tokens.size(0)); + const int32_t num_val_tokens = num_speculative_tokens + 1; + CHECK_EQ(sample_output.next_tokens.size(1), num_val_tokens) + << "validate output width mismatch"; + CHECK_EQ(pruned_prefix_lengths.size(), static_cast(batch_size)) + << "adaptive pruning prefix length batch mismatch"; + CHECK(target_next_tokens.defined()) + << "target output tokens are required for adaptive pruning"; + torch::Tensor target_next_tokens_view = + target_next_tokens.view({batch_size, num_val_tokens}); + torch::Tensor prefix_lengths = + safe_to(make_cpu_int_tensor(pruned_prefix_lengths), + torch::TensorOptions() + .dtype(torch::kLong) + .device(sample_output.next_tokens.device()), + /*non_blocking=*/true) + .clamp(0, num_speculative_tokens); + torch::Tensor positions = + torch::arange(num_val_tokens, prefix_lengths.options()); + torch::Tensor keep_mask = + positions.unsqueeze(/*dim=*/0).le(prefix_lengths.unsqueeze(/*dim=*/1)); + torch::Tensor cut_mask = + positions.unsqueeze(/*dim=*/0) + .eq(prefix_lengths.unsqueeze(/*dim=*/1)) + .logical_and( + prefix_lengths.unsqueeze(/*dim=*/1).lt(num_speculative_tokens)); + torch::Tensor target_tokens = safe_to(target_next_tokens_view, + sample_output.next_tokens.options(), + /*non_blocking=*/true); + torch::Tensor next_tokens = + torch::where(cut_mask, target_tokens, sample_output.next_tokens); + sample_output.next_tokens = + torch::where(keep_mask, next_tokens, -torch::ones_like(next_tokens)); + + if (sample_output.logprobs.defined()) { + CHECK_EQ(sample_output.logprobs.dim(), 2) + << "validate output logprobs should be [batch, width]"; + CHECK_EQ(sample_output.logprobs.size(0), batch_size) + << "validate logprob batch mismatch"; + CHECK_EQ(sample_output.logprobs.size(1), num_val_tokens) + << "validate logprob width mismatch"; + torch::Tensor output_logprobs = safe_to(sample_output.logprobs, + sample_output.logprobs.options(), + /*non_blocking=*/true); + sample_output.logprobs = + torch::where(keep_mask.to(sample_output.logprobs.device()), + output_logprobs, + torch::zeros_like(output_logprobs)); + } + + if (sample_output.top_tokens.defined()) { + CHECK(sample_output.top_logprobs.defined()) + << "top_logprobs must be defined when top_tokens are defined"; + CHECK_EQ(sample_output.top_tokens.dim(), 3) + << "validate top_tokens should be [batch, width, top_k]"; + CHECK_EQ(sample_output.top_logprobs.dim(), 3) + << "validate top_logprobs should be [batch, width, top_k]"; + CHECK_EQ(sample_output.top_tokens.size(0), batch_size) + << "validate top_tokens batch mismatch"; + CHECK_EQ(sample_output.top_tokens.size(1), num_val_tokens) + << "validate top_tokens width mismatch"; + CHECK_EQ(sample_output.top_logprobs.size(0), batch_size) + << "validate top_logprobs batch mismatch"; + CHECK_EQ(sample_output.top_logprobs.size(1), num_val_tokens) + << "validate top_logprobs width mismatch"; + torch::Tensor top_keep_mask = + keep_mask.to(sample_output.top_tokens.device()).unsqueeze(/*dim=*/-1); + sample_output.top_tokens = + torch::where(top_keep_mask, + sample_output.top_tokens, + torch::zeros_like(sample_output.top_tokens)); + sample_output.top_logprobs = + torch::where(top_keep_mask.to(sample_output.top_logprobs.device()), + sample_output.top_logprobs, + torch::zeros_like(sample_output.top_logprobs)); + } + + if (!sample_output.embeddings.defined()) { + return; + } + CHECK_EQ(sample_output.embeddings.dim(), 3) + << "validate output embeddings should be [batch, width, hidden]"; + CHECK_EQ(sample_output.embeddings.size(0), batch_size) + << "validate embedding batch mismatch"; + CHECK_EQ(sample_output.embeddings.size(1), num_val_tokens) + << "validate embedding width mismatch"; + + torch::Tensor embedding_keep_mask = + keep_mask.to(sample_output.embeddings.device()).unsqueeze(/*dim=*/-1); + sample_output.embeddings = + torch::where(embedding_keep_mask, + sample_output.embeddings, + torch::zeros_like(sample_output.embeddings)); +} + +void sync_pruned_boundary_logprobs( + SampleOutput& sample_output, + const ForwardOutput& target_output, + int32_t batch_size, + int32_t num_val_tokens, + int32_t num_speculative_tokens, + const std::vector& pruned_prefix_lengths) { + if (!sample_output.logprobs.defined()) { + return; + } + + CHECK(target_output.sample_output.logprobs.defined()) + << "target output logprobs are required for adaptive pruning"; + CHECK_EQ( + target_output.sample_output.logprobs.numel(), + static_cast(batch_size) * static_cast(num_val_tokens)) + << "target logprob count mismatch"; + torch::Tensor prefix_lengths = + safe_to(make_cpu_int_tensor(pruned_prefix_lengths), + torch::TensorOptions() + .dtype(torch::kLong) + .device(sample_output.logprobs.device()), + /*non_blocking=*/true) + .clamp(0, num_speculative_tokens); + torch::Tensor positions = + torch::arange(num_val_tokens, prefix_lengths.options()); + torch::Tensor cut_mask = + positions.unsqueeze(/*dim=*/0) + .eq(prefix_lengths.unsqueeze(/*dim=*/1)) + .logical_and( + prefix_lengths.unsqueeze(/*dim=*/1).lt(num_speculative_tokens)); + torch::Tensor target_logprobs = safe_to( + target_output.sample_output.logprobs.view({batch_size, num_val_tokens}), + sample_output.logprobs.options(), + /*non_blocking=*/true); + sample_output.logprobs = + torch::where(cut_mask, target_logprobs, sample_output.logprobs); +} + +void sync_pruned_boundary_top_logprobs( + SampleOutput& sample_output, + const ForwardOutput& target_output, + int32_t batch_size, + int32_t num_val_tokens, + int32_t num_speculative_tokens, + const std::vector& pruned_prefix_lengths) { + if (!sample_output.top_tokens.defined()) { + return; + } + + CHECK(sample_output.top_logprobs.defined()) + << "top_logprobs must be defined when top_tokens are defined"; + CHECK(target_output.sample_output.top_tokens.defined()) + << "target top_tokens are required for adaptive pruning"; + CHECK(target_output.sample_output.top_logprobs.defined()) + << "target top_logprobs are required for adaptive pruning"; + CHECK_EQ( + target_output.sample_output.top_tokens.size(0), + static_cast(batch_size) * static_cast(num_val_tokens)) + << "target top_tokens count mismatch"; + CHECK_EQ( + target_output.sample_output.top_logprobs.size(0), + static_cast(batch_size) * static_cast(num_val_tokens)) + << "target top_logprobs count mismatch"; + + torch::Tensor prefix_lengths = + safe_to(make_cpu_int_tensor(pruned_prefix_lengths), + torch::TensorOptions() + .dtype(torch::kLong) + .device(sample_output.top_tokens.device()), + /*non_blocking=*/true) + .clamp(0, num_speculative_tokens); + torch::Tensor positions = + torch::arange(num_val_tokens, prefix_lengths.options()); + torch::Tensor cut_mask = + positions.unsqueeze(/*dim=*/0) + .eq(prefix_lengths.unsqueeze(/*dim=*/1)) + .logical_and( + prefix_lengths.unsqueeze(/*dim=*/1).lt(num_speculative_tokens)) + .unsqueeze(/*dim=*/-1); + const int64_t top_k = sample_output.top_tokens.size(2); + CHECK_EQ(target_output.sample_output.top_tokens.size(1), top_k) + << "target top_tokens top_k mismatch"; + CHECK_EQ(target_output.sample_output.top_logprobs.size(1), top_k) + << "target top_logprobs top_k mismatch"; + torch::Tensor target_top_tokens = target_output.sample_output.top_tokens.view( + {batch_size, num_val_tokens, top_k}); + torch::Tensor target_top_logprobs = + target_output.sample_output.top_logprobs.view( + {batch_size, num_val_tokens, top_k}); + sample_output.top_tokens = + torch::where(cut_mask, + safe_to(target_top_tokens, + sample_output.top_tokens.options(), + /*non_blocking=*/true), + sample_output.top_tokens); + sample_output.top_logprobs = + torch::where(cut_mask.to(sample_output.top_logprobs.device()), + safe_to(target_top_logprobs, + sample_output.top_logprobs.options(), + /*non_blocking=*/true), + sample_output.top_logprobs); +} + +void sync_pruned_boundary_outputs( + SampleOutput& sample_output, + const ForwardOutput& target_output, + int32_t batch_size, + int32_t num_val_tokens, + int32_t num_speculative_tokens, + const std::vector& pruned_prefix_lengths) { + sync_pruned_boundary_logprobs(sample_output, + target_output, + batch_size, + num_val_tokens, + num_speculative_tokens, + pruned_prefix_lengths); + sync_pruned_boundary_top_logprobs(sample_output, + target_output, + batch_size, + num_val_tokens, + num_speculative_tokens, + pruned_prefix_lengths); +} + void clear_ready_events(ForwardInput& input) { input.metadata_ready_event.reset(); } @@ -483,24 +800,38 @@ bool is_mimo_target_model_type(const std::string& model_type) { MTPWorkerImpl::MTPWorkerImpl(const ParallelArgs& parallel_args, const torch::Device& device, const runtime::Options& options) - : MTPWorkerImpl(parallel_args, - device, - options, - MTPTargetOptions(options), - MTPDraftOptions(options), - ::xllm::SpeculativeConfig::get_instance() - .enable_opt_validate_probs()) {} + : MTPWorkerImpl( + parallel_args, + device, + options, + MTPTargetOptions(options), + MTPDraftOptions(options), + ::xllm::SpeculativeConfig::get_instance().enable_opt_validate_probs(), + /*enable_adaptive_speculative_decode=*/true) {} MTPWorkerImpl::MTPWorkerImpl(const ParallelArgs& parallel_args, const torch::Device& device, const runtime::Options& options, const runtime::Options& target_options, const runtime::Options& draft_options, - bool enable_opt_validate_probs) + bool enable_opt_validate_probs, + bool enable_adaptive_speculative_decode) : SpeculativeWorkerImpl(parallel_args, device, options, target_options), enable_opt_validate_probs_(enable_opt_validate_probs) { draft_impl_ = std::make_unique(parallel_args, device, draft_options); + const bool enable_parallel_adaptive_sl = + parallel_args.dp_size() <= 1 && parallel_args.ep_size() <= 1; + if (enable_adaptive_speculative_decode && enable_parallel_adaptive_sl) { + adaptive_spec_controller_ = + std::make_unique(options); + } else if (enable_adaptive_speculative_decode && + options.enable_adaptive_speculative_decode()) { + LOG(WARNING) + << "Adaptive speculative decode is disabled for DP/EP parallelism " + << "in v1. dp_size=" << parallel_args.dp_size() + << ", ep_size=" << parallel_args.ep_size(); + } } bool MTPWorkerImpl::init_model(const std::string& model_weights_path, @@ -939,6 +1270,12 @@ std::optional MTPWorkerImpl::step_decode( stabilize_decode_host_tensors(input); } const int32_t num_speculative_tokens = options_.num_speculative_tokens(); + // Adaptive is enabled only after profile completes (registry has predictor), + // ensuring profiling warmup never triggers the adaptive HCCL broadcast. + const bool use_adaptive_speculative_decode = + adaptive_spec_controller_ != nullptr && + adaptive_spec_controller_->enabled() && adaptive_enabled() && + SpeculativeProfileRegistry::get_instance().has_validate_time_predictor(); std::vector draft_outputs; ForwardInput current_draft_input, validate_input, next_step_input; @@ -999,10 +1336,12 @@ std::optional MTPWorkerImpl::step_decode( enable_schedule_overlap()); update_decode_step_input(input, last_states); prepare_draft_extend_inputs(input, last_states, current_draft_input); + draft_outputs.reserve(num_speculative_tokens); const bool reuse_mtp_topk_indices = should_reuse_mtp_topk_indices(draft_impl_->context_.get_model_args()); torch::Tensor mtp_topk_indices; + timer.reset(); for (int32_t draft_idx = 0; draft_idx < num_speculative_tokens; ++draft_idx) { if (reuse_mtp_topk_indices) { current_draft_input.input_params.dsa_topk_indices = mtp_topk_indices; @@ -1016,7 +1355,7 @@ std::optional MTPWorkerImpl::step_decode( // Overlap next-step input preparation with async draft forward. if (draft_idx == num_speculative_tokens - 1) { - prepare_validate_inputs(input, validate_input); + prepare_validate_inputs(input, validate_input, num_speculative_tokens); } else { prepare_draft_inputs(input, next_step_input, draft_idx + 1); } @@ -1057,46 +1396,185 @@ std::optional MTPWorkerImpl::step_decode( *compute_stream_); } } + const double draft_latency_ms = timer.elapsed_milliseconds(); COUNTER_ADD(speculative_execution_latency_seconds_draft, - timer.elapsed_seconds()); - return run_validate(input, draft_outputs, validate_input); + draft_latency_ms / 1000.0); + + if (use_adaptive_speculative_decode) { + // Adaptive pruning: decide per-seq how many draft tokens to validate. + // Score = expected_emitted / estimated_time; tokens are added greedily + // by descending path probability until score stops improving. + const double target_step_time_ms = + adaptive_step_time_estimate(input, num_speculative_tokens); + const int32_t batch_size = input.input_params.meta.num_sequences; + std::vector per_seq_kv_lens(static_cast(batch_size), 0.0); + const Slice kv_seq_lens = + input.input_params.attention.host.kv_seq_lens; + for (int32_t i = 0; i < batch_size; ++i) { + if (static_cast(i) < kv_seq_lens.size()) { + per_seq_kv_lens[static_cast(i)] = static_cast( + specBuilder::calc_kv_len(kv_seq_lens, i, /*offset=*/0)); + } + } + // All ranks compute pruning independently (inputs are consistent across + // ranks: draft probs from deterministic sampling, kv_lens from shared + // input, predictor coefficients from profile broadcast). No sync needed. + std::vector prefix_lengths; + if (has_selected_probs_by_step(draft_outputs)) { + prefix_lengths = adaptive_spec_controller_->select_pruned_prefix_lengths( + selected_probs_by_step(draft_outputs), + 0.0, + target_step_time_ms, + per_seq_kv_lens); + } else { + prefix_lengths.assign(static_cast(batch_size), + num_speculative_tokens); + } + clamp_prefix_lengths(prefix_lengths, + input.input_params.meta.num_sequences, + num_speculative_tokens); + int32_t effective_speculative_tokens = + max_pruned_prefix_length(prefix_lengths, num_speculative_tokens); + if (effective_speculative_tokens <= 0) { + effective_speculative_tokens = 1; + } + std::vector per_seq_val_tokens(static_cast(batch_size)); + for (int32_t i = 0; i < batch_size; ++i) { + per_seq_val_tokens[static_cast(i)] = + std::max(prefix_lengths[static_cast(i)], 1) + 1; + } + std::vector pruned_draft_outputs = + truncate_draft_outputs(draft_outputs, effective_speculative_tokens); + prepare_validate_inputs(input, validate_input, per_seq_val_tokens); + return run_validate(input, + pruned_draft_outputs, + validate_input, + effective_speculative_tokens, + per_seq_val_tokens, + &prefix_lengths); + } + + return run_validate( + input, draft_outputs, validate_input, num_speculative_tokens); } void MTPWorkerImpl::fill_validate_input_from_draft_outputs( const std::vector& draft_outputs, ForwardInput& validate_input, + int32_t num_speculative_tokens, Stream& compute_stream) { - const int32_t num_speculative_tokens = options_.num_speculative_tokens(); const int32_t num_val_tokens = num_speculative_tokens + 1; - CHECK_EQ(draft_outputs.size(), static_cast(num_speculative_tokens)) - << "draft output count mismatch"; CHECK(validate_input.token_ids.defined()) << "validate token_ids must be prepared before draft token fill"; CHECK_EQ(validate_input.token_ids.dim(), 1) << "validate token_ids must be flat"; CHECK_EQ(validate_input.token_ids.numel() % num_val_tokens, 0) << "validate token_ids size must be divisible by validation width"; + const int32_t num_sequences = + static_cast(validate_input.token_ids.numel() / num_val_tokens); + std::vector per_seq_val_tokens(static_cast(num_sequences), + num_val_tokens); + fill_validate_input_from_draft_outputs( + draft_outputs, validate_input, per_seq_val_tokens, compute_stream); +} - const int64_t total_num_val_tokens = validate_input.token_ids.numel(); - const int64_t num_sequences = total_num_val_tokens / num_val_tokens; - const auto token_options = validate_input.token_ids.options(); +void MTPWorkerImpl::fill_validate_input_from_draft_outputs( + const std::vector& draft_outputs, + ForwardInput& validate_input, + const std::vector& per_seq_val_tokens, + Stream& compute_stream) { + CHECK(!per_seq_val_tokens.empty()) << "per_seq_val_tokens must not be empty"; + const int32_t num_sequences = static_cast(per_seq_val_tokens.size()); + CHECK(validate_input.token_ids.defined()) + << "validate token_ids must be prepared before draft token fill"; + CHECK_EQ(validate_input.token_ids.dim(), 1) + << "validate token_ids must be flat"; + + const torch::TensorOptions token_options = validate_input.token_ids.options(); c10::StreamGuard stream_guard = compute_stream.set_stream_guard(); wait_metadata_ready_event(validate_input, compute_stream); - torch::Tensor validate_token_rows = - validate_input.token_ids.view({num_sequences, num_val_tokens}); validate_input.device_tensors_ready = false; - for (int32_t i = 0; i < num_speculative_tokens; ++i) { - const auto& draft_output = draft_outputs[i]; - const torch::Tensor& next_tokens = draft_output.sample_output.next_tokens; - CHECK(next_tokens.defined()) - << "draft next_tokens must be defined for validate token fill"; - torch::Tensor draft_tokens = - safe_to(next_tokens.flatten(), token_options, /*non_blocking=*/true); - CHECK_EQ(draft_tokens.numel(), num_sequences) - << "draft token count must match validate sequence count"; - validate_token_rows.select(/*dim=*/1, /*index=*/i + 1) - .copy_(draft_tokens, /*non_blocking=*/true); + + const int32_t max_val_tokens = + *std::max_element(per_seq_val_tokens.begin(), per_seq_val_tokens.end()); + const int32_t total_val_tokens = + static_cast(validate_input.token_ids.numel()); + const bool is_uniform = (total_val_tokens == num_sequences * max_val_tokens); + + if (is_uniform) { + // Fast path: all seqs have same val_tokens, use vectorized view+copy. + const int32_t num_draft_tokens = max_val_tokens - 1; + torch::Tensor validate_token_rows = validate_input.token_ids.view( + {static_cast(num_sequences), max_val_tokens}); + for (int32_t i = 0; i < num_draft_tokens; ++i) { + CHECK(static_cast(i) < draft_outputs.size()) + << "draft_outputs index out of range for step " << i; + const torch::Tensor& next_tokens = + draft_outputs[static_cast(i)].sample_output.next_tokens; + CHECK(next_tokens.defined()) + << "draft next_tokens must be defined for validate token fill"; + torch::Tensor draft_tokens = + safe_to(next_tokens.flatten(), token_options, /*non_blocking=*/true); + validate_token_rows.select(/*dim=*/1, /*index=*/i + 1) + .copy_(draft_tokens, /*non_blocking=*/true); + } + } else { + // Slow path: per-seq variable-length, group by draft step. + std::vector dst_idx_vec; + std::vector src_idx_vec; + std::vector step_vec; + int32_t offset = 0; + for (int32_t seq_id = 0; seq_id < num_sequences; ++seq_id) { + const int32_t seq_val_tokens = + per_seq_val_tokens[static_cast(seq_id)]; + const int32_t seq_draft_tokens = seq_val_tokens - 1; + for (int32_t draft_idx = 0; draft_idx < seq_draft_tokens; ++draft_idx) { + dst_idx_vec.push_back(offset + draft_idx + 1); + src_idx_vec.push_back(seq_id); + step_vec.push_back(draft_idx); + } + offset += seq_val_tokens; + } + + if (!dst_idx_vec.empty()) { + const int32_t max_draft_step = + *std::max_element(step_vec.begin(), step_vec.end()); + for (int32_t step = 0; step <= max_draft_step; ++step) { + CHECK(static_cast(step) < draft_outputs.size()) + << "draft_outputs index out of range for step " << step; + const torch::Tensor& next_tokens = + draft_outputs[static_cast(step)].sample_output.next_tokens; + CHECK(next_tokens.defined()) + << "draft next_tokens must be defined for validate token fill"; + torch::Tensor flat_tokens = safe_to( + next_tokens.flatten(), token_options, /*non_blocking=*/true); + + std::vector step_dst_indices; + std::vector step_src_indices; + for (size_t k = 0; k < step_vec.size(); ++k) { + if (step_vec[k] == step) { + step_dst_indices.push_back(dst_idx_vec[k]); + step_src_indices.push_back(src_idx_vec[k]); + } + } + if (step_dst_indices.empty()) { + continue; + } + torch::Tensor dst_idx = + torch::tensor(step_dst_indices, + torch::TensorOptions() + .dtype(torch::kLong) + .device(validate_input.token_ids.device())); + torch::Tensor src_idx = + torch::tensor(step_src_indices, + torch::TensorOptions() + .dtype(torch::kLong) + .device(flat_tokens.device())); + torch::Tensor gathered = flat_tokens.index_select(/*dim=*/0, src_idx); + validate_input.token_ids.index_copy_(/*dim=*/0, dst_idx, gathered); + } + } } validate_input.device_tensors_ready = true; } @@ -1104,31 +1582,110 @@ void MTPWorkerImpl::fill_validate_input_from_draft_outputs( std::optional MTPWorkerImpl::run_validate( const ForwardInput& input, const std::vector& draft_outputs, - ForwardInput& validate_input) { - // run the target model to get the verification scores + ForwardInput& validate_input, + int32_t num_speculative_tokens, + const std::vector* pruned_prefix_lengths) { + const int32_t batch_size = input.input_params.meta.num_sequences; + const int32_t val_tokens = num_speculative_tokens + 1; + std::vector per_seq_val_tokens(static_cast(batch_size), + val_tokens); + return run_validate(input, + draft_outputs, + validate_input, + num_speculative_tokens, + per_seq_val_tokens, + pruned_prefix_lengths); +} + +// Run target model validate with per-seq variable-length support. +// When all seqs have the same val_tokens (uniform case), uses zero-copy view. +// When variable-length, pads logits to [batch * max_val_tokens, vocab] for +// RejectionSampler compatibility, with -inf at padding positions. +std::optional MTPWorkerImpl::run_validate( + const ForwardInput& input, + const std::vector& draft_outputs, + ForwardInput& validate_input, + int32_t num_speculative_tokens, + const std::vector& per_seq_val_tokens, + const std::vector* pruned_prefix_lengths) { Timer timer; ForwardInput target_prepared; fill_validate_input_from_draft_outputs( - draft_outputs, validate_input, *compute_stream_); + draft_outputs, validate_input, per_seq_val_tokens, *compute_stream_); ForwardOutput target_output = run_llm_no_sync_impl(*impl_, validate_input, *prepare_stream_, *compute_stream_, target_prepared) .value(); + const double target_latency_ms = timer.elapsed_milliseconds(); COUNTER_ADD(speculative_execution_latency_seconds_target, - timer.elapsed_seconds()); + target_latency_ms / 1000.0); + + const int32_t batch_size = static_cast(per_seq_val_tokens.size()); + const int32_t max_val_tokens = num_speculative_tokens + 1; + const int32_t total_tokens = + static_cast(target_output.logits.size(0)); + const int32_t vocab_size = + static_cast(target_output.logits.size(-1)); + const int64_t padded_total = + static_cast(batch_size) * max_val_tokens; + + ForwardOutput padded_target_output = target_output; + if (total_tokens == static_cast(padded_total)) { + // Fast path: all seqs have the same val_tokens (= max), no padding needed. + padded_target_output.logits = + target_output.logits.view({padded_total, vocab_size}); + } else { + // Slow path: per-seq variable-length, scatter into padded layout. + std::vector dst_indices_vec; + dst_indices_vec.reserve(static_cast(total_tokens)); + for (int32_t i = 0; i < batch_size; ++i) { + const int32_t seq_tokens = per_seq_val_tokens[static_cast(i)]; + for (int32_t j = 0; j < seq_tokens; ++j) { + dst_indices_vec.push_back(static_cast(i) * max_val_tokens + j); + } + } + torch::Tensor dst_indices = + torch::tensor(dst_indices_vec, + torch::TensorOptions() + .dtype(torch::kLong) + .device(target_output.logits.device())); + + torch::Tensor padded_logits = torch::full( + {padded_total, vocab_size}, -1e9, target_output.logits.options()); + padded_logits.index_copy_(/*dim=*/0, dst_indices, target_output.logits); + padded_target_output.logits = padded_logits; + + torch::Tensor padded_next_tokens = torch::zeros( + {padded_total}, target_output.sample_output.next_tokens.options()); + padded_next_tokens.index_copy_( + /*dim=*/0, dst_indices, target_output.sample_output.next_tokens); + padded_target_output.sample_output.next_tokens = padded_next_tokens; + + if (target_output.sample_output.embeddings.defined()) { + const int32_t hidden_size = + static_cast(target_output.sample_output.embeddings.size(-1)); + torch::Tensor padded_embeddings = + torch::zeros({padded_total, hidden_size}, + target_output.sample_output.embeddings.options()); + padded_embeddings.index_copy_( + /*dim=*/0, dst_indices, target_output.sample_output.embeddings); + padded_target_output.sample_output.embeddings = padded_embeddings; + } + } - // verify the proposals with target and update the batch timer.reset(); - SampleOutput val_output = - validate(input.sampling_params, draft_outputs, target_output); + SampleOutput val_output = validate(input.sampling_params, + draft_outputs, + padded_target_output, + num_speculative_tokens, + pruned_prefix_lengths); + record_validate_metrics( + val_output, num_speculative_tokens, pruned_prefix_lengths); COUNTER_ADD(speculative_execution_latency_seconds_validation, timer.elapsed_seconds()); - // Catch-all for cross-rank RNG divergence: unify the accepted next_tokens to - // the consensus group's rank 0 so all_draft_accepted and the next - // draft-extend row layout agree across ranks. if (get_optimization_config().enable_spec_token_broadcast && enable_schedule_overlap()) { c10::StreamGuard stream_guard = compute_stream_->set_stream_guard(); @@ -1138,7 +1695,7 @@ std::optional MTPWorkerImpl::run_validate( compute_stream_->synchronize(); val_output.next_tokens = val_output.next_tokens.to(torch::kCPU); - write_target_context_to_cache(input, val_output); + write_target_context_to_cache(input, val_output, num_speculative_tokens); if (!enable_schedule_overlap() && !driver_ && !dp_driver_) { return std::nullopt; @@ -1151,7 +1708,8 @@ std::optional MTPWorkerImpl::run_validate( void MTPWorkerImpl::write_target_context_to_cache( const ForwardInput& input, - const SampleOutput& validate_output) { + const SampleOutput& validate_output, + int32_t num_speculative_tokens) { CHECK(embedding_cache_ != nullptr) << "embedding_cache_ must be initialized before target cache write"; CHECK(!input.input_params.embedding.embedding_ids.empty()) @@ -1161,7 +1719,100 @@ void MTPWorkerImpl::write_target_context_to_cache( input.input_params.embedding.request_ids, validate_output.next_tokens, validate_output.embeddings, - options_.num_speculative_tokens()); + num_speculative_tokens); +} + +void MTPWorkerImpl::record_validate_metrics( + const SampleOutput& validate_output, + int32_t num_speculative_tokens, + const std::vector* pruned_prefix_lengths) const { + CHECK(validate_output.next_tokens.defined()) + << "validate output tokens are undefined"; + CHECK_EQ(validate_output.next_tokens.dim(), 2) + << "validate output tokens should be [batch, width]"; + const int32_t batch_size = + static_cast(validate_output.next_tokens.size(0)); + CHECK_EQ(validate_output.next_tokens.size(1), num_speculative_tokens + 1) + << "validate output width mismatch"; + + torch::Tensor next_tokens_cpu = validate_output.next_tokens.to(torch::kCPU) + .to(torch::kInt64) + .contiguous(); + const int64_t* token_data = next_tokens_cpu.const_data_ptr(); + int64_t num_draft_tokens = 0; + int64_t accepted_count = 0; + for (int32_t seq_id = 0; seq_id < batch_size; ++seq_id) { + int32_t prefix_len = num_speculative_tokens; + if (pruned_prefix_lengths != nullptr) { + CHECK_EQ(pruned_prefix_lengths->size(), static_cast(batch_size)) + << "adaptive pruning prefix length batch mismatch"; + prefix_len = + std::clamp((*pruned_prefix_lengths)[static_cast(seq_id)], + 0, + num_speculative_tokens); + } + num_draft_tokens += prefix_len; + + const int64_t row_offset = static_cast(seq_id) * + static_cast(num_speculative_tokens + 1); + int32_t emitted_len = 0; + for (int32_t token_idx = 0; token_idx <= num_speculative_tokens; + ++token_idx) { + if (token_data[row_offset + token_idx] < 0) { + break; + } + ++emitted_len; + } + accepted_count += std::min(prefix_len, std::max(emitted_len - 1, 0)); + } + COUNTER_ADD(speculative_num_draft_tokens_total, num_draft_tokens); + COUNTER_ADD(speculative_num_accepted_tokens_total, accepted_count); +} + +bool MTPWorkerImpl::adaptive_enabled() const { + return adaptive_spec_controller_ != nullptr && + adaptive_spec_controller_->enabled(); +} + +double MTPWorkerImpl::adaptive_step_time_estimate( + const ForwardInput& input, + int32_t max_speculative_tokens) const { + const int32_t batch_size = input.input_params.meta.num_sequences; + const Slice kv_seq_lens = + input.input_params.attention.host.kv_seq_lens; + if (batch_size <= 0 || max_speculative_tokens <= 0) { + return 1.0; + } +#if defined(USE_NPU) + const size_t min_kv_seq_lens_size = static_cast(batch_size); +#else + const size_t min_kv_seq_lens_size = static_cast(batch_size) + 1; +#endif + if (kv_seq_lens.size() < min_kv_seq_lens_size) { + return 1.0; + } + + double total_prefix_len = 0.0; + for (int32_t seq_id = 0; seq_id < batch_size; ++seq_id) { + total_prefix_len += static_cast( + specBuilder::calc_kv_len(kv_seq_lens, seq_id, /*offset=*/0)); + } + const double avg_prefix_len = + total_prefix_len / static_cast(batch_size); + + SpeculativeProfileRegistry& registry = + SpeculativeProfileRegistry::get_instance(); + if (!registry.has_validate_time_predictor()) { + return 1.0; + } + + const double max_speculative_target_time = registry.predict_validate_time_ms( + batch_size, max_speculative_tokens + 1, avg_prefix_len); + if (max_speculative_target_time <= 0.0 || + !std::isfinite(max_speculative_target_time)) { + return 1.0; + } + return max_speculative_target_time; } void MTPWorkerImpl::process_draft_sample_output(SampleOutput& sample_output) { @@ -1296,7 +1947,8 @@ void MTPWorkerImpl::update_decode_step_input( } void MTPWorkerImpl::prepare_validate_inputs(const ForwardInput& input, - ForwardInput& validate_input) { + ForwardInput& validate_input, + int32_t num_speculative_tokens) { c10::StreamGuard stream_guard = prepare_stream_->set_stream_guard(); validate_input = input; clear_ready_events(validate_input); @@ -1306,7 +1958,6 @@ void MTPWorkerImpl::prepare_validate_inputs(const ForwardInput& input, torch::TensorOptions token_options = validate_input.token_ids.options(); torch::TensorOptions position_options = validate_input.positions.options(); - const int32_t num_speculative_tokens = options_.num_speculative_tokens(); const int32_t num_sequences = input_params.meta.num_sequences; const int32_t num_val_tokens = num_speculative_tokens + 1; const int32_t total_num_val_tokens = num_sequences * num_val_tokens; @@ -1443,6 +2094,138 @@ void MTPWorkerImpl::prepare_validate_inputs(const ForwardInput& input, finish_metadata_prepare(*prepare_stream_, validate_input); } +// Per-seq variable-length validate input construction. +// Each seq gets per_seq_val_tokens[i] tokens in the ATB kernel input, +// reducing attention/FFN computation for seqs with fewer speculative tokens. +void MTPWorkerImpl::prepare_validate_inputs( + const ForwardInput& input, + ForwardInput& validate_input, + const std::vector& per_seq_val_tokens) { + c10::StreamGuard stream_guard = prepare_stream_->set_stream_guard(); + validate_input = input; + clear_ready_events(validate_input); + validate_input.device_tensors_ready = false; + auto& input_params = validate_input.input_params; + input_params.embedding.input_embedding = torch::Tensor(); + torch::TensorOptions token_options = validate_input.token_ids.options(); + torch::TensorOptions position_options = validate_input.positions.options(); + + const int32_t num_sequences = input_params.meta.num_sequences; + CHECK_EQ(per_seq_val_tokens.size(), static_cast(num_sequences)) + << "per_seq_val_tokens size mismatch with num_sequences"; + int32_t total_num_val_tokens = 0; + int32_t max_val_tokens = 0; + for (int32_t i = 0; i < num_sequences; ++i) { + total_num_val_tokens += per_seq_val_tokens[static_cast(i)]; + max_val_tokens = + std::max(max_val_tokens, per_seq_val_tokens[static_cast(i)]); + } + const int32_t block_size = options_.block_size(); + specBuilder::DecodeRowContext row_ctx = + specBuilder::make_decode_row_context(input); + Slice token_ids = { + input.token_ids_host.data_ptr(), + static_cast(input.token_ids_host.numel())}; + Slice positions = { + input.positions_host.data_ptr(), + static_cast(input.positions_host.numel())}; + Slice kv_seq_lens = input.input_params.attention.host.kv_seq_lens; + const bool use_atb_spec_kernel = + ::xllm::SpeculativeConfig::get_instance().enable_atb_spec_kernel() || + use_chunked_prefill_spec_verify_path(); + specBuilder::DecodeBuildBuffers buf; + buf.out_token_ids.reserve(total_num_val_tokens); + buf.out_positions.reserve(total_num_val_tokens); + buf.out_new_cache_slots.reserve(total_num_val_tokens); + if (!use_atb_spec_kernel) { + buf.out_kv_seq_lens.reserve(total_num_val_tokens); + buf.out_q_seq_lens.reserve(total_num_val_tokens); + buf.out_q_cu_seq_lens.reserve(total_num_val_tokens); + buf.out_block_tables.reserve(static_cast(total_num_val_tokens) * + row_ctx.block_table_stride); + } + + std::vector atb_kv_seq_lens_vec; + std::vector atb_q_seq_lens_vec; + std::vector atb_q_cu_seq_lens_vec; + int32_t atb_kv_max_seq_len = 0; + for (int32_t seq_id = 0; seq_id < num_sequences; ++seq_id) { + const int32_t seq_val_tokens = + per_seq_val_tokens[static_cast(seq_id)]; + const int32_t start_position = positions[seq_id]; + const int32_t kv_len = + specBuilder::calc_kv_len(kv_seq_lens, seq_id, /*offset=*/0); + CHECK_EQ(start_position + 1, kv_len) + << "validate position/kv_len mismatch, seq_id=" << seq_id + << ", start_position=" << start_position << ", kv_len=" << kv_len; + + for (int32_t val_idx = 0; val_idx < seq_val_tokens; ++val_idx) { + specBuilder::RowSpec row; + row.seq_id = seq_id; + row.token_id = val_idx == 0 ? token_ids[seq_id] : -val_idx; + row.position_offset = val_idx; + row.append_kv_len = !use_atb_spec_kernel; + row.append_q_len_one = !use_atb_spec_kernel; + row.append_block_table = !use_atb_spec_kernel; + specBuilder::append_decode_row(row_ctx, row, block_size, buf); + } + + if (use_atb_spec_kernel) { + const int32_t kv_len_after_validation = kv_len + seq_val_tokens - 1; + specBuilder::update_kv_seq_lens_and_max( + atb_kv_seq_lens_vec, kv_len_after_validation, atb_kv_max_seq_len); + specBuilder::append_q_seq_len( + atb_q_seq_lens_vec, atb_q_cu_seq_lens_vec, seq_val_tokens); + } + } + + CHECK_EQ(buf.out_new_cache_slots.size(), buf.out_token_ids.size()) + << "validate kv slots/tokens mismatch"; + CHECK_EQ(buf.out_positions.size(), buf.out_token_ids.size()) + << "validate positions/tokens mismatch"; + + set_token_position_tensors(validate_input, + buf.out_token_ids, + buf.out_positions, + token_options, + position_options); + if (!use_atb_spec_kernel) { + input_params.meta.num_sequences = total_num_val_tokens; + input_params.meta.batch_forward_type = BatchForwardType::DECODE; + } else { + input_params.meta.batch_forward_type = BatchForwardType::CHUNKED_PREFILL; + } + if (use_atb_spec_kernel) { + specBuilder::update_input_params(input_params, + buf, + max_val_tokens, + std::move(atb_q_seq_lens_vec), + std::move(atb_q_cu_seq_lens_vec), + atb_kv_max_seq_len, + std::move(atb_kv_seq_lens_vec)); + } else { + specBuilder::update_input_params(input_params, + buf, + 1, + std::move(buf.out_q_seq_lens), + std::move(buf.out_q_cu_seq_lens), + buf.meta.kv_max_seq_len, + std::move(buf.out_kv_seq_lens), + /*update_block_tables=*/true); + } + + update_sampling_params( + validate_input.sampling_params, per_seq_val_tokens, total_num_val_tokens); + + for (int32_t& token_num : input_params.parallel.dp_global_token_nums) { + token_num = total_num_val_tokens; + } + + input_params.attention.rebuild_device_buffer(device_); + validate_input.device_tensors_ready = true; + finish_metadata_prepare(*prepare_stream_, validate_input); +} + void MTPWorkerImpl::prepare_draft_extend_inputs( const ForwardInput& base_input, const std::vector& last_states, @@ -1721,10 +2504,12 @@ void MTPWorkerImpl::prepare_draft_inputs(const ForwardInput& input, SampleOutput MTPWorkerImpl::validate( const SamplingParameters& sampling_params, const std::vector& draft_outputs, - const ForwardOutput& target_output) { + const ForwardOutput& target_output, + int32_t num_speculative_tokens, + const std::vector* pruned_prefix_lengths) { const int32_t num_target_tokens = target_output.sample_output.next_tokens.numel(); - const int32_t num_val_tokens = options_.num_speculative_tokens() + 1; + const int32_t num_val_tokens = num_speculative_tokens + 1; CHECK_EQ(num_target_tokens % num_val_tokens, 0); const int32_t batch_size = num_target_tokens / num_val_tokens; const int32_t vocab_size = target_output.logits.size(/*dim=*/-1); @@ -1733,44 +2518,52 @@ SampleOutput MTPWorkerImpl::validate( std::vector draft_probs_steps; draft_token_ids_steps.reserve(draft_outputs.size()); draft_probs_steps.reserve(draft_outputs.size()); - for (const auto& draft_output : draft_outputs) { + for (const ForwardOutput& draft_output : draft_outputs) { draft_token_ids_steps.emplace_back(draft_output.sample_output.next_tokens); draft_probs_steps.emplace_back(draft_output.sample_output.probs); } - auto [draft_token_ids, draft_probs] = + std::pair validate_tensors = specBuilder::draftProbs::build_validate_tensors( draft_token_ids_steps, draft_probs_steps, batch_size, vocab_size, enable_opt_validate_probs_); - return validate(sampling_params, draft_token_ids, draft_probs, target_output); + return validate(sampling_params, + validate_tensors.first, + validate_tensors.second, + target_output, + num_speculative_tokens, + pruned_prefix_lengths); } -SampleOutput MTPWorkerImpl::validate(const SamplingParameters& sampling_params, - const torch::Tensor& draft_token_ids, - const torch::Tensor& draft_probs, - const ForwardOutput& target_output) { +SampleOutput MTPWorkerImpl::validate( + const SamplingParameters& sampling_params, + const torch::Tensor& draft_token_ids, + const torch::Tensor& draft_probs, + const ForwardOutput& target_output, + int32_t num_speculative_tokens, + const std::vector* pruned_prefix_lengths) { const int32_t num_target_tokens = target_output.sample_output.next_tokens.numel(); - const int32_t num_val_tokens = options_.num_speculative_tokens() + 1; + const int32_t num_val_tokens = num_speculative_tokens + 1; CHECK_EQ(num_target_tokens % num_val_tokens, 0); const int32_t batch_size = num_target_tokens / num_val_tokens; const int32_t vocab_size = target_output.logits.size(/*dim=*/-1); using torch::indexing::None; using ISlice = torch::indexing::Slice; - auto bonus_token_ids = + torch::Tensor bonus_token_ids = target_output.sample_output.next_tokens .index({"...", ISlice(num_val_tokens - 1, None, num_val_tokens)}) .view({-1, 1}); - auto target_logits = + torch::Tensor target_logits = target_output.logits.view({batch_size, num_val_tokens, vocab_size}); // prepare input for rejection sampling - auto rejection_sampler = + std::unique_ptr rejection_sampler = std::make_unique(sampling_params.do_sample, sampling_params.all_random_sample, sampling_params.all_greedy_sample, @@ -1787,9 +2580,21 @@ SampleOutput MTPWorkerImpl::validate(const SamplingParameters& sampling_params, /*mask_out_rejected_tokens=*/true); // process embedding - auto embeddings = target_output.sample_output.embeddings; + torch::Tensor embeddings = target_output.sample_output.embeddings; sample_output.embeddings = embeddings.view({batch_size, num_val_tokens, embeddings.size(-1)}); + if (pruned_prefix_lengths != nullptr) { + sync_pruned_boundary_outputs(sample_output, + target_output, + batch_size, + num_val_tokens, + num_speculative_tokens, + *pruned_prefix_lengths); + apply_pruned_prefix_lengths(sample_output, + target_output.sample_output.next_tokens, + num_speculative_tokens, + *pruned_prefix_lengths); + } return sample_output; } diff --git a/xllm/core/runtime/mtp_worker_impl.h b/xllm/core/runtime/mtp_worker_impl.h index e961019cee..edf5b2e9c9 100644 --- a/xllm/core/runtime/mtp_worker_impl.h +++ b/xllm/core/runtime/mtp_worker_impl.h @@ -15,11 +15,17 @@ limitations under the License. #pragma once +#include +#include +#include +#include + #include "framework/kv_cache/embedding_cache.h" #include "framework/kv_cache_transfer/kv_cache_transfer.h" #if defined(USE_NPU) #include "framework/kv_cache_transfer/spec_kv_cache_transfer.h" #endif +#include "runtime/adaptive_speculative_controller.h" #include "runtime/speculative_worker_impl.h" namespace xllm { @@ -49,7 +55,8 @@ class MTPWorkerImpl : public SpeculativeWorkerImpl { const runtime::Options& options, const runtime::Options& target_options, const runtime::Options& draft_options, - bool enable_opt_validate_probs = false); + bool enable_opt_validate_probs = false, + bool enable_adaptive_speculative_decode = false); public: bool init_model(const std::string& model_weights_path, @@ -77,24 +84,45 @@ class MTPWorkerImpl : public SpeculativeWorkerImpl { void fill_validate_input_from_draft_outputs( const std::vector& draft_outputs, ForwardInput& validate_input, + int32_t num_speculative_tokens, + Stream& compute_stream); + void fill_validate_input_from_draft_outputs( + const std::vector& draft_outputs, + ForwardInput& validate_input, + const std::vector& per_seq_val_tokens, Stream& compute_stream); std::optional run_validate( const ForwardInput& input, const std::vector& draft_outputs, - ForwardInput& validate_input); + ForwardInput& validate_input, + int32_t num_speculative_tokens, + const std::vector* pruned_prefix_lengths = nullptr); + std::optional run_validate( + const ForwardInput& input, + const std::vector& draft_outputs, + ForwardInput& validate_input, + int32_t num_speculative_tokens, + const std::vector& per_seq_val_tokens, + const std::vector* pruned_prefix_lengths = nullptr); - virtual SampleOutput validate(const SamplingParameters& sampling_params, - const std::vector& draft_outputs, - const ForwardOutput& target_output); + virtual SampleOutput validate( + const SamplingParameters& sampling_params, + const std::vector& draft_outputs, + const ForwardOutput& target_output, + int32_t num_speculative_tokens, + const std::vector* pruned_prefix_lengths = nullptr); // Hook for algorithm-specific draft output post-processing during decode. // Default MTP behavior always compresses probs for cache storage. virtual void process_draft_sample_output(SampleOutput& sample_output); - SampleOutput validate(const SamplingParameters& sampling_params, - const torch::Tensor& draft_token_ids, - const torch::Tensor& draft_probs, - const ForwardOutput& target_output); + SampleOutput validate( + const SamplingParameters& sampling_params, + const torch::Tensor& draft_token_ids, + const torch::Tensor& draft_probs, + const ForwardOutput& target_output, + int32_t num_speculative_tokens, + const std::vector* pruned_prefix_lengths = nullptr); // PD separation: placeholder size for empty embedding slot. Default: 1x // hidden_size. Eagle3 overrides to 3 * target_hidden_size. @@ -111,7 +139,11 @@ class MTPWorkerImpl : public SpeculativeWorkerImpl { // Prepare target validate input from cached target context. void prepare_validate_inputs(const ForwardInput& inputs, - ForwardInput& validate_inputs); + ForwardInput& validate_inputs, + int32_t num_speculative_tokens); + void prepare_validate_inputs(const ForwardInput& inputs, + ForwardInput& validate_inputs, + const std::vector& per_seq_val_tokens); // prepare inputs for draft model at Decode phase. void prepare_draft_inputs(const ForwardInput& inputs, @@ -128,7 +160,15 @@ class MTPWorkerImpl : public SpeculativeWorkerImpl { ForwardInput& extend_input); void write_target_context_to_cache(const ForwardInput& input, - const SampleOutput& validate_output); + const SampleOutput& validate_output, + int32_t num_speculative_tokens); + void record_validate_metrics( + const SampleOutput& validate_output, + int32_t num_speculative_tokens, + const std::vector* pruned_prefix_lengths = nullptr) const; + bool adaptive_enabled() const; + double adaptive_step_time_estimate(const ForwardInput& input, + int32_t max_speculative_tokens) const; protected: // Draft model worker @@ -140,6 +180,7 @@ class MTPWorkerImpl : public SpeculativeWorkerImpl { // Whether validation directly uses selected-only draft_probs [B, S]. // If false, selected-only cache values are restored to dense [B, S, V]. bool enable_opt_validate_probs_ = false; + std::unique_ptr adaptive_spec_controller_; #if defined(USE_NPU) || defined(USE_MLU) std::shared_ptr kv_cache_transfer_; diff --git a/xllm/core/runtime/options.h b/xllm/core/runtime/options.h index 161b9200a8..69fdc78537 100644 --- a/xllm/core/runtime/options.h +++ b/xllm/core/runtime/options.h @@ -83,6 +83,10 @@ struct Options { PROPERTY(bool, speculative_suffix_use_tree_spec) = false; + PROPERTY(bool, enable_adaptive_speculative_decode) = false; + + PROPERTY(double, adaptive_speculative_min_gain) = 0.0; + // enable speculative decode PROPERTY(bool, enable_speculative_decode) = false; diff --git a/xllm/core/runtime/speculative_profile_registry.cpp b/xllm/core/runtime/speculative_profile_registry.cpp new file mode 100644 index 0000000000..875c0170be --- /dev/null +++ b/xllm/core/runtime/speculative_profile_registry.cpp @@ -0,0 +1,91 @@ +/* Copyright 2026 The xLLM Authors. All Rights Reserved. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + https://github.com/jd-opensource/xllm/blob/main/LICENSE + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +==============================================================================*/ + +#include "runtime/speculative_profile_registry.h" + +#include + +#include +#include +#include + +namespace xllm { +namespace { + +double sanitize_non_negative(double value) { + if (!std::isfinite(value)) { + return 0.0; + } + return std::max(value, 0.0); +} + +} // namespace + +SpeculativeProfileRegistry& SpeculativeProfileRegistry::get_instance() { + static SpeculativeProfileRegistry registry; + return registry; +} + +void SpeculativeProfileRegistry::set_validate_time_predictor( + const ValidateTimePredictor& predictor) { + ValidateTimePredictor sanitized_predictor = predictor; + sanitized_predictor.intercept_ms = + sanitize_non_negative(predictor.intercept_ms); + sanitized_predictor.batch_ms = sanitize_non_negative(predictor.batch_ms); + sanitized_predictor.query_token_ms = + sanitize_non_negative(predictor.query_token_ms); + sanitized_predictor.query_prefix_ms = + sanitize_non_negative(predictor.query_prefix_ms); + std::lock_guard lock(mutex_); + validate_time_predictor_ = sanitized_predictor; +} + +void SpeculativeProfileRegistry::reset_validate_time_predictor() { + std::lock_guard lock(mutex_); + validate_time_predictor_.reset(); +} + +bool SpeculativeProfileRegistry::has_validate_time_predictor() const { + std::lock_guard lock(mutex_); + return validate_time_predictor_.has_value(); +} + +std::optional +SpeculativeProfileRegistry::validate_time_predictor() const { + std::lock_guard lock(mutex_); + return validate_time_predictor_; +} + +double SpeculativeProfileRegistry::predict_validate_time_ms( + int32_t batch_size, + int32_t query_len, + double avg_prefix_len) const { + std::optional predictor = validate_time_predictor(); + if (!predictor.has_value() || batch_size <= 0 || query_len <= 0 || + !std::isfinite(avg_prefix_len)) { + return 0.0; + } + + const double batch = static_cast(batch_size); + const double query = static_cast(query_len); + const double prefix = std::max(avg_prefix_len, 0.0); + const double predicted_time = + predictor->intercept_ms + predictor->batch_ms * batch + + predictor->query_token_ms * batch * query + + predictor->query_prefix_ms * batch * query * prefix; + return sanitize_non_negative(predicted_time); +} + +} // namespace xllm diff --git a/xllm/core/runtime/speculative_profile_registry.h b/xllm/core/runtime/speculative_profile_registry.h new file mode 100644 index 0000000000..4d41e77a20 --- /dev/null +++ b/xllm/core/runtime/speculative_profile_registry.h @@ -0,0 +1,56 @@ +/* Copyright 2026 The xLLM Authors. All Rights Reserved. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + https://github.com/jd-opensource/xllm/blob/main/LICENSE + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +==============================================================================*/ + +#pragma once + +#include +#include +#include + +namespace xllm { + +// Thread-safe singleton storing the validate time predictor coefficients +// fitted by ProfileManager. Used by AdaptiveSpeculativeController to estimate +// per-seq validate cost. Also serves as the gate for enabling the adaptive +// path: adaptive decode only activates after the predictor is set. +class SpeculativeProfileRegistry final { + public: + struct ValidateTimePredictor { + double intercept_ms = 0.0; + double batch_ms = 0.0; + double query_token_ms = 0.0; + double query_prefix_ms = 0.0; + }; + + static SpeculativeProfileRegistry& get_instance(); + + void set_validate_time_predictor(const ValidateTimePredictor& predictor); + void reset_validate_time_predictor(); + + bool has_validate_time_predictor() const; + std::optional validate_time_predictor() const; + + double predict_validate_time_ms(int32_t batch_size, + int32_t query_len, + double avg_prefix_len) const; + + private: + SpeculativeProfileRegistry() = default; + + mutable std::mutex mutex_; + std::optional validate_time_predictor_; +}; + +} // namespace xllm diff --git a/xllm/core/runtime/speculative_worker_impl.cpp b/xllm/core/runtime/speculative_worker_impl.cpp index e7cf4b1cd5..f624c87d0b 100644 --- a/xllm/core/runtime/speculative_worker_impl.cpp +++ b/xllm/core/runtime/speculative_worker_impl.cpp @@ -201,6 +201,40 @@ void SpeculativeWorkerImpl::update_sampling_params( TENSOR_REPEAT(sampling_params.do_sample, num_val_tokens); } +void SpeculativeWorkerImpl::update_sampling_params( + SamplingParameters& sampling_params, + const std::vector& per_seq_val_tokens, + const int32_t total_num_val_tokens) { + std::vector selected_token_idxes_vec; + selected_token_idxes_vec.reserve(total_num_val_tokens); + for (int32_t i = 0; i < total_num_val_tokens; i++) { + selected_token_idxes_vec.emplace_back(i); + } + torch::Tensor selected_token_idxes = torch::tensor(selected_token_idxes_vec); + sampling_params.selected_token_idxes = selected_token_idxes.to(device_); + sampling_params.sample_idxes = selected_token_idxes.to(device_); + + torch::Tensor repeats_tensor = + torch::tensor(std::vector(per_seq_val_tokens.begin(), + per_seq_val_tokens.end()), + torch::kLong) + .to(device_); + auto repeat_per_seq = [&](torch::Tensor& tensor) { + if (!tensor.defined()) return; + tensor = tensor.repeat_interleave(repeats_tensor, /*dim=*/0); + }; + repeat_per_seq(sampling_params.frequency_penalties); + repeat_per_seq(sampling_params.presence_penalties); + repeat_per_seq(sampling_params.repetition_penalties); + repeat_per_seq(sampling_params.temperatures); + repeat_per_seq(sampling_params.top_p); + repeat_per_seq(sampling_params.top_k); + repeat_per_seq(sampling_params.unique_token_ids); + repeat_per_seq(sampling_params.unique_token_counts); + repeat_per_seq(sampling_params.unique_token_ids_lens); + repeat_per_seq(sampling_params.do_sample); +} + void SpeculativeWorkerImpl::prepare_validate_inputs( const ForwardInput& input, ForwardInput& validate_input) { diff --git a/xllm/core/runtime/speculative_worker_impl.h b/xllm/core/runtime/speculative_worker_impl.h index 3c4e4075a7..8845709e1f 100644 --- a/xllm/core/runtime/speculative_worker_impl.h +++ b/xllm/core/runtime/speculative_worker_impl.h @@ -124,6 +124,9 @@ class SpeculativeWorkerImpl : public WorkerImpl { void update_sampling_params(SamplingParameters& sampling_params, const int32_t num_val_tokens, const int32_t total_num_val_tokens); + void update_sampling_params(SamplingParameters& sampling_params, + const std::vector& per_seq_val_tokens, + const int32_t total_num_val_tokens); // prepare inputs for target model at Decode phase (validation). void prepare_validate_inputs(const ForwardInput& inputs, diff --git a/xllm/core/runtime/worker.cpp b/xllm/core/runtime/worker.cpp index 0d9e614567..b289e23276 100644 --- a/xllm/core/runtime/worker.cpp +++ b/xllm/core/runtime/worker.cpp @@ -88,6 +88,13 @@ bool Worker::allocate_kv_cache(const KVCacheShape& kv_cache_shape) { return impl_->allocate_kv_cache(kv_cache_shape); } +bool Worker::set_speculative_validate_time_predictor( + const SpeculativeProfileRegistry::ValidateTimePredictor& predictor) { + SpeculativeProfileRegistry::get_instance().set_validate_time_predictor( + predictor); + return true; +} + void Worker::get_cache_info(uint64_t& cluster_id, std::string& addr, uint16_t& port) { diff --git a/xllm/core/runtime/worker.h b/xllm/core/runtime/worker.h index 5b7e30fc40..7a4f2fc18d 100644 --- a/xllm/core/runtime/worker.h +++ b/xllm/core/runtime/worker.h @@ -29,6 +29,7 @@ limitations under the License. #include "framework/state_dict/state_dict.h" #include "runtime/executor.h" #include "runtime/options.h" +#include "runtime/speculative_profile_registry.h" #include "runtime/worker_impl.h" #include "util/threadpool.h" @@ -70,6 +71,9 @@ class Worker { // allocate kv cache. blocking call bool allocate_kv_cache(const KVCacheShape& kv_cache_shape); + bool set_speculative_validate_time_predictor( + const SpeculativeProfileRegistry::ValidateTimePredictor& predictor); + void get_cache_info(uint64_t& cluster_id, std::string& addr, uint16_t& port); bool link_cluster(const std::vector& cluster_ids, diff --git a/xllm/core/runtime/worker_client.cpp b/xllm/core/runtime/worker_client.cpp index 0759598e4e..17dc7059b5 100644 --- a/xllm/core/runtime/worker_client.cpp +++ b/xllm/core/runtime/worker_client.cpp @@ -42,6 +42,11 @@ bool WorkerClient::allocate_kv_cache(const KVCacheShape& kv_cache_shape) { return worker_->allocate_kv_cache(kv_cache_shape); } +bool WorkerClient::set_speculative_validate_time_predictor( + const SpeculativeProfileRegistry::ValidateTimePredictor& predictor) { + return worker_->set_speculative_validate_time_predictor(predictor); +} + void WorkerClient::get_cache_info(uint64_t& cluster_id, std::string& addr, uint16_t& port) { diff --git a/xllm/core/runtime/worker_client.h b/xllm/core/runtime/worker_client.h index 8186527ccc..fae87902fb 100644 --- a/xllm/core/runtime/worker_client.h +++ b/xllm/core/runtime/worker_client.h @@ -27,6 +27,7 @@ limitations under the License. #include "framework/quant_args.h" #include "framework/state_dict/state_dict.h" #include "runtime/executor.h" +#include "runtime/speculative_profile_registry.h" #include "runtime/worker.h" #include "util/threadpool.h" @@ -62,6 +63,9 @@ class WorkerClient { // allocate kv cache. blocking call virtual bool allocate_kv_cache(const KVCacheShape& kv_cache_shape); + virtual bool set_speculative_validate_time_predictor( + const SpeculativeProfileRegistry::ValidateTimePredictor& predictor); + virtual void get_cache_info(uint64_t& cluster_id, std::string& addr, uint16_t& port); diff --git a/xllm/core/scheduler/profile/CMakeLists.txt b/xllm/core/scheduler/profile/CMakeLists.txt index dd8fd05eec..a37689b7cd 100644 --- a/xllm/core/scheduler/profile/CMakeLists.txt +++ b/xllm/core/scheduler/profile/CMakeLists.txt @@ -14,6 +14,7 @@ cc_library( time_predictor.cpp DEPS :batch + :distributed_runtime :request :runtime glog::glog diff --git a/xllm/core/scheduler/profile/profile_manager.cpp b/xllm/core/scheduler/profile/profile_manager.cpp index e19a5950b6..6c4bd07ee4 100644 --- a/xllm/core/scheduler/profile/profile_manager.cpp +++ b/xllm/core/scheduler/profile/profile_manager.cpp @@ -19,8 +19,11 @@ limitations under the License. #include #include +#include #include +#include #include +#include #include #include #include @@ -36,6 +39,7 @@ limitations under the License. #include "core/framework/config/speculative_config.h" #include "framework/batch/batch_factory.h" #include "framework/request/request_state.h" +#include "runtime/speculative_profile_registry.h" #include "scheduler/profile/graph_warmup.h" #include "util/rec_model_utils.h" @@ -53,6 +57,7 @@ ProfileManager::ProfileManager(Engine* engine, const Options& options) if (options.enable_profile_step_time()) { LOG(INFO) << "Starting profiliing step time."; profile_step_time(false); + profile_speculative_validate_time(); // test accuracy // eval_sequence_latency_prediction(); // eval_batch_latency_prediction("only_prefill"); @@ -371,6 +376,178 @@ void ProfileManager::train_decode_time_predictor( decode_time_predictor_->fit_for_decode(time_profiling_data); } +void ProfileManager::train_speculative_validate_time_predictor( + const std::vector>& + time_profiling_data) { + if (time_profiling_data.empty()) { + return; + } + + constexpr int32_t kNumCoefficients = 4; + Eigen::MatrixXd matrix(time_profiling_data.size(), kNumCoefficients); + Eigen::VectorXd target(time_profiling_data.size()); + for (int32_t i = 0; i < static_cast(time_profiling_data.size()); + ++i) { + const int32_t batch_size = std::get<0>(time_profiling_data[i]); + const int32_t query_len = std::get<1>(time_profiling_data[i]); + const int32_t prefix_len = std::get<2>(time_profiling_data[i]); + const double batch = static_cast(batch_size); + const double query = static_cast(query_len); + const double prefix = static_cast(prefix_len); + matrix(i, 0) = 1.0; + matrix(i, 1) = batch; + matrix(i, 2) = batch * query; + matrix(i, 3) = batch * query * prefix; + target(i) = std::get<3>(time_profiling_data[i]); + } + + Eigen::VectorXd coefficients = matrix.colPivHouseholderQr().solve(target); + double sum_abs_error = 0.0; + double sum_percentage_error = 0.0; + for (int32_t i = 0; i < static_cast(time_profiling_data.size()); + ++i) { + const double actual = std::get<3>(time_profiling_data[i]); + const double prediction = matrix.row(i).dot(coefficients); + const double abs_error = std::abs(prediction - actual); + sum_abs_error += abs_error; + if (actual > 0.0) { + sum_percentage_error += abs_error / actual; + } + } + const double mae = + sum_abs_error / static_cast(time_profiling_data.size()); + const double mape = sum_percentage_error / + static_cast(time_profiling_data.size()) * 100.0; + + for (int32_t i = 0; i < kNumCoefficients; ++i) { + if (coefficients(i) < 0.0) { + LOG(ERROR) << "Negative speculative validate coefficient: " + << coefficients(i) << ", set it to 0."; + coefficients(i) = 0.0; + } + } + + SpeculativeProfileRegistry::ValidateTimePredictor predictor; + predictor.intercept_ms = coefficients(0); + predictor.batch_ms = coefficients(1); + predictor.query_token_ms = coefficients(2); + predictor.query_prefix_ms = coefficients(3); + SpeculativeProfileRegistry::get_instance().set_validate_time_predictor( + predictor); + if (!engine_->set_speculative_validate_time_predictor(predictor)) { + LOG(WARNING) + << "Failed to broadcast speculative validate predictor to workers. " + << "Adaptive speculative decode will fallback to runtime EWMA on " + << "workers without the profile."; + } + + LOG(INFO) << "Fitted speculative validate equation: time = " + << predictor.batch_ms << " * batch_size + " + << predictor.query_token_ms << " * batch_size * query_len + " + << predictor.query_prefix_ms + << " * batch_size * query_len * prefix_len + " + << predictor.intercept_ms << ", MAE: " << mae << ", MAPE: " << mape + << "%"; +} + +void ProfileManager::profile_speculative_validate_time() { + const SpeculativeConfig& speculative_config = + ::xllm::SpeculativeConfig::get_instance(); + std::string speculative_algorithm = + speculative_config.speculative_algorithm(); + std::transform( + speculative_algorithm.begin(), + speculative_algorithm.end(), + speculative_algorithm.begin(), + [](unsigned char c) { return static_cast(std::tolower(c)); }); + if (speculative_config.num_speculative_tokens() <= 1 || + speculative_algorithm != "mtp") { + return; + } + LOG(INFO) << "Starting speculative validate profile for MTP, " + << "adaptive_enabled=" + << speculative_config.enable_adaptive_speculative_decode(); + + auto& model_args = engine_->model_args(); + const int32_t max_context_len = model_args.max_position_embeddings(); + const int32_t profile_max_prompt_length = + std::min(max_context_len, options_.profile_max_prompt_length()); + if (profile_max_prompt_length <= 16) { + LOG(WARNING) + << "Skip speculative validate profile because prompt length is too " + << "small: " << profile_max_prompt_length; + return; + } + + const int32_t max_query_len = + std::min(speculative_config.num_speculative_tokens() + 1, 10); + const int32_t max_batch_size = + std::min(options_.max_seqs_per_batch(), 256); + std::vector query_lens; + query_lens.push_back(1); + if (max_query_len > 2) { + query_lens.push_back((max_query_len + 1) / 2); + } + if (max_query_len > 1) { + query_lens.push_back(max_query_len); + } + query_lens.erase(std::unique(query_lens.begin(), query_lens.end()), + query_lens.end()); + + constexpr int32_t kMaxProfileBatchSize = 32; + std::vector candidate_batch_sizes = {1, 16, 32}; + std::vector batch_sizes; + for (const int32_t batch_size : candidate_batch_sizes) { + if (batch_size <= max_batch_size && batch_size <= kMaxProfileBatchSize) { + batch_sizes.push_back(batch_size); + } + } + if (batch_sizes.empty()) { + batch_sizes.push_back(std::min( + std::max(max_batch_size, 1), kMaxProfileBatchSize)); + } + + std::vector prefix_lens; + prefix_lens.push_back(profile_max_prompt_length); + if (profile_max_prompt_length >= 64) { + prefix_lens.push_back(profile_max_prompt_length / 4); + } + prefix_lens.push_back(16); + for (int32_t& prefix_len : prefix_lens) { + prefix_len = std::clamp(prefix_len, 16, profile_max_prompt_length); + } + std::sort(prefix_lens.begin(), prefix_lens.end()); + prefix_lens.erase(std::unique(prefix_lens.begin(), prefix_lens.end()), + prefix_lens.end()); + + std::vector> + time_profiling_data; + const int32_t total_blocks = + static_cast(block_manager_pool_->num_blocks()); + const auto block_size = block_manager_pool_->options().block_size(); + for (const int32_t prefix_len : prefix_lens) { + for (const int32_t query_len : query_lens) { + for (const int32_t batch_size : batch_sizes) { + const int32_t token_length = prefix_len + query_len; + const int32_t blocks_per_seq = + (prefix_len + block_size - 1) / block_size + + (token_length + block_size - 1) / block_size; + if (batch_size * blocks_per_seq > total_blocks * 9 / 10) { + continue; + } + double latency_mean = 0.0; + for (int32_t k = 0; k < profile_count_per_step_; ++k) { + latency_mean += run_request(token_length, prefix_len, batch_size); + } + latency_mean /= static_cast(profile_count_per_step_); + time_profiling_data.emplace_back( + batch_size, query_len, prefix_len, latency_mean); + } + } + } + train_speculative_validate_time_predictor(time_profiling_data); +} + // ----------------------predict step time----------------------- std::vector ProfileManager::get_coefficients(bool is_prefill) { if (is_prefill) { diff --git a/xllm/core/scheduler/profile/profile_manager.h b/xllm/core/scheduler/profile/profile_manager.h index bd1c22846e..1ab8aa7527 100644 --- a/xllm/core/scheduler/profile/profile_manager.h +++ b/xllm/core/scheduler/profile/profile_manager.h @@ -17,6 +17,7 @@ limitations under the License. #include #include +#include #include #include "common/macros.h" @@ -114,6 +115,10 @@ class ProfileManager { void train_prefill_time_predictor( std::vector> time_profiling_data); + void train_speculative_validate_time_predictor( + const std::vector>& + time_profiling_data); + double get_constant_overhead(); int32_t get_quadratic_root(Sequence* sequence, double budget); @@ -146,6 +151,8 @@ class ProfileManager { void profile_token_budget(); + void profile_speculative_validate_time(); + // Warmup ACL graph executor according to the instance role. void warmup_for_graph(); diff --git a/xllm/proto/worker.proto b/xllm/proto/worker.proto index ed95104d06..e168d6044b 100644 --- a/xllm/proto/worker.proto +++ b/xllm/proto/worker.proto @@ -56,6 +56,13 @@ message AllocateKVCacheRequest { KVCacheShape kv_cache_shape = 1; } +message SpeculativeValidateTimePredictor { + double intercept_ms = 1; + double batch_ms = 2; + double query_token_ms = 3; + double query_prefix_ms = 4; +} + message SleepRequest { MasterStatus master_status = 1; } @@ -354,6 +361,8 @@ service DistributeWorker { rpc ProcessGroupTest (Empty) returns (Status); rpc ProfileDeviceMemory (Empty) returns (DeviceMemory); rpc AllocateKVCache (AllocateKVCacheRequest) returns (Status); + rpc SetSpeculativeValidateTimePredictor(SpeculativeValidateTimePredictor) + returns (Status) {} rpc AllocateKVCacheWithTransfer(AllocateKVCacheRequest) returns (Status) {} rpc PullKVCache(PullKVCacheRequest) returns (Status) {} rpc GetCacheInfo(Empty) returns (CacheInfo) {} diff --git a/xllm/xllm.cpp b/xllm/xllm.cpp index a923f5f278..d4514ed01d 100644 --- a/xllm/xllm.cpp +++ b/xllm/xllm.cpp @@ -146,6 +146,10 @@ Options create_options(const std::string& instance_name, bool is_local) { speculative_config.speculative_suffix_max_cached_requests()) .speculative_suffix_use_tree_spec( speculative_config.speculative_suffix_use_tree_spec()) + .enable_adaptive_speculative_decode( + speculative_config.enable_adaptive_speculative_decode()) + .adaptive_speculative_min_gain( + speculative_config.adaptive_speculative_min_gain()) .num_request_handling_threads( service_config.num_request_handling_threads()) .communication_backend(parallel_config.communication_backend()) From e19508c2d99745f30748325b63479b79ce73b0b1 Mon Sep 17 00:00:00 2001 From: huangweizhe1 Date: Thu, 9 Jul 2026 11:36:22 +0800 Subject: [PATCH 2/2] refactor: move adaptive speculative files to framework/speculative. Move AdaptiveSpeculativeController and SpeculativeProfileRegistry from runtime/ to framework/speculative/ to keep runtime/ focused on worker/executor concerns. Update all include paths and CMake targets. --- tests/core/runtime/CMakeLists.txt | 6 +++--- .../adaptive_speculative_controller_test.cpp | 4 ++-- .../speculative_profile_registry_test.cpp | 2 +- xllm/core/distributed_runtime/comm_channel.h | 2 +- xllm/core/distributed_runtime/engine.h | 2 +- xllm/core/framework/CMakeLists.txt | 1 + xllm/core/framework/speculative/CMakeLists.txt | 16 ++++++++++++++++ .../adaptive_speculative_controller.cpp | 4 ++-- .../adaptive_speculative_controller.h | 0 .../speculative_profile_registry.cpp | 2 +- .../speculative}/speculative_profile_registry.h | 0 xllm/core/runtime/CMakeLists.txt | 5 +---- xllm/core/runtime/mtp_worker_impl.cpp | 2 +- xllm/core/runtime/mtp_worker_impl.h | 2 +- xllm/core/runtime/worker.h | 2 +- xllm/core/runtime/worker_client.h | 2 +- xllm/core/scheduler/profile/profile_manager.cpp | 2 +- 17 files changed, 34 insertions(+), 20 deletions(-) create mode 100644 xllm/core/framework/speculative/CMakeLists.txt rename xllm/core/{runtime => framework/speculative}/adaptive_speculative_controller.cpp (98%) rename xllm/core/{runtime => framework/speculative}/adaptive_speculative_controller.h (100%) rename xllm/core/{runtime => framework/speculative}/speculative_profile_registry.cpp (97%) rename xllm/core/{runtime => framework/speculative}/speculative_profile_registry.h (100%) diff --git a/tests/core/runtime/CMakeLists.txt b/tests/core/runtime/CMakeLists.txt index 0774aa2a4d..ce44363313 100644 --- a/tests/core/runtime/CMakeLists.txt +++ b/tests/core/runtime/CMakeLists.txt @@ -82,8 +82,8 @@ cc_test( adaptive_speculative_controller_test SRCS adaptive_speculative_controller_test.cpp - "${PROJECT_SOURCE_DIR}/xllm/core/runtime/adaptive_speculative_controller.cpp" - "${PROJECT_SOURCE_DIR}/xllm/core/runtime/speculative_profile_registry.cpp" + "${PROJECT_SOURCE_DIR}/xllm/core/framework/speculative/adaptive_speculative_controller.cpp" + "${PROJECT_SOURCE_DIR}/xllm/core/framework/speculative/speculative_profile_registry.cpp" DEPS torch glog::glog @@ -96,7 +96,7 @@ cc_test( speculative_profile_registry_test SRCS speculative_profile_registry_test.cpp - "${PROJECT_SOURCE_DIR}/xllm/core/runtime/speculative_profile_registry.cpp" + "${PROJECT_SOURCE_DIR}/xllm/core/framework/speculative/speculative_profile_registry.cpp" DEPS glog::glog GTest::gtest_main diff --git a/tests/core/runtime/adaptive_speculative_controller_test.cpp b/tests/core/runtime/adaptive_speculative_controller_test.cpp index 7a4f859f70..dd681d9599 100644 --- a/tests/core/runtime/adaptive_speculative_controller_test.cpp +++ b/tests/core/runtime/adaptive_speculative_controller_test.cpp @@ -13,15 +13,15 @@ See the License for the specific language governing permissions and limitations under the License. ==============================================================================*/ -#include "runtime/adaptive_speculative_controller.h" +#include "core/framework/speculative/adaptive_speculative_controller.h" #include #include #include +#include "core/framework/speculative/speculative_profile_registry.h" #include "runtime/options.h" -#include "runtime/speculative_profile_registry.h" namespace xllm { namespace { diff --git a/tests/core/runtime/speculative_profile_registry_test.cpp b/tests/core/runtime/speculative_profile_registry_test.cpp index e9c8b45404..2abc877c5f 100644 --- a/tests/core/runtime/speculative_profile_registry_test.cpp +++ b/tests/core/runtime/speculative_profile_registry_test.cpp @@ -13,7 +13,7 @@ See the License for the specific language governing permissions and limitations under the License. ==============================================================================*/ -#include "runtime/speculative_profile_registry.h" +#include "core/framework/speculative/speculative_profile_registry.h" #include diff --git a/xllm/core/distributed_runtime/comm_channel.h b/xllm/core/distributed_runtime/comm_channel.h index 68bff66580..59ce198fde 100644 --- a/xllm/core/distributed_runtime/comm_channel.h +++ b/xllm/core/distributed_runtime/comm_channel.h @@ -24,11 +24,11 @@ limitations under the License. #include #include "common/types.h" +#include "core/framework/speculative/speculative_profile_registry.h" #include "framework/kv_cache/kv_cache_shape.h" #include "framework/xtensor/xtensor.h" #include "runtime/forward_params.h" #include "runtime/params_utils.h" -#include "runtime/speculative_profile_registry.h" #include "worker.pb.h" namespace xllm { diff --git a/xllm/core/distributed_runtime/engine.h b/xllm/core/distributed_runtime/engine.h index a1524b8891..f626d5e677 100644 --- a/xllm/core/distributed_runtime/engine.h +++ b/xllm/core/distributed_runtime/engine.h @@ -19,13 +19,13 @@ limitations under the License. #include +#include "core/framework/speculative/speculative_profile_registry.h" #include "framework/batch/batch.h" #include "framework/block/block_manager_pool.h" #include "framework/model/model_args.h" #include "framework/tokenizer/tokenizer.h" #include "framework/tokenizer/tokenizer_args.h" #include "runtime/options.h" -#include "runtime/speculative_profile_registry.h" namespace xllm { class Engine { diff --git a/xllm/core/framework/CMakeLists.txt b/xllm/core/framework/CMakeLists.txt index 770b5d5345..96b48b9e8f 100644 --- a/xllm/core/framework/CMakeLists.txt +++ b/xllm/core/framework/CMakeLists.txt @@ -24,6 +24,7 @@ add_subdirectory(state_dict) add_subdirectory(tokenizer) add_subdirectory(eplb) add_subdirectory(dit_cache) +add_subdirectory(speculative) cc_library( diff --git a/xllm/core/framework/speculative/CMakeLists.txt b/xllm/core/framework/speculative/CMakeLists.txt new file mode 100644 index 0000000000..99771e3f25 --- /dev/null +++ b/xllm/core/framework/speculative/CMakeLists.txt @@ -0,0 +1,16 @@ +include(cc_library) + +cc_library( + NAME + speculative + HDRS + adaptive_speculative_controller.h + speculative_profile_registry.h + SRCS + adaptive_speculative_controller.cpp + speculative_profile_registry.cpp + DEPS + torch + glog::glog + proto::xllm_proto +) diff --git a/xllm/core/runtime/adaptive_speculative_controller.cpp b/xllm/core/framework/speculative/adaptive_speculative_controller.cpp similarity index 98% rename from xllm/core/runtime/adaptive_speculative_controller.cpp rename to xllm/core/framework/speculative/adaptive_speculative_controller.cpp index fc672cc73b..223f65d71f 100644 --- a/xllm/core/runtime/adaptive_speculative_controller.cpp +++ b/xllm/core/framework/speculative/adaptive_speculative_controller.cpp @@ -13,7 +13,7 @@ See the License for the specific language governing permissions and limitations under the License. ==============================================================================*/ -#include "runtime/adaptive_speculative_controller.h" +#include "core/framework/speculative/adaptive_speculative_controller.h" #include @@ -23,7 +23,7 @@ limitations under the License. #include #include -#include "runtime/speculative_profile_registry.h" +#include "core/framework/speculative/speculative_profile_registry.h" #include "util/tensor_helper.h" namespace xllm { diff --git a/xllm/core/runtime/adaptive_speculative_controller.h b/xllm/core/framework/speculative/adaptive_speculative_controller.h similarity index 100% rename from xllm/core/runtime/adaptive_speculative_controller.h rename to xllm/core/framework/speculative/adaptive_speculative_controller.h diff --git a/xllm/core/runtime/speculative_profile_registry.cpp b/xllm/core/framework/speculative/speculative_profile_registry.cpp similarity index 97% rename from xllm/core/runtime/speculative_profile_registry.cpp rename to xllm/core/framework/speculative/speculative_profile_registry.cpp index 875c0170be..8af7d9a99e 100644 --- a/xllm/core/runtime/speculative_profile_registry.cpp +++ b/xllm/core/framework/speculative/speculative_profile_registry.cpp @@ -13,7 +13,7 @@ See the License for the specific language governing permissions and limitations under the License. ==============================================================================*/ -#include "runtime/speculative_profile_registry.h" +#include "core/framework/speculative/speculative_profile_registry.h" #include diff --git a/xllm/core/runtime/speculative_profile_registry.h b/xllm/core/framework/speculative/speculative_profile_registry.h similarity index 100% rename from xllm/core/runtime/speculative_profile_registry.h rename to xllm/core/framework/speculative/speculative_profile_registry.h diff --git a/xllm/core/runtime/CMakeLists.txt b/xllm/core/runtime/CMakeLists.txt index 55b89ab0ae..506c496d94 100644 --- a/xllm/core/runtime/CMakeLists.txt +++ b/xllm/core/runtime/CMakeLists.txt @@ -32,8 +32,6 @@ cc_library( mm_embed_vlm_worker_impl.h worker_client.h xservice_client.h - adaptive_speculative_controller.h - speculative_profile_registry.h speculative_worker_impl.h mtp_worker_impl.h suffix_worker_impl.h @@ -66,8 +64,6 @@ cc_library( worker_client.cpp xservice_client.cpp params_utils.cpp - adaptive_speculative_controller.cpp - speculative_profile_registry.cpp speculative_worker_impl.cpp mtp_worker_impl.cpp suffix_worker_impl.cpp @@ -82,6 +78,7 @@ cc_library( :platform :model_context :request + :speculative :encoder_cache :state_dict :dit_cache diff --git a/xllm/core/runtime/mtp_worker_impl.cpp b/xllm/core/runtime/mtp_worker_impl.cpp index 996aad1899..437afa6e38 100644 --- a/xllm/core/runtime/mtp_worker_impl.cpp +++ b/xllm/core/runtime/mtp_worker_impl.cpp @@ -39,7 +39,7 @@ limitations under the License. #include "core/framework/config/speculative_config.h" #include "core/framework/kv_cache/kv_cache_estimation.h" #include "core/framework/multimodal/mm_data.h" -#include "runtime/speculative_profile_registry.h" +#include "core/framework/speculative/speculative_profile_registry.h" #include "spec_input_builder.h" #include "util/env_var.h" #include "util/pretty_print.h" diff --git a/xllm/core/runtime/mtp_worker_impl.h b/xllm/core/runtime/mtp_worker_impl.h index edf5b2e9c9..5c4f5618f1 100644 --- a/xllm/core/runtime/mtp_worker_impl.h +++ b/xllm/core/runtime/mtp_worker_impl.h @@ -25,7 +25,7 @@ limitations under the License. #if defined(USE_NPU) #include "framework/kv_cache_transfer/spec_kv_cache_transfer.h" #endif -#include "runtime/adaptive_speculative_controller.h" +#include "core/framework/speculative/adaptive_speculative_controller.h" #include "runtime/speculative_worker_impl.h" namespace xllm { diff --git a/xllm/core/runtime/worker.h b/xllm/core/runtime/worker.h index 7a4f2fc18d..ee0ac8332e 100644 --- a/xllm/core/runtime/worker.h +++ b/xllm/core/runtime/worker.h @@ -20,6 +20,7 @@ limitations under the License. #include #include "common/types.h" +#include "core/framework/speculative/speculative_profile_registry.h" #include "forward_params.h" #include "framework/kv_cache/kv_cache_shape.h" #include "framework/model/causal_lm.h" @@ -29,7 +30,6 @@ limitations under the License. #include "framework/state_dict/state_dict.h" #include "runtime/executor.h" #include "runtime/options.h" -#include "runtime/speculative_profile_registry.h" #include "runtime/worker_impl.h" #include "util/threadpool.h" diff --git a/xllm/core/runtime/worker_client.h b/xllm/core/runtime/worker_client.h index fae87902fb..55915b9b59 100644 --- a/xllm/core/runtime/worker_client.h +++ b/xllm/core/runtime/worker_client.h @@ -19,6 +19,7 @@ limitations under the License. #include #include "common/types.h" +#include "core/framework/speculative/speculative_profile_registry.h" #include "forward_params.h" #include "framework/kv_cache/kv_cache_shape.h" #include "framework/model/causal_lm.h" @@ -27,7 +28,6 @@ limitations under the License. #include "framework/quant_args.h" #include "framework/state_dict/state_dict.h" #include "runtime/executor.h" -#include "runtime/speculative_profile_registry.h" #include "runtime/worker.h" #include "util/threadpool.h" diff --git a/xllm/core/scheduler/profile/profile_manager.cpp b/xllm/core/scheduler/profile/profile_manager.cpp index 6c4bd07ee4..26765ade54 100644 --- a/xllm/core/scheduler/profile/profile_manager.cpp +++ b/xllm/core/scheduler/profile/profile_manager.cpp @@ -37,9 +37,9 @@ limitations under the License. #include "core/framework/config/scheduler_config.h" #include "core/framework/config/service_config.h" #include "core/framework/config/speculative_config.h" +#include "core/framework/speculative/speculative_profile_registry.h" #include "framework/batch/batch_factory.h" #include "framework/request/request_state.h" -#include "runtime/speculative_profile_registry.h" #include "scheduler/profile/graph_warmup.h" #include "util/rec_model_utils.h"