Gọi 20.000 request LLM trong Python mà không gãy vì 429 hay bị treo

Khách hàng gửi bạn 20.000 ticket hỗ trợ và muốn có kết quả phân loại bằng LLM trước sáng thứ Hai. Chạy tuần tự bằng vòng for thì ước tính mất cả ngày, nên bạn chuyển sang asyncio.gather và bắn hết một lượt. Màn hình lập tức kín lỗi 429, sau đó đến PoolTimeout, rồi tới 97% thì script đứng im.

Gần như FDE nào cũng từng gặp cảnh này, dù bên kia là API của OpenAI, của Anthropic hay một hệ thống nội bộ cũ của khách. Chậm chưa phải vấn đề, vì chậm thì cuối cùng job vẫn chạy xong.

Thứ làm hỏng việc là script không có phanh. Bài này đi qua bốn cái phanh cần lắp, một pipeline hoàn chỉnh, và cách đọc đúng lỗi 429 để biết lúc nào nên đợi, lúc nào nên dừng.

Vì sao “bắn hết một lượt” lại chậm hơn?

API nào cũng có giới hạn. Tài liệu của OpenAI ghi rõ: request vượt giới hạn tạm thời sẽ nhận về lỗi 429. Khi 20.000 coroutine cùng xếp hàng, phần lớn chỉ đi gom 429 về, rồi cùng retry vào một thời điểm, rồi lại cùng bị chặn.

Phía client cũng tắc. httpx giữ một pool kết nối có giới hạn. Request nào chờ lấy kết nối quá thời hạn pool sẽ bị ném PoolTimeout, nghĩa là bạn đang tạo nhiều request đồng thời hơn số kết nối mà pool cấp được.

Còn treo ở 97% thường là do một request bị kẹt mà không có gì cắt nó đi, hoặc một vòng retry không có điểm dừng. Lỗi kiểu này không văng ra. Script chỉ đứng yên và bạn không biết nó kẹt ở đâu.

Semaphore chưa đủ, bạn còn cần giữ nhịp

Cái phanh đầu tiên là asyncio.Semaphore. Tài liệu Python mô tả nó như một bộ đếm không bao giờ xuống dưới 0. Khi bộ đếm về 0, task gọi acquire() phải chờ đến lúc có task khác gọi release(). Đặt Semaphore(20) thì lúc nào cũng chỉ có tối đa 20 request đang chạy.

Nhiều người bỏ sót phần tiếp theo. Anthropic giải thích rằng API của họ dùng thuật toán token bucket: dung lượng được nạp lại liên tục chứ không reset theo mốc phút. Vì thế giới hạn 60 RPM có thể được áp thành 1 request mỗi giây, và một đợt request dồn lại trong thời gian ngắn vẫn có thể vượt giới hạn.

Thử hình dung 20 request chạy song song, mỗi request chỉ mất 300 ms. Chỉ trong một giây bạn đã gửi khoảng 60 request, tức là dùng hết quota của cả phút, dù chưa lúc nào vượt quá 20 request đồng thời.

Vì vậy cái phanh thứ hai là bộ điều nhịp (pacer), giữ khoảng cách tối thiểu giữa hai lần gửi. Anthropic còn khuyên tăng lưu lượng từ từ và giữ mức dùng ổn định để không chạm acceleration limit. Khi làm thật, bạn bắt đầu với khoảng cách lớn rồi rút ngắn dần trong vài phút đầu.

Hai cái phanh này làm hai việc khác nhau, và nên tính trước khi chạy. Pacer đặt trần cho thông lượng: ở mức 1 request/giây, 20.000 ticket mất 20.000 giây, tức khoảng 5,5 giờ, dù bạn có cho bao nhiêu request chạy song song.

Semaphore thì giới hạn số request đang bay. Theo cách tính đơn giản, số đó xấp xỉ nhịp gửi nhân với thời gian của một request: mỗi LLM call mất 10 giây thì có khoảng 10 request đang bay, nên Semaphore(20) chủ yếu là lưới an toàn khi API chậm bất thường.

Nếu 5,5 giờ vẫn kịp hạn thứ Hai thì cứ để job chạy qua đêm. Nếu không kịp, đừng tăng Semaphore, vì làm vậy không giúp job nhanh hơn. Việc cần làm là hỏi khách về hạn mức cao hơn.

Pipeline phân loại 20.000 ticket

Dưới đây là khung gọi API phân loại bằng httpx. Hai cái phanh còn lại là timeout và retry có trần. Hãy đọc kỹ các con số vì mỗi con số nằm ở một tầng khác nhau.

import asyncio, random, time
import httpx

MAX_CONCURRENT = 20
MIN_INTERVAL = 1.0      # 60 RPM -> 1 request/giây -> 20.000 ticket ~ 5,5 giờ
RETRY_BUDGET_S = 120    # tổng thời gian được phép retry
PER_ITEM_S = 150        # trần cứng cho mỗi ticket

# Cờ theo nhà cung cấp:
# Anthropic: 429 do chạm trần chi tiêu không có retry-after, retry vô ích -> True
# OpenAI: thiếu header thì backoff mũ + jitter -> False
STOP_ON_BARE_429 = True

class SpendLimitError(Exception):
    """429 không kèm retry-after kiểu trần chi tiêu của Anthropic: retry vô ích."""

class Pacer:
    def __init__(self, interval: float):
        self.interval = interval
        self._lock = asyncio.Lock()
        self._next = 0.0

    async def wait(self):
        async with self._lock:
            now = time.monotonic()
            if self._next > now:
                await asyncio.sleep(self._next - now)
            self._next = max(now, self._next) + self.interval

def retry_after_seconds(resp):
    try:
        return float(resp.headers["retry-after"])
    except (KeyError, ValueError):
        return None

async def call_with_retry(client, pacer, payload, max_attempts=5):
    start = time.monotonic()
    for attempt in range(max_attempts):
        await pacer.wait()
        resp = await client.post("/classify", json=payload)
        if resp.status_code not in (429, 503):
            resp.raise_for_status()
            return resp.json()
        if (resp.status_code == 429 and STOP_ON_BARE_429
                and "retry-after" not in resp.headers):
            raise SpendLimitError("429 không có retry-after")  # dừng, không retry
        wait = retry_after_seconds(resp)
        if wait is None:                      # header thiếu/hỏng (cờ tắt) hoặc 503
            wait = min(2 ** attempt, 30)      # backoff mũ
        wait += random.uniform(0, 1)          # jitter
        if time.monotonic() - start + wait > RETRY_BUDGET_S:
            break
        await asyncio.sleep(wait)
    raise RuntimeError(f"hết lượt retry, status cuối {resp.status_code}")

async def worker(client, sem, pacer, ticket):
    async with sem:
        try:
            async with asyncio.timeout(PER_ITEM_S):
                result = await call_with_retry(client, pacer, ticket)
            return {"id": ticket["id"], "ok": True, "result": result}
        except SpendLimitError:
            raise                             # để TaskGroup huỷ cả job
        except Exception as e:
            return {"id": ticket["id"], "ok": False, "error": repr(e)}

async def main(tickets):
    sem = asyncio.Semaphore(MAX_CONCURRENT)
    pacer = Pacer(MIN_INTERVAL)
    limits = httpx.Limits(max_connections=MAX_CONCURRENT)
    timeout = httpx.Timeout(60.0, pool=30.0)
    try:
        async with httpx.AsyncClient(base_url="https://api.khach.example",
                                     limits=limits, timeout=timeout) as client:
            async with asyncio.TaskGroup() as tg:
                tasks = [tg.create_task(worker(client, sem, pacer, t))
                         for t in tickets]
    except* SpendLimitError:
        print("Chạm trần chi tiêu: dừng job, báo người giữ tài khoản phía khách")
        raise
    return [t.result() for t in tasks]

Đoạn retry làm đúng những gì tài liệu OpenAI hướng dẫn. Giá trị Retry-After chỉ là mức tối thiểu: chờ ít nhất chừng đó rồi cộng thêm một khoảng ngẫu nhiên nhỏ.

Header thiếu hoặc hỏng thì quay về exponential backoff có jitter, để các client không cùng retry một lúc. Retry phải có trần cho cả số lần thử lẫn tổng thời gian, ở đây là 5 lần và 120 giây.

Nhánh dừng hẳn khi gặp 429 không kèm header được bật bằng cờ STOP_ON_BARE_429, và cờ này phải đặt theo từng nhà cung cấp. Với Anthropic, tài liệu mô tả 429 do chạm trần chi tiêu không có retry-after và retry vô ích, nên bật cờ là hợp lý.

Với OpenAI, hướng dẫn là thiếu header thì backoff, nên hãy tắt cờ. Nếu gọi API của khách, hãy hỏi họ báo hiệu trường hợp “đừng thử lại nữa” bằng cách nào rồi sửa đúng dòng điều kiện đó.

Timeout được đặt theo ba tầng. httpx mặc định ném TimeoutException sau 5 giây không có hoạt động mạng. Mức đó quá ngắn cho một LLM call phải sinh câu trả lời dài, nên ở đây nâng lên 60 giây. Ngân sách retry 120 giây nằm ở giữa, còn asyncio.timeout(150) bao ngoài cùng và đổi mọi trường hợp kẹt thành TimeoutError, để không ticket nào treo mãi.

Lỗi nào giữ lại, lỗi nào để lọt ra?

Điểm dễ bỏ sót nằm ở chỗ try/except được đặt bên trong worker. Với TaskGroup, lần đầu một task lỗi bằng exception khác CancelledError, các task còn lại trong nhóm sẽ bị huỷ. Một ticket hỏng không được phép huỷ 19.999 ticket kia, nên worker tự bắt lỗi thường và trả về bản ghi ok: False.

Cơ chế huỷ đó lại đúng là thứ bạn cần khi gặp SpendLimitError. Worker ném lỗi này ra ngoài, TaskGroup huỷ toàn bộ các task còn lại, và main nhận lỗi qua except*. Job dừng ngay thay vì để hàng nghìn ticket lần lượt đập vào cùng một bức tường.

Vì sao không dùng gather? Ở chế độ mặc định, gather ném lỗi đầu tiên ra ngoài, còn các awaitable khác không bị huỷ và vẫn chạy tiếp.

Hàm main của bạn đã thoát mà request vẫn đang bay, và đó là nguồn của những lỗi rất khó lần. Với TaskGroup, khi có lỗi lọt ra thì các task còn lại bị huỷ, nên không còn request nào chạy ngoài tầm kiểm soát.

Không phải lỗi 429 nào cũng nên retry

Nếu gọi qua SDK chính thức, đừng viết lại vòng retry ở trên. Tài liệu OpenAI cho biết mỗi SDK chính thức đã tự retry các phản hồi 429 và 503 đủ điều kiện, theo cấu hình retry của nó.

Thử làm phép tính: SDK retry vài lần cho mỗi lần gọi, vòng của bạn lại thử 5 lần, thì số request thật gửi đi bằng tích của hai con số đó, chứ không phải tổng. Hãy giữ Semaphore, pacer và timeout, còn retry thì chỉnh trong cấu hình của SDK.

Còn một loại 429 nữa. Theo tài liệu Anthropic, khi tổ chức chạm trần chi tiêu, phản hồi 429 không có header retry-after, và retry, kể cả retry tự động của SDK, đều thất bại cho đến khi quyền truy cập được mở lại. Đó là lý do khi chạy với Anthropic, code ở trên bật cờ và ném SpendLimitError thay vì tăng backoff.

Nếu bạn dùng SDK của Anthropic, hãy kiểm tra header này trong log hoặc trong exception mà SDK trả về. Gặp đúng trường hợp đó thì dừng job và báo người giữ tài khoản phía khách.

Những lỗi hay gặp ở customer site

Các request thừa ngồi chờ trong pool rồi nhận PoolTimeout, trông giống lỗi mạng nhưng thật ra là lỗi cấu hình. Hãy giữ hai con số này khớp nhau.

Lỗi thứ hai là retry không có ngân sách thời gian. Năm lần thử, mỗi lần chờ Retry-After 60 giây, đã thành năm phút cho một ticket. Lỗi thứ ba là chỉ log tổng số lỗi. Khi kết quả có id và error cho từng ticket, bạn chỉ cần chạy lại đúng 312 ticket hỏng chứ không phải cả 20.000.

Lỗi cuối là chạy hết công suất ngay phút đầu tiên trên hệ thống của khách. API nội bộ của họ có thể không trả 429 lịch sự như OpenAI mà sập luôn. Hãy hỏi trước giới hạn thật của họ, rồi tăng tải dần.

Cách kỹ năng này hiện ra trên CV

Khi đọc job description của FDE, hãy để ý những dòng nhắc đến tích hợp với API của khách hay xử lý dữ liệu số lượng lớn, vì đó là chỗ kỹ năng này được dùng tới.

Trên CV, thay vì ghi “thành thạo asyncio”, hãy mô tả kết quả: xử lý N bản ghi qua một API có rate limit, có pacing và retry có trần, báo lỗi theo từng bản ghi và không job nào bị treo.

Khi phỏng vấn, nếu bạn giải thích được vì sao TaskGroup khác gather, hay vì sao với Anthropic một lỗi 429 không kèm retry-after là lúc phải dừng chứ không retry, thì sẽ thuyết phục hơn nhiều so với việc kể tên thư viện.

Script batch tốt không cần nhanh nhất. Nó cần chạy xong và cho bạn biết chính xác ticket nào chưa xong, vì sao, trước khi khách hàng hỏi.