三种 API 后端的 L2 transform

文档摘要

三种 API 后端的 L2 transform 本节摘要:上一节我们看到 sampler 吐出统一的 SamplingEvent,但模型 API 的原始响应格式却因后端而异。把「后端私有的原始格式」转换成「统一事件」的工作,由 L2 transform 完成(源码在 )。Grok Build 支持三种后端:OpenAI 风格的 responses、chatcompletions,以及 Anthropic 风格的 messages。每种有自己的 transform,把自家的 chunk 流转换成统一的 SamplingEvent。本节会讲清这三种后端的格式差异、L2 transform 的工作方式,以及「collect」这一步如何把事件流聚合成最终响应。

三种 API 后端的 L2 transform

本节摘要:上一节我们看到 sampler 吐出统一的 SamplingEvent,但模型 API 的原始响应格式却因后端而异。把「后端私有的原始格式」转换成「统一事件」的工作,由 L2 transform 完成(源码在 xai-grok-sampler/src/stream/)。Grok Build 支持三种后端:OpenAI 风格的 responses、chat_completions,以及 Anthropic 风格的 messages。每种有自己的 transform,把自家的 chunk 流转换成统一的 SamplingEvent。本节会讲清这三种后端的格式差异、L2 transform 的工作方式,以及「collect」这一步如何把事件流聚合成最终响应。理解这一节,你就彻底明白 sampler 内部「适配」是如何发生的。

一、为什么有三种后端

在讲 transform 之前,先理解为什么 Grok Build 要支持三种后端。

历史与现实的原因

LLM 的 API 生态没有统一标准。最早 OpenAI 用 Chat Completions API,后来推出更新的 Responses API;Anthropic 用自己的 Messages API;各家厂商(包括 xAI)往往兼容其中一种或多种。一个想兼容多种模型的工具,必须支持多种后端协议。

Grok Build 的支持

Grok Build 默认用 responses 后端(从 default_models.json 看到 "api_backend": "responses")。但同时支持 chat_completions 与 messages,这让用户可以:

接入 OpenAI 兼容端点(很多模型提供商都兼容)
接入 Anthropic 端点
接入自托管的兼容服务

第 6 章会讲「自定义模型集成」,届时你会看到如何配置一个 OpenAI 或 Anthropic 兼容的端点。本节聚焦后端格式如何被 sampler 处理。

二、三种后端的格式差异

三种后端都是基于 HTTP 的流式响应(通常是 SSE,Server-Sent Events),但具体的 chunk 结构不同。下面用高度简化的方式说明差异(实际格式更复杂,以各后端官方文档为准):

responses 后端(OpenAI Responses API)

chunk 大致长这样(简化):


event: response.output_text.delta
data: {"delta": "Hello"}

event: response.function_call_arguments.delta
data: {"tool_index": 0, "delta": "{\"comm"}

event: response.completed
data: {"response": {... 完整响应 ...}}

特征:

每行以 event: 标明事件类型
data: 是 JSON
不同事件类型对应不同的字段(delta、tool_index、response 等)
完成时给一个完整的 response 对象

chat_completions 后端(OpenAI Chat Completions API)

chunk 大致长这样(简化):


data: {"choices": [{"delta": {"content": "Hello"}}]}

data: {"choices": [{"delta": {"tool_calls": [{"index": 0, "function": {"arguments": "{\"comm"}}]}}]}

data: [DONE]

特征:

没有 event 类型,只有 data
数据在 choices 数组的 delta 里
工具调用在 delta.tool_calls,带 index
流结束标志是 [DONE]

messages 后端(Anthropic Messages API)

chunk 大致长这样(简化):


event: content_block_delta
data: {"delta": {"text": "Hello"}}

event: content_block_delta
data: {"delta": {"type": "input_json_delta", "partial_json": "{\"comm"}}

event: message_stop
data: {}

特征:

又是 event + data 结构,但事件类型与 responses 不同
文本与工具调用的 delta 用不同的 type 区分
流结束是 message_stop

核心观察:三种后端都在表达「文本增量、工具调用增量、完成」这些相同概念,但用完全不同的字段名、结构、结束标志。这就是需要 transform 的原因。

三、L2 transform 的工作

L2 transform 是「把后端私有格式翻译成统一 SamplingEvent」的函数。每种后端一个:


stream/mod.rs:
    pub mod chat_completions;   // stream_chat_completions
    pub mod messages;            // stream_messages
    pub mod responses;           // stream_responses
    pub mod collect;             // collect_response(聚合成最终响应)

每个 transform 的核心任务是:


async fn stream_<backend>(raw_chunk_stream) -> Stream<SamplingEvent> {
    for raw_chunk in raw_chunk_stream {
        # ① 解析原始 chunk(按本后端的格式)
        let parsed = parse(raw_chunk)
        
        # ② 按解析结果,映射成零或多个 SamplingEvent
        let events = match parsed {
            文本增量 => vec![ChannelToken { channel: Text, text: ... }],
            工具调用增量 => vec![ToolCallDelta { tool_index: ..., arguments_delta: ... }],
            完成事件 => vec![Completed { response: ..., metrics: ... }],
            错误 => vec![Failed { error: ... }],
            其他(忽略/元信息) => vec![],
        }
        
        # ③ 把事件吐出去
        for ev in events {
            yield ev
        }
    }
}

关键点:

输入:后端私有的原始 chunk 流
输出:统一的 SamplingEvent 流
映射规则:每个 transform 内部有自己的映射逻辑,把本后端的字段名、结构翻译成统一事件

举个例子,三种后端都产生 ChannelToken,但来源不同:


responses 后端:
    event: response.output_text.delta → ChannelToken(Text)

chat_completions 后端:
    choices[0].delta.content → ChannelToken(Text)

messages 后端:
    content_block_delta 里 delta.text → ChannelToken(Text)

transform 内部就是这些「字段到事件」的映射规则。

四、每请求 task 如何使用 transform

回顾第 01 节的 request_task,它根据 config.api_backend 选 transform:


async fn request_task(request_id, request, config, ...):
    let response_stream = match config.api_backend {
        Responses => client.conversation_stream_responses(request),
        ChatCompletions => client.conversation_stream_chat_completions(request),
        Messages => client.conversation_stream_messages(request),
    }
    
    # response_stream 已经是「原始 chunk 流」
    # 接下来用对应的 transform 把它变成 SamplingEvent 流
    
    let event_stream = match config.api_backend {
        Responses => stream_responses(response_stream),
        ChatCompletions => stream_chat_completions(response_stream),
        Messages => stream_messages(response_stream),
    }
    
    # 现在 event_stream 吐的是统一 SamplingEvent
    for ev in event_stream {
        发出 Retrying 事件前,先检查 cancel_token
        event_tx.send(ev).await
    }

两层抽象:

第一层:conversation_stream_*(request)—— 发起 HTTP 请求,拿到原始 chunk 流
第二层:stream_*(chunk_stream)—— 把原始 chunk 流转换成统一事件流

这种「先拿原始,再 transform」的两层设计,让 HTTP 通信与格式解析分离——HTTP 层不必关心格式,格式层不必关心网络。

五、collect:从事件流到最终响应

事件流虽然统一,但它是「增量」的——一个个 token、一个个工具调用片段。最终的 ConversationResponse 需要「完整」的对象。这个「从增量到完整」的聚合工作,由 collect 模块完成:


async fn collect_response(event_stream) -> ConversationResponse {
    let mut text = String::new()
    let mut reasoning = String::new()
    let mut tool_calls = Vec::new()
    let mut usage = None
    
    for ev in event_stream {
        match ev {
            ChannelToken { channel: Text, text: t } => text.push_str(&t),
            ChannelToken { channel: Reasoning, text: t } => reasoning.push_str(&t),
            ToolCallDelta { tool_index, arguments_delta, ... } => {
                # 按 tool_index 累积参数
                tool_calls[tool_index].arguments.push_str(&arguments_delta)
            }
            Completed { response, metrics } => {
                # 用服务端给的完整响应(如果后端在完成时给了)
                return response
            }
            ...
        }
    }
    
    # 若后端没在 Completed 里给完整响应,自己拼一个
    return ConversationResponse {
        text, reasoning, tool_calls, usage, ...
    }
}

collect 的两种来源:

服务端给的完整响应:有些后端在流结束时会给一个完整的 response 对象(如 responses 的 response.completed),collect 直接用它
自己聚合:有些后端只给增量,collect 把累积的 text、tool_calls 拼成完整响应

这种「优先用服务端的,否则自己拼」的策略,既保证准确性(服务端的更权威),又保证健壮性(后端不给也能工作)。

六、transform 的几个设计要点

把三种 transform 看下来,有几个设计要点值得琢磨:

要点一:transform 是纯函数式的映射

理想情况下,transform 不持有状态——每个原始 chunk 独立映射成事件。但实际上,某些后端的工具调用增量需要按 tool_index 累积,transform 内部会有少量状态。即便如此,这种状态是局部的、与单次请求绑定的,不影响其他请求。

要点二:transform 不做重试

transform 只负责「格式转换」,不做「错误恢复」。遇到原始流里的错误,transform 把它翻译成 Failed 事件;重试决策在 request_task 里(根据错误 kind 与 RetryPolicy)。这种分离让 transform 简单聚焦。

要点三:transform 不响应取消

取消由 request_task 的 select! 处理(响应 cancel_token)。transform 本身不感知取消——它只是被 request_task 调用,task 取消时 transform 也随之结束。这让 transform 不必关心取消逻辑。

要点四:transform 与 collect 分离

transform 把 chunk 变事件,collect 把事件变最终响应。这两步分离,让它们各自聚焦——transform 关心「这一刻发生了什么」,collect 关心「整体意味着什么」。

七、统一带来的具体好处

把 L2 transform 的设计放回更大的语境,「统一」带来的具体好处:

好处一:新增后端只改 sampler

想支持一个新的后端(如 Google Gemini 的 API),只需在 sampler 里加一个 stream_gemini transform 与对应的 conversation_stream_gemini,shell 层一行代码都不用改。这种可扩展性是统一事件的核心价值。

好处二:后端格式变化被吸收

某个后端改了字段名(如 delta 改成 chunk),只需改对应 transform,上层无感。这种「变化局部化」让系统对后端演进具有韧性。

好处三:工具调用、推理过程等概念统一处理

不同后端对「工具调用」「推理过程」的表达不同,但都映射到 ToolCallDelta、ChannelToken(Reasoning)等统一事件。上层处理这些概念时,代码只有一份。

好处四:错误处理统一

不同后端的错误格式各异,但都翻译成 SamplingErrorKind 的同一套类型。这让 sampler 的重试决策与 shell 的翻译逻辑只写一份。

关键概念:L2 transform 是「适配器模式」的教科书式应用。它把「变化的(后端格式)」与「不变的(统一事件)」分离,让变化被局部吸收,让不变得以复用。这是应对「多个外部系统,各自演进」这类问题的经典解法。

八、与第 6 章的衔接

本节讲的三种后端,直接关联第 6 章的「自定义模型集成」。届时你会看到:

如何在配置里指定一个 OpenAI 兼容端点(用 chat_completions 后端)
如何指定一个 Anthropic 兼容端点(用 messages 后端)
如何带自己的 API key(Bring-your-own-key)

理解了本节的 transform,第 6 章的自定义模型集成就是「配置一个端点 + 选一个后端」的简单操作——因为 sampler 已经支持了这些后端的解析。

本节要点回顾

三种后端:responses(OpenAI 风格,Grok Build 默认)、chat_completions(OpenAI 兼容)、messages(Anthropic 风格)。
格式各异但概念相同:都在表达文本增量、工具调用增量、完成、错误,但字段名、结构、结束标志不同。
L2 transform 做翻译:每后端一个 transform,把私有 chunk 流映射成统一 SamplingEvent 流。
两层抽象:conversation_stream_(发 HTTP 拿原始流)→ stream_(transform 成事件流),HTTP 与格式分离。
collect 聚合:从增量事件流聚合成完整 ConversationResponse,优先用服务端给的完整响应,否则自己拼。
transform 是纯映射:不做重试、不响应取消、状态局部,聚焦格式转换。
统一的好处:新增后端只改 sampler、后端格式变化被吸收、概念统一处理、错误统一分类。
衔接第 6 章:理解 transform,自定义模型集成就是「配置端点 + 选后端」的简单操作。

下一节,我们看清默认模型从哪里来、运行中如何切换模型——default_models.json 与 UpdateConfig 命令、models_manager 广播机制。