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

> Script batch thường gãy vì không có phanh chứ hiếm khi vì chạy chậm: không giới hạn số request chạy cùng lúc, không giữ nhịp gửi, không có timeout và retry không có điểm dừng.

Bản gốc: https://fdetimes.net/vi/bach-khoa/python-asyncio-goi-api-llm-hang-loat/

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.

**Điểm mấu chốt:** Giới hạn số request chạy cùng lúc chưa đủ; bạn còn phải giữ nhịp gửi đều tay.

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.

```python
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.

**Thử ngay tuần này:**

- Lấy một script batch đang dùng asyncio.gather không giới hạn, thêm Semaphore, bộ điều nhịp và asyncio.timeout, rồi chạy thử trên 500 bản ghi.
- Rà code đang dùng SDK chính thức của OpenAI hoặc Anthropic, xoá mọi vòng retry tự viết chồng lên SDK và chỉ đặt max_retries ở một chỗ.
- Cho job xuất file kết quả có cột ok/error cho từng bản ghi, rồi đếm lỗi theo loại để quyết định có chạy lại hay không.

## Nguồn

- [Rate limits (OpenAI API docs)](https://developers.openai.com/api/docs/guides/rate-limits)

- [Synchronization Primitives — Python 3.15.0 documentation](https://docs.python.org/3/library/asyncio-sync.html)

- [Coroutines and tasks — Python 3.15.0 documentation](https://docs.python.org/3/library/asyncio-task.html)

- [Timeouts - HTTPX](https://www.python-httpx.org/advanced/timeouts/)

- [Rate limits (Claude API docs)](https://platform.claude.com/docs/en/api/rate-limits)