张芷铭的个人博客

生成器

一、模型流式处理返回的 result 是否是生成器？

大概率是生成器（或类生成器对象，如异步生成器）。 模型流式返回（如大模型逐字输出、数据分批处理）的核心需求是“边生成边返回，避免一次性加载全部数据”，而生成器的“惰性计算、逐次产出”特性完美匹配这个场景——每次调用仅生成/返回一小段数据，既节省内存，又能实时响应。

除了原生生成器，也可能是 asyncio 异步生成器（async def + yield），或框架封装的类生成器对象（如 OpenAI SDK 的 AsyncStream、LangChain 的 Iterator），但核心逻辑与生成器一致。

二、Python 生成器的核心用法

生成器是一种“可迭代、可遍历但不可重复遍历”的对象，通过 yield 关键字定义，而非 return，分为生成器函数和生成器表达式两类：

1. 基础用法：生成器函数（最常用）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

# 定义：用 yield 替代 return，函数调用后返回生成器对象（不立即执行）
def stream_data():
    for i in range(3):
        # 暂停执行，返回值；下次调用时从此处继续
        yield f"流式数据 {i}"  

# 使用1：遍历（最推荐，自动触发 yield）
gen = stream_data()
for data in gen:
    print(data)  # 输出：流式数据 0 → 流式数据 1 → 流式数据 2

# 使用2：手动调用 next()（逐次获取，需捕获 StopIteration）
gen = stream_data()
print(next(gen))  # 流式数据 0
print(next(gen))  # 流式数据 1
print(next(gen))  # 流式数据 2
# next(gen)  # 无数据时抛出 StopIteration 异常

# 使用3：转为列表（一次性获取所有数据，失去流式优势）
print(list(stream_data()))  # ['流式数据 0', '流式数据 1', '流式数据 2']

2. 简化写法：生成器表达式（类似列表推导式）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# 语法：(表达式 for 变量 in 可迭代对象)，而非 []
gen = (x * 2 for x in range(3))
for num in gen:
    print(num)  # 0 → 2 → 4

# 对比列表推导式：生成器不占内存，列表一次性加载
import sys
lst = [x * 2 for x in range(10000)]
gen = (x * 2 for x in range(10000))
print(sys.getsizeof(lst))  # ~80040 字节（占用内存）
print(sys.getsizeof(gen))  # ~112 字节（仅保存生成逻辑）

3. 进阶：流式处理实战（模拟模型返回）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

def llm_stream_response(prompt):
    """模拟大模型流式返回结果"""
    response = "您好！很高兴为您解答问题。"
    for char in response:
        # 模拟模型逐字生成（可加延迟）
        yield char
        # time.sleep(0.1)

# 实时接收流式数据
for char in llm_stream_response("你好"):
    print(char, end="", flush=True)  # 逐字输出：您→好→！→...

4. 异步生成器（适配异步框架，如 FastAPI/AsyncIO）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

import asyncio

async def async_stream_data():
    """异步生成器：async def + yield"""
    for i in range(3):
        await asyncio.sleep(0.5)  # 模拟异步IO（如网络请求）
        yield f"异步流式数据 {i}"

# 异步遍历
async def main():
    async for data in async_stream_data():
        print(data)

asyncio.run(main())

三、生成器的关键注意事项

1. 核心特性：惰性计算 + 一次性遍历

  • 生成器不提前生成所有数据，仅在 next()/遍历/async for 时执行到 yield 才产出值；
  • 遍历一次后就“耗尽”，再次遍历无输出（需重新创建生成器对象）：
    1
2
3

    gen = stream_data()
list(gen)  # 第一次遍历：['流式数据 0', '流式数据 1', '流式数据 2']
list(gen)  # 第二次遍历：[]（已耗尽）

2. 异常处理：捕获 StopIteration

手动调用 next() 时，无数据会抛出 StopIteration，建议用 next(gen, 默认值) 规避：

1
2
3
4
5

gen = stream_data()
print(next(gen, "无数据"))  # 流式数据 0
print(next(gen, "无数据"))  # 流式数据 1
print(next(gen, "无数据"))  # 流式数据 2
print(next(gen, "无数据"))  # 无数据（避免报错）

3. 内存与性能：优势与限制

  • 优势：处理超大数据集（如百万行日志、大模型长文本）时，生成器仅占用固定内存，远优于列表；
  • 限制：不支持索引/切片（gen[0] 报错）、不支持 len()（无法直接获取长度），需遍历才能统计。

4. 避免“隐式耗尽”：警惕无意遍历

生成器是可迭代对象，若被 list()/sum()/max() 等函数一次性遍历，会直接耗尽，后续使用无数据：

1
2
3
4

gen = (x for x in range(3))
sum(gen)  # 0+1+2=3（已遍历耗尽）
for x in gen:
    print(x)  # 无输出

5. 线程/协程安全：生成器非线程安全

  • 单个生成器对象不能被多线程/多协程同时调用，否则会导致数据错乱；
  • 解决方案：为每个线程/协程创建独立的生成器实例。

6. yield vs yield from：简化嵌套生成器

若需调用另一个生成器，用 yield from 替代嵌套遍历，更简洁：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

def sub_gen():
    yield 1
    yield 2

def main_gen():
    yield 0
    yield from sub_gen()  # 等价于 for x in sub_gen(): yield x
    yield 3

list(main_gen())  # [0,1,2,3]

7. 生成器关闭：close() 与 GeneratorExit

可主动关闭生成器，触发 GeneratorExit 异常（常用于清理资源）：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

def gen_with_cleanup():
    try:
        yield 1
        yield 2
    except GeneratorExit:
        print("生成器被关闭，清理资源")

gen = gen_with_cleanup()
print(next(gen))  # 1
gen.close()       # 输出：生成器被关闭，清理资源
# next(gen)       # 抛出 StopIteration

四、总结

  1. 模型流式返回的 result 通常是生成器（或异步生成器），核心是“逐次产出、惰性计算”；
  2. 生成器核心用法：yield 函数、表达式、async for 异步遍历；
  3. 核心注意：一次性遍历、无索引/长度、手动调用需处理异常、非线程安全。

在实际开发中，生成器是处理流式数据（大模型输出、分批读取文件/数据库）的首选方案，既能提升响应速度，又能大幅降低内存占用。

💬 评论