10.1 迭代器与生成器
目标
- 理解迭代协议：可迭代对象（
__iter__）与迭代器（__next__、StopIteration）。 - 读懂含
yield的生成器函数，理解它“暂停—恢复”的执行方式。 - 知道惰性求值的好处与“只能遍历一次”的限制；认识
yield from。
核心概念
迭代协议
for x in obj 背后发生的事：
- 调用
iter(obj)（即obj.__iter__()）得到一个迭代器； - 反复调用
next(迭代器)（即__next__()）取下一个值； - 迭代器抛出
StopIteration时循环结束。
Python · 可运行
python
items = ["a", "b"]
it = iter(items) # list 是可迭代对象，iter() 返回它的迭代器
print(next(it), next(it))
try:
next(it)
except StopIteration:
print("耗尽：StopIteration")
# for 循环的等价展开
it = iter(items)
while True:
try:
x = next(it)
except StopIteration:
break
print("取到", x)生成器函数
函数体中出现 yield，调用它就不会执行函数体，而是返回一个生成器对象（一种迭代器）。每次 next() 执行到下一个 yield 并暂停，下次从暂停处继续。
Python · 可运行
python
def countdown(n):
print("开始")
while n > 0:
yield n # 产出一个值并暂停
n -= 1
print("结束")
gen = countdown(3)
print(type(gen).__name__) # generator：函数体还没执行
print(next(gen)) # 开始 → 3
print(list(gen)) # 继续执行剩余部分：[2, 1]，并打印“结束”惰性求值
生成器按需产出，不需要先把所有结果放进内存；可以表示无限序列，也可以把处理步骤串成流水线：
Python · 可运行
python
def read_lines(text):
for line in text.splitlines():
yield line
def non_empty(lines):
for line in lines:
if line.strip():
yield line.strip()
def numbered(lines):
for i, line in enumerate(lines, 1):
yield f"{i}: {line}"
text = "第一行\n\n第二行\n \n第三行"
pipeline = numbered(non_empty(read_lines(text))) # 此时什么都还没执行
for row in pipeline:
print(row)yield from 与生成器表达式
Python · 可运行
python
def walk(tree):
"""深度优先遍历嵌套列表。"""
for node in tree:
if isinstance(node, list):
yield from walk(node) # 把子生成器的值逐个转交出去
else:
yield node
print(list(walk([1, [2, [3, 4]], 5])))
squares = (n * n for n in range(5)) # 生成器表达式（4.4）
print(sum(squares))自定义可迭代类
Python · 可运行
python
class Conversation:
def __init__(self, messages):
self.messages = messages
def __iter__(self): # 用生成器实现 __iter__：最简洁的写法
for m in self.messages:
yield m["role"], m["content"]
conv = Conversation([{"role": "user", "content": "hi"}, {"role": "assistant", "content": "hello"}])
for role, content in conv:
print(role, content)
print(dict(conv)) # 任何接受可迭代对象的函数都能用易错点
- 生成器只能遍历一次：
list(gen)后再遍历为空。需要多次遍历时先转成list。 - 调用生成器函数不会执行函数体：
countdown(3)不打印“开始”，必须迭代它。 - 对生成器用
len()或下标：抛TypeError；先list()。 - 在
return和yield混用时：生成器中的return 值不会作为普通返回值返回，而是结束迭代（值附在StopIteration.value上）。
Python · 可运行演示报错
python
gen = (n for n in range(3))
print(len(gen))对照
| 概念 | Python | JS | Java |
|---|---|---|---|
| 迭代协议 | __iter__ / __next__ + StopIteration | [Symbol.iterator]() / next() 返回 {value, done} | Iterable / Iterator.hasNext() |
| 生成器 | def f(): yield x | function* f() { yield x } | 无（Stream 部分替代） |
| 委托 | yield from g | yield* g | — |
| 惰性流水线 | 生成器链 | 生成器链 / iterator helpers | Stream |
读代码
SDK 的流式接口就是迭代器：模型逐段生成文本，客户端逐段产出事件。典型用法（示意）：
python
with client.messages.stream(model=model, max_tokens=1024, messages=messages) as stream:
for text in stream.text_stream: # 每次迭代得到新生成的一小段文本
print(text, end="", flush=True)
final = stream.get_final_message()读到 for x in something: 而 something 不是列表时，问：它每次产出什么？什么时候结束？能遍历几次？
minimal_agent.py 中的 for turn in range(...)、for block in resp.content、推导式里的 for b in resp.content 都依赖迭代协议：range 是惰性的可迭代对象，resp.content 是列表。
练习
练习 1：分块生成器
实现生成器函数 chunks(text, size)：按 size 个字符依次 yield 文本片段（最后一段可能更短）。
练习
def chunks(text, size):
return [text]
快捷键：⌘/Ctrl + Enter 提交检查
练习 2：模拟流式输出
events 是事件列表，每个事件是 {"type": ..., "text": ...}。实现生成器 text_stream(events)：只产出 type == "text_delta" 的 text；遇到 type == "message_stop" 立即结束（即使后面还有事件）。
练习
def text_stream(events):
for e in events:
yield e["text"]
快捷键：⌘/Ctrl + Enter 提交检查
小结
for背后是iter()+next()+StopIteration。- 含
yield的函数返回生成器：按需执行、暂停恢复、只能遍历一次。 - 生成器适合流水线与流式数据；
yield from转交子生成器。