绝不浪费一个token（15分钟阅读）

# 绝不浪费一个 token 来源：https://sunilpai.dev/posts/never-waste-a-token/ *\(这篇帖子本身是 LLM 垃圾食品，但味道还行\)* tl;dr - 在您的 AI Agent 和大模型供应商之间放置一个持久化的缓冲区。现在，供应商连接的生命周期比您的进程更长，因此，在流式传输过程中，一次部署不会让您为已经支付的 token 再次买单。而且，同一个让断开的浏览器重新跟上的缓冲区，也能让崩溃的轮次恢复。一个日志，两个读者。 --- 过去几周，我一直在思考一个问题：当一个 Agent 在执行一轮任务中途进程死亡时，会发生什么？这个问题深入得很快：可能已经触发的工具调用、子 Agent、等待人类响应的半成品流。我打算把这些内容单独写出来（持久化 Agent 循环，即将推出）。但其中一小部分足够独立，值得单独拎出来：**当您的进程在推理中途死亡时，您不仅会丢失进度，还会损失金钱。** ## 很容易忽略的问题 您的 Agent 向模型发起一个流式请求，模型开始生成。输出 token 一旦生成，您就需要付费。然后，您的进程被替换了。可能是部署、被驱逐、或者内存溢出。通常的安慰是“别担心，状态是持久的。”确实，您的对话历史保存下来了。但是**向供应商发起的正在进行的 HTTP 请求**却没有。它存在于刚刚死亡的进程内存中。所以，当您恢复时，您唯一的选择是**再次发起调用**。您不得不为那些输出 token 再付一次钱。 现在，考虑一个真正的 Agent。它在一个轮次中可能会发出多次工具调用：`用户消息 → 流式输出文本 → 工具调用 → 工具结果 → 流式输出更多文本 → 工具调用 → 工具结果 → 流式输出最终答案`。每一次中断都会丢弃该轮次中已经生成的所有输出 token。而且，这还取决于您实际想用的模型：`gpt-5.5` 的输出 token 价格为每百万个 30 美元，而 `gpt-5.5-mini` 仅为 2 美元，因此旗舰模型的一次重试成本大约是 mini 模型的 15 倍。模型越好，损失越惨重。部署频繁发生，进程驱逐也频繁发生，每一次落在活跃流上的事件，都意味着直接损失金钱。快乐路径掩盖了这个问题。只有当您在事故后开始统计 token，发现数字对不上时，您才会意识到它的存在。 ## 解决方法：将请求与进程解耦 崩溃浪费 token 的原因是，供应商连接存在于**崩溃的对象内部**。所以，把它移出来。在 Agent 和供应商之间放置一个缓冲区，并将其作为一个**独立的部署**：它自己的 Worker，它自己的 Durable Object。 DiagramDiagram 当一个请求进来时，缓冲区按顺序做三件事：重置状态以准备新的流；启动一个后台任务，将供应商的连接数据存入 SQLite；然后立即向调用方返回一个流，该流会追随这些行数据到达。 ```typescript async proxyAndBuffer(req: ProviderRequest): Promise { this.resetBuffer(); // status = "streaming", chunkCount = 0 const reader = (await fetch(req.url, req)).body!.getReader(); // 在后台消费供应商数据。故意不 await - 下面的响应会立即返回，而此任务继续运行。 this.keepAliveWhile(() => this.consumeProvider(reader)); // 给调用方一个流，它追随写入的行数据。 return new Response(this.tailFrom(0), { headers: { "X-Buffer-Status": "streaming" } }); } private async consumeProvider(reader: Reader) { for (let i = 0; ; i++) { const { done, value } = await reader.read(); if (done) break; this.sql`INSERT INTO buffer_chunks VALUES (${i}, ${decode(value)})`; this.notify(); // 唤醒任何追随者（更多见下文） } this.setStatus("completed"); } ``` 关键在于，`consumeProvider` **不**依附于 Agent。它在这里运行，在一个未被 Agent 部署影响的独立部署中。所以，当 Agent 在流中间被驱逐，其追随连接被取消时，消费循环会继续读取。您已支付的 token 会继续落入 SQLite，无论是否有人在监听。`keepAliveWhile` 负责在消费期间保持缓冲区存活。长时间生成会有静默期，而 Durable Object 可能因看起来空闲而被驱逐。`keepAliveWhile` 会在消费期间通过心跳维持 alarm，并在任务完成或抛出时立即移除它，从而让缓冲区在静默期存活，而不会之后泄露心跳。当 Agent 重启时，它会调用 `/resume?from=N` 并获取它遗漏的块。没有浪费的 token，没有重复的供应商调用。 ## 一个日志，两个读者 在构建这个功能时，我总觉得我以前解决过类似的问题。确实如此。这和**可恢复的流式传输**是同一个问题。你应该知道：用户正在接收响应，关上了笔记本电脑，从 Wi-Fi 切换到蜂窝网络，回来，然后流只是继续……实现方式是，在流式传输时将每个块持久化到一个持久的日志中，当客户端重新连接时，它会读取已存储的块，直到赶上实时光标。恢复正是**完全相同的日志**。缓冲区将每个块以索引为键存储在 SQLite 中： ```sql CREATE TABLE buffer_chunks ( chunk_index INTEGER PRIMARY KEY, data TEXT NOT NULL ) ``` 读取这些块是一个函数。实时代理调用 `tailFrom(0)`；恢复中的 Agent 调用 `tailFrom(N)`，其中 N 是它看到的最后一个块索引。唯一的区别是游标： ```typescript tailFrom(cursor: number): ReadableStream { return new ReadableStream({ pull: async (c) => { const rows = this.rowsFrom(cursor); // 自 `cursor` 以来存储的所有内容 if (rows.length) { c.enqueue(rows); cursor += rows.length; return; } if (this.isDone()) return c.close(); // completed / interrupted / error await this.signal.promise; // 否则等待下一次 notify() } }); } ``` 所以 `/resume?from=N` 就是 `tailFrom(N)`。它只会在两种情况下被调用： - **生产者还活着。** 更多块即将到来。它追随：提供已存储的内容，等待下一个，重复，直到流完成。这是浏览器重新连接的情况。 - **生产者已消失。** 它随旧进程一起死亡，因此流变成**孤儿**，不再有块到来。它不是向实时光标追随，而是重建所有已存储的内容，完成它，然后从那里继续该轮次。这是崩溃恢复的情况。同一个持久化日志。**唯一**的区别是是否仍有活的生产者连接着。 可恢复的流式传输回答了“客户端重新连接：让它赶上”。恢复回答了“生产者死亡：完成它正在写的内容”。为了重新连接而持久化块，几乎免费地为您提供了大部分崩溃恢复功能。缓冲区在其状态机中明确区分了这两种情况。重启时，如果它发现自己的状态仍标记为 `streaming`，它就**知道**前一个实例在途中死亡，将自己翻转为 `interrupted`，调用方就知道他们得到的是不完整的数据： ``` idle → streaming → completed → (ack / TTL) → idle │ [DO evicted] ▼ interrupted ← "生产者已消失，这是我所拥有的数据" ``` ## 无轮询的追随 有一个值得记住的细节。追随读取器从不轮询 SQLite。在热循环中轮询数据库以查看一个 token 是否到达，在演示中感觉不错，在生产中则是一种痛苦。相反，消费循环的 `notify()` 在每次插入后解析一个共享的 promise，而追随器只是 `await` 它。这是可行的，因为 Durable Object 是单线程运行的：插入和通知发生在一个同步块中，所以唤醒的追随器总是能看到新的行已经提交。没有竞态条件，没有需要调整的轮询间隔。运行时为您完成了最难的部分，这也是在 Durable Object 上构建 Agent 的主要卖点之一。 ## 无需编写 SSE 解析器的回放 好的，您已经恢复了运行。现在，您需要将其转换为 Agent 循环可以继续执行的东西。显而易见的方法是自行解析缓冲的 SSE。这是一个陷阱。您会永远维护一个针对 OpenAI 格式、Anthropic 格式、Google 格式的定制解析器，并追逐他们每一次的线路格式变更。所以，不要这样做。存储**原始字节**，并在返回路径上重用供应商**自己的**解析器。这正是 `workers-ai-provider` 中的模式：一个提供者通过 AI Gateway 路由所有模型，每个插件携带其原生线路格式，并且 resume 默认开启。 ```typescript import { createWorkersAI } from "workers-ai-provider"; import { openai } from "workers-ai-provider/openai"; import { anthropic } from "workers-ai-provider/anthropic"; const workersai = createWorkersAI({ binding: env.AI, providers: [openai, anthropic] // 每个插件自带其 SSE 解析器 }); const result = streamText({ model: workersai("openai/gpt-5.5", { resume: true }) }); // result.response.headers["cf-aig-run-id"] 标识要重新连接的运行。 ``` `streamText()` 解析、运行工具循环、处理推理，并原生地渲染响应，就像一次全新的调用。任何地方都没有自定义的 SSE 解析，而且提供者更改其格式对您没有成本，因为您使用的是**他们的**解析器。进程驱逐情况是整个的重点。当流运行时，您持久化 `{ runId, eventOffset }`（通过 `onDispatch` 和 `onProgress`）；当 Agent 回来时，您重新连接到同一个运行，而不是重新调用模型： ```typescript const stream = createResumableStream({ binding: env.AI, gateway: "my-gateway", runId, // 从 cf-aig-run-id 保存 fromEvent: savedOffset, // 从 onProgress 保存 onResumeExpired: "accept-partial" // 一旦约 5.5 分钟的缓冲区 TTL 过期 }); ``` 没有重新计费，没有重复调用，不需要维护解析器。 ## 等等，其他人也这么做吗？ 我去查了一下，因为“绝不浪费一个 token”看起来像是已经有人构建过的东西。结果发现，有一个供应商已经构建了几乎完全一样的东西（针对他们自己的 API），而其他人都把它留给了您。需要关心的两个问题是： 1. 当您的连接断开后，提供者**是否会继续生成**？（还是 token 会消失？） 2. 您能否通过游标**恢复同一个流**，而无需为已生成的内容重新计费？ | | 断开后继续生成？ | 通过游标恢复，无需重新计费？ | 如何实现 | |---|---|---|---| | **OpenAI** (Responses, background mode) | 是 | 是 | `background: true` + `stream: true`，通过 `?starting_after={sequence_number}` 恢复 | | **Anthropic** | 否 | 否 | 提示模型“继续” - 重新计费 token，可能产生偏差 | | **Google Gemini** | 否 | 否 | 相同的“从这里继续”提示 hack | | **OpenRouter** | 部分（取消停止计费） | 否（仅全响应缓存） | 响应缓存 + 取消；自行构建缓冲区 | | **Vercel AI SDK** (`resumable-stream`) | 是（通过 `waitUntil` 保持生产者存活） | 是，但仅限于页面刷新 | 应用层 Redis 缓冲区 - 相同的技巧，范围更窄 | | **本文 / AI Gateway** | 是 | 是 | 基础设施层持久缓冲区，供应商无关，能抵御您的部署 | 有几个突出的点。**OpenAI 已经证明了这一想法。** Responses API 的 Background mode (https://developers.openai.com/api/docs/guides/background) 即使在您断开连接后，也能在服务器端保持作业运行，并且您可以通过跟踪 `sequence_number` 游标来恢复：`GET /v1/responses/{id}?stream=true&starting_after={n}`。这是持久化推理，供应商原生，现已推出。但它仅限于 OpenAI 自己的 API（需要 `store: true`，并且 TTFT 更高）。**Anthropic 和 Gemini 让您重新付费。** 两者都不支持服务器端恢复。有记录的恢复方法是捕获部分响应，然后发送一个**新**请求，要求模型从它离开的地方继续。这需要再次支付输出 token，并且续接不保证一致。Anthropic 的文档 (https://platform.claude.com/docs/en/build-with-claude/streaming) 甚至指出工具使用和思考块无法部分恢复。这正是本文开头所说的代价。**Vercel 的 `resumable-stream` 是最接近的表亲。** 它独立地得出了相同的核心技巧：即使原始读者消失，生产者也会完成流，并且第二个消费者可以跟随。但它是应用层的，所以您必须运行 Redis，并且生产者存在于**您的**进程中——因此，它无法在您重新部署代码时存活。这最后一种情况正是我需要独立部署的全部原因。所以，并不是没有人做这个。OpenAI 表明，对一个 API 来说这是值得做的。其他所有人都让您重新提示并重新付费。而且，没有人涵盖**您自己的进程**死亡的情况。这就是空白。 ## 点睛之笔：这即将在 AI Gateway 中推出 那么，它应该放在哪里？对比直接指明了方向。OpenAI 的版本有效，因为它运行在您无需部署的基础设施上；Vercel 的版本不足，因为它做不到。您希望缓冲区位于已经在请求路径中、已经前置所有提供者、并且在您部署代码时永远不会被重新部署的地方。这就是 AI Gateway。我将其原型化为一个 Durable Object (RFC #1257 (https://github.com/cloudflare/agents/issues/1257))，但它始终旨在存在于托管基础设施中，而不是您的 Agent 中。将其放入网关，您就获得了 OpenAI 的 background-mode 理念，并将其提供给**每一个**提供者，包括今天让您重新付费的那些。好消息是：持久化恢复**即将在 Cloudflare AI Gateway 中推出**。它尚未广泛发布，但我已经用真实流量运行了它。每次运行都会返回一个 `cf-aig-run-id`，并且您可以要求网关从某个事件索引开始回放。我通过它运行了六个模型，在中间点切断了每个流，然后请求尾部： | 模型 | 事件数 | 字节数 | 从中部恢复 → 尾部匹配？ | |---|---|---|---| | gpt-4o-mini | 63 | 20,604 | ✅ 字节精确 | | gpt-5.4 | 25 | 7,350 | ✅ 字节精确 | | claude-haiku-4.5 | 11 | 1,924 | ✅ 字节精确 | | claude-sonnet-4.5 | 10 | 1,868 | ✅ 字节精确 | | gemini-3-flash | 4 | 2,253 | ✅ 字节精确 | 从 gpt-4o-mini 运行的事件索引 31 处恢复，准确地返回了流的后半部分，逐字节匹配。（一个注意事项：`from` 是**事件索引**，不是字节偏移。字节偏移没有解决。当它落地时，值得知道这一点。）这就是我希望人们带走的信息。您不必永远手动构建这个 Durable Object hack。目标是将其作为一等、可选的功能——我们瞄准的形态，即将在聊天 Agent 基类（`AIChatAgent` 和 `Think`）中推出，大致如下： ```typescript export class MyAgent extends Think { override durableBuffer = true; // 通过持久缓冲区路由推理（即将推出） } ``` 翻转一个开关，您就永远不必为同一个 token 支付两次。 ## 要点 - **输出 token 一旦生成，您就需要付费。** 如果崩溃迫使您重试，您需要再次付费，并且在一个 Agent 循环中，这会通过轮次中的每次工具调用而叠加。 - **不要将供应商连接与打开它的进程绑定。** 一个独立的、永不重新部署的缓冲区能在您的部署过程中保持流存活，而 `keepAliveWhile` 能在运行时保持后台消费不中断。 - **可恢复的流式传输和崩溃恢复是同一个机制。** 同一个持久的块日志；唯一的问题是是否仍有活的生产者连接着。 - **存储原始字节，通过真实供应商的解析器回放。** 不要自己编写 SSE 解析器；让每个供应商自己的插件进行格式转换，并通过 ID 重新连接到运行。 - **这应该放在网关中。** 永不重新部署的托管基础设施是它的正确位置，并且持久化恢复即将在 Cloudflare AI Gateway 中推出。 更大的故事是它自己的帖子：一旦您获得了恢复的流，该如何处理，以及 Agent 循环恢复决策树的其余部分。这是题外话。但这是一个能在每次部署时为您省钱的题外话，所以我认为值得单独拿出来。绝不浪费一个 token。