返回博客首页
Python asyncio 并发调用多模型：批量请求与智能重试实战
本站为独立第三方技术服务平台，提供多模型 API 聚合接入服务，与 Anthropic、OpenAI、Google 等模型提供商无任何关联、授权或合作关系。
TL;DR
- 批量处理几百上千条请求时，串行一条条跑慢到无法接受，
asyncio并发能把耗时从"分钟级"压到"秒级"。 - 关键三件事：用
Semaphore控并发数、给每个请求加超时、失败做指数退避重试。 - Code0 的 OpenAI 兼容接口支持
AsyncOpenAI，代码和官方 SDK 一模一样，只改base_url。 - 还能顺手做"多模型分流"——不同任务并发跑不同模型。
前置准备
- 注册 Code0 账号并获取 API Key：https://console.code0.ai/console/dashboard
- 安装 SDK：
pip install openai - Python 3.8+（建议 3.10+）
为什么要用 asyncio
假设你要给 500 条用户评论做情感分类。每条请求往返 1 秒，串行就是 500 秒——8 分多钟干等。
但这些请求彼此独立，完全可以同时发出去。网络 I/O 等待的时间里，程序可以去发下一个请求。asyncio 正是干这个的：用单线程把大量 I/O 等待重叠起来。并发跑 20 条，500 条的总耗时能压到 30 秒以内。
步骤拆解
步骤 1：用 AsyncOpenAI 发一个异步请求
import asyncio
from openai import AsyncOpenAI
client = AsyncOpenAI(
base_url="https://hk.code0.ai/v1",
api_key="sk-你的Key", # 在 console.code0.ai 获取
)
async def ask(prompt):
resp = await client.chat.completions.create(
model="claude-opus-4-8",
messages=[{"role": "user", "content": prompt}],
)
return resp.choices[0].message.content
print(asyncio.run(ask("你好")))
注意用的是 AsyncOpenAI 和 await，其余和同步版本一致。
步骤 2：控制并发数（Semaphore）
一次性把 500 个请求全丢出去，会撞上速率限制、也可能拖垮本地资源。用信号量把同时在飞的请求数控制在一个合理值：
sem = asyncio.Semaphore(20) # 最多 20 个并发
async def ask_limited(prompt):
async with sem: # 拿到令牌才能发，发完自动释放
return await ask(prompt)
步骤 3：加超时与指数退避重试
网络会抖、偶发错误在所难免。给每个请求加超时，失败后按指数退避重试，是生产级批处理的标配：
async def ask_robust(prompt, max_retries=3):
async with sem:
for attempt in range(max_retries):
try:
return await asyncio.wait_for(ask(prompt), timeout=30)
except Exception as e:
if attempt == max_retries - 1:
return f"[失败] {e}"
wait = 2 ** attempt # 1s, 2s, 4s 指数退避
await asyncio.sleep(wait)
指数退避的好处：偶发抖动很快就重试成功；如果是服务端临时压力大，逐渐拉长的间隔能避免"越重试越拥堵"。而且请求失败在 Code0 是不计费的，重试不会让你为失败的调用付钱。
步骤 4：批量并发跑起来
async def batch(prompts):
tasks = [ask_robust(p) for p in prompts]
return await asyncio.gather(*tasks)
prompts = [f"把这条评论分类为正面/负面：{c}" for c in comments] # 你的 500 条数据
results = asyncio.run(batch(prompts))
asyncio.gather 会并发跑所有任务（受 Semaphore 限流），全部完成后按原顺序返回结果。
完整代码示例
把上面几块拼起来，就是一个可直接跑的批处理脚本：
import asyncio
from openai import AsyncOpenAI
client = AsyncOpenAI(base_url="https://hk.code0.ai/v1", api_key="sk-你的Key")
sem = asyncio.Semaphore(20)
async def ask(prompt, model="claude-opus-4-8"):
resp = await client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}],
)
return resp.choices[0].message.content
async def ask_robust(prompt, model="claude-opus-4-8", max_retries=3):
async with sem:
for attempt in range(max_retries):
try:
return await asyncio.wait_for(ask(prompt, model), timeout=30)
except Exception as e:
if attempt == max_retries - 1:
return f"[失败] {e}"
await asyncio.sleep(2 ** attempt)
async def batch(prompts, model="claude-opus-4-8"):
return await asyncio.gather(*[ask_robust(p, model) for p in prompts])
if __name__ == "__main__":
data = ["评论1", "评论2", "评论3"] # 换成你的数据
prompts = [f"把这条评论分类为正面/负面：{c}" for c in data]
for c, r in zip(data, asyncio.run(batch(prompts))):
print(f"{c} -> {r}")
进阶：多模型分流
因为 Code0 一个 Key 通所有模型，你可以让不同任务并发跑不同模型——简单任务用便宜快的模型，难任务用旗舰：
async def mixed():
tasks = [
ask_robust("这条评论什么情感：还不错", model="deepseek-v3"), # 简单，用轻量模型
ask_robust("分析这份财报的风险点并给建议", model="claude-opus-4-8"), # 复杂，用旗舰
]
return await asyncio.gather(*tasks)
常见问题
- Q: 并发数设多少合适？ → 从 10–20 起步，观察成功率和响应，再逐步调。不是越高越好，撞速率限制反而更慢。
- Q:
asyncio.gather里一个任务报错会中断全部吗？ → 会。所以上面把异常在ask_robust内部消化成返回值；或给gather传return_exceptions=True。 - Q: 结果顺序会乱吗？
→
gather保证返回顺序和输入顺序一致，放心。 - Q: 想换模型？
→ 改
model参数即可，同一套并发框架适用于 Code0 上的全部模型。
小结
asyncio + 并发控制 + 智能重试，是把 AI 批处理从"能跑"提升到"能上生产"的关键三件套。配合 Code0 的 OpenAI 兼容异步接口和"一 Key 通模"，你还能在同一批并发任务里灵活分流不同模型，兼顾速度与成本。去 控制台 拿 Key，把上面的脚本换上你的数据就能跑。
