13.4 Python 异步编程实战
生活类比：你去食堂打饭，同步操作是排一个窗口等前面的人打完你再打；异步操作是同时排三个窗口——先在 A 窗口点菜，趁师傅炒菜的空隙去 B 窗口拿饮料，再趁等饮料的空隙去 C 窗口拿餐具。三个窗口几乎同时完成，总时间只比最慢的那个多一点。
Agent 开发中最常见的性能瓶颈就是"等 IO"——等大模型 API 返回、等数据库查询、等外部接口响应。异步编程让这些等待不再串行，而是并发进行。本节从 asyncio 基础讲到 Agent 实战中的并发 LLM 调用。
13.4.1 asyncio 快速入门
python
import asyncio
# 定义协程：async def 声明这是一个异步函数
async def say_hello(name: str) -> str:
await asyncio.sleep(1) # 模拟耗时 IO（比如网络请求），await 表示"等它完成"
return f"Hello, {name}!"
# 并发执行多个协程
async def main():
# asyncio.gather 同时启动多个协程，等它们全部完成
results = await asyncio.gather(
say_hello("Alice"), # 这三个调用不是排队执行
say_hello("Bob"), # 而是同时开始
say_hello("Charlie") # 总耗时约 1 秒，而非 3 秒
)
print(results) # ['Hello, Alice!', 'Hello, Bob!', 'Hello, Charlie!']
# 运行事件循环
asyncio.run(main()) # asyncio.run 是程序入口，启动事件循环为什么并发后只要 1 秒？ 因为 asyncio.sleep(1) 本质是"告诉事件循环：我这里要等 1 秒，你先去忙别的"。事件循环就趁这 1 秒去推进其他协程。如果是真正的 CPU 计算（不是 IO），asyncio 不能加速——它是给 IO 密集型任务设计的。
13.4.2 异步 HTTP 客户端：httpx
python
import httpx
import asyncio
async def fetch_url(client: httpx.AsyncClient, url: str) -> dict:
"""异步获取一个 URL 的内容"""
try:
# await 等待 HTTP 响应，设置 30 秒超时
response = await client.get(url, timeout=30.0)
return {
"url": url,
"status": response.status_code, # HTTP 状态码
"size": len(response.text) # 响应体大小（字符数）
}
except httpx.TimeoutException: # 超时
return {"url": url, "error": "timeout"}
except Exception as e: # 其他异常
return {"url": url, "error": str(e)}
async def fetch_all(urls: list[str]) -> list[dict]:
"""并发获取多个 URL——核心在于复用同一个 client"""
# async with 确保用完关闭连接池
async with httpx.AsyncClient() as client:
# 为每个 URL 创建一个协程任务
tasks = [fetch_url(client, url) for url in urls]
# gather 并发执行所有任务，返回结果列表
return await asyncio.gather(*tasks) # *tasks 展开为多个参数13.4.3 并发调用 LLM API：Agent 的核心性能优化
这是 Agent 开发中最实用的一段代码。假设你要批量处理 100 条用户消息，串行调用要等 100 次 LLM 返回，并发调用可以同时发 5-10 个请求，速度提升数倍。
python
from openai import AsyncOpenAI # 异步版本的 OpenAI 客户端
import asyncio
import time
client = AsyncOpenAI() # 初始化异步客户端
async def call_llm(prompt: str, model: str = "gpt-4o-mini") -> dict:
"""单次异步 LLM 调用"""
start = time.time() # 记录开始时间
try:
# 异步调用 chat completions API
response = await client.chat.completions.create(
model=model, # 指定模型
messages=[{"role": "user", "content": prompt}], # 用户消息
temperature=0.7 # 采样温度
)
elapsed = time.time() - start # 计算耗时
return {
"prompt": prompt[:50], # 截取前 50 字符做标识
"answer": response.choices[0].message.content[:100], # 回复前 100 字
"tokens": response.usage.total_tokens, # 消耗的 token 数
"time": elapsed # 耗时（秒）
}
except Exception as e:
return {"prompt": prompt[:50], "error": str(e)}
async def batch_llm_calls(prompts: list[str], max_concurrent: int = 5) -> list[dict]:
"""
使用信号量控制并发的批量 LLM 调用
为什么要控制并发？因为 API 有 rate limit（速率限制），
同时发太多请求会被 429 拒绝。Semaphore 就像红绿灯——
最多放行 max_concurrent 个请求同时通过。
"""
semaphore = asyncio.Semaphore(max_concurrent) # 创建信号量
async def bounded_call(prompt: str):
async with semaphore: # 获取信号量（满了就等）
return await call_llm(prompt) # 执行调用
# 离开 with 块后自动释放信号量，下一个等待者可以进入
tasks = [bounded_call(p) for p in prompts] # 为每个 prompt 创建任务
return await asyncio.gather(*tasks) # 并发执行
# 性能对比：串行 vs 并发
async def compare_performance():
prompts = [f"用一句话介绍{i}" for i in range(10)] # 10 个测试 prompt
# 串行调用：一个接一个，总耗时 = 10 x 单次耗时
start = time.time()
for p in prompts:
await call_llm(p) # 串行：等前一个完成才发下一个
serial_time = time.time() - start
# 并发调用：同时发 5 个请求
start = time.time()
await batch_llm_calls(prompts, max_concurrent=5) # 并发：5 路同时进行
parallel_time = time.time() - start
print(f"串行: {serial_time:.1f}s, 并发: {parallel_time:.1f}s, "
f"加速: {serial_time/parallel_time:.1f}x")
# 典型输出：串行: 15.2s, 并发: 3.4s, 加速: 4.5x13.4.4 异步流式 LLM 调用
流式调用让用户看到"逐字输出"的效果，而不是等 10 秒后一次性蹦出一大段文字。这是提升用户体验的关键。
python
async def stream_llm(prompt: str) -> str:
"""异步流式调用 LLM——逐 chunk 接收"""
full_response = "" # 累积完整回复
# stream=True 开启流式模式
stream = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
stream=True # 关键参数：流式返回
)
# 异步遍历流中的每个 chunk
async for chunk in stream: # 每个 chunk 包含一小段文本
if chunk.choices[0].delta.content: # delta.content 是增量内容
content = chunk.choices[0].delta.content
full_response += content # 累积到完整回复
print(content, end="", flush=True) # 立即打印，不换行
print() # 最后换行
return full_response13.4.5 LLM 并发调用池：生产级封装
把前面的知识封装成一个可复用的并发调用池，支持统计、并发控制和错误处理：
python
import asyncio
from openai import AsyncOpenAI
class LLMPool:
"""
LLM 并发调用池
就像银行窗口：有 N 个窗口（max_concurrent），
客户排队取号，空出窗口就放一个进去。
同时记录每天的办理数量、成功/失败次数。
"""
def __init__(self, max_concurrent: int = 5, model: str = "gpt-4o-mini"):
self.client = AsyncOpenAI() # 异步客户端
self.semaphore = asyncio.Semaphore(max_concurrent) # 信号量控制并发
self.model = model # 默认模型
self.stats = { # 运行统计
"total": 0, # 总调用次数
"success": 0, # 成功次数
"error": 0, # 失败次数
"total_tokens": 0 # 总 token 消耗
}
async def call(self, messages: list, **kwargs) -> dict:
"""单次调用（带并发控制）"""
async with self.semaphore: # 获取信号量（控制并发）
self.stats["total"] += 1 # 计数
try:
response = await self.client.chat.completions.create(
model=self.model, # 模型
messages=messages, # 消息列表
**kwargs # 其他参数（temperature 等）
)
self.stats["success"] += 1 # 成功计数
self.stats["total_tokens"] += response.usage.total_tokens # token 统计
return {
"content": response.choices[0].message.content, # 回复内容
"tokens": response.usage.total_tokens # token 数
}
except Exception as e:
self.stats["error"] += 1 # 失败计数
return {"error": str(e)}
async def batch_call(self, messages_list: list) -> list:
"""批量调用：对一组消息列表并发处理"""
tasks = [self.call(msgs) for msgs in messages_list] # 创建所有任务
return await asyncio.gather(*tasks) # 并发执行
# 使用示例
# pool = LLMPool(max_concurrent=3)
# messages_list = [
# [{"role": "user", "content": f"用一句话介绍{i}"}]
# for i in range(5)
# ]
# results = await pool.batch_call(messages_list)
# print(pool.stats) # 查看统计：总调用数、成功率、token 消耗13.4.6 带超时和重试的异步调用
生产环境中，网络不稳定、API 偶尔抽风是常态。超时和重试是必备的"保险措施"：
python
async def call_with_timeout(prompt: str, timeout: float = 30.0) -> dict:
"""带超时的 LLM 调用——超过 30 秒就不等了"""
try:
# asyncio.wait_for 给协程加一个超时限制
return await asyncio.wait_for(call_llm(prompt), timeout=timeout)
except asyncio.TimeoutError:
return {"prompt": prompt[:50], "error": "timeout"}
async def call_with_retry(prompt: str, max_retries: int = 3) -> dict:
"""
带指数退避重试的调用
指数退避：第一次失败等 1 秒重试，第二次失败等 2 秒，第三次等 4 秒。
就像打电话没人接：第一次等 1 分钟再打，还没人接等 2 分钟，再没人接等 4 分钟。
越等越久，给对方（服务器）喘息恢复的时间。
"""
for attempt in range(max_retries):
result = await call_llm(prompt)
if "error" not in result: # 成功就返回
return result
wait = 2 ** attempt # 指数退避：1, 2, 4 秒
print(f"重试 {attempt + 1}/{max_retries}, 等待 {wait}s")
await asyncio.sleep(wait) # 等待后重试
return {"error": "all retries failed"} # 全部重试失败13.4.7 综合实战：带超时 + 重试 + 并发控制的完整方案
python
import asyncio
from openai import AsyncOpenAI
class RobustLLMCaller:
"""
生产级 LLM 调用器：集并发控制、超时、重试于一体
就像一个靠谱的外卖系统：
- 并发控制：同时最多派 N 个骑手
- 超时：30 分钟没送到就标记超时
- 重试：送失败了换个骑手再试，最多试 3 次
"""
def __init__(self, max_concurrent: int = 5, timeout: float = 30.0,
max_retries: int = 3, model: str = "gpt-4o-mini"):
self.client = AsyncOpenAI()
self.semaphore = asyncio.Semaphore(max_concurrent) # 并发控制
self.timeout = timeout # 超时时间
self.max_retries = max_retries # 最大重试次数
self.model = model
async def call(self, messages: list, **kwargs) -> dict:
"""带全套保护的调用"""
async with self.semaphore: # 控制并发
for attempt in range(self.max_retries): # 重试循环
try:
# 把 API 调用包装在超时里
response = await asyncio.wait_for(
self.client.chat.completions.create(
model=self.model,
messages=messages,
**kwargs
),
timeout=self.timeout
)
return {
"content": response.choices[0].message.content,
"tokens": response.usage.total_tokens,
"attempts": attempt + 1
}
except asyncio.TimeoutError: # 超时
if attempt == self.max_retries - 1:
return {"error": "timeout", "attempts": self.max_retries}
await asyncio.sleep(2 ** attempt) # 指数退避
except Exception as e: # 其他错误
if attempt == self.max_retries - 1:
return {"error": str(e), "attempts": self.max_retries}
await asyncio.sleep(2 ** attempt)
return {"error": "unknown", "attempts": self.max_retries}
async def batch_call(self, messages_list: list, **kwargs) -> list:
"""批量调用"""
tasks = [self.call(msgs, **kwargs) for msgs in messages_list]
return await asyncio.gather(*tasks)
# 使用示例
# caller = RobustLLMCaller(max_concurrent=5, timeout=30, max_retries=3)
# messages_list = [
# [{"role": "user", "content": "你好"}],
# [{"role": "user", "content": "介绍下 Python"}],
# ]
# results = await caller.batch_call(messages_list)常见误区
- "async 一定比同步快"：async 的优势在 IO 等待时可以切换去做别的。如果是 CPU 密集型任务（大量计算），async 不会加速，反而因为协程切换有微小开销。
- "在 async 函数里用了 requests 库"：
requests是同步库，在 async 函数里调用它会阻塞整个事件循环。必须用httpx或aiohttp等异步库。 - "并发数越大越好"：并发太高会触发 API 的 rate limit（429 错误），反而导致大量重试。一般 5-10 的并发数比较安全。
- "忘了 asyncio.run"：直接调用
async def函数只会返回一个协程对象，不会执行。必须用asyncio.run()或在已有事件循环中await。 - "在 Jupyter 里直接 asyncio.run"：Jupyter 已经有自己的事件循环，直接调
asyncio.run()会报错。在 Jupyter 里直接await即可。
本节小结
| 要点 | 说明 |
|---|---|
async/await | 协程语法，await 挂起等待 IO，事件循环趁机执行其他协程 |
asyncio.gather | 并发执行多个协程，等全部完成 |
Semaphore | 控制并发上限，防止触发 API rate limit |
httpx.AsyncClient | 异步 HTTP 客户端，复用连接池提升性能 |
AsyncOpenAI | OpenAI 异步客户端，支持并发和流式调用 |
| 指数退避 | 失败重试时等待时间按 2 的幂次增长，给服务器恢复时间 |
asyncio.wait_for | 给协程加超时限制，防止无限等待 |
核心心法：Agent 开发中 80% 的性能优化来自把"串行调用 LLM"改成"并发调用 LLM"。先把这个改了，再去想其他优化。
延伸阅读
- Python asyncio 官方文档：https://docs.python.org/3/library/asyncio.html
- OpenAI Async Python Client：https://github.com/openai/openai-python#async-usage
- httpx 异步文档：https://www.python-httpx.org/async/