Skip to content
Ephizen logo ephizen / wiki
CS 502

asyncio for LLM Batch Calls

Fan out 10,000 LLM calls without spinning up 10,000 threads — the production pattern for I/O-bound ML work.

Software Engineering Practices intermediate #python #async #llms

TL;DR

You have 10,000 prompts. Each LLM call takes ~1.5 seconds, mostly spent waiting for the model to generate tokens. Run them serially: 4 hours. Run them on 50 OS threads: ~5 minutes, but you’ve spawned 50 threads, paid the per-thread memory, and might still hit OS limits at larger scales. Run them with asyncio and a semaphore: ~5 minutes, zero threads, one process, trivially scales to thousands of in-flight calls.

This is the production pattern for any I/O-bound LLM workload: batch classification, dataset enrichment, eval runs, agent swarms. It generalises to any HTTP-fan-out problem.

The complete shape:

import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI()
SEM = asyncio.Semaphore(20)         # at most 20 concurrent calls

async def call_one(prompt):
    async with SEM:                  # bound concurrency
        resp = await client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
        )
        return resp.choices[0].message.content

async def main(prompts):
    results = await asyncio.gather(
        *(call_one(p) for p in prompts),
        return_exceptions=True,
    )
    return results

Three lines of bookkeeping (semaphore, gather, return_exceptions) on top of one async client call. That’s the entire pattern.

The picture in your head

Each client.chat.completions.create(...) is a coroutine that sends the request and awaits the response. While one is awaiting, the event loop runs others. The semaphore caps how many can be in flight at once — above the cap, additional calls block at async with SEM: until a slot frees up.

The math: with concurrency = 20 and latency = 1.5s, you process ~13 calls per second. 10,000 calls → ~12 minutes. Crank concurrency to 50 → ~5 minutes. The ceiling is whatever rate limit your provider enforces (request-per-minute or tokens-per-minute), not Python.

Why asyncio over threads

Threads (ThreadPoolExecutor)Asyncio
Memory per concurrent unit~8MB / thread~few KB / coroutine
Max practical concurrency~few hundredthousands+
Provider-side rate limitSameSame
Code stylesync; uses requestsasync; uses httpx / async client
Cancellation / timeouthardeasy via asyncio.wait_for
Stack tracescleanslightly noisier

Both work for “make 100 concurrent HTTP calls.” Beyond ~500 concurrent, asyncio scales noticeably better. For LLM workloads specifically, every modern SDK ships an async client (AsyncOpenAI, AsyncAnthropic) — use it.

The complete pattern — with retries, timeouts, and rate limiting

import asyncio
import logging
from typing import Iterable

from openai import AsyncOpenAI, RateLimitError, APIError
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

logger = logging.getLogger(__name__)
client = AsyncOpenAI(timeout=60)

# Concurrency cap matched to provider tier
sem = asyncio.Semaphore(50)

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=1, max=30),
    retry=retry_if_exception_type((RateLimitError, APIError, asyncio.TimeoutError)),
    reraise=True,
)
async def call_once(prompt: str, model: str = "gpt-4o-mini") -> str:
    async with sem:
        resp = await asyncio.wait_for(
            client.chat.completions.create(
                model=model,
                messages=[{"role": "user", "content": prompt}],
            ),
            timeout=45,
        )
        return resp.choices[0].message.content


async def batch(prompts: Iterable[str]) -> list[str | Exception]:
    tasks = [call_once(p) for p in prompts]
    results: list[str | Exception] = []
    for i, fut in enumerate(asyncio.as_completed(tasks)):
        try:
            r = await fut
            results.append(r)
        except Exception as e:
            logger.exception("call %d failed", i)
            results.append(e)
        if i % 100 == 0:
            logger.info("progress: %d/%d", i, len(tasks))
    return results

What each piece does:

  • asyncio.Semaphore(50) — caps in-flight calls. Set to your provider’s safe concurrency tier.
  • @retry from tenacity — exponential-backoff retries on rate-limit / API errors. Don’t write this yourself.
  • asyncio.wait_for(..., timeout=45) — per-call hard timeout. Without it, a stuck call holds a semaphore slot forever.
  • asyncio.as_completed — yields results as they finish, in arbitrary order. Lets you log progress.
  • Per-task exception handling — record failures rather than cancelling everything (what gather would do without return_exceptions=True).

This is approximately the production loop shape used by every batch LLM job at scale.

Rate limiting — semaphore is for concurrency, not throughput

A semaphore caps concurrent calls. It does not cap calls-per-second. If your provider says “10,000 requests per minute” (RPM), and your calls average 100ms, then 50 concurrent calls means 500 RPS = 30,000 RPM — over the limit.

For RPM/TPM (tokens-per-minute) limits, use a token-bucket rate limiter:

from aiolimiter import AsyncLimiter

# 10,000 requests per 60 seconds
rate = AsyncLimiter(max_rate=10_000, time_period=60)

async def call_one(prompt):
    async with rate:                  # acquires a token from the bucket
        async with sem:                # caps concurrency
            return await client.chat.completions.create(...)

The two limiters compose: rate caps throughput; semaphore caps concurrency. Both matter — provider rate limits care about the first; TCP/socket exhaustion cares about the second.

Streaming — token-by-token responses

For long generations, you can stream tokens as they arrive instead of waiting for the full response. Useful for UIs and for long-running calls where you want partial results.

async def stream_one(prompt):
    stream = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        stream=True,
    )
    chunks = []
    async for chunk in stream:
        if chunk.choices[0].delta.content:
            chunks.append(chunk.choices[0].delta.content)
            print(chunks[-1], end="", flush=True)
    return "".join(chunks)

The async for over the stream is the canonical async iteration form — same idea as a for loop but each step awaits the next chunk.

Common gotchas

  • Calling client = OpenAI() instead of AsyncOpenAI(). The sync client blocks the event loop on every call. The whole async benefit evaporates.
  • No semaphore. gather(*[call(p) for p in 10_000_prompts]) opens 10,000 connections instantly, hits provider rate limits, exhausts file descriptors. Always cap.
  • No timeout. A stuck call holds its semaphore slot forever, starving other coroutines.
  • No retries. Transient 503s and rate-limit errors will tank your batch job. Always wrap with tenacity or equivalent.
  • gather without return_exceptions=True. First failure cancels the whole batch and you lose all completed results. Almost never what you want for batch jobs.
  • Logging from inside coroutines that share a queue. Standard logging is thread-safe but each log call is sync; for very high log rates use a queue handler.
  • asyncio.run() inside a notebook. Jupyter already has a loop running. Use await main() directly.

A worked example — classifying 100,000 reviews

import asyncio
import logging
from pathlib import Path

import pandas as pd
from openai import AsyncOpenAI
from pydantic import BaseModel, Field
from tenacity import retry, stop_after_attempt, wait_exponential

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger(__name__)

client = AsyncOpenAI(timeout=60)
sem = asyncio.Semaphore(50)


class Sentiment(BaseModel):
    label: str = Field(pattern="^(positive|negative|neutral)$")
    confidence: float = Field(ge=0, le=1)


SYSTEM_PROMPT = (
    "Classify sentiment. Reply with JSON only, matching this schema: "
    '{"label": "positive" | "negative" | "neutral", "confidence": 0..1}'
)


@retry(stop=stop_after_attempt(5), wait=wait_exponential(min=1, max=30))
async def classify(text: str) -> Sentiment:
    async with sem:
        resp = await asyncio.wait_for(
            client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[
                    {"role": "system", "content": SYSTEM_PROMPT},
                    {"role": "user", "content": text},
                ],
                response_format={"type": "json_object"},
            ),
            timeout=45,
        )
        return Sentiment.model_validate_json(resp.choices[0].message.content)


# For stricter schema enforcement (additionalProperties: false, all fields
# required), use the `instructor` library, which patches the SDK to accept a
# Pydantic model directly and handles the strict-mode schema rewrite for you.


async def run(in_path: Path, out_path: Path):
    df = pd.read_parquet(in_path)
    logger.info("scoring %d rows", len(df))

    tasks = [classify(t) for t in df["review_text"]]
    results: list[Sentiment | Exception] = []
    for i, fut in enumerate(asyncio.as_completed(tasks), start=1):
        try:
            results.append(await fut)
        except Exception as e:
            logger.warning("row %d failed: %s", i, e)
            results.append(e)
        if i % 1000 == 0:
            logger.info("progress: %d / %d", i, len(tasks))

    df["label"] = [r.label if isinstance(r, Sentiment) else None for r in results]
    df["confidence"] = [r.confidence if isinstance(r, Sentiment) else None for r in results]
    df.to_parquet(out_path)
    logger.info("done. %d failures", sum(1 for r in results if isinstance(r, Exception)))


if __name__ == "__main__":
    asyncio.run(run(Path("reviews.parquet"), Path("scored.parquet")))

Counting parts: Pydantic-validated structured output, semaphore-bounded concurrency, exponential-backoff retries, per-call timeout, per-row exception handling, progress logging. ~80 lines for a production-grade batch scoring job.

Where this shows up in real ML work

  • LLM-as-judge eval. Score thousands of generations on quality rubrics; partial failures expected and handled.
  • Dataset enrichment. Take a dataset, run an LLM over each row, attach a column.
  • Synthetic data generation. Fan-out prompt → model → filter → store, all async.
  • Agent fan-out. Multiple agents working in parallel on independent subtasks.
  • RAG ingestion. Async chunking + embedding + vector-DB upserts.
  • Backfilling production scores. Score N million existing rows with a new model.

The defensive habit: every batch LLM job has a semaphore, retries, timeouts, per-call error handling, and progress logging. Anything missing is a future production incident.

Resources