📖 Day 3 学习页:生成器与上下文管理器

Day 3:生成器、异步生成器与上下文管理器

🎯 今日目标:实现一个"逐字吐出回复"的流式 LLM 客户端。你平时看到的 ChatGPT 打字机效果、 Agent 流式输出,底层就是一个异步生成器。

📚 本日资源:📝 练习模板 | ✅ 验收脚本

⏱ 今日安排(约 2 小时)

时间内容方式
30min理论学习:yield 的暂停/恢复机制 + 上下文管理器阅读
30min编码理解:同步/异步生成器各写一遍编码
60min动手实操:完成练习模板 11 个 TODO编码

📖 第一步:理论学习

1. 生成器:能暂停的函数

普通函数 return 后就结束了;带 yield 的函数每次吐出一个值就暂停, 下次迭代从暂停处继续。好处:不一次性把所有数据放进内存,要多少吐多少。

def 计数器(上限):
    n = 0
    while n < 上限:
        yield n      # 吐出 n 并暂停,下次从这里继续
        n += 1

for i in 计数器(3):   # 0 1 2 —— 一次只存在一个值
    print(i)

两个关键性质:① 只能遍历一次( exhausted 就没了,需要重复用就转 list()); ② 惰性求值(没人要就不计算——处理大文件/长流式响应不爆内存的根本原因)。

2. 异步生成器 = async def + yield(LLM 流式输出的原理)

LLM API 是一个 token 一个 token 返回的。客户端用异步生成器逐块接收,界面就能打字机式刷新:

import asyncio
from typing import AsyncGenerator

async def 流式响应(提示词: str) -> AsyncGenerator[str, None]:
    # 注意:async def 里必须有 yield,才是"异步生成器"
    for chunk in [提示词[:i+1] for i in range(len(提示词))]:
        await asyncio.sleep(0.1)   # 模拟网络到达一个 chunk
        yield chunk

async def main():
    async for chunk in 流式响应("你好世界"):   # 用 async for 消费
        print(chunk, end="", flush=True)

⚠️ 高频错误:async def 函数体里没有 yield,它就不是异步生成器, 而是普通协程——用 async for 迭代它会直接 TypeError。练习模板里特意留了这个坑给你踩。

3. 上下文管理器:自动收尾的资源管家

with 语句的底层。四种写法按需选择:

from contextlib import contextmanager, asynccontextmanager
import time

# 写法 1:类实现(最完整,能保存状态)
class 计时器:
    def __enter__(self):
        self.开始 = time.time()
        return self                      # 约定:返回 self
    def __exit__(self, exc_type, exc_val, exc_tb):
        print(f"耗时 {time.time()-self.开始:.3f}s")
        return False                     # False = 异常继续往外抛

# 写法 2:@contextmanager(简单的场景两行搞定)
@contextmanager
def 计时器简版(名称):
    开始 = time.time()
    yield                                # ← yield 前是 __enter__,后是 __exit__
    print(f"{名称} 耗时 {time.time()-开始:.3f}s")

with 计时器简版("查询"):        # 用法和写法 1 完全一样
    time.sleep(0.5)

# 异步版:__aenter__/__aexit__ 或 @asynccontextmanager(把 time.sleep 换成 await asyncio.sleep)

4. 组合技:今天的终极目标

把「异步生成器(流式)」+「异步上下文管理器(自动计时)」组合成一个 流式LLM客户端——async with 计时,async for 逐块 yield。这就是 Day 5/10 里 Agent 流式输出的雏形。

🛠 第二步:动手实操

  1. 打开模板:day03_practice_template.py,11 个 TODO:
    • 任务 1/2:同步、异步流式生成器(照本页示例写,注意异步版函数体里必须有 yield)
    • 任务 3/4:四种上下文管理器(类版 __enter__ 要 return self)
    • 任务 5:流式LLM客户端.chat() —— 签名已经是 async def,你要在方法体里加 yield 并用 async with 包计时器
  2. 运行自测:python day03_practice_template.py,能看到逐块打字机输出。
  3. 验收:python day03_practice_validator.py —— 验收会真的迭代你的生成器数 chunk 数、 真的进入你的 with 块检查 __enter__ 是否返回 self。

⚠️ 避坑

  • ❌ 生成器只能遍历一次,第二次遍历是空的;要重复使用先转 list
  • ❌ async def 忘写 yield → 不是异步生成器,async for 报 TypeError
  • ❌ @contextmanager 版本忘写 yield → RuntimeError: generator didn't yield
  • ✅ 大文件/流式数据必须用生成器,一次性读进内存会爆

🔗 延伸资源