企业官网建设流程全解析

本系列讲实现 Agent harness 时会反复碰到的 Node / JS 运行时能力。默认读者：会一点 JS，但还没系统用过异步与「流式」写法。
上一篇：（2）子进程
示例仓库：react-agent-mini
相关前作：150 行搞懂 Agent 主循环

场景：为什么 Agent 离不开「等」和「一段段出来」

做 Agent 时，程序经常在干两件和「时间」有关的事：

1. 等外部结果：读文件、跑命令、调大模型 API——都不是立刻返回的。
2. 边等边吐：模型回复往往是流式的，字一个个（或一小段一小段）过来；CLI 要立刻打到屏幕上，而不是等整段说完才显示。

如果用「普通同步函数」硬等：

constreply=callModelAndWaitForever(...);// 假想：卡住直到全部说完console.log(reply);

用户会感觉界面假死；中间也无法取消；工具跑很久时整条进程都堵着。

所以主循环不是「一个函数算完返回字符串」，而是：

一边跑，一边往外 yield 事件（text_delta、消息……） 外层用 for await 一段段接住

本篇把这条链从零讲清楚：async/await→ 异步生成器 →for await→ 对照callModel/query。

说明：async/ 生成器首先是JavaScript 语言能力；在 Node 里写 Agent 时几乎天天用。本系列仍按「做 Agent 会踩到的运行时基础」来写。

1. 同步 vs 异步：一句话直觉

上一篇的spawn也是异步味道：先起子进程，再用事件/Promise等它结束，而不是函数直接返回命令输出。

2. Promise：一张「稍后兑现」的欠条

constp=readFile("a.txt","utf-8");// 立刻返回的是 Promise，不是文件内容consttext=awaitp;// 等到读完，text 才是字符串

可以记：

• Promise= 「这件事还在进行 / 最终会成功或失败」
• await= 「停在这条async函数里，等这张欠条兑现；兑现前把控制权交回事件循环」

只有async function（以及后面的async function*）里才能用await。

最小例子：

asyncfunctionload(){consttext=awaitreadFile("a.txt","utf-8");returntext;}// 调用方：consttext=awaitload();

Agent 的工具call、读盘、等子进程，几乎都是这种「async+await」形状。

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 之前可以 await

asyncfunction*ticks(){yield"a";awaitdelay(100);// 假想的等待yield"b";}forawait(constxofticks()){console.log(x);}

注意外层变成了for await (... of ...)：因为下一项可能还要等 IO。

可以对照记：

Agent 的query、callModel、runTools用的就是第三种。

4. 消费方：for await在干什么

forawait(constitemofquery({messages,tools,toolUseContext})){if(item.type==="text_delta"){process.stdout.write(item.text);// 立刻打到终端}// 其它类型：完整消息等}

循环每转一圈：

1. 向生成器要「下一个值」
2. 若生成器卡在某个await（例如还在等模型 chunk），就继续等
3. 一旦yield出来，进入循环体处理
4. 生成器结束（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 / UI

5.1callModel：自己也是异步生成器

export async function* callModel( params: CallModelParams, ): AsyncGenerator<StreamEvent | 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.2parseOpenAIStream：for await读网流，yield成内部事件

export async function* parseOpenAIStream( stream: AsyncIterable<ChatCompletionChunk>, ): AsyncGenerator<StreamEvent | AssistantMessage> { let text = '' const toolCalls = new Map<number, 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 await+yield

query本身是async function*。每一轮里它会：

1. for await消费callModel，把text_delta/assistant再 yield 给外层
2. 若有工具，再for await消费runTools，把tool_result消息 yield 出去
3. 追加历史，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()循环，直到done：

async function drainNestedQuery( params: Parameters<typeof 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) } } }

对比：

同一套query生成器，外层怎么消费决定了产品形态：REPL 流式展示，子代理则同步等摘要。

8. REPL 侧：用户输入也可以是异步迭代

会话循环对「一行行用户输入」同样用for await：

for 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 类即可。

常见坑

和主循环的关系

用户一句输入 → query（async function*） → callModel（async function*） → parseOpenAIStream（for await 网流 + yield） → 若有 tool_use → runTools（async function*） → 多轮直到结束 return Terminal → REPL for await 打印 text_delta / 消息

前作讲的 ReAct「模型 ↔ 工具」循环，落到 JS 里就是：异步生成器管道。学这部分，是在学主循环怎样「转」而不堵死进程。

本系列下一篇预告

（4）取消与 AbortController——用户按 Ctrl+C、工具超时、嵌套子代理中止时，信号怎么往下传、流式请求怎么停。

你可以带走什么？

1. await等一次结果；async function*+yield多次往外送。
2. for await是消费异步生成器 / 异步可迭代的标准姿势。
3. yield*用来对接生成器管道（如callModel→parseOpenAIStream）。
4. 流式体验 = 尽早 yield 增量（text_delta），最后再给完整assistant。
5. 同一生成器可以流式展示，也可以 drain 只要结果——子代理常用后者。

仓库与延伸

• GitHub：react-agent-mini
• 前作主循环：150 行搞懂 Agent 主循环
• 源码：query.ts · client.ts · stream.ts · AgentTool.ts

欢迎 Star、Issue 和 PR。

本文为「做 Agent 会用到的 Node API」系列第 3 篇；示例基于 react-agent-mini。