第 1 章 · Python 异步编程与 FastAPI 生产进阶
本章目标：
- 掌握 asyncio 核心概念：event loop、Task、gather、async with、async for
- 理解 aiohttp 异步 HTTP 客户端，用于 LLM API 调用
- 实现生产级限流、重试（tenacity）和熔断模式
- 区分 BackgroundTasks 与消息队列的适用边界
1.1 asyncio 核心概念
Python 的 asyncio 是单线程并发模型，通过 event loop 调度协程。理解以下概念是构建高并发 Agent 服务的基础：
python
import asyncio
async def fetch_data(name: str, delay: float) -> str:
"""模拟异步数据获取"""
await asyncio.sleep(delay)
return f"{name}: {delay}s"
async def main():
# 创建多个任务并发执行
tasks = [fetch_data("A", 0.5), fetch_data("B", 0.3), fetch_data("C", 0.8)]
# gather 并发运行所有任务，返回结果列表
results = await asyncio.gather(*tasks)
for r in results:
print(r)
# 单个任务创建与等待
task = asyncio.create_task(fetch_data("D", 1.0))
result = await task
print(result)
asyncio.run(main())关键 API：
asyncio.run(coro)：运行顶层协程asyncio.create_task(coro)：调度协程并发执行asyncio.gather(*coros)：并发运行多个协程，收集结果asyncio.wait_for(coro, timeout)：带超时的等待async with/async for：异步上下文管理与迭代
1.2 aiohttp：异步 HTTP 客户端
Agent 服务需要异步调用 LLM API（如 Vercel AI Gateway），aiohttp 是首选：
python
import aiohttp
import asyncio
async def call_llm_async(prompt: str) -> str:
"""异步调用 LLM（示例用 httpbin 替代实际 API）"""
async with aiohttp.ClientSession() as session:
async with session.post(
"https://httpbin.org/post",
json={"prompt": prompt},
timeout=aiohttp.ClientTimeout(total=30)
) as resp:
data = await resp.json()
return data.get("json", {}).get("prompt", "")
# 并发调用多个 LLM
async def main():
prompts = ["你好", "解释量子计算", "写一首诗"]
tasks = [call_llm_async(p) for p in prompts]
results = await asyncio.gather(*tasks, return_exceptions=True)
for r in results:
print(r if isinstance(r, str) else f"错误: {r}")
asyncio.run(main())1.3 生产级限流：Token Bucket 实现
python
import asyncio
import time
from typing import Dict
class RateLimiter:
"""Token Bucket 限流器"""
def __init__(self, rate: float, burst: int):
self.rate = rate # 每秒允许的请求数
self.burst = burst # 最大突发量
self._tokens = burst
self._last_refill = time.monotonic()
self._lock = asyncio.Lock()
async def acquire(self) -> None:
async with self._lock:
now = time.monotonic()
elapsed = now - self._last_refill
self._tokens = min(self.burst, self._tokens + elapsed * self.rate)
self._last_refill = now
if self._tokens < 1:
wait_time = (1 - self._tokens) / self.rate
await asyncio.sleep(wait_time)
self._tokens = 0
else:
self._tokens -= 1
async def limited_llm_call(prompt: str, limiter: RateLimiter) -> str:
await limiter.acquire() # 等待令牌
return f"response to: {prompt}"
async def main():
limiter = RateLimiter(rate=5.0, burst=10) # 5 req/s，最大突发 10
tasks = [limited_llm_call(f"prompt-{i}", limiter) for i in range(20)]
results = await asyncio.gather(*tasks)
print(f"完成 {len(results)} 个请求")
asyncio.run(main())1.4 重试策略：tenacity
python
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import asyncio
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10),
retry=retry_if_exception_type((asyncio.TimeoutError, ConnectionError)),
before_sleep=lambda retry_state: print(f"重试 {retry_state.attempt_number}...")
)
async def call_llm_with_retry(prompt: str, timeout: float = 5.0) -> str:
"""带指数退避重试的 LLM 调用"""
try:
return await asyncio.wait_for(_actual_call(prompt), timeout=timeout)
except asyncio.TimeoutError:
raise # 超时触发重试
async def _actual_call(prompt: str) -> str:
await asyncio.sleep(0.1)
return f"result for: {prompt}"
async def main():
try:
result = await call_llm_with_retry("hello")
print(result)
except Exception as e:
print(f"最终失败: {e}")
asyncio.run(main())1.5 熔断器模式
python
import asyncio
import time
from enum import Enum
class CircuitState(Enum):
CLOSED = "closed" # 正常
OPEN = "open" # 熔断
HALF_OPEN = "half_open" # 半开
class CircuitBreaker:
def __init__(self, failure_threshold: int = 5, reset_timeout: float = 30.0):
self.failure_threshold = failure_threshold
self.reset_timeout = reset_timeout
self._state = CircuitState.CLOSED
self._failures = 0
self._last_failure_time = 0.0
self._lock = asyncio.Lock()
@property
def state(self) -> CircuitState:
return self._state
async def call(self, func, *args, **kwargs):
async with self._lock:
if self._state == CircuitState.OPEN:
if time.monotonic() - self._last_failure_time > self.reset_timeout:
self._state = CircuitState.HALF_OPEN
print("熔断器：半开状态，允许一次试探请求")
else:
raise RuntimeError("熔断器已打开，服务不可用")
try:
result = await func(*args, **kwargs)
async with self._lock:
if self._state == CircuitState.HALF_OPEN:
self._state = CircuitState.CLOSED
self._failures = 0
print("熔断器：恢复关闭状态")
return result
except Exception as e:
async with self._lock:
self._failures += 1
self._last_failure_time = time.monotonic()
if self._failures >= self.failure_threshold:
self._state = CircuitState.OPEN
print(f"熔断器：达到 {self.failure_threshold} 次失败，打开熔断")
raise
async def main():
cb = CircuitBreaker(failure_threshold=3, reset_timeout=2.0)
async def flaky_service():
raise ConnectionError("服务不可用")
for i in range(5):
try:
await cb.call(flaky_service)
except Exception as e:
print(f"请求 {i+1} 失败: {e}")
await asyncio.sleep(0.1)
asyncio.run(main())1.6 BackgroundTasks vs 消息队列的边界
| 场景 | 方案 | 原因 |
|---|---|---|
| 轻量异步任务（发送通知邮件） | FastAPI BackgroundTasks | 同进程内，简单快速 |
| 长时间 LLM 调用 | Celery Worker + Redis Queue | 进程隔离、可重试、持久化 |
| 高并发 Agent 任务 | SQS + Fargate Worker | 水平扩展、分布式协调 |
💡 详见 FastAPI ch16 关于 BackgroundTasks 的完整讲解。
本章小结
asyncio是单线程并发模型，核心 API 包括gather、create_task、sleepaiohttp用于异步 HTTP 调用，适合 LLM API 并发请求- 生产级限流用 Token Bucket，重试用
tenacity（指数退避），熔断用三态机 - 区分
BackgroundTasks（轻量）与消息队列（重任务）的使用边界
🛠️ 动手实践
- 实现一个异步 LLM 调用服务，使用 aiohttp + 自定义 RateLimiter，限制每秒最多 3 个请求
- 给上述服务加上 tenacity 重试（最多 3 次，指数退避 1-8 秒）
- 实现 CircuitBreaker 熔断器，连续 3 次失败后打开，30 秒后自动半开
🧪 随堂测验
点击你认为正确的选项。答错时会展示正确答案与原因解析。
1. asyncio.gather() 的主要作用是什么？
2. Token Bucket 限流器中，burst 参数表示什么？
3. tenacity 的 wait_exponential 如何实现退避？
4. CircuitBreaker 的 HALF_OPEN 状态用于什么场景？