执行摘要

此 PR 为 vLLM Rust 前端的增量解码器建立了模型无关的 token 归属合约，通过重构核心接口和引入归属缓冲机制，解决了流式解码中文本生成与 token 采样在不同“时钟”上推进的脱节问题。变更涉及 Rust 前端的 tokenizer、text 和 server 多个模块，是未来实现推理 token 统计等高级功能的基础性重构。

功能与动机

为什么做：在流式推理中，解码的文本片段（delta）和引擎采样的 token 元数据（token_ids, logprobs）并不同步。一个 token 可能跨引擎更新仍无法解码，而一段可见文本可能由多个 token 共同产生。原有的 TextDelta 事件将两者混合在 delta、token_ids 字段中，丢失了这种关键区分。本 PR 的动机是建立一个清晰的合约来分离这两种数据流，为下游消费者提供准确的归属信息。PR 描述中明确指出：“Decoded text and sampled token metadata advance on different clocks... This PR establishes the metadata contract only.”

实现拆解

变更遵循从底层数据结构到上层事件流的自底向上路径：

1. 建立归属模型基础：在 rust/src/tokenizer/src/incremental/attribution.rs（新增文件）中，定义了 TokenAnchor（Visible/ZeroWidth）、TokenAttribution 和 DecodedText 结构。AttributedTextBuffer 是核心状态机，管理累积文本、待解析（PendingAnchor::Unresolved）和已解析的归属记录，支持安全地按字节偏移切分输出块。

2. 改造解码器接口与实现：在 rust/src/tokenizer/src/incremental.rs 中，IncrementalDecoder trait 的 next_chunk 和 flush 方法返回类型从 String 升级为 DecodedText。DecodeStream 实现用 AttributedTextBuffer 替代了原有的两个字符串字段，并在 push_token 中通过 record_token 方法根据 is_special_id 记录归属。

3. 重构输出事件流以分离时钟：在 rust/src/text/src/output/decoded.rs 中，DecodedTextEvent::TextDelta 枚举被重构。原来的 delta、token_ids、logprobs 字段被拆分为 decoded: DecodedText（跟随解码器时钟）和 sampled: SampledDelta（跟随引擎更新时钟，包含 token_ids 和 logprobs）。decoded_text_event_stream 异步流内部逻辑相应调整，分别累积并生成两个时钟的数据。

4. 适配下游消费者与测试：更新了所有直接使用 DecodedTextEvent 的模块：

   • rust/src/server/src/routes/openai/completions.rs（OpenAI 兼容接口）：在事件匹配中提取 decoded.text 作为 delta。
   • rust/src/server/src/grpc/inference.rs（gRPC 推理层）：同样适配新的字段结构。
   • rust/src/chat/src/output/default/unified.rs 和 rust/src/chat/src/output/harmony/mod.rs（Chat 处理器）：更新模式匹配。
   • 多个测试文件（harmony/tests.rs, decoded.rs 测试模块）：新增辅助函数（如 decoded_tokens、decoded_delta）以构造符合新合约的测试事件。

5. 配套调整：在 rust/Cargo.toml 和 rust/src/tokenizer/Cargo.toml 中添加 smallvec 依赖，用于高效存储内联归属记录。更新模块的 pub use 导出以暴露新的公开类型。

rust/src/tokenizer/src/incremental/attribution.rs

引入了整个 token 归属功能的核心数据结构 TokenAnchor, TokenAttribution, DecodedText 和内部状态管理类 AttributedTextBuffer。是新合约的基础。

/// Position of one generated token in decoded text.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum TokenAnchor {
    /// The token contributed to visible decoded text.
    Visible { byte_offset: u32 },
    /// The token produced no visible decoded bytes.
    ZeroWidth { byte_offset: u32 },
}
​
impl TokenAnchor {
    /// Return a new anchor offset by `delta` bytes.
    /// 用于在拼接解码文本块时，重新计算归属偏移量。
    #[must_use]
    pub(super) fn offset_by(self, delta: i64) -> Self {
        let byte_offset = i64::from(self.byte_offset())
            .checked_add(delta)
            .and_then(|offset| u32::try_from(offset).ok())
            .expect("token anchor byte offset out of range");
        match self {
            Self::Visible { .. } => Self::Visible { byte_offset },
            Self::ZeroWidth { .. } => Self::ZeroWidth { byte_offset },
        }
    }
}
​
/// One generated token and its position in decoded text.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct TokenAttribution {
    /// Token ID passed to [`super::IncrementalDecoder::push_token`].
    pub token_id: u32,
    /// Position of the token in the decoded text.
    pub anchor: TokenAnchor,
}
​
/// Decoded text and the generated tokens attributed to it.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct DecodedText {
    /// Decoded UTF-8 text.
    pub text: String,
    /// One entry per generated token, in generation order.
    /// Offsets are local to `text`. Visible offsets may repeat when multiple
    /// tokens jointly decode into one character. A zero-width offset may equal
    /// `text.len()`.
    pub attributions: SmallVec<[TokenAttribution; 4]>,
}
​
impl DecodedText {
    /// Append another decoded fragment, rebasing its token anchors.
    /// 核心操作：将另一个解码片段的文本和归属记录追加到当前实例。
    /// 所有新归属的字节偏移量会加上当前文本的长度，实现偏移量的重基。
    pub fn append(&mut self, other: Self) {
        if self.text.is_empty() && self.attributions.is_empty() {
            *self = other;
            return;
        }
​
        let byte_offset = offset_as_u32(self.text.len());
        self.text.push_str(&other.text);
        self.attributions.extend(other.attributions.into_iter().map(|attribution| {
            TokenAttribution {
                token_id: attribution.token_id,
                anchor: attribution.anchor.offset_by(i64::from(byte_offset)),
            }
        }));
    }
}

rust/src/tokenizer/src/incremental.rs

修改了 IncrementalDecoder trait 接口，并重写了 DecodeStream 实现以使用新的归属缓冲机制。是新合约的执行者。

/// Stateful incremental decoder that emits text chunks one token at a time.
pub trait IncrementalDecoder: Send {
    /// Push one generated token and return how many new string bytes were
    /// added.
    fn push_token(&mut self, token_id: u32) -> Result<usize>;
​
    /// Consume any text which is currently ready.
    /// 接口变更：返回 `DecodedText` 而非 `String`。
    fn next_chunk(&mut self) -> Option<DecodedText>;
​
    /// Flush any remaining buffered text that has not yet been emitted.
    /// 接口变更：返回带有完整归属的 `(Option<DecodedText>, DecodedText)`。
    fn flush(&mut self, truncate_output_to: Option<usize>) -> Result<(Option<DecodedText>, DecodedText)>;
​
    /// Return cumulative decoded text so far.
    fn output(&self) -> &str;
}
​
/// [`IncrementalDecoder`] built on [`Tokenizer::decode()`] with prefix-diffing.
pub(crate) struct DecodeStream<'a, T: Tokenizer + ?Sized> {
    // ... 其他字段保持不变 ...
    // 核心变更：用 `AttributedTextBuffer` 替代原有的两个字符串状态字段。
    decoded: AttributedTextBuffer,
}
​
impl<T: Tokenizer + ?Sized> DecodeStream<'_, T> {
    // 新增方法：根据 token 是否是特殊 token 来记录零宽度或待定归属。
    fn record_token(&mut self, token_id: u32) {
        if self.skip_special_tokens && self.tokenizer.is_special_id(token_id) {
            self.decoded.record_zero_width_token(token_id);
        } else {
            self.decoded.record_pending_token(token_id);
        }
    }
    // ... seed_prefix 方法保持不变 ...
}
​
impl<T: Tokenizer + ?Sized> IncrementalDecoder for DecodeStream<'_, T> {
    fn push_token(&mut self, token_id: u32) -> Result<usize> {
        // ... 前置逻辑不变 ...
        self.ids.push(token_id);
        self.record_token(token_id); // 记录归属
        let string = self.tokenizer.decode(&self.ids, self.skip_special_tokens)?;
        let prefix_len = self.prefix.len();
        if string.len() <= prefix_len || string.ends_with('\u{FFFD}') {
            return Ok(0);
        }
        let new_chunk = &string[string.floor_char_boundary(prefix_len)..];
        self.decoded.append_visible_text(new_chunk); // 将可见文本追加到归属缓冲区
        self.ids.drain(..self.prefix_index);
        self.prefix = self.tokenizer.decode(&self.ids, self.skip_special_tokens)?;
        self.prefix_index = self.ids.len();
        Ok(new_chunk.len())
    }
​
    fn next_chunk(&mut self) -> Option<DecodedText> {
        let cutoff = self.decoded.len().saturating_sub(self.min_bytes_to_buffer);
        let cutoff = self.decoded.text().floor_char_boundary(cutoff);
        self.decoded.take_ready(cutoff) // 从归属缓冲区取出发射好的块
    }
​
    fn flush(&mut self, truncate_output_to: Option<usize>) -> Result<(Option<DecodedText>, DecodedText)> {
        // ... 清空状态，处理剩余缓冲 ...
        self.decoded.resolve_pending_zero_width(); // 将所有待定归属解析为零宽度
        // ... 截断等逻辑 ...
        Ok(self.decoded.finish()) // 返回最后一个块和完整的解码文本及归属
    }
}

rust/src/text/src/output/decoded.rs

将新的归属合约集成到文本输出事件流中，通过分离 decoded 和 sampled 字段，实现了时钟解耦。是合约的上层集成点。

/// Sample metadata emitted by one engine output update.
/// Token IDs and logprobs share the engine-update clock.
#[derive(Debug, Clone, Default, PartialEq)]
pub struct SampledDelta {
    /// Token IDs produced by this engine update, after stop-string truncation.
    pub token_ids: Vec<u32>,
    /// Per-position logprobs aligned with `token_ids`, when requested.
    pub logprobs: Option<DecodedLogprobs>,
}
​
/// Internal decoded-text event emitted before higher-level assistant
/// adaptation.
#[derive(Debug, Clone, PartialEq)]
pub enum DecodedTextEvent {
    // ... Start variant 不变 ...
    /// Decoded text and sampled metadata observed during one output update.
    TextDelta {
        /// follows the decoder clock. May be empty or contain attributions from earlier updates.
        decoded: DecodedText,
        /// follows the engine-update clock. token_ids and logprobs align with each other.
        sampled: SampledDelta,
        /// Heap-backed because terminal metadata is large and appears once per request.
        finished: Option<Box<Finished>>,
    },
}
​
// 在 `decoded_text_event_stream` 异步流中：
// 对 `decodable_token_ids` 的循环内部变更：
// let mut decoded = DecodedText::default();
// // ...
// if intermediate && let Some(chunk) = decoder.next_chunk() {
// decoded.append(chunk); // 将归属文本块追加到事件级的 `decoded` 结构
// }
// 
// 最终事件生成时：
// y.yield_ok(DecodedTextEvent::TextDelta {
// decoded, // 包含本次更新产生的所有归属文本
// sampled: SampledDelta {
// token_ids: sampled_token_ids, // 引擎本次更新产生的 token IDs
// logprobs: sampled_logprobs, // 对应的 logprobs
// },
// finished: Some(Box::new(Finished { ... })),
// }).await;

评论区精华

本 PR 的公开讨论极少：

• claude[bot]: “This pull request is from a fork — automated review is disabled. A repository maintainer can comment @claude review to run a one-time review.”
• njhill（Approver）: “Thanks @BugenZhao”
  从 commit 历史（8 commits）看，变更经历了多次内部重构和分支合并，表明作者进行了谨慎的迭代开发。

风险与影响

风险：

1. 内部 API 破坏性变更：IncrementalDecoder trait 和 DecodedTextEvent 的结构变更属于破坏性修改。尽管所有已知的内部消费者都已适配，但任何依赖这些未稳定内部接口的外部代码（如插件或分支）都将失效。
2. 状态管理复杂性：AttributedTextBuffer 内部状态（PendingAnchor、偏移计算）逻辑较为复杂。尽管有详尽的单元测试覆盖，但仍是潜在的正确性风险点。
3. 轻微性能开销：引入的归属跟踪结构和 DecodedText::append 中的偏移重计算带来了额外的内存与 CPU 开销。对于极高吞吐量的流式场景，可能成为优化目标。

影响：

• 系统：重构了 Rust 前端文本处理的核心数据流，增强了数据处理的清晰度和可扩展性。
• 用户：对外部 API 行为无直接影响。但为未来提供更精细的 token 级别统计（如推理 token 消耗）奠定了基础。
• 团队：为 Rust 前端建立了统一的 token 归属模型，简化了未来相关功能的设计和实现。

关联脉络

此 PR 与 PR #45802（[Python] Implement reasoning token usage accounting for Kimi-K3）明确相关。后者是 Python 端对推理 token 消耗统计的实现，是本 PR 所建立合约的直接下游用例之一。本 PR 通过在 Rust 前端提供相同的元数据能力，使得类似功能可以在 Rust 前端路径中实现，体现了跨语言架构的一致性演进。