Python + AI 之异步推理网关——从 47 QPS 到 620 QPS 的调优纪实

原创

IT大佬 jzit-top

发布于 2026-10-02 10:12:03

我们给一家电商做商品文案生成服务，第一版用 Flask + 同步 requests 调 LLM API，压测 47 QPS，P99 延迟 4.8s。高峰期用户排队超过 30s，客服投诉不断。本文只讲怎么用纯 Python 把 QPS 拉到 620、P99 压到 720ms。

1. 同步阻塞是原罪

第一版代码长这样：

# 反例：每个请求阻塞一个线程
@app.post("/generate")
def generate():
    resp = requests.post(LLM_URL, json=payload, timeout=30)
    return resp.json()

Gunicorn 开 8 worker × 4 线程 = 32 并发。每次调用 LLM 平均 680ms，理论上限 32/0.68 ≈ 47 QPS，实测完全吻合。问题不是线程不够，而是线程在等 IO。

2. 换成 asyncio + aiohttp，先解决 IO 等待

import asyncio, aiohttp
from fastapi import FastAPI

app = FastAPI()
session: aiohttp.ClientSession | None = None
SEM = asyncio.Semaphore(256)  # 并发上限，保护下游

@app.on_event("startup")
async def start():
    global session
    conn = aiohttp.TCPConnector(limit=512, limit_per_host=512, ttl_dns_cache=300)
    session = aiohttp.ClientSession(connector=conn)

@app.post("/generate")
async def generate(body: dict):
    async with SEM:
        async with session.post(LLM_URL, json=body, timeout=30) as r:
            return await r.json()

单进程 uvicorn 压测直接到 210 QPS。但 P99 还在 1.8s，因为 LLM 端对高并发有排队。

3. 微批处理：把 200 个请求合成一次调用

真正的瓶颈在下游 LLM 服务，它支持 batch 但没被我们用上。我们做了一个 20ms 的微批次聚合器：

import asyncio
from collections import deque

class Batcher:
    def __init__(self, max_size: int = 32, max_wait_ms: int = 20):
        self.max_size = max_size
        self.max_wait = max_wait_ms / 1000
        self.q: deque = deque()
        self._lock = asyncio.Lock()

    async def submit(self, item: dict) -> dict:
        fut = asyncio.get_running_loop().create_future()
        async with self._lock:
            self.q.append((item, fut))
            if len(self.q) == 1:
                asyncio.create_task(self._flush())
        return await asyncio.wait_for(fut, timeout=35)

    async def _flush(self):
        await asyncio.sleep(self.max_wait)
        async with self._lock:
            batch, self.q = list(self.q), deque()
        try:
            results = await call_llm_batch([x for x, _ in batch])
            for (_, fut), r in zip(batch, results):
                if not fut.done():
                    fut.set_result(r)
        except Exception as e:
            for _, fut in batch:
                if not fut.done():
                    fut.set_exception(e)

微批把下游调用次数从 620 次/秒降到 24 次/秒，P99 掉到 720ms，QPS 到 620。

4. 重试必须带抖动，否则雪崩

很多人重试用 sleep(2 ** n)，结果下游一抖动，所有客户端同时重试。我们用 decorrelated jitter：

import random
async def retry_call(fn, *, attempts=3, base=0.2, cap=3.0):
    for i in range(attempts):
        try:
            return await fn()
        except (aiohttp.ClientError, asyncio.TimeoutError):
            if i == attempts - 1:
                raise
            sleep = min(cap, random.uniform(base, base * 3 ** i))
            await asyncio.sleep(sleep)

抖动比指数退避的关键差异是：重试间隔随机化，避免同步冲击。

5. 结果与踩坑

最终数据：QPS 47 → 620（13×），P99 4.8s → 720ms，错误率 2.1% → 0.14%，单实例内存从 4GB 降到 900MB（因为不再需要 32 个线程栈）。

踩过的坑：

asyncio.Semaphore 加在错误的层级，导致同进程内所有请求共用，应该按上游来源限流；
aiohttp 默认 DNS 缓存 10s，LLM 域名切换后端时短暂 5xx，改成 300s；
微批处理没做超时兜底，某次下游 hang 住，整个批次 200 个请求全部超时，加了 wait_for 才稳定。

Python + AI 的核心不是调模型，是把 IO 密集的调用链做成有背压、有批量、有抖动重试的异步系统。这一步做不好，再好的模型也扛不住流量。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

python