【从零到一实现一个 AI Agent 框架 · 第十篇】流式输出：让 Agent 边想边说

09. 流式输出让 Agent 边想边说从零到一实现一个 AI Agent 框架 · 第九篇1. 为什么需要 Streaming先看一个普通的 LLM 调用。用户帮我分析这个项目的工具系统 ↓ 等待 15 秒 ↓ LLM这是完整分析……这 15 秒里用户看到的是空白。问题不只是“慢”。更麻烦的是用户不知道 Agent 是卡住了还是正在生成。如果方向错了用户只能等完整回答出来才发现。长回答会让终端像死了一样。CtrlC 中断时用户没有足够反馈判断该不该中断。加上流式输出之后体验变成这样用户帮我分析这个项目的工具系统 ↓ LLM工具系统的核心是…… ↓ LLM它主要分为注册、权限、分发…… ↓ LLM下面看 Axon 的实现……模型还在生成用户已经开始阅读。这就是Streaming流式输出的价值把“等待完整结果”变成“持续接收增量结果”。2. 从零开始最小 Streaming非流式调用像这样constresponseawaitclient.chat.completions.create({model,messages,stream:false,})console.log(response.choices[0].message.content)模型生成完整回答后服务端一次性返回。流式调用则不同conststreamawaitclient.chat.completions.create({model,messages,stream:true,})forawait(constchunkofstream){consttextchunk.choices[0]?.delta?.contentif(text)process.stdout.write(text)}这里的关键变化是非流式流式一次性拿到完整 message多次拿到 chunkmessage.contentdelta.content最后统一打印边收到边打印用户等待完整结果用户持续看到进展最小 streaming 其实不复杂。真正复杂的是Agent 不只会输出文字还会调用工具。3. 工具调用为什么让 Streaming 变复杂OpenAI-compatible 的流式响应里工具调用不是一次性到达的。例如模型想调用{name:read_file,arguments:{\path\:\src/agent.ts\}}流式过程中可能拆成这样chunk 1: tool_call index0, nameread_file, arguments{\path\ chunk 2: tool_call index0, arguments:\src/ chunk 3: tool_call index0, argumentsagent.ts\}也就是说Agent 必须一边接收 chunk一边把工具调用拼起来。consttoolCallMap:Recordnumber,ToolCall{}forawait(constchunkofstream){constdeltachunk.choices[0]?.deltaif(delta?.tool_calls){for(consttcofdelta.tool_calls){if(!toolCallMap[tc.index]){toolCallMap[tc.index]{id:tc.id??,name:tc.function?.name??,arguments:,}}if(tc.function?.arguments){toolCallMap[tc.index].argumentstc.function.arguments}}}}为什么用index因为一次模型响应里可能有多个工具调用而且它们的参数 chunk 可能交错到达。index是把碎片拼回正确工具调用的锚点。4. Streaming 在 Agent Loop 里的位置Streaming 不改变 Agent Loop 的本质。原来的循环是是否用户输入调用 LLM有工具调用?执行工具返回最终回答加了 streaming 之后只是把“调用 LLM”这一格拆细文本 delta工具 deltausage否是调用 LLM stream接收 chunkchunk 类型输出给用户累积 tool_call记录 metricsstream 结束?返回完整 content toolCalls注意最后仍然要返回完整结果。因为 Agent Loop 后续还需要把 assistant message 写入历史。判断finishReason是不是tool_calls。执行工具。把工具结果塞回 messages。进入下一轮 LLM 调用。所以 streaming 的职责不是替代 Agent Loop而是让 LLM 调用过程变成可观察的增量事件。5. Axon 的 Streaming 实现Axon 的核心代码在src/agent.ts。简化后是这样privateasynccallApiOnce():PromiseApiResult{conststreamawaitthis.client.chat.completions.create({model:this.model,messages:[{role:system,content:this.systemPrompt},...this.messages,],tools:DEFINITIONS,stream:true,stream_options:{include_usage:true},},{signal:this.abortController?.signal})letcontentletfinishReasonstopletusageconsttoolCallMap:Recordnumber,ToolCall{}forawait(constchunkofstream){if(chunk.usage)usagechunk.usageconstchoicechunk.choices[0]if(!choice)continueif(choice.finish_reason){finishReasonchoice.finish_reason}constdeltachoice.deltaif(delta.content){this.events.onTextDelta?.(delta.content)contentdelta.content}if(delta.tool_calls){// 按 index 拼接工具调用}}this.events.onTextDelta?.(\n)return{content,toolCalls:Object.values(toolCallMap),finishReason,usage}}这里有几个关键点。5.1stream: truestream:true告诉 provider 返回 AsyncIterable而不是普通 response。5.2stream_options.include_usagestream_options:{include_usage:true}让最后的 chunk 带上 token usage。这样/metrics才能统计本次输入 token本次输出 token累计输入 token累计输出 token5.3AbortSignal{signal:this.abortController?.signal}让 CtrlC 可以中断正在进行的 API stream。这和 CLI 里的 CtrlC 双语义配合Agent 正在处理 → abort 当前 stream Agent 空闲 → 再按一次退出进程6. 输出事件不要把 runtime 绑死到 stdout最早的实现可以直接写process.stdout.write(delta.content)这很直观但有个问题Session被绑死到了终端输出。以后如果想做这些事情就会很别扭测试 streaming 输出。把 Axon 接到 Web UI。做更复杂的 TUI。录制完整 transcript。给工具调用状态做统一渲染。所以 Axon 把输出抽成了SessionEventsexportinterfaceSessionEvents{onTextDelta?:(text:string)voidonToolCallDelta?:(delta:{name?:stringargumentsDelta?:string})voidonRetry?:(event:RetryEvent)void}默认事件仍然写终端{onTextDelta:(text)process.stdout.write(text),onToolCallDelta:(delta){if(delta.name)process.stdout.write(\n⚙${delta.name})if(delta.argumentsDelta)process.stdout.write(.)},}但测试或其他 UI 可以传自己的实现constdeltas:string[][]constsessionnewSession(apiKey,model,agentsContext,memoryContext,hooks,baseURL,client,skillLoader,{onTextDelta:(text)deltas.push(text),},)这个变化很小但边界很重要Agent runtime 负责产生事件CLI 负责渲染事件。7. API 重试不是所有失败都该重试流式请求跑到一半也可能失败HTTP 429 速率限制 HTTP 503 服务暂时不可用 ECONNRESET 连接被重置 ETIMEDOUT 网络超时 overloaded 模型服务过载这些失败通常是临时的可以重试。但下面这些就不应该重试HTTP 400 请求参数错 HTTP 401 API key 错 HTTP 404 模型名错重试不会让错误配置突然变对只会浪费时间和 token。Axon 的判断逻辑是exportfunctionisRetryableApiError(error:any):boolean{conststatuserror?.status??error?.statusCodeif([429,503,529].includes(status))returntrueif([ECONNRESET,ETIMEDOUT].includes(error?.code))returntrueconstmessageString(error?.message??).toLowerCase()returnmessage.includes(overloaded)||message.includes(temporarily unavailable)}然后用指数退避constdelayMsMath.min(1000*2**attempt,30_000)Math.floor(Math.random()*1000)等待时间大致是第 1 次重试1s jitter 第 2 次重试2s jitter 第 3 次重试4s jitter为什么要加随机抖动如果很多客户端都在服务过载后固定 1 秒重试它们会同时冲回去形成“重试风暴”。随机抖动可以把请求打散。8. Anthropic 适配统一接口背后的差异Axon 的主循环使用 OpenAI-compatible chunk 格式。但 Anthropic 的原生 stream 事件长得不一样content_block_start content_block_delta message_delta message_stop所以 Axon 有一个 adapterAnthropic stream event ↓ providers/anthropic.ts ↓ OpenAI-compatible chunk ↓ agent.ts callApiOnce()例如Anthropic 事件OpenAI-compatible chunktext_deltadelta.contenttool_usestartdelta.tool_calls[].function.nameinput_json_deltadelta.tool_calls[].function.argumentsmessage_delta.stop_reasonfinish_reasonmessage_stop.usageusage这样Session不需要关心 provider 差异。这次优化里还补了一个细节Anthropic adapter 需要透传AbortSignal。create(params,options){returnanthropicToOpenAIStream(anthropic,model,system,messages,tools,options,)}否则 CLI 里按 CtrlC 时OpenAI provider 能中断Anthropic provider 却可能继续跑。9. 为什么不做“提前执行工具”有一种更激进的优化当流式响应里某个工具调用 block 已经完整接收时立刻执行这个工具不等整个模型响应结束。听起来很好因为可以把工具执行时间藏在模型继续生成的时间里。但 Axon 暂时没有做。原因有三个OpenAI-compatible 抽象会被打破提前执行工具更依赖 Anthropic 原生content_block_stop事件。Axon 当前选择把 provider 统一成 OpenAI chunk主循环更简单。收益有限Axon 已经支持并发安全工具批量执行。对于read_file、search_files这类快速工具响应结束后并发跑已经够快。权限边界更复杂提前执行必须保证只执行无需确认、只读、并发安全的工具。否则用户还没看到完整模型意图工具已经动手了。所以第一阶段更稳的路线是先把 streaming 事件、retry、abort、metrics 做扎实 再考虑 provider-specific 的提前工具执行这是一个很典型的工程取舍不是能做就马上做而是看它是否值得增加复杂度。10. 测试 StreamingStreaming 测试的关键是构造一个假的 AsyncIterable。asyncfunction*streamChunks(chunks:any[]){for(constchunkofchunks){yieldchunk}}然后 mock clientconstclient{chat:{completions:{create:vi.fn().mockReturnValue(streamChunks([{choices:[{delta:{content:hello}}]},{choices:[{delta:{tool_calls:[{index:0,id:call-1,function:{name:read_file,arguments:{\path\,},}],},}],},{choices:[{delta:{tool_calls:[{index:0,function:{arguments::\README.md\}},}],},finish_reason:stop,}],},])),},},}这样可以验证三件事文本 delta 会触发onTextDelta。工具参数碎片会触发onToolCallDelta。usage chunk 会进入 metrics。重试也可以单独测letcalls0constresultawaitwithApiRetry(async(){callsif(calls1){consterror:anynewError(rate limited)error.status429throwerror}returnok})expect(result).toBe(ok)expect(calls).toBe(2)测试不需要真的连模型服务。只要协议边界测住runtime 行为就稳很多。11. 小结Streaming 的表面作用是“逐字输出”但对 Agent 来说它更像一层 runtime 事件系统。它让三件事变得可见模型正在生成什么。模型准备调用什么工具。API 请求是否失败、是否重试、用了多少 token。Axon 的实现遵循一个简单边界Session负责接收流式事件并维护 Agent LoopCLI 负责把事件渲染给用户。这让 streaming 不只是一个终端打印技巧而是未来接 TUI、Web UI、日志、metrics、transcript 的基础。第一版不需要追求最炫的提前工具执行。先把流式输出、AbortSignal、usage、retry 和事件解耦做好Agent 就已经从“等结果的黑盒”变成了“持续可观察的运行时”。