张芷铭的个人博客

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

模型流式返回（如大模型逐字输出、数据分批处理）的核心需求是"边生成边返回"，生成器的"惰性计算、逐次产出"特性完美匹配此场景。

大概率是生成器或类生成器对象。除原生生成器外，也可能是 asyncio 异步生成器（async def + yield），或框架封装对象（如 OpenAI SDK 的 AsyncStream）。

生成器函数（最常用）

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

def stream_data():
    for i in range(3):
        yield f"流式数据 {i}"  # 暂停执行，返回值；下次调用时从此处继续

# 遍历（推荐）
gen = stream_data()
for data in gen:
    print(data)  # 流式数据 0 → 流式数据 1 → 流式数据 2

# 手动调用 next()
gen = stream_data()
print(next(gen))  # 流式数据 0
print(next(gen, "无数据"))  # 流式数据 1（带默认值避免 StopIteration）

# 转为列表（失去流式优势）
print(list(stream_data()))  # ['流式数据 0', '流式数据 1', '流式数据 2']

生成器表达式

1
2
3
4
5
6
7

# 语法：(表达式 for 变量 in 可迭代对象)
gen = (x * 2 for x in range(3))

# 内存对比
import sys
lst = [x * 2 for x in range(10000)]  # ~80040 字节
gen = (x * 2 for x in range(10000))  # ~112 字节（仅保存生成逻辑）

异步生成器

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

import asyncio

async def async_stream_data():
    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())

核心特性

特性说明
惰性计算仅在 next()/遍历时执行到 yield 才产出值
一次性遍历遍历后"耗尽"，再次遍历无输出
无索引/长度不支持 gen[0]、len(gen)
非线程安全单个生成器不能被多线程同时调用

实战：模拟模型流式返回

1
2
3
4
5
6
7
8

def llm_stream_response(prompt):
    """模拟大模型流式返回"""
    response = "您好！很高兴为您解答问题。"
    for char in response:
        yield char

for char in llm_stream_response("你好"):
    print(char, end="", flush=True)  # 逐字输出

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]

生成器关闭

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

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

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

生成器是处理流式数据（大模型输出、分批读取文件/数据库）的首选方案，兼顾响应速度与内存效率。

Comments