模型客户端 · Responses API 的流式调用

模型客户端 · Responses API 的流式调用

上一篇里，try_run_sampling_request 只调用了一句 client_session.stream(...)，拿回一个事件流。这一篇看这句话背后的东西。涉及四个 crate：codex-core 的 client.rs 负责拼请求、选传输；codex-api 实现 Responses 的 HTTP 与 WebSocket 端点、SSE 解析、错误与限额头的翻译；codex-client 提供传输抽象与请求级重试；codex-http-client 是底层的 reqwest 封装（代理、自定义 CA 等）。

用户看到的样子

对用户来说，模型客户端几乎是透明的。能碰到的配置是提供方的 request_max_retries（默认 4）、stream_max_retries（默认 5）、stream_idle_timeout_ms（默认 300000）、websocket_connect_timeout_ms（默认 15000）与 supports_websockets；wire_api 只剩 responses，写 chat 会在加载配置时报错（详见模型与推理、配置项速查）。能看到的现象是 WebSocket 连续失败时的一条 “Falling back from WebSockets to HTTPS transport.” 警告（此后整个会话都走 HTTPS），以及界面上的额度用量，它来自每次响应头里的限额信息。

两层对象：ModelClient 与 ModelClientSession

ModelClient（codex-rs/core/src/client.rs:265）在 Session::new 里创建，放进 SessionServices，整个会话共用。它持有提供方、认证、会话来源、若干功能开关，以及两份跨轮状态：一个 disable_websockets 原子布尔（WebSocket 一旦回落到 HTTP，整个会话都不再尝试），一个缓存起来的 WebSocket 会话（让下一轮能复用上一轮的连接）。每轮开始时，run_turn 用 new_session() 从它派生一个 ModelClientSession：

pub struct ModelClientSession {
    client: ModelClient,
    websocket_session: WebsocketSession,
    /// Turn state for sticky routing.
    ///
    /// This is an `OnceLock` that stores the turn state value received from the server
    /// on turn start via the `x-codex-turn-state` response header. Once set, this value
    /// should be sent back to the server in the `x-codex-turn-state` request header for
    /// all subsequent requests within the same turn to maintain sticky routing.
    ///
    /// This is a contract between the client and server: we receive it at turn start,
    /// keep sending it unchanged between turn requests (e.g., for retries, incremental
    /// appends, or continuation requests), and must not send it between different turns.
    /// An auth ownership change clears it so the new owner gets fresh routing state.
    turn_state: Arc<OnceLock<String>>,
}

（codex-rs/core/src/client.rs:290）

类型上方的注释要求每轮新建一个 ModelClientSession，跨轮复用会把上一轮的路由令牌带进下一轮。字段注释把“轮”与“会话”的边界讲得很清楚：x-codex-turn-state 是服务端在一轮第一个响应里下发的粘性路由令牌，同一轮内的重试、增量追加都要原样带回，换一轮就必须作废。ModelClientSession 被 drop 时，会把 WebSocket 会话还给 ModelClient 缓存；下一轮新建会话时若发现账号换了主人，就丢弃旧连接。能用 WebSocket 时，会话启动后还有一次预热：后台用空输入发一个 generate=false 的请求，把连接和工具清单都准备好，第一轮直接领用这个预热过的会话。

请求怎么拼

run_turn 这一侧，build_prompt（codex-rs/core/src/session/turn.rs:1552）把一次 step 需要的东西装进 Prompt：历史生成的输入、StepContext 里的模型可见工具、基础指令、可选的输出 schema。ModelClient::build_responses_request 再把它翻成线上格式：

        let request = ResponsesApiRequest {
            model: model_info.slug.clone(),
            instructions,
            input,
            tools,
            tool_choice: "auto".to_string(),
            parallel_tool_calls: prompt.parallel_tool_calls && !model_info.use_responses_lite,
            reasoning: Some(reasoning),
            store: false,
            stream: true,
            stream_options,
            include,
            service_tier,
            prompt_cache_key,
            text,
            client_metadata: Some(client_metadata),
            access_programs: None,
        };

（codex-rs/core/src/client.rs:999）

几个字段值得停一下：

• store: false：请求不在服务端留存会话状态，走 HTTP 时每次都要带上完整历史，这正是下一篇“只追加”原则的出发点。include 固定为 reasoning.encrypted_content，推理内容以密文形式回到客户端，随历史一起再发回去。
• Responses Lite：模型元数据里 use_responses_lite 为真时，instructions 与 tools 字段留空，工具清单改成一条 AdditionalTools 条目、基础指令改成一条 developer 消息，插到输入最前面；两者的 ID 用线程 ID 做命名空间的 UUIDv5 由内容算出，注释说这样“重试与恢复的会话能保持它们的身份”。此时 parallel_tool_calls 为假，推理的 context 设为 all_turns。随附模型目录里，除 gpt-5.5 外的模型都开启了它（codex-rs/models-manager/models.json）。
• service_tier：Fast 档位在线上是 priority；default 表示不带档位；Amazon Bedrock 一律不带（见模型目录与提供方）。
• prompt_cache_key：默认取会话 ID，也就是根线程的 ID，同一线程的所有请求共用一个缓存键；非 OpenAI 提供方收到的请求还会被剥掉内部元数据。

HTTP 头与 client_metadata 里还带着一串轮级元数据：x-codex-turn-metadata、x-codex-window-id、会话与线程 ID、子代理标记等，由 responses_metadata.rs 统一生成，并用 RESERVED_METADATA_KEYS 挡住客户端自带元数据覆盖内核字段。在 ChatGPT 登录、OpenAI 提供方下，HTTP 请求体默认用 zstd 压缩（功能开关 enable_request_compression，默认开启）。

两种传输

stream()（codex-rs/core/src/client.rs:2215）的选择很简单：提供方声明 supports_websockets（内置的 OpenAI 提供方为真，Bedrock、本地模型为假）且本会话没回落过，就走 WebSocket；握手返回 426 时当场改走 HTTP。

WebSocket 的价值在于增量。连接在一轮内复用，还会跨轮缓存；每次请求前，get_incremental_items 判断这次请求是不是上一次的延续：除输入外的字段（模型、指令、工具、推理设置、缓存键等）完全相同，且新输入以“上次输入加上次输出”为前缀。满足就只发多出来的那几项，并带上上次的响应 ID：

            let mut ws_payload = ResponseCreateWsRequest {
                previous_response_id,
                input: incremental_items.as_deref().unwrap_or(&request.input),
                generate: if warmup { Some(false) } else { None },
                client_metadata: response_create_client_metadata(
                    Some(client_metadata),
                    request_trace.as_ref(),
                ),
                ..ResponseCreateWsRequest::from(&request)
            };

（codex-rs/core/src/client.rs:2013）

不满足（比如压缩改写了历史、换了工具清单）就发完整请求。一轮里模型调工具、结果送回、再采样时，模型自己的输出已算在“上次输出”里，增量通常只有工具输出等几项，请求因此很小。服务端返回 previous_response_not_found 时，codex-api 把它当成可重试错误（codex-rs/codex-api/src/endpoint/responses_websocket.rs:633），出错的连接被关掉，重试时重建连接、发完整请求。

HTTP 路径由 codex-api 的 ResponsesClient::stream_request 发出 POST /responses。建立连接这一步由 codex-client 的 run_with_retry 按提供方的 request_max_retries 重试：起始间隔 200 毫秒倍增，重试 5xx 与传输错误，不重试 429（codex-rs/model-provider-info/src/lib.rs:441）。响应一旦开始，就交给 spawn_response_stream。

流事件与错误翻译

spawn_response_stream（codex-rs/codex-api/src/sse/responses.rs:38）先读响应头：限额快照、X-Models-Etag（模型目录变了就刷新）、OpenAI-Model（服务端实际用的模型与请求不同时，界面给出警告）、X-Reasoning-Included、x-codex-turn-state，然后在一个任务里逐条解析 SSE。每次等下一个事件都受空闲超时约束，超时报 “idle timeout waiting for SSE”；流在 response.completed 之前断开，报 “stream closed before response.completed”。这些都是 CodexErr::Stream，由上一篇的流级重试接手。

response.failed 里的错误码被翻译成语义化的错误。context_length_exceeded 变成上下文超窗，insufficient_quota 一类变成配额耗尽，cyber_policy、bio_policy 等策略拒绝各有专门的类型，server_is_overloaded 映射为过载，这些都不重试。剩下的走这一段：

                        let retry_after =
                            try_parse_retry_delay(&error).and_then(RetryAfter::from_delay);
                        let message = error.message.unwrap_or_default();
                        response_error = match error.code.as_deref() {
                            Some("rate_limit_exceeded" | "slow_down") => {
                                ApiError::RateLimitExceeded {
                                    message,
                                    retry_after,
                                }
                            }
                            _ => ApiError::Retryable {
                                message,
                                retry_after,
                            },
                        };

（codex-rs/codex-api/src/sse/responses.rs:473）

try_parse_retry_delay 用正则从 “Please try again in 11.054s” 这样的消息里抠出等待时间，作为服务端建议交给重试逻辑。HTTP 状态码的翻译在 api_bridge.rs：429 且类型是 usage_limit_reached 时，会顺带从响应头解析出限额快照，用来告诉用户额度何时重置。

codex-core 的 map_response_events（codex-rs/core/src/client.rs:2370）再包一层：起一个任务把 codex-api 的事件转发进容量 1600 的通道，同时记下本次响应产出的全部输出项，供 WebSocket 下次判断增量；消费者先放手（比如用户中断）时，这个任务会记下“流被丢弃”并退出。

限额头的格式是 x-codex-primary-used-percent、x-codex-primary-window-minutes、x-codex-primary-reset-at 以及对应的 secondary 版本，另有 x-<限额名>-... 形式的其他限额族（codex-rs/codex-api/src/rate_limits.rs:28）。每一族解析成一个 RateLimitSnapshot，作为 ResponseEvent::RateLimits 流进 run_turn；SessionState::set_rate_limits 把新旧快照合并（新快照缺套餐、积分信息时沿用旧值），等本次请求的 token 用量到了，再随 TokenCount 事件一起发给前端。

认证失败也在客户端内部消化：HTTP 请求或 WebSocket 握手返回 401 时，handle_unauthorized（codex-rs/core/src/client.rs:2621）先尝试一次提供方自己的认证恢复（例如 Bedrock 刷新 AWS 凭据），再按认证管理器的恢复步骤逐步尝试（例如刷新 ChatGPT 令牌），每成功一步就重建请求再发一次，步骤用完才把错误交给上层。

和《从 LLM 到 Coding Agent》对照

一次 LLM API 调用讲的“无状态”在这里被 store: false 明确选择，WebSocket 的 previous_response_id 只是传输层的续接，历史的主人仍然是客户端。流式处理拆解的事件序列，对应 codex-api 里 response.output_item.added、response.output_text.delta、response.output_item.done、response.completed 的翻译。Prompt 缓存一篇用显式的缓存断点，Codex 的请求里没有断点，靠的是稳定的 prompt_cache_key 加上处处维护字节稳定：确定性的条目 ID、只追加的历史、推理强度改动在开发中的 reasoning_effort_override 功能下改为追加配置条目而不改请求参数。

上一篇：run_turn 主循环 · 一轮对话是怎么转起来的 · 下一篇：上下文与历史 · 只追加、不改写