asyncio 并发调用 40 个 LLM：代码少但坑多

一次模型调用几乎全部时间都花在等待上。40 个顺序调用，每个两秒，共需八十秒；40 个并发调用只需要约两秒，再加上限速器的限制。代码很短——但让它真正并发运行，才是花掉整个下午的地方。

为什么并发是这里的全部优势

在客户端这边，模型调用完全不受 CPU 约束。你只需要序列化一个小 JSON 请求体，等待一到六十秒，再反序列化一个小 JSON 响应体。在等待期间你的进程无事可做，这正是 asyncio 存在的场景——这意味着关于 GIL 的常见争论在这里毫无意义。多线程也能工作；但对每个飞行中的请求来说，async 的成本更低，而且几千个并发等待的上限远远高于任何限速允许的范围。

应该追求的数字不是"越多越好"，而是提供商标定的最大容忍数量，通常在五到五十之间。超过这个数字，你只是在制造 429 错误并为重试付出代价。

模式：一个客户端，一个信号量

三个组件：一个被所有任务共享的 AsyncClient、一个限制飞行中任务数量的 asyncio.Semaphore，以及等待所有任务完成的 asyncio.gather。

# fanout.py
import asyncio
import os
from dataclasses import dataclass

import httpx

BASE_URL = os.environ["LLM_BASE_URL"].rstrip("/")
API_KEY = os.environ["LLM_API_KEY"]
MODEL = os.environ.get("LLM_MODEL", "openai/gpt-4o-mini")

CONCURRENCY = 8

@dataclass
class Result:
    index: int
    prompt: str
    text: str | None
    error: str | None

async def one_call(
    client: httpx.AsyncClient,
    sem: asyncio.Semaphore,
    index: int,
    prompt: str,
) -> Result:
    async with sem:                          # acquire a slot, release on exit
        try:
            response = await client.post(
                "/chat/completions",
                json={
                    "model": MODEL,
                    "messages": [{"role": "user", "content": prompt}],
                    "max_tokens": 300,
                },
            )
            response.raise_for_status()
            body = response.json()
            return Result(index, prompt, body["choices"][0]["message"]["content"], None)
        except httpx.HTTPStatusError as exc:
            return Result(index, prompt, None, f"{exc.response.status_code}: {exc.response.text[:200]}")
        except httpx.HTTPError as exc:
            return Result(index, prompt, None, f"{type(exc).__name__}: {exc}")

async def run_all(prompts: list[str]) -> list[Result]:
    limits = httpx.Limits(max_connections=CONCURRENCY,
                          max_keepalive_connections=CONCURRENCY)
    timeout = httpx.Timeout(connect=5.0, read=90.0, write=10.0, pool=30.0)
    sem = asyncio.Semaphore(CONCURRENCY)

    async with httpx.AsyncClient(
        base_url=BASE_URL,
        headers={"Authorization": f"Bearer {API_KEY}"},
        limits=limits,
        timeout=timeout,
    ) as client:
        tasks = [one_call(client, sem, i, p) for i, p in enumerate(prompts)]
        return await asyncio.gather(*tasks)

if __name__ == "__main__":
    prompts = [f"Give one fact about the number {n}." for n in range(40)]
    results = asyncio.run(run_all(prompts))
    ok = [r for r in results if r.error is None]
    print(f"{len(ok)}/{len(results)} succeeded")

信号量和 httpx.Limits 都故意设置为 CONCURRENCY。如果连接池小于信号量，任务会通过信号量，然后在连接池上隐性排队——你配置并发数和实际获得的并发数不符，症状是 PoolTimeout，看起来像是提供商变慢了。保持两个数字相等，让信号量成为决定并发的唯一地方。

让结果与输入关联

asyncio.gather 按任务传入顺序返回结果，而非完成顺序，所以简单的 zip(prompts, results) 是正确的。这种方式容易依赖，也容易在第一次有人加入过滤器时破坏，这也是上面的 Result 数据类携带自己的索引和提示词的原因。

另一方面是失败处理。默认情况下，如果一个任务抛出异常，gather 会传播该异常并丢失其他三十九个成功任务的结果。有两种解决方案，选择很重要：

在任务内部捕获并返回带错误字段的结果对象，如上所示。每个任务总是返回一些东西，类型保持诚实，调用者看到的是一个可分割的列表。这是应该采用的方法。

gather(*tasks, return_exceptions=True) 在结果位置返回异常对象。虽然方便，但列表变成了 list[Result | BaseException]，每个使用者都得记得检查。适用于一次性脚本。

如果一次失败应该停止所有操作——Python 3.11 及更高版本中的 asyncio.TaskGroup 在任务抛出异常时取消其兄弟任务，这正是后续阶段依赖前面所有阶段的管线的正确行为。

事件循环卡住的三个地方

每一种情况都让程序正确但变成串行执行。这比出错更糟糕，因为唯一的症状就是慢。

1. 任务内部创建客户端

在 one_call 内部使用 async with httpx.AsyncClient() 会为每个请求创建自己的连接池和自己的 TLS 握手。它仍然并发运行，但每次调用都要付出完整的握手代价，而且 keep-alive 消失后，并发节省的开销在设置阶段就消耗殆尽了。创建一次客户端并传入。

2. async 函数中的阻塞调用

一次 time.sleep()、一次 requests.post()、一次大文件的 open(...).read()，或一次对同步数据库驱动的调用，都会导致整个事件循环停止——包括所有其他任务。这是经典问题，而且容易隐藏在别人写的辅助函数里。

# wrong: blocks the loop for two seconds, for every task
async def one_call(...):
    time.sleep(2)

# right: yields control
async def one_call(...):
    await asyncio.sleep(2)

# unavoidable blocking code goes to a thread
text = await asyncio.to_thread(pdf_to_text, path)

asyncio.to_thread（Python 3.9 及更高版本）是 CPU 密集型或固执同步函数的逃生舱口。开发时用 asyncio.run(main(), debug=True) 运行循环：调试模式会对占用循环超过 100 毫秒的任何回调记录警告，这样无需二分查找就能发现这些问题。

3. 开始前构建整个任务列表

对 50,000 个协程执行 gather 会立即创建 50,000 个任务对象。信号量正确限制了调用网络的数量，但每个任务和每个提示词都是常驻的，如果每个任务持有一行 dataframe，你现在就复制了整个 dataframe。对于大量输入，分块处理：

async def run_chunked(prompts: list[str], size: int = 500) -> list[Result]:
    out: list[Result] = []
    for start in range(0, len(prompts), size):
        chunk = prompts[start:start + size]
        out.extend(await run_all(chunk))
        print(f"{start + len(chunk)}/{len(prompts)} done", flush=True)
    return out

进度与部分失败

gather 在所有任务完成前什么都不返回，这对于运行一个小时的作业来说很不友好。asyncio.as_completed 按完成顺序 yield awaitables，让你可以在每个结果到达时立即写出：

import json

async def run_streaming_results(prompts: list[str], out_path: str) -> None:
    limits = httpx.Limits(max_connections=CONCURRENCY,
                          max_keepalive_connections=CONCURRENCY)
    sem = asyncio.Semaphore(CONCURRENCY)
    async with httpx.AsyncClient(base_url=BASE_URL,
                                 headers={"Authorization": f"Bearer {API_KEY}"},
                                 limits=limits, timeout=90.0) as client:
        tasks = [one_call(client, sem, i, p) for i, p in enumerate(prompts)]
        done = 0
        with open(out_path, "a", encoding="utf-8") as fh:
            for future in asyncio.as_completed(tasks):
                result = await future
                fh.write(json.dumps(result.__dict__) + "\n")
                fh.flush()
                done += 1
                if done % 25 == 0:
                    print(f"{done}/{len(tasks)}", flush=True)

每到达一个结果就追加一行 JSON 对象意味着崩溃只会损失飞行中的调用，其他不受影响。这与将 50,000 行分类构建分解为可恢复作业的原则相同。

截止时间与取消

httpx 的读取超时限制的是字节之间的间隔，而非调用本身。一个稳定流式传输十分钟的模型永远不会触发九十秒的读取超时，所以没有其他限制的 fan-out 可能远远超过调用者设定的任何截止时间。

import asyncio

# Python 3.11+: a real wall-clock deadline around one task
async def one_call_bounded(client, sem, index, prompt, *, seconds: float = 45.0):
    try:
        async with asyncio.timeout(seconds):
            return await one_call(client, sem, index, prompt)
    except TimeoutError:
        return Result(index, prompt, None, f"deadline of {seconds}s exceeded")

# Python 3.10 and earlier
async def one_call_bounded_310(client, sem, index, prompt, *, seconds: float = 45.0):
    try:
        return await asyncio.wait_for(one_call(client, sem, index, prompt), seconds)
    except asyncio.TimeoutError:
        return Result(index, prompt, None, f"deadline of {seconds}s exceeded")

请注意，asyncio.timeout 和 wait_for 都是通过取消任务来工作的，而在 asyncio 中取消是协作式的：它在下一个 await 处抛出 CancelledError。一个卡在阻塞调用中的任务——上面第二种卡住——根本无法取消，超时根本不会触发。如果截止时间没有被遵守，在怀疑超时之前先找找阻塞代码。

两条规则随之而来。永远不要用 bare except Exception 吞掉 CancelledError——在 Python 3.8 及更高版本中，它继承自 BaseException 正是为了让这种情况不发生，但捕获 BaseException 进行清理的代码必须重新抛出它。并且要慎重决定截止时间放在信号量的哪一侧：上面的包装器同时限制了队列等待和请求，当调用者有总体截止时间时这是正确的，但当你意思是"45 秒的提供商时间"时就错了——对于后者，把 asyncio.timeout 移到 one_call 内部，放在 async with sem 之后，否则一个排队三分钟的任务会在从未与任何人交谈的时间里失败。

什么时候 asyncio 不是正确的工具

当限制是他们的，不是你的。 并发不能长期超过速率限制。超过限制后，你是在把吞吐量转换成 429 错误，重试使 burst 更糟。将此与客户端令牌桶结合，使并发设置成为一个下限而非奢望。

当工作真正是 CPU 密集型时。 在本地嵌入 100,000 个向量、解析 5,000 个 PDF 或对语料库进行分词都不是 I/O 等待。这些需要 ProcessPoolExecutor，async 毫无帮助。

当提供商提供批处理端点时。 许多提供商对异步提交并有较长完成窗口的工作收取低得多的费用。如果工作可以等待数小时，这比任何并发设置都更有价值——batch inference APIs 涵盖了这种权衡。

当管线必须经受重启时。 asyncio fan-out 与进程共存亡。一旦工作时间长到部署会中断它，就需要队列——用于长时间 AI 任务的后台作业。

用 tenacity 重试模型调用

分类 50,000 行而不烧穿预算