Python 异步编程学习笔记：生成器、异步生成器、async for、asyncio.wait 与 timeout

0. 为什么大模型对话需要异步？

在大模型对话里，尤其是流式输出时，服务端往往不会一次性返回完整答案，而是不断返回一小段内容。

例如 SSE（Server-Sent Events）场景中，模型可能按 token 或 chunk 持续返回：

1
2
3
4
5
data: 你好
data: ，
data: 我是
data: AI
data: [DONE]

这种数据流的特点是：

  1. 数据不是一次性准备好的；
  2. 每次获取下一段数据都可能需要等待网络；
  3. 等待期间不能阻塞整个程序；
  4. 最终适合用异步迭代的方式处理。

所以学习 Python 异步时，可以重点掌握下面几类工具：

  1. 生成器；
  2. 异步生成器；
  3. 异步循环 async for；
  4. asyncio.wait()；
  5. async with asyncio.timeout()。

1. 生成器：同步的数据流

生成器是使用 yield 的函数。

它的特点是：不会一次性返回所有结果，而是每次产出一个值。

1
2
3
4
5
6
def gen():
    yield "hello"
    yield "world"

for item in gen():
    print(item)

输出：

1
2
hello
world

生成器适合表示“逐步产生的数据”。

比如：

1
2
3
4
def fake_tokens():
    yield "你"
    yield "好"
    yield "呀"

这很像大模型 token 流，但它是同步的，不能在里面直接 await 网络请求。

生成器的核心理解：

1
yield = 产出一个值，然后暂停，下次继续

2. 异步生成器：异步数据流的常用写法

异步生成器是：

1
2
3
async def 函数名():
    await ...
    yield ...

例子：

1
2
3
4
5
6
7
8
9
10
11
import asyncio

async def async_tokens():
    await asyncio.sleep(1)
    yield "你"

    await asyncio.sleep(1)
    yield "好"

    await asyncio.sleep(1)
    yield "呀"

它和普通生成器最大的区别是：

普通生成器只能同步产出数据；

1
2
def gen():
    yield value

异步生成器可以一边等待异步操作，一边产出数据：

1
2
3
async def async_gen():
    await something()
    yield value

这非常适合网络流式数据，比如 SSE、WebSocket、大模型 token streaming。

3. 手写异步迭代器 vs 异步生成器

传统异步迭代器需要手写 __aiter__() 和 __anext__()。

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

class AsyncCounter:
    def __init__(self, n):
        self.n = n
        self.current = 0

    def __aiter__(self):
        return self

    async def __anext__(self):
        if self.current >= self.n:
            raise StopAsyncIteration

        await asyncio.sleep(1)
        self.current += 1
        return self.current

使用：

1
2
async for item in AsyncCounter(3):
    print(item)

这套写法比较麻烦。

大多数情况下，可以用异步生成器替代：

1
2
3
4
5
6
import asyncio

async def async_counter(n):
    for i in range(1, n + 1):
        await asyncio.sleep(1)
        yield i

使用方式一样：

1
2
async for item in async_counter(3):
    print(item)

所以可以这样记：

1
2
简单异步数据流：优先用异步生成器
复杂状态对象：再考虑手写异步迭代器

异步生成器可以看作是手写异步迭代器的简洁替代方案，但不是所有场景的绝对上位替代。

如果对象内部有复杂状态、多个控制方法、重置逻辑、资源管理逻辑，手写异步迭代器类仍然更合适。

4. 异步循环：async for

async for 是用来遍历异步迭代器的循环语法。

普通循环：

1
2
for item in iterable:
    ...

异步循环：

1
2
async for item in async_iterable:
    ...

完整例子：

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

async def async_tokens():
    for token in ["你", "好", "呀"]:
        await asyncio.sleep(1)
        yield token

async def main():
    async for token in async_tokens():
        print(token)

asyncio.run(main())

async for 的特点是：每次获取下一个值时，都可能需要 await。

它大概等价于：

1
2
3
4
5
6
7
8
9
iterator = async_iterable.__aiter__()

while True:
    try:
        item = await iterator.__anext__()
    except StopAsyncIteration:
        break

    print(item)

所以它和普通 for 的区别是：

1
2
for       -> next(iterator)
async for -> await iterator.__anext__()

适合 async for 的场景：

  1. SSE 流式响应；
  2. WebSocket 消息；
  3. 异步下载；
  4. 异步数据库游标；
  5. 大模型 token streaming；
  6. 网络请求分块读取。

5. SSE / 大模型流式对话中的异步生成器

大模型流式输出很适合抽象成异步生成器。

伪代码：

1
2
3
4
5
6
async def stream_llm_response():
    async for event in sse_client:
        if event.data == "[DONE]":
            break

        yield event.data

使用：

1
2
3
async def main():
    async for chunk in stream_llm_response():
        print(chunk, end="")

这就形成了很自然的数据流：

1
SSE 网络事件 -> async for 接收 -> yield chunk -> 外部 async for 消费

在 FastAPI 里也常见类似结构：

1
2
3
async def generate():
    async for chunk in stream_llm_response():
        yield f"data: {chunk}\n\n"

也就是说，异步生成器是连接“大模型流式接口”和“上层业务代码”的常见桥梁。

6. asyncio.wait()：等待多个任务

asyncio.wait() 用来等待一组任务。

它的基本形式是：

1
done, pending = await asyncio.wait(tasks)

返回两个集合：

1
2
done     # 已完成任务
pending  # 未完成任务

例子：

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

async def request(name, delay):
    await asyncio.sleep(delay)
    return f"{name} finished"

async def main():
    tasks = [
        asyncio.create_task(request("A", 1)),
        asyncio.create_task(request("B", 2)),
        asyncio.create_task(request("C", 3)),
    ]

    done, pending = await asyncio.wait(tasks)

    for task in done:
        print(task.result())

asyncio.run(main())

asyncio.wait() 适合处理“多个任务的状态”。

常见参数：

1
2
3
4
5
done, pending = await asyncio.wait(
    tasks,
    timeout=3,
    return_when=asyncio.FIRST_COMPLETED
)

return_when 常见取值：

1
2
3
asyncio.ALL_COMPLETED      # 默认，等全部完成
asyncio.FIRST_COMPLETED    # 有一个完成就返回
asyncio.FIRST_EXCEPTION    # 有一个异常就返回

需要注意：

asyncio.wait() 返回后，pending 里的任务不会自动取消。

如果不想让它们继续运行，需要手动取消：

1
2
3
4
for task in pending:
    task.cancel()

await asyncio.gather(*pending, return_exceptions=True)

所以可以这样记：

1
asyncio.wait() 不是直接拿结果，而是查看一组任务的完成状态。

Python 官方文档也把 wait() 归类为“monitor for completion”，而 gather() 更偏向“并发调度并等待结果”。官方高层 API 索引中也说明 wait()、gather()、wait_for()、timeout() 都属于任务与超时相关 API。

7. asyncio.wait() 和 gather() 的区别

gather() 更适合：

1
我想并发运行多个任务，并拿到所有结果。
1
results = await asyncio.gather(task1, task2, task3)

wait() 更适合：

1
我想知道哪些任务完成了，哪些还没完成。
1
done, pending = await asyncio.wait(tasks)

对比：

1
2
gather() -> 返回结果列表
wait()   -> 返回 done / pending 任务集合

所以：

1
2
3
4
5
要结果：gather()
要状态：wait()
谁先完成就处理谁：as_completed()
要给单个 awaitable 加超时：wait_for()
要给一段代码加超时：asyncio.timeout()

8. async with asyncio.timeout()：超时控制

Python 3.11 新增了 asyncio.timeout()，它是一个异步上下文管理器。

写法：

1
2
async with asyncio.timeout(3):
    await something()

意思是：这段异步代码最多运行 3 秒。

完整例子：

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

async def slow_task():
    await asyncio.sleep(5)
    return "done"

async def main():
    try:
        async with asyncio.timeout(3):
            result = await slow_task()
            print(result)
    except TimeoutError:
        print("timeout")

asyncio.run(main())

输出：

1
timeout

它适合限制一整段异步逻辑：

1
2
3
4
5
async with asyncio.timeout(10):
    await connect()
    await send_request()
    async for chunk in stream:
        print(chunk)

这比 asyncio.wait_for() 更适合包住多个 await 操作。

9. asyncio.timeout() 和 wait_for() 的区别

asyncio.wait_for() 是给一个 awaitable 加超时：

1
await asyncio.wait_for(fetch(), timeout=3)

asyncio.timeout() 是给一段异步代码块加超时：

1
2
3
4
async with asyncio.timeout(3):
    await step1()
    await step2()
    await step3()

所以可以这样记：

1
2
wait_for()        -> 管一个 awaitable
asyncio.timeout() -> 管一段 async 代码块

在大模型流式场景里，asyncio.timeout() 经常更舒服，因为流式过程往往不是一个单独的 await，而是多个步骤加一个 async for。

例如：

1
2
3
4
5
6
7
async def consume_stream(stream):
    try:
        async with asyncio.timeout(30):
            async for chunk in stream:
                print(chunk, end="")
    except TimeoutError:
        print("stream timeout")

10. 大模型 SSE 场景中的整体结构

可以把一次大模型流式对话理解成：

1
2
3
4
5
6
7
1. 发起 HTTP 请求
2. 服务端通过 SSE 持续返回数据
3. 客户端 async for 一段一段读取
4. 每读到一个 chunk，就 yield 给上层
5. 上层再把 chunk 推给前端或终端
6. 外层用 timeout 控制超时
7. 多个模型请求时，可以用 wait / gather 管理任务

伪代码：

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

async def llm_sse_stream():
    async with asyncio.timeout(60):
        async for event in sse_client:
            if event.data == "[DONE]":
                break

            yield event.data

async def main():
    async for chunk in llm_sse_stream():
        print(chunk, end="")

如果同时请求多个模型：

1
2
3
4
5
6
7
8
9
tasks = [
    asyncio.create_task(call_model_a()),
    asyncio.create_task(call_model_b()),
]

done, pending = await asyncio.wait(
    tasks,
    return_when=asyncio.FIRST_COMPLETED
)

比如“谁先返回就用谁的结果”，这时 asyncio.wait() 就很合适。

11. 最终记忆版

可以按这条线记：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
生成器：
def + yield
同步地一段一段产出数据。

异步生成器：
async def + yield
可以 await 网络、数据库、SSE，然后一段一段产出数据。

async for：
遍历异步迭代器。
每次取下一个值时，背后可能发生 await。

asyncio.wait：
等待一组任务，返回 done 和 pending。
适合看多个任务的完成状态。

asyncio.timeout：
Python 3.11+ 的超时上下文管理器。
适合限制一整段异步代码块。

对于大模型流式对话，最重要的是：

1
2
3
4
5
SSE / token streaming 本质上就是一个异步数据流。
异步生成器负责生产 chunk。
async for 负责消费 chunk。
asyncio.timeout 负责控制超时。
asyncio.wait / gather 负责管理多个并发任务。
许可协议

本文采用 署名-非商业性使用-相同方式共享 4.0 国际 许可协议，转载请注明出处。

快来参与讨论吧~

本站由 楠瓜 使用 Stellar 1.33.1 主题创建。
风起于青萍之末，浪成于微澜之间。