asyncio 不加速模型推理，但能提升 ML API 并发吞吐量，因其将 I/O 等待时间用于处理其他请求；需复用 aiohttp.ClientSession、用 run_in_executor 加载/推理模型、用 Semaphore 限流，并优先选用原生异步接口或批量请求。
asyncio 本身不加速模型推理，但能显著提升机器学习 API 的并发吞吐量——关键在于把 I/O 等待时间“腾出来”做别的事。
为什么 async/await 对 ML API 有效？
大模型或传统 ML 接口的瓶颈几乎从不来自 Python 计算，而是：网络请求（调外部模型服务）、数据库查特征、文件读取（加载 embedding）、缓存访问（Redis）、甚至本地模型的 torch.load() 或 joblib.load()。这些全是 I/O 密集型操作。
同步代码中，每个请求独占一个线程，等 300ms 的 Redis 响应时，线程就挂在那里；异步代码中，同一个线程在等待期间立刻去处理下一个请求的预处理逻辑。
- 不是 CPU 更快，是单位时间完成的请求数翻倍
- 100 个并发请求，同步可能要开 100 个线程（内存爆涨），异步只需 1–4 个 OS 线程
- Python 3.12 的
TaskGroup和更快的任务调度让这种并发更稳定、错误传播更干净
用 aiohttp 调用远程大模型 API 时必须共享 session
常见错误：aiohttp.ClientSession() 每次都新建，导致连接池失效、TIME_WAIT 爆满、DNS 解析重复开销。
立即学习“Python免费学习笔记（深入）”；
正确做法是在整个生命周期内复用一个 ClientSession 实例，通常通过依赖注入或全局单例管理：
import asyncio
import aiohttp
<h1>✅ 正确：session 复用，连接池生效</h1><p>async def call_llm(session, prompt):
url = "<a href="https://www.php.cn/link/12414eb5c95aa700701f8a776ed91914">https://www.php.cn/link/12414eb5c95aa700701f8a776ed91914</a>"
payload = {"messages": [{"role": "user", "content": prompt}]}
async with session.post(url, json=payload) as resp:
return await resp.json()</p><p>async def main():</p><h1>创建一次，传给所有协程</h1><pre class="brush:php;toolbar:false;">async with aiohttp.ClientSession() as session:
tasks = [call_llm(session, p) for p in ["hi", "summarize", "translate"]]
results = await asyncio.gather(*tasks)
- 不复用 session → 平均耗时增加 20–40%，尤其在高并发下连接建立失败率上升
- 若需设置超时或自定义 headers，统一在
ClientSession(...)构造时传入，别在每次请求里重复写 - 不要在循环里
await session.close()——async with自动处理
本地模型加载和推理不能直接 await，得用 run_in_executor
torch.load()、transformers.AutoModel.from_pretrained()、joblib.load() 全是阻塞调用，会卡死事件循环。
必须把它们移出主线程，用 loop.run_in_executor() 包裹：
import asyncio
import concurrent.futures
import torch
<p>model = None</p><p>async def load_model():
global model
loop = asyncio.get_running_loop()</p><h1>✅ 在线程池中加载，不阻塞事件循环</h1><pre class="brush:php;toolbar:false;">with concurrent.futures.ThreadPoolExecutor() as pool:
model = await loop.run_in_executor(pool, torch.load, "model.pt")async def predict(input_data): loop = asyncio.get_running_loop() with concurrent.futures.ThreadPoolExecutor() as pool:
✅ 推理也进线程池
result = await loop.run_in_executor(pool, model.forward, input_data)
return result
- 别用
ProcessPoolExecutor—— 大多数 ML 模型（尤其是 PyTorch）在多进程下有 CUDA 上下文问题或 pickle 失败 - 加载一次就够了，后续预测复用已加载模型；避免每次请求都 reload
- 如果模型支持异步推理（如 vLLM 的
AsyncLLMEngine），优先用原生异步接口，比run_in_executor更高效
限制并发数防止下游服务被打垮
无节制并发（比如 1000 个请求同时发给一个 LLM API）大概率触发限流、429 错误，甚至让对方服务熔断。
Python 3.12 推荐用 asyncio.Semaphore 控制并发度：
sem = asyncio.Semaphore(5) # 最多 5 个并发请求 <p>async def safe_call_llm(session, prompt): async with sem: # 进入前自动 acquire，退出时自动 release return await call_llm(session, prompt)</p><h1>然后用 gather 并发发起</h1><p>results = await asyncio.gather(*[safe_call_llm(session, p) for p in prompts])
- 数值选多少？参考下游 API 的 rate limit（如每分钟 60 次 → 每秒约 1 次 →
Semaphore(1)） - 别用
asyncio.wait(..., return_when=asyncio.FIRST_COMPLETED)替代限流 —— 它不控制并发总数，只控制“谁先返回” - 如果下游支持批量请求（如一次传 10 条 prompt），优先用 batch 接口，比并发 10 次单条更省资源
真正的难点不在语法，而在判断哪段代码该异步、哪段必须挪到线程池、哪段其实根本不该并发——这取决于你调用的具体模型服务类型和部署方式。
