大话 Pi

Appearance

🐣 小a的第一块砖

工坊开张,造 Agent 的第一件事——得能跟大模型说上话。看起来不就是个 HTTP 请求?但有几个坑:流式怎么收?token 怎么算?出错怎么处理?把这几个搞定,你的『翻译局』地基就稳了。

这一章,我们写最小的 LLM 客户端——一个 streamText() 函数。

25.1 目标:一个 streamText 函数 ​

我们要实现的:

typescript
// 用法
for await (const event of streamText({
  model: "claude-sonnet-4-5",
  messages: [{ role: "user", content: "你好" }],
})) {
  if (event.type === "text") process.stdout.write(event.text);
}
  • 流式输出(一个字一个字收)
  • 返回事件(文字增量、工具调用、完成)
  • 处理 token 用量

25.2 最简实现:调 Anthropic ​

先写最简版,调 Anthropic(裸 fetch,不用 SDK):

typescript
// llm.ts
const ANTHROPIC_API_URL = "https://api.anthropic.com/v1/messages";

export interface StreamTextOptions {
  model: string;
  messages: Array<{ role: "user" | "assistant"; content: string }>;
  system?: string;
  maxTokens?: number;
  apiKey?: string;  // 不传则读环境变量
}

export async function* streamText(options: StreamTextOptions) {
  const apiKey = options.apiKey ?? process.env.ANTHROPIC_API_KEY;
  if (!apiKey) throw new Error("缺少 ANTHROPIC_API_KEY");

  const response = await fetch(ANTHROPIC_API_URL, {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      "x-api-key": apiKey,                    // Anthropic 用这个头
      "anthropic-version": "2023-06-01",
    },
    body: JSON.stringify({
      model: options.model,
      messages: options.messages,
      system: options.system,
      max_tokens: options.maxTokens ?? 4096,
      stream: true,                            // 关键:开流式
    }),
  });

  if (!response.ok) {
    throw new Error(`LLM 请求失败: ${response.status} ${await response.text()}`);
  }

  // 解析 SSE 流(第 3 章讲的流式格式)
  const reader = response.body!.getReader();
  const decoder = new TextDecoder();
  let buffer = "";

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    buffer += decoder.decode(value, { stream: true });

    // SSE 按 "\n\n" 分事件
    const events = buffer.split("\n\n");
    buffer = events.pop() ?? "";  // 最后一段可能不完整,留着

    for (const evt of events) {
      const data = evt.split("\n").find(l => l.startsWith("data: "))?.slice(6);
      if (!data) continue;
      const parsed = JSON.parse(data);

      // 把 Anthropic 的事件翻译成我们的统一事件
      if (parsed.type === "content_block_delta" && parsed.delta?.text) {
        yield { type: "text" as const, text: parsed.delta.text };
      } else if (parsed.type === "message_stop") {
        yield { type: "done" as const };
      }
      // ... 还能处理 usage 等
    }
  }
}

🧙 老z带小a读这段代码:

  1. fetch 裸调——你能看到每个 HTTP 请求长什么样(x-api-key 头、stream: true body)
  2. async generator——async function* + yield,天然适合流式(产出一个事件,调用方消费一个)
  3. SSE 解析——按 \n\n 分事件(第 3 章讲过 SSE 格式),buffer 处理半包
  4. 翻译——把 Anthropic 的原生事件(content_block_delta)翻译成我们的统一事件({ type: "text" })

25.3 关键点一:流式怎么收 ​

🧙 流式收发的核心:

  • 请求里加 stream: true,服务器就流式返回(SSE)
  • 响应 body 是个 ReadableStream,要边读边解析
  • 不能 await response.json()——那只返回完整结果,失去了流式

reader.read() 一次次读 chunk,累加到 buffer,按事件分隔符切。这是流式处理的标准套路。

25.4 关键点二:统一事件格式 ​

我们定义自己的"统一事件":

typescript
export type LLMEvent =
  | { type: "text"; text: string }           // 文字增量
  | { type: "tool_use"; id: string; name: string; input: any }  // 工具调用
  | { type: "usage"; input: number; output: number }            // token 用量
  | { type: "done" };                       // 完成

🧙 为什么翻译成统一格式?

呼应第 3 章——各家厂商流式格式不同(Anthropic 的 content_block_delta、OpenAI 的 choices[0].delta)。我们定义统一事件,把厂商格式翻译过来。

这样上层(第 26 章的循环)只认统一事件,不关心底层是哪家。这就是"翻译局"的雏形——虽然我们现在只支持 1 家,但接口已经留好了扩展空间。

25.5 关键点三:错误处理 ​

🧙 三个常见错误:

  1. 401 Unauthorized:API key 错了 → 检查环境变量
  2. 429 Too Many Requests:限流了(第 3 章)→ 退避重试
  3. 网络中断:流读到一半断了 → 处理(简化版:报错;完整版:断点续传)

简化版错误处理(够用):

typescript
if (!response.ok) {
  const body = await response.text();
  if (response.status === 429) {
    throw new Error("限流了,请稍后重试(附录 E.9)");
  }
  throw new Error(`LLM 错误 ${response.status}: ${body}`);
}

🐣 小a问:"429 不重试吗?"

🧙 "生产环境要重试(退避策略)。但学习阶段,先抛错让你看到。 真要做重试,加个循环+setTimeout 即可。重点是别静默吞掉错误——fail loud。"

25.6 对照 pi:我们省了什么 ​

🧙 我们 vs pi-ai:

我们的 llm.tspi-ai
支持厂商1 家(Anthropic)近 40 家
Provider 抽象无有(createProvider)
代码行数~50 行几千行
auth 责任链无(直接读环境变量)OAuth + 4 路(第 17 章)

我们省了 95% 的代码。 因为我们只要 1 家,不需要 Provider 抽象(第 24 章的 YAGNI)。

但核心思想一致:都是"翻译"——把厂商格式翻成统一格式。我们只是简化到只翻 1 家。

25.7 测试一下 ​

写完 llm.ts,测一下:

typescript
// test.ts
import { streamText } from "./llm.ts";

for await (const event of streamText({
  model: "claude-sonnet-4-5",
  messages: [{ role: "user", content: "用一句话介绍你自己" }],
})) {
  if (event.type === "text") process.stdout.write(event.text);
}

跑:tsx test.ts。如果看到文字一个个蹦出来——恭喜,你的 LLM 调用层通了!

本章小结 ​

🐣 小a的第二十五课

┌──────── LLM 调用层 ────────┐
│                            │
│  • streamText() 函数       │
│    async generator + yield │
│                            │
│  • 流式:fetch + stream:true│
│    reader.read() 边读边解析 │
│    SSE 按 \n\n 分事件       │
│                            │
│  • 统一事件格式:           │
│    text / tool_use /       │
│    usage / done            │
│    → 翻译厂商格式          │
│                            │
│  • 错误处理:401/429/中断   │
│    → fail loud,不吞错     │
│                            │
│  • 对照 pi:省了 95%        │
│    (只 1 家,不要抽象)    │
└────────────────────────────┘

关键认知:LLM 调用层的核心是"流式收 + 统一事件"。我们用 ~50 行搞定,pi 用几千行是因为它要支持 近 40 家。本质一样,复杂度不同。

课后实验 ​

  1. 跑通:照着 25.2 的代码,写出 llm.ts,跑通"用一句话介绍你自己"。
  2. 看请求:在代码里 console.log 一下发出的请求 body,看清"调 LLM 到底发了什么"。
  3. 换厂商:如果你想用 OpenAI 而非 Anthropic,改哪里?(提示:URL、headers、body 格式、事件解析都不同——这就是"方言")

下一章:能调 LLM 了,但它还只会说话。装上循环,让它会调工具——这是 Agent 的核心。 → 第 26 章 · Agent 循环