执行摘要

本 PR 修复了 Rust benchmark 工具（vllm-bench）中 SSE 流式处理对非 ASCII 文本的破坏：原先每个网络 chunk 独立用 String::from_utf8_lossy 解码，多字节 UTF-8 字符被 TCP 或 HTTP 分片切断时会产生 U+FFFD 乱码，并可能污染保存的 benchmark 输出与多轮对话历史。修复方案是把 StreamedResponseHandler 的缓冲从 String 改为 Vec<u8>，先累积原始字节，只在完整 SSE 消息边界才解码；投机 JSON 解析改用临时解码视图，不破坏字节累积。改动集中于 rust/src/bench/src/backends/streaming.rs（+30/-10），并新增两个针对性测试，影响范围限定在 benchmark 工具链，风险低。

功能与动机

PR body 描述了明确的缺陷：

The Rust benchmark SSE handler decoded every incoming network chunk independently with String::from_utf8_lossy. When TCP or HTTP chunking split a multibyte UTF-8 character, each incomplete byte sequence was replaced with U+FFFD before the following chunk arrived. For example, splitting 中（E4 B8 AD）after E4 corrupted the streamed text.

问题不仅影响终端显示，还影响 benchmark 数据的可信度——损坏的响应被保存或被复用为多轮会话历史时会污染后续测试。作者同时明确要求保留对真正非法 UTF-8 的 lossy 行为，避免修复引入语义回退。

实现拆解

1. 缓冲结构改造：StreamedResponseHandler.buffer 由 String 改为 Vec<u8>，new() 对应改为 Vec::with_capacity(4096)。本质上把“增量解码”推迟为“整条消息解码”。
2. add_chunk 先累积后切分：入口处不再对 chunk_bytes 做 from_utf8_lossy，而是 self.buffer.extend_from_slice(chunk_bytes)；切分 SSE 消息时用 `windows(2).position(|w| w == b"

")在字节层定位双换行，切出的消息段才解码为字符串；drain(..pos + 2)` 消费逻辑不变。

1. 投机 JSON 解析用临时视图：无结尾分隔符的 speculative 路径（对应 Python StreamedResponseHandler 的 json.loads()，关乎 TTFT/ITL 统计准确性）通过 String::from_utf8_lossy(&self.buffer) 生成临时 Cow 视图完成查找与解析，self.buffer 中的原始字节不受影响，半截多字节字符可留待下一 chunk 补齐。
2. 测试配套：同文件 mod tests 新增 test_split_multibyte_utf8（逐字节投喂含 中 与 😀 的消息，验证只产出一条完整消息）与 test_invalid_utf8_remains_lossy（\xFF 仍输出 U+FFFD，锁定兼容行为）。没有单独的测试文件变更。

rust/src/bench/src/backends/streaming.rs

唯一变更文件，承载全部核心逻辑：StreamedResponseHandler 的 buffer 从 String 改为 Vec，add_chunk 不再按 chunk 独立解码，改为累积原始字节并在完整 SSE 消息边界解码；投机 JSON 解析改用临时 lossy 视图。同文件内新增两个针对性测试。

关键源码片段

rust/src/bench/src/backends/streaming.rs

唯一变更文件，承载全部核心逻辑：StreamedResponseHandler 的 buffer 从 String 改为 Vec，add_chunk 不再按 chunk 独立解码，改为累积原始字节并在完整 SSE 消息边界解码；投机 JSON 解析改用临时 lossy 视图。同文件内新增两个针对性测试。

// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright contributors to the vLLM project
​
/// SSE 流式响应处理器。
///
/// 以原始字节累积网络 chunk，只在完整 SSE 消息边界才做 UTF-8 解码，
/// 避免 TCP 或 HTTP 分片把多字节字符切成两半时，前一半立刻被
/// `String::from_utf8_lossy` 替换为 U+FFFD，导致流式文本乱码。
pub struct StreamedResponseHandler {
    // buffer 存原始字节而不是 String：跨 chunk 的多字节字符保持完整
    buffer: Vec<u8>,
    /// 可复用消息缓冲，避免每次 `add_chunk` 都新分配 Vec
    messages: Vec<String>,
}
​
impl StreamedResponseHandler {
    pub fn new() -> Self {
        Self {
            buffer: Vec::with_capacity(4096),
            messages: Vec::with_capacity(4),
        }
    }
​
    /// 加入一个字节 chunk，返回完整 SSE 消息。
    ///
    /// 返回值借用 `self`，只在下次 `add_chunk` 前有效。
    pub fn add_chunk(&mut self, chunk_bytes: &[u8]) -> &[String] {
        self.messages.clear();
​
        // 1. 先累积原始字节，不做任何解码，保证不完整多字节字符跨 chunk 保留
        self.buffer.extend_from_slice(chunk_bytes);
​
        // 2. 按 SSE 消息分隔符（双换行）切分，切出的完整消息才解码；
        // 这里在字节层面用 windows(2) 定位 b"
​
"，等价于原 String 里 find
        while let Some(pos) = self.buffer.windows(2).position(|window| window == b"
​
") {
            let message = String::from_utf8_lossy(&self.buffer[..pos]).trim().to_string();
            // 移除已消费字节，剩余数据前移
            self.buffer.drain(..pos + 2);
            if !message.is_empty() {
                self.messages.push(message);
            }
        }
​
        // 3. 尾部没有 
​
 的“投机 JSON 解析”路径，对齐 Python 端
        // StreamedResponseHandler 里的 speculative json.loads()，保证
        // TTFT/ITL 指标准确性：data 消息与其结束分隔符分属两个 TCP 段时，
        // 可以在第一个段到达时上报，而不是等第二个段。
        //
        // 这里用 `String::from_utf8_lossy` 构建临时 Cow 视图用于查找和解析；
        // 原始字节仍留在 `self.buffer`，不完整的多字节字符可等下一个 chunk 补齐。
        let buffer = String::from_utf8_lossy(&self.buffer);
        let data_start = if buffer.starts_with("data: ") {
            Some(0)
        } else {
            // 兼容多字段 SSE 事件（如 Dynamo 前端发的 "event: ...\ndata: ..."）
            buffer.find("\ndata: ").map(|p| p + 1)
        };
        if let Some(offset) = data_start {
            let content = buffer[offset + 6..].trim();
            if content == "[DONE]"
                || (!content.is_empty()
                    && serde_json::from_str::<&serde_json::value::RawValue>(content).is_ok())
            {
                self.messages.push(buffer.trim().to_string());
                // 已消费整条消息，清空累积的原始字节
                self.buffer.clear();
            }
        }
​
        &self.messages
    }
}

#[cfg(test)]
mod tests {
    use super::*;
​
    #[test]
    fn test_split_multibyte_utf8() {
        // 逐字节投喂含 CJK 与 emoji 的 SSE 消息，端到端结果必须与一次性投喂
        // 完全一致，证明跨 chunk 的 UTF-8 序列没有被破坏。
        // 中 是 3 字节（E4 B8 AD），😀 是 4 字节，chunks(1) 会任意切割。
        let mut handler = StreamedResponseHandler::new();
        let message = "data: {\"test\":\"中😀\"}
​
";
        let mut messages = Vec::new();
​
        for chunk in message.as_bytes().chunks(1) {
            messages.extend_from_slice(handler.add_chunk(chunk));
        }
​
        assert_eq!(messages, &[message.trim().to_string()]);
    }
​
    #[test]
    fn test_invalid_utf8_remains_lossy() {
        // 真正非法的 UTF-8 字节仍按原行为替换为 U+FFFD，不因本修复改变
        let mut handler = StreamedResponseHandler::new();
        let msgs = handler.add_chunk(b"data: {\"test\":\"\xFF\"}
​
");
        assert_eq!(msgs, &["data: {\"test\":\"�\"}".to_string()]);
    }
}

评论区精华

• esmeetu 直接给出 “LGTM!” 并批准，无进一步技术讨论。
• claude[bot] 仅说明：“This pull request is from a fork — automated review is disabled.”，提示维护者可用 @claude review 触发一次性 review。

本 PR 的技术论证主要由 PR body 的质量自述与同文件测试用例承担，review 过程未暴露设计分歧。

风险与影响

• 影响面：仅限 vllm-bench 的 SSE 流式后端，不触及推理运行时、调度器、KV cache 等核心路径，回归风险低。
• 行为一致性：投机 JSON 解析改为基于临时 lossy 视图，若多字节字符与 `

分隔符交错，windows(2)` 仍可能切在多字节字符中间，导致该消息解码出现 U+FFFD；但 SSE 帧内 JSON 字符串不允许裸换行，实际触发概率极低，且该边界未被测试覆盖，属于上下文不足时可标注的不确定性。

• 兼容性：对真正非法 UTF-8 字节的 lossy 行为通过 test_invalid_utf8_remains_lossy 锁定，不会因修复而改变。
• 性能：Vec<u8> 累积与原有 String 累积复杂度一致，windows(2) 逐字节扫描对短消息可忽略。

关联脉络

本 PR 与近期 Rust 侧的正确性修复形成同一条维护脉络：如 PR#50746 拒绝空 gRPC stop 字符串以避免解码器 panic，PR#48048 在 Rust 前端做 session id 全链路贯穿。这些变更共同表明 vLLM 的 Rust 组件（前端与基准工具）正在系统性地加固输入边界处理。更宏观地看，该修复为 Rust benchmark 工具在非 ASCII 与多轮对话场景下的可信度补齐了关键一环。