Vibe Rust · AI 原生 Rust 阅读课

简体中文

English

简体中文

English

Appearance

Provider 抽象与一次真实调用 ​

第 7 章你精读过 trait Provider 的签名;这一章把它接到真实的 HTTP 上。读完你应能回答:

  • 一次连接超时和一次 HTTP 400,分别变成 ProviderError 的哪个变体?哪个值得换下一个模型,依据各是哪几行代码?
  • API key 是哪一刻从环境变量里读出来的?为什么流式请求刻意不设总超时?
  • 一条流式响应被网络切在 Hel 与 lo 中间,jackdaw 靠哪个结构拼回完整的行?

主线是一个请求:body 为 {"model":"group/coder","messages":[…]}(第 11 章配置里的那个组),要发给 OpenAI 兼容端点;model 字段先被改写成上游 id——改写者是路由层(第 13 章展开),本章只管把它发出去、诚实地接回来。

契约:三个方法,一个供应商 ​

没有这层抽象,ask 与路由循环就得各自拼 HTTP;第 13 章测故障转移时,还需要一个不联网的假供应商替身。特征 Provider 把"一个上游"固定成契约。先看它搬运的东西:

rust
/// A chat request body, passed through untouched except for `model`.
#[derive(Debug, Clone)]
pub struct ChatRequest(pub Value);

impl ChatRequest {
    // ...
    /// A copy of the body with `model` set to the upstream model id.
    ///
    /// One deliberate `clone` per candidate: the original request must stay
    /// intact so the failover loop can rewrite it for the next model.
    pub fn with_model(&self, model_id: &str) -> ChatRequest {
        let mut body = self.0.clone();
        // Insert through the object map (not `body["model"] = ...`, which
        // panics on non-objects); callers guarantee an object body.
        if let Value::Object(map) = &mut body {
            map.insert("model".to_string(), Value::String(model_id.to_string()));
        }
        ChatRequest(body)
    }
    // ...
}

读法:从 pub struct ChatRequest(pub Value) 看起——它是对 serde_json::Value(第 6 章)的 newtype,.0 就是原始 JSON;不给 messages/tools 建强类型是刻意的,模块注释说网关不理解 tools、temperature 也能忠实转发,唯一改写的字段是 model。with_model(&self, …) 只读借用(第 5 章),先 .0.clone() 复制整份 body——这是故意的一次克隆,原始请求要留给故障转移循环改写下一个候选;注释还解释了为何用对象表 insert 而非下标赋值(后者对非对象会 panic)。返回的新 ChatRequest 才是真正上路的请求。契约本身:

rust
/// Byte chunks of a streaming (SSE) response body, in arrival order.
pub type ByteStream = Pin<Box<dyn Stream<Item = Result<Bytes, ProviderError>> + Send>>;

/// One upstream LLM API. [`crate::openai`] implements it for any
/// OpenAI-compatible endpoint; unit tests implement it with in-memory fakes.
#[async_trait::async_trait]
pub trait Provider: Send + Sync {
    /// Provider name, as used in `[providers.<name>]` and model references.
    fn name(&self) -> &str;

    /// Send the request and wait for the full JSON response.
    async fn complete(&self, request: ChatRequest) -> Result<ChatResponse, ProviderError>;

    /// Send the request and return the raw SSE byte stream. A successful
    /// return means the upstream already *accepted* the request (the HTTP
    /// status was checked); failures after the first byte are delivered
    /// through the stream itself — at that point failover is impossible.
    async fn complete_stream(&self, request: ChatRequest) -> Result<ByteStream, ProviderError>;
}

读法:先认 ByteStream——Pin<Box<dyn Stream<…>>> 读作"装着任意字节流的指针"。三个方法各回答一个问题:name() 用 [providers.X] 的表名给日志对账;complete 等完整 JSON;complete_stream 返回一条裸字节流。#[async_trait] 与 Send + Sync 是第 7 章的老朋友;文档注释里那句"首字节之后转移已不可能",第四节回来讲。

TIP

Pin 是给编译器看的自引用安全标记。第一次读认出"一个装着流的指针"就够了,不必展开。

一次真实的 POST:reqwest 与超时 ​

OpenAiProvider 四个字段:name;base_url(构造时剪掉结尾斜杠,endpoint() 于是只是一次 format!("{}/chat/completions", …));api_key_env——存环境变量名而非值,第 11 章的规矩;一个构造一次、反复使用的 http: Client,构造时把 CONNECT_TIMEOUT 的 10 秒装了进去。非流式与流式共用私有方法 send:

rust
/// How long to wait for a TCP connection to the upstream.
const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
/// Total budget for one non-streaming round trip. Streams deliberately have
/// no overall timeout — a long answer is not a stuck one.
const REQUEST_TIMEOUT: Duration = Duration::from_secs(120);
// ...
/// Fire the POST. Network-level failures (DNS, connect, timeout) are
/// retryable by definition: nothing was learned about the request.
async fn send(
    &self,
    request: &ChatRequest,
    streaming: bool,
) -> Result<reqwest::Response, ProviderError> {
    let mut builder = self
        .http
        .post(self.endpoint())
        .bearer_auth(self.api_key()?)
        .json(&request.0);
    if !streaming {
        builder = builder.timeout(REQUEST_TIMEOUT);
    }
    builder.send().await.map_err(|error| {
        ProviderError::Retryable(format!("request to {} failed: {error}", self.endpoint()))
    })
}

读法:从 self.http.post(self.endpoint()) 看数据流——URL 拼好,bearer_auth(self.api_key()?) 把 key 放进 Authorization: Bearer 头,? 在这一步就把"环境变量没设"(api_key() 返回的 Retryable 错误)提前送出去,.json(&request.0) 把主线那个 body 序列化进请求体。key 刻意在调用时才读而不是启动时——注释说 key 会轮换,缺 key 是单个 provider 的问题,不该让整个进程起不来。超时分支只管非流式:120 秒是完整往返的预算,流式不加,注释原话 "a long answer is not a stuck one"(回答长不等于卡住);最后 map_err 把 DNS、连接、超时一律归 Retryable——此刻还没人检验过请求本身,换条路也许就通。

complete 的骨架随之可读:send(&request, false) → 状态不成功交给 classify(下一节)→ 成功则 response.json() 装进 ChatResponse,连"上游回了非法 JSON"也归 Retryable。complete_stream 完全对称,只是 streaming = true、返回逐块字节流。

Retryable 还是 Fatal:错误的二分 ​

像寄快递:地址写错,换哪家公司都送不到(Fatal);公司爆仓,换一家也许就到(Retryable)。失配在于:jackdaw 对 Fatal 不"改地址重寄",只把上游判决原文透传。这个二分是故障转移的唯一依据,所以边界写在类型上:

rust
/// How an upstream failure is classified. The router's failover decision
/// rests entirely on this split, so the boundary is drawn carefully:
///
/// * **Retryable** — network errors, timeouts, HTTP 429, HTTP 5xx. The
///   request may be perfectly fine; a different model might succeed.
/// * **Fatal** — every other 4xx. The request itself is the problem
///   (unknown model, malformed call, bad key); switching models cannot
///   help, so the gateway passes the upstream verdict through unchanged.
#[derive(Debug, Clone, thiserror::Error)]
pub enum ProviderError {
    #[error("retryable: {0}")]
    Retryable(String),
    #[error("upstream rejected the request with HTTP {status}")]
    Fatal { status: u16, body: String },
}

读法:两个变体的形状就是语义——Retryable(String) 只带一句话,反正要换人,够日志就行;Fatal { status, body } 把上游状态码与响应体原样留住,因为要原封不动还给调用方。thiserror 的 #[error(…)] 自动生成 Display(第 11 章讲过它与 anyhow 的分工);文档注释把边界说死:网络错误、超时、429、5xx 归 Retryable,其余 4xx 说明请求本身有问题,换模型无益。做判断的是 classify:

rust
/// Read the error body once, then classify 429/5xx as retryable and
/// every other non-2xx as fatal.
async fn classify(&self, response: reqwest::Response) -> ProviderError {
    let status = response.status().as_u16();
    let body = response.text().await.unwrap_or_default();
    if status == 429 || (500..600).contains(&status) {
        // Keep a short excerpt: enough to act on, not enough to drown logs.
        let excerpt: String = body.chars().take(200).collect();
        ProviderError::Retryable(format!("HTTP {status}: {excerpt}"))
    } else {
        ProviderError::Fatal { status, body }
    }
}

读法:从 let status = …as_u16() 看起,response.text() 先把错误体读出来——分类同时要用两者,且读完后 response 就消费掉了。if 这行是本章题眼:429 与 5xx 归 Retryable 并截前 200 字符防日志被淹,其余落 Fatal。回收开头第一问:连接超时在 send 的 map_err 就成了 Retryable;HTTP 400 走到这里的 else,成为 Fatal。

SSE 流:攒够一行才算一条 ​

流式的线上格式(源码注释原文):

text
data: {"choices":[{"delta":{"content":"Hel"}}]}
data: {"choices":[{"delta":{"content":"lo"}}]}
data: [DONE]

data: 行是载荷,data: [DONE] 是收尾哨兵;传输层不做解析——complete_stream 的注释说字节原样透传,"SSE 解析属于需要文本的人(ask),不属于传输"。解析侧:

rust
/// Parse a single SSE line. Blank lines, comments and non-`data` lines
/// carry nothing jackdaw needs, so they map to `None`.
pub fn parse_sse_line(line: &str) -> Option<SseData> {
    // Accept both "data: x" and "data:x" spellings.
    let payload = line.strip_prefix("data:")?.trim_start();
    if payload == "[DONE]" {
        return Some(SseData::Done);
    }
    serde_json::from_str(payload).ok().map(SseData::Payload)
}

/// The delta text inside one stream chunk: `choices[0].delta.content`.
pub fn delta_content(payload: &Value) -> Option<&str> {
    payload
        .pointer("/choices/0/delta/content")
        .and_then(Value::as_str)
}

读法:strip_prefix("data:")? 起步——不是 data 行就地返回 None(? 作用在 Option 上,第 6 章),注释说同时容忍 data: x 与 data:x 两种拼法。[DONE] 映射 SseData::Done,其余交给 serde_json,失败则 .ok() 成 None,坏行直接丢;delta_content 用 JSON Pointer /choices/0/delta/content 直取增量文本,取不到(如首块只带 role)就 None。网络给的块不按行对齐,可能切在一个 UTF-8 字符中间——第三问的答案在这里:

rust
#[derive(Default)]
pub struct SseAggregator {
    /// Complete-but-unparsed bytes of the current line.
    pending: Vec<u8>,
    /// Assistant text assembled so far.
    text: String,
    /// How much of `text` has already been emitted by [`Self::take_recent`].
    emitted: usize,
}

impl SseAggregator {
    /// Feed one raw chunk from the stream.
    pub fn push(&mut self, chunk: &[u8]) {
        self.pending.extend_from_slice(chunk);
        while let Some(newline) = self.pending.iter().position(|byte| *byte == b'\n') {
            let line_bytes: Vec<u8> = self.pending.drain(..=newline).collect();
            let line = String::from_utf8_lossy(&line_bytes);
            self.consume(line.trim_end_matches(['\r', '\n']));
        }
    }
    // ...
    /// Everything assembled since the previous call — this is what lets
    /// `ask` print deltas live while `--no-stream` waits for the whole text.
    pub fn take_recent(&mut self) -> String {
        let start = self.emitted;
        self.emitted = self.text.len();
        self.text[start..].to_string()
    }
    // ...
}

读法:push 先把新块追加进 pending,while let 反复找换行——drain(..=newline) 连行带换行符取走、转成字符串、剪掉 \r\n 尾巴,交给 consume 解析并把增量文本拼进 text;换行字节不会出现在 UTF-8 多字节字符内部,所以攒到换行顺带保证了行不会断在字符中间。take_recent 用 emitted 记住已吐出的位置、每次只返回新增——ask 靠它边收边打印,--no-stream 则等 text() 的全文。

INFO

这里简化了:完整 SSE 规范还有 event:、retry: 等行,但这些端点不发;源码注释写明 jackdaw 只读 data: 行——"that is all these endpoints send"。

WARNING

流到一半断了,错误以 Retryable("stream interrupted…") 的形状从流里送出,但这时不换下一个模型:部分内容可能已经转发出去,重发会重复。这就是 trait 注释那句"首字节之后转移不可能";网关侧的完整语义第 14 章展开。

小结 ​

超时在 send 的 map_err 变 Retryable,400 在 classify 的 else 变 Fatal;key 在调用时读,流式不设总超时;SseAggregator::push 攒到换行才算一行。下一章把候选列表和这个二分接成 group/<id> 的故障转移循环。

自检 ​

  1. 给 jackdaw 加一个讲 Anthropic 协议的 provider,哪些文件要动?ProviderError 的分类要跟着改吗?
  2. parse_sse_line 依次收到 "data: [DONE]"、"event: ping"、"data: not-json",各返回什么?
  3. 环境变量没设导致的失败,为什么归为 Retryable 而不是 Fatal?
参考答案
  1. 新建模块实现 impl Provider for …;provider.rs 与 openai.rs 一行不动。分类依据是"问题出在请求还是上游",与协议无关,不需要改。真正要动的是装配:build_providers 目前为每张 [providers.X] 表都造 OpenAiProvider,区分协议需要在配置里加类型标记——第 16 章扩展练习"Anthropic 协议翻译"正从此处入口。
  2. Some(SseData::Done)、None、None:第二行没有 data: 前缀;第三行前缀对但不是合法 JSON,.ok() 把失败折叠成 None。openai.rs 的测试 parses_the_done_sentinel 与 ignores_non_data_lines 断言的就是这几条。
  3. 缺的是"这一家"的 key:组里下一个成员可能读的是另一个已设置的环境变量,换人也许就通过——Retryable 的含义正是"换个模型或许能成";错误消息同时写明配置存的是变量名而非值,读日志的人知道该去 export 什么。