2.6 异步迭代器与 for await

本章解决什么问题：上一章的 Promise 解决了「一个还没到来的值」怎么表达；但模型的流式回复不是一个值，而是一串值——一小块一小块的文字，隔几十毫秒到一块。本章讲 JavaScript 为「多个陆续异步到来的值」准备的标准形状：异步迭代器（Async Iterator），以及生产它的 async function* 和消费它的 for await..of。Pi 里模型回复、Agent 事件全都以这个形状流动——这一章是读懂 Pi 流式代码的钥匙。

前置知识：2.5 异步编程：Promise 与 async/await（Promise、await）；2.3 union、字面量与可辨识联合（Pi 的事件类型会用到）；2.4 泛型、class 与类型收窄（本章末尾读 Pi 源码时会用到泛型 class）。

学习目标：读完本章后你能

• 说出迭代器协议的两个要素——next 方法和 { value, done } 结果对象——并解释 for..of 只是这套协议的简写；
• 写出异步生成器（async function*），用 yield 一块一块地交出值，用 for await..of 消费；
• 读懂 AsyncIterable / AsyncIterator 这两个 interface 的形状，说出 Symbol.asyncIterator 在其中扮演的角色；
• 解释在 for await 循环里 break 时生成器会发生什么，知道清理代码应该写在哪里；
• 在 Pi 源码里认出这套机制：EventStream 类与 Provider 里的事件转换生成器。

建立直觉

想象你问大语言模型（LLM，Large Language Model）一个问题，完整回复要 30 秒才能生成完。表达这次交互的结果，有两种形状：

• 形状甲：Promise<string>——等 30 秒，一次性拿到整段文字。类型上毫无问题，体验上是灾难：用户面对 30 秒白屏，不知道程序是在思考还是已经卡死。
• 形状乙：流式输出（Streaming）——模型每生成一小块就立刻发给你，屏幕上文字一点点长出来。总时间一秒没省，但第一块几百毫秒就到了。

形状乙的数据不再是「一个未来值」，而是「多个未来值，一块一块地异步到来」——Promise 装不下它。把「一个还是多个」「同步还是异步」两个维度交叉，会得到一张四格表：

	一个值	多个值
同步拿到	普通值 T	Iterable<T>（数组、迭代器）
异步到来	Promise<T>	AsyncIterable<T>（本章主角）

前三格你都已经认识，JavaScript 直到 2018 年（ES2018）才补齐第四格：异步迭代器加上配套的消费语法 for await..of。一个方便记忆的类比：Promise 像一张「单件包裹的取件码」，凭码等一件、取一件、完事；异步迭代器像一条传送带——包裹一件一件送到面前，每件之间要等待，直到传送带宣布「没有了」。

图加载中…

图 2.6-1 Promise 与异步迭代器：一个未来值 vs 一串未来值
阅读顺序：两行各自从左到右。上排是 2.5 章的世界：一次等待换一个结果。下排是本章的世界：等待和取值交替出现任意多次，最后以一个「done」信号收尾——这正是模型流式回复在网络上的真实形状，也是 Pi 事件流的形状。

快速铺垫：迭代器与 for..of

先回到同步世界打个底。for..of 你在前面章节已经用过：

for (const n of [10, 20, 30]) {
  console.log(n);
}

它对数组、字符串、Set、Map 都有效，靠的不是魔法，而是这些类型都遵守同一套迭代器协议（Iterator Protocol）。协议内容非常小：

1. **迭代器（Iterator）**是一个带 next 方法的对象；
2. 每次调用 next()，返回一个 { value, done } 对象：done: false 表示 value 是下一个值；done: true 表示序列到头了。

手写这样的对象有些啰嗦，所以 JavaScript 提供了生成器函数（Generator Function）：在 function 后加一个星号，函数体里就可以用 yield 交出值。存成 countdown.ts：

function* countdown(from: number) {
  for (let i = from; i >= 1; i--) {
    yield i;
  }
}

const it = countdown(3);
console.log(it.next());
console.log(it.next());
console.log(it.next());
console.log(it.next());

for (const n of countdown(3)) {
  console.log(n);
}

用 1.2 章的方式 tsx countdown.ts 运行，真实输出：

{ value: 3, done: false }
{ value: 2, done: false }
{ value: 1, done: false }
{ value: undefined, done: true }
3
2
1

两个要点值得停下来体会：

• yield 像「能用多次的 return」：交出一个值并暂停在原地，下次再被要值时，从暂停处继续执行。调用 countdown(3) 本身不执行函数体，只是拿到一个待命的迭代器；每次 next() 才推动它走到下一个 yield。
• 迭代器是拉取式（pull）的：消费者要一次，生产者才动一步；不要，就永远停着。这个性质本章后面会变得非常重要。

而 for..of 只是这套协议的简写（俗称「语法糖」——不添加新能力、只让既有写法更顺手的语法）：它替你反复调用 next()、把 value 交给循环变量、见到 done: true 就停。

异步迭代器：让值一块一块地异步到来

没有它会怎样

同步迭代器的 next() 必须立刻返回 { value, done }。可是流式回复的下一块此刻还在网络上，根本给不出来。只用已有工具硬凑，剩下两个都不理想的选项：

• 退回 Promise<string[]>：等所有块到齐再一次性返回——流式的意义全没了；
• 为「每到一块」注册回调函数：能用，但取值逻辑被拆碎在回调里，也没有统一的「循环 + 自然结束」的写法（这种推送式风格什么时候合适，是 2.7 章的话题）。

缺的那块拼图其实只有一句话：让 next() 返回「未来才兑现的 { value, done }」，也就是 Promise<IteratorResult>。这就是异步迭代器的全部定义——协议还是那套协议，只是每一步都变成了异步。

📘 概念异步迭代器（Async Iterator）

带 next 方法的对象，但 next 返回的是 Promise，兑现后才得到 { value, done }。它表达「多个值陆续异步到来」的序列：取每个值都可能需要等待，序列以 done: true 结束。生产端的便捷写法是异步生成器 async function*，消费端的配套语法是 for await..of。

async function*：既能 await 又能 yield

生产异步迭代器的便捷写法，是把上一章的 async 和上一节的 * 合在一起——异步生成器（Async Generator）。函数体里两种能力同时具备：await（等一会儿）和 yield（交出一个值）。存成 fake-reply.ts：

function sleep(ms: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, ms));
}

async function* fakeModelReply(): AsyncGenerator<string> {
  const chunks = ["你好", "，我是", "一个", "会流式输出的", "模型"];
  for (const chunk of chunks) {
    await sleep(200);
    yield chunk;
  }
}

async function main(): Promise<void> {
  for await (const chunk of fakeModelReply()) {
    process.stdout.write(chunk);
  }
  process.stdout.write("\n");
}

main();

逐段读：

• sleep 是实验 08 里那位老朋友：把 setTimeout 包装成「ms 毫秒后兑现的 Promise」，用来模拟网络延迟；
• fakeModelReply 每块之间先 await sleep(200) 再 yield——「等一会儿，给一块」，循环五次。返回类型 AsyncGenerator<string> 表示「逐个吐出 string 的异步生成器」；
• 消费端换用 for await..of：与 for..of 的唯一区别是每圈循环会先等下一个值到来。它只能出现在 async 函数里，所以套了一层 main（与实验 08 相同的做法）；
• process.stdout.write 是不自动换行版的打印（console.log 会换行），后一块紧接着前一块，形成打字机效果。

真实输出（终端上是每 200 毫秒长出一块，静态书页展示不出过程，值得亲手跑一次）：

你好，我是一个会流式输出的模型

掀开 for await 的引擎盖

for await 同样只是协议的简写。手动转一圈引擎，能看清它每步拿到的东西——把 main 换成：

const it = fakeModelReply();
console.log(await it.next());
console.log(await it.next());

真实输出：

{ value: '你好', done: false }
{ value: '，我是', done: false }

和同步版一模一样的 { value, done }，只是每个都要 await 之后才拿得到。for await..of 替你做的就是：循环调用 next()、await 它、把 value 交给循环变量、见到 done: true 停止。

AsyncIterable 的接口形状

上面一直在说「协议」，现在把它写成 2.2 章学过的 interface。教学用的简化版声明长这样（官方内置声明还有几个可选成员，日常写代码不需要背全）：

interface AsyncIterator<T> {
  next(): Promise<IteratorResult<T>>;
}

interface AsyncIterable<T> {
  [Symbol.asyncIterator](): AsyncIterator<T>;
}

AsyncIterator 是迭代器本体：一个 next，返回 Promise。AsyncIterable 则是「可以被异步迭代的东西」：它只要求一件事——通过一把特殊的钥匙，能取出一个迭代器。

那把钥匙就是 Symbol.asyncIterator。Symbol 是 JavaScript 的一种基本类型，特点是每个 Symbol 值都独一无二，适合当「绝不会撞名」的属性键；语言内置了几个众所周知的 Symbol 作为协议的「门牌号」，Symbol.asyncIterator 就是异步迭代协议的门牌号。方括号写法 [Symbol.asyncIterator]() 叫计算属性名：属性的名字不是写死的字符串，而是这个 Symbol 值。for await 拿到对象后做的第一件事，就是敲这扇门要迭代器。

协议的意义在于：for await 认门牌不认出身。不用 async function*，徒手攒一个对象也完全合法。存成 shape.ts：

const threeTicks: AsyncIterable<number> = {
  [Symbol.asyncIterator](): AsyncIterator<number> {
    let n = 0;
    return {
      async next(): Promise<IteratorResult<number>> {
        n += 1;
        if (n <= 3) {
          return { value: n, done: false };
        }
        return { value: undefined, done: true };
      },
    };
  },
};

async function main(): Promise<void> {
  for await (const n of threeTicks) {
    console.log(n);
  }
}

main();

真实输出：

1
2
3

异步生成器造出来的对象、这样手写的对象、实现了该协议的 class 实例（本章末尾 Pi 的 EventStream 就是），在 for await 眼里一律平等。

🌱 初学者提示AsyncGenerator 与 AsyncIterable 的关系

async function* 的返回类型标注写 AsyncGenerator<T>。这个类型同时满足 AsyncIterable（它有 Symbol.asyncIterator 方法——返回它自己）和 AsyncIterator（它有 next）。所以生成器对象既能直接放进 for await..of，也能像上一节那样手动调 next。函数参数想「接受任何异步序列」时，习惯把类型写成更宽的 AsyncIterable<T>——Pi 源码正是这么做的。

提前 break 与清理

流式场景有个绕不开的现实：消费者随时可能不想要了。用户看到回复开头就按下停止键，循环 break 之后，生产者那头会发生什么？

用一个「无限流」把问题推到极端——它永远不会主动结束，如果 break 管不住它，程序就麻烦了。存成 early-break.ts：

function sleep(ms: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, ms));
}

async function* endlessNumbers(): AsyncGenerator<number> {
  try {
    for (let n = 1; ; n++) {
      await sleep(100);
      yield n;
    }
  } finally {
    console.log("清理：生成器已关闭，不会再产出新值");
  }
}

async function main(): Promise<void> {
  for await (const n of endlessNumbers()) {
    console.log(n);
    if (n === 3) break;
  }
  console.log("break 之后的第一行");
}

main();

真实输出：

1
2
3
清理：生成器已关闭，不会再产出新值
break 之后的第一行

程序正常退出了，没有失控的死循环。三件事在 break 的瞬间发生：

1. for await 在退出循环前，自动调用生成器的 return 方法——这是协议里那几个「可选成员」之一，意思是「不要了，请收尾」；
2. 生成器从暂停的 yield 处被强制走向函数结束。正因如此，把清理逻辑写在 try/finally 的 finally 块里，就能保证无论正常走完、被 break，还是循环体抛了异常，它都执行——输出里「清理」一行印在「break 之后的第一行」之前，说明循环退出前会先等清理完成；
3. 由于生成器是拉取式的，剩下的值从未被生产：不是「做完了被扔掉」，而是根本没做。第 4、5……个数字对应的 sleep 一次都不会再执行。

所以规则很简单：流式代码的资源清理（关闭网络连接、释放文件句柄、停掉计时器）写在生成器的 finally 里。Pi 的 Provider 代码里就有教科书式的一例：packages/ai/src/api/anthropic-messages.ts 读取网络字节流的生成器，正是在 finally 里释放流的读取器（第 441–443 行）。

⚠️ 常见误解以为 break 之后生成器还在后台继续跑

不会。生成器不是独立线程，它只在消费者调用 next 时才前进；break 触发 return 之后，它连暂停点都没有了。但要分清楚层次：finally 清理的是「生成器函数体」这一层。如果底层还有一个需要主动喊停的操作——比如一个正在传输的 HTTP 请求——那需要取消信号配合，这正是下一章 AbortController 的主题。

把本章的完整机制画在一张图上：

图加载中…

图 2.6-2 for await 与异步生成器的一次完整对话（含提前 break）
阅读顺序：从上到下。每对「实线去、虚线回」是一次 next 调用与一个 Promise 的兑现；生成器只在被要值时前进（拉取式）。最下面三行是 break 的收尾流程：return、finally、done——对应 early-break.ts 输出的最后两行，也对应本章实验第 4 部分你将亲手观察到的现象。

Pi 中哪里用到了它

Pi 与各家模型服务对接的代码都在 packages/ai 这个 package 里（6.1 章精读）。本章机制在那里不是配角，而是主干道：一次流式请求的返回值，就是一个实现了 AsyncIterable 的对象。

第一处：EventStream——手写协议实现的真实样本。

packages/ai/src/utils/event-stream.ts · EventStream

earendil-works/pi@c13ffe1第 4–67 行在 GitHub 查看 ↗

通用事件流 class：implements AsyncIterable，内部用队列做「推转拉」的缓冲，Symbol.asyncIterator 方法本身就是一个异步生成器。

export class EventStream<T, R = T> implements AsyncIterable<T> {
	private queue: T[] = [];
	private waiting: ((value: IteratorResult<T>) => void)[] = [];
	private done = false;
	// …（省略：最终结果相关的字段与构造函数）

	push(event: T): void {
		// …（省略：有消费者在等就直接交给它，否则先排进 queue）
	}

	async *[Symbol.asyncIterator](): AsyncIterator<T> {
		while (true) {
			if (this.queue.length > 0) {
				yield this.queue.shift()!;
			} else if (this.done) {
				return;
			} else {
				const result = await new Promise<IteratorResult<T>>((resolve) => this.waiting.push(resolve));
				if (result.done) return;
				yield result.value;
			}
		}
	}
	// …（省略：end 与 result 方法）
}

这二十来行浓缩了近三章的知识：泛型 class（T 是事件类型、R 是最终结果类型，2.4 章）、implements 与 private（2.4 章）、计算属性名方法 [Symbol.asyncIterator]——而且它前面带 async *，也就是说这个方法本身就是一个异步生成器，类型签名里的 IteratorResult 正是本章反复出现的 { value, done }（shift()! 末尾的 ! 是非空断言，2.4 章 as 的近亲：刚检查过队列非空，告诉编译器这里取不出 undefined）。从源码结构看，这个 class 是一个「推转拉」的适配器：网络层每收到一块数据就调用 push 推进来；Agent 侧用 for await 按自己的节奏拉走；两头速度不一致时靠 queue 与 waiting 缓冲。同文件第 69 行起的子类 AssistantMessageEventStream 把 T 填成 AssistantMessageEvent——那是一个 2.3 章式的可辨识联合（types.ts 第 501–513 行），成员有 text_delta（文字增量）、toolcall_start、done 等，3.3 流式输出与5.4 流式事件如何传播到界面会顺着它讲完整条链路。

第二处：Provider 内部的生成器流水线。模型服务端通过 SSE（Server-Sent Events，一种服务器沿着 HTTP 连接持续推送文本消息的约定）送回数据，Pi 的 Anthropic 对接代码用两个异步生成器首尾相接来消化它：

packages/ai/src/api/anthropic-messages.ts · iterateAnthropicEvents

earendil-works/pi@c13ffe1第 446–485 行在 GitHub 查看 ↗

异步生成器消费另一个异步生成器：for await 读取 SSE 消息流，解析后 yield 出类型化的事件。

async function* iterateAnthropicEvents(
	response: Response,
	signal?: AbortSignal,
): AsyncGenerator<RawMessageStreamEvent> {
	// …（省略：response.body 缺失时抛错、完整性标记的初始化）
	for await (const sse of iterateSseMessages(response.body, signal)) {
		// …（省略：跳过无关事件、解析 JSON、解析失败时抛错）
		yield event;
	}
	// …（省略：流没有正常收尾时抛错）
}

注意它的结构：函数自己是 async function*，函数体里又用 for await 消费上游的 iterateSseMessages（同文件第 387 行，也是异步生成器：把网络字节流切成一条条 SSE 消息），解析后把类型化的事件 yield 给下游——下游在同文件第 573 行再用一个 for await 接住。异步生成器就这样像水管一样一节节接起来，每一节只做一种转换，数据一块进、一块出，全程没有任何一处「攒齐了再处理」。参数表里的 signal?: AbortSignal 是贯穿这条水管的取消信号，正是下一章的主角。

实践任务

🛠 实践任务用异步生成器做一个打字机labs/typescript-basics/09-async-iterators

目标：亲手写并观察本章全部机制：同步生成器与 for..of、「一次性等整句」与「逐词流式」的体验对比、async function* + for await..of 的打字机效果、提前 break 触发 finally 清理。实验目录：labs/typescript-basics/09-async-iterators（全书实验索引见实践任务索引）。

步骤：

1. 进入实验目录，安装依赖并运行：

   cd labs/typescript-basics/09-async-iterators
   npm install
   npm start

2. 盯着第 3 部分的输出：词是一个一个蹦出来的。对照 src/main.ts 里的 streamWords，指出「等一会儿」和「交出一个词」分别是哪一行。

3. 对比第 2、3 部分打印的毫秒数：总耗时几乎相同，「看到第一个词」的时间差了约 7 倍——用一句话说出流式输出到底改善了什么。

4. 把第 4 部分 break 的条件从 count === 3 改成 count === 5，预测输出再运行验证；然后把 streamWords 里的 finally 块整个删掉，观察少了哪两行输出，改回去。

5. 思考题：第 4 部分 break 之后，剩下四个词对应的 sleep 执行了吗？在 streamWords 的 await sleep(delayMs) 前加一行 console.log("生产中…") 数一数次数，验证你的答案。

预期现象：npm start 输出与实验目录下 expected-output.txt 一致（三处毫秒数每次运行略有浮动，属正常现象）：

== 第 1 部分：同步生成器热身（for..of） ==
倒计时: 3 2 1 发射

== 第 2 部分：没有流式——等够了才能看到第一个字 ==
等了 701ms 才看到第一个字: 异步迭代器是理解Pi流式输出的钥匙

== 第 3 部分：for await 逐词消费——打字机效果 ==
异步迭代器 是 理解 Pi 流式输出 的 钥匙
[streamWords] finally 执行：资源已清理
第一个词 101ms 就到了，整句 708ms 收完

== 第 4 部分：提前 break——消费者说停就停 ==
只要前三个词: 异步迭代器 是 理解
[streamWords] finally 执行：资源已清理
循环已退出：剩下的词不会再产生，对应的计时也不会再走

如何判断成功：肉眼看到第 3 部分逐词出现；能说出「finally 的两次输出」分别由什么触发（正常走完 / break）；第 5 步数出「生产中…」只出现 3 次。

常见错误：

• 第 3 部分一整行同时出现：多半是把输出重定向到了文件，或在会缓冲输出的 IDE 面板里运行——换标准终端直接 npm start；
• 在普通函数里写 for await 报错：它只能出现在 async 函数（或异步生成器）里；
• 把 for await (const word of streamWords(...)) 写成 for (const word of ...)：for..of 走的是同步协议，编译器会告诉你这个对象身上没有 Symbol.iterator。

对应源码位置：packages/ai/src/utils/event-stream.ts 的 EventStream（上一节的 SourceRef）——读懂了实验里的 streamWords，它就只是「多了个缓冲队列」的版本。

本章小结

• 迭代器协议只有两个要素：next() 方法与 { value, done } 结果；for..of 是替你循环调 next 的简写；生成器函数 function* 用 yield（可多次使用的 return，交值并暂停）来便捷地产出迭代器。
• 异步迭代器 = 同一套协议，next() 改为返回 Promise<IteratorResult>。它填上了「多个值 × 异步到来」那格空缺，正是大语言模型流式输出的数据形状。
• 生产端用 async function*（既能 await 又能 yield），消费端用 for await..of（每圈循环等一个值，只能写在 async 函数里）。
• 协议的入口是 Symbol.asyncIterator：for await 认门牌不认出身——生成器、手写对象、class 实例一律平等。
• 生成器是拉取式的：消费者 break 时，for await 自动调用生成器的 return，finally 里的清理代码保证执行，剩余的值根本不会被生产；需要主动喊停底层操作时，则要配合下一章的取消信号。
• Pi 中 EventStream 是 implements AsyncIterable + async *[Symbol.asyncIterator] 的「推转拉」适配器；Provider 内部用异步生成器首尾相接组成解析流水线。

关键术语：迭代器（Iterator）、迭代器协议、生成器（Generator）、yield、异步迭代器（Async Iterator）、异步生成器（async function*）、for await..of、Symbol.asyncIterator、AsyncIterable / AsyncIterator、拉取式（pull）、流式输出（Streaming）

关键源码索引：packages/ai/src/utils/event-stream.ts 的 EventStream / AssistantMessageEventStream；packages/ai/src/api/anthropic-messages.ts 的 iterateSseMessages / iterateAnthropicEvents；packages/ai/src/types.ts 的 AssistantMessageEvent

自测问题：

1. Promise<string[]> 和 AsyncIterable<string> 都能表示「多个字符串的异步结果」，它们在「什么时候能拿到第一个值」上有什么本质区别？各适合什么场景？
2. 手动调用异步生成器的 next() 会拿到什么类型的值？要经过什么操作才能得到 { value, done }？
3. for await 循环里 break 后，生成器暂停在 yield 处的代码会怎样？为什么说清理逻辑应该写在生成器的 finally 里而不是循环后面？
4. Pi 的 EventStream 为什么需要 queue 和 waiting 两个数组？提示：想一想「网络推数据的节奏」和「消费者拉数据的节奏」不一致时的两种情况。

下一章预告：2.7 事件、回调与取消（AbortController）——异步迭代器是拉取式的：不去要，就不会来。但有些事情天生是推送式的：用户敲下按键、子进程打出一行日志。下一章讲 Node.js 的 EventEmitter，以及贯穿 Pi 全部异步链路的「停止按钮」——AbortController 与 AbortSignal，你已经在 iterateAnthropicEvents 的参数表里见过它了。