本系列讲实现 Agent harness 时会反复碰到的 Node / JS 运行时能力。默认读者会一点 JS但还没系统用过异步与「流式」写法。上一篇2子进程示例仓库react-agent-mini相关前作150 行搞懂 Agent 主循环场景为什么 Agent 离不开「等」和「一段段出来」做 Agent 时程序经常在干两件和「时间」有关的事等外部结果读文件、跑命令、调大模型 API——都不是立刻返回的。边等边吐模型回复往往是流式的字一个个或一小段一小段过来CLI 要立刻打到屏幕上而不是等整段说完才显示。如果用「普通同步函数」硬等constreplycallModelAndWaitForever(...);// 假想卡住直到全部说完console.log(reply);用户会感觉界面假死中间也无法取消工具跑很久时整条进程都堵着。所以主循环不是「一个函数算完返回字符串」而是一边跑一边往外 yield 事件text_delta、消息…… 外层用 for await 一段段接住本篇把这条链从零讲清楚async/await→ 异步生成器 →for await→ 对照callModel/query。说明async/ 生成器首先是JavaScript 语言能力在 Node 里写 Agent 时几乎天天用。本系列仍按「做 Agent 会踩到的运行时基础」来写。1. 同步 vs 异步一句话直觉同步异步调用时立刻做完或卡住直到做完先登记任务稍后再拿结果期间进程很难顺便干别的可以继续跑事件循环收别的 IO、定时器典型写法readFileSync、死循环等await readFile、await fetch……上一篇的spawn也是异步味道先起子进程再用事件/Promise等它结束而不是函数直接返回命令输出。2. Promise一张「稍后兑现」的欠条constpreadFile(a.txt,utf-8);// 立刻返回的是 Promise不是文件内容consttextawaitp;// 等到读完text 才是字符串可以记Promise 「这件事还在进行 / 最终会成功或失败」await 「停在这条async函数里等这张欠条兑现兑现前把控制权交回事件循环」只有async function以及后面的async function*里才能用await。最小例子asyncfunctionload(){consttextawaitreadFile(a.txt,utf-8);returntext;}// 调用方consttextawaitload();Agent 的工具call、读盘、等子进程几乎都是这种「asyncawait」形状。3. 普通async function不够还要「多次往外送」async function只能 return 一次一个最终值。但 Agent 主循环需要先送出一小段text_delta供打字机效果再送出完整的assistant消息工具跑完再送出tool_result……多轮很多次这就是异步生成器async function*yield。3.1 普通生成器同步版先建立直觉function*count(){yield1;yield2;yield3;}for(constnofcount()){console.log(n);// 1然后 2然后 3}function*生成器函数yield暂停并把一个值交给外层外层用for...of一次次拿走3.2 异步生成器每次 yield 之前可以 awaitasyncfunction*ticks(){yielda;awaitdelay(100);// 假想的等待yieldb;}forawait(constxofticks()){console.log(x);}注意外层变成了for await (... of ...)因为下一项可能还要等 IO。可以对照记写法往外给几个值中途能否 awaitasync function最多 1 个return能function*多个yield不能同步async function*多个yield能Agent 的query、callModel、runTools用的就是第三种。4. 消费方for await在干什么forawait(constitemofquery({messages,tools,toolUseContext})){if(item.typetext_delta){process.stdout.write(item.text);// 立刻打到终端}// 其它类型完整消息等}循环每转一圈向生成器要「下一个值」若生成器卡在某个await例如还在等模型 chunk就继续等一旦yield出来进入循环体处理生成器结束return后for await退出示例仓库入口注释里写的也是这套用法* example * ts * for await (const item of query({ messages, tools, toolUseContext })) { * if (item.type text_delta) process.stdout.write(item.text) * } * const { value: terminal } await gen.next() // 需手动 next 获取 return * 若用for await只遍历 yield 出的值生成器的return值——例如终止原因Terminal——要另用gen.next()在done时取。测试里常见「drain」辅助函数就是干这个。5. 流式调模型从 API chunk 到text_delta生产路径大致是OpenAI 兼容接口stream: true → 一串 ChatCompletionChunk异步可迭代 → parseOpenAIStream拆成 text_delta / 最终 assistant → callModel再 yield 出去 → query继续 yield 给 REPL / UI5.1callModel自己也是异步生成器export async function* callModel( params: CallModelParams, ): AsyncGeneratorStreamEvent | AssistantMessage { // ... const stream await client.chat.completions.create( { model: config.model, messages, tools: tools.length 0 ? tools : undefined, stream: true, }, { signal: params.signal }, ) yield* parseOpenAIStream(stream) }这里的yield*很重要yield x自己产出一个值yield* otherGenerator把另一个生成器产出的值原样转发出去管道对接于是callModel不必手写一遍「解析 chunk」的循环解析逻辑集中在parseOpenAIStream。5.2parseOpenAIStreamfor await读网流yield成内部事件export async function* parseOpenAIStream( stream: AsyncIterableChatCompletionChunk, ): AsyncGeneratorStreamEvent | AssistantMessage { let text const toolCalls new Mapnumber, ToolCallAccumulator() for await (const chunk of stream) { const choice chunk.choices[0] if (!choice) continue const delta choice.delta if (delta.content) { text delta.content yield { type: text_delta, text: delta.content } }直觉网络上每次来一小片delta.content立刻yield { type: text_delta, text: ... }上层就能打印同时在本地text ...攒全文流结束或出现 tool_calls时再yield一条完整的assistant消息供主循环判断有没有tool_use「流」在这里不是 Node 的fs.createReadStream那种 Stream 类那是另一套 API而是更宽的意思异步可迭代AsyncIterable——用for await一段段拿。HTTP 流式响应、异步生成器都落在这个心智里。6. 主循环query套娃式的for awaityieldquery本身是async function*。每一轮里它会for await消费callModel把text_delta/assistant再 yield 给外层若有工具再for await消费runTools把tool_result消息 yield 出去追加历史continue下一轮或return终止原因核心片段调模型for await (const chunk of deps.callModel({ messages: outbound, tools: params.tools, systemPrompt: params.systemPrompt, signal: abortSignal, })) { if (abortSignal?.aborted) { trace(query.turn_end, { reason: aborted, turn: turnCount }) return { reason: aborted } } if (chunk.type text_delta) { yield chunk satisfies StreamEvent continue } if (chunk.type assistant) { assistantMessages.push(chunk) yield chunk工具阶段同理for await (const update of runTools( toolUseBlocks, parentMessage, params.toolUseContext, )) { if (update.message) { yield update.message toolResults.push(update.message) } }画成管道parseOpenAIStream yield text_delta / assistant ↑ yield* callModel ↑ for await … yield query ↑ for await REPL / UI打印、渲染主循环的「转起来」在代码形态上就是异步生成器层层对接而不是一个巨大的回调金字塔。入口也可以写成return yield* queryLoop(...)把内部循环生成器的产出与最终return一并交给外层——又是yield*管道。7. 另一种消费法把生成器「抽干」drain有时不需要把子过程的每个text_delta都转给用户只想跑完整个query拿到最终Terminal顺便收集几条assistant做摘要子代理工具里就是这种模式手动gen.next()循环直到doneasync function drainNestedQuery( params: Parameterstypeof query[0], ): Promise{ terminal: Terminal assistants: AssistantMessage[] } { const assistants: AssistantMessage[] [] const gen query(params) while (true) { const { value, done } await gen.next() if (done) { return { terminal: value, assistants } } if (value.type assistant) { assistants.push(value) } } }对比方式适合for await (const x of gen)关心每一次 yield打字、更新 UI手动next抽干嵌套跑完要结果或只要部分事件 最终 return 值同一套query生成器外层怎么消费决定了产品形态REPL 流式展示子代理则同步等摘要。8. REPL 侧用户输入也可以是异步迭代会话循环对「一行行用户输入」同样用for awaitfor await (const line of deps.lines) {lines可以是把readline包成的异步生成器。这样「等用户打字」和「等模型吐字」是同一种消费模型测试时也能塞进假的异步 iterable不必真连终端。9. 和「Node Stream 类」的关系避免名词混淆Node 还有Readable/Writable等Stream 类例如fs.createReadStream、HTTPIncomingMessage。它们也能变成异步可迭代在较新的 Node 里常可以直接forawait(constchunkofreadable){...}本篇 Agent 主路径里你更常直接写的是async function*yield/yield*for await消费不必先精通整个 Stream 管道pipe、backpressure才能读懂query。等真要处理大文件字节流时再单独补 Stream 类即可。常见坑坑说明建议写成普通async function却想多次推送只能 return 一次需要多次推送就用async function*for...of去套异步生成器拿不到异步下一项用for await...of忘记消费生成器生成器不跑惰性必须for await或反复next()把callModel()的返回值当「最终字符串」返回的是生成器对象要迭代或抽干后再用结果只yield最终全文不yielddeltaCLI/UI 无法流式显示有增量就尽早 yield嵌套子query却把所有 delta 盲目外抛父 UI 可能被刷屏按产品决定转发还是 drain和主循环的关系用户一句输入 → queryasync function* → callModelasync function* → parseOpenAIStreamfor await 网流 yield → 若有 tool_use → runToolsasync function* → 多轮直到结束 return Terminal → REPL for await 打印 text_delta / 消息前作讲的 ReAct「模型 ↔ 工具」循环落到 JS 里就是异步生成器管道。学这部分是在学主循环怎样「转」而不堵死进程。本系列下一篇预告4取消与 AbortController——用户按 CtrlC、工具超时、嵌套子代理中止时信号怎么往下传、流式请求怎么停。你可以带走什么await等一次结果async function*yield多次往外送。for await是消费异步生成器 / 异步可迭代的标准姿势。yield*用来对接生成器管道如callModel→parseOpenAIStream。流式体验 尽早 yield 增量text_delta最后再给完整assistant。同一生成器可以流式展示也可以 drain 只要结果——子代理常用后者。仓库与延伸GitHubreact-agent-mini前作主循环150 行搞懂 Agent 主循环源码query.ts · client.ts · stream.ts · AgentTool.ts欢迎 Star、Issue 和 PR。本文为「做 Agent 会用到的 Node API」系列第 3 篇示例基于 react-agent-mini。
