Skip to content
Multigrid

Forty Requests at Once With asyncio

11 min read · updated August 4, 2026

A model call spends almost all of its wall-clock time waiting. Forty sequential calls at two seconds each take eighty seconds; forty concurrent ones take about two, plus whatever the rate limiter allows. The code is short. Getting it to actually run concurrently is where the afternoon goes.

Why concurrency is the whole win here

Nothing about a model call is CPU-bound on your side. You serialise a small JSON body, wait between one and sixty seconds, and deserialise a small JSON body. During the wait your process has nothing to do, which is exactly the case asyncio exists for — and it means the usual argument about the GIL is irrelevant here. Threads would work too; async is simply cheaper per in-flight request, and the ceiling of a few thousand concurrent waits is far above what any rate limit will let you use.

The number to aim for is not “as many as possible”. It is the largest number your provider’s limits tolerate, which is usually between five and fifty. Above that you are just generating 429s and paying for retries.

The pattern: one client, one semaphore

Three components: a single AsyncClient shared by every task, an asyncio.Semaphore that caps how many are in flight, and asyncio.gather to wait for all of them.

# 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")

The semaphore and httpx.Limits are both set to CONCURRENCY on purpose. If the pool is smaller than the semaphore, tasks pass the semaphore and then queue invisibly on the pool — the concurrency you configured is not the concurrency you get, and the symptom is a PoolTimeout that looks like the provider being slow. Keeping the two numbers equal makes the semaphore the only place concurrency is decided.

Keeping results attached to inputs

asyncio.gather returns results in the order the tasks were passed in, not the order they completed, so a plain zip(prompts, results) is correct. That is easy to rely on and easy to break the first time somebody adds a filter, which is why the Result dataclass above carries its own index and prompt.

The other half is failure. By default, if one task raises, gather propagates that exception and you lose the results of the thirty-nine that worked. There are two ways out, and the choice matters:

  • Catch inside the task and return a result object with an error field, as above. Every task always returns something, the types stay honest, and the caller sees a list it can partition. This is the one to reach for.
  • gather(*tasks, return_exceptions=True) returns exception objects in place of results. Convenient, but the list is now list[Result | BaseException] and every consumer has to remember to check. Useful for a throwaway script.

If one failure should stop everything — asyncio.TaskGroup in Python 3.11 and later cancels its siblings when a task raises, which is the right behaviour for a pipeline where later stages depend on all of the earlier ones.

Three places the event loop stalls

Each of these makes the program correct and serial. That is worse than being wrong, because the only symptom is that it is slow.

1. A client created inside the task

async with httpx.AsyncClient() as client: inside one_call gives every request its own connection pool and its own TLS handshake. It still runs concurrently, but you pay a full handshake per call, and with keep-alive gone the saving from concurrency is largely spent on setup. Build the client once and pass it in.

2. A blocking call in an async function

One time.sleep(), one requests.post(), one open(...).read() of a large file, or one call into a synchronous database driver, and the entire event loop stops — every other task included. This is the classic, and it hides well inside a helper somebody else wrote.

# 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 and later) is the escape hatch for a CPU-bound or stubbornly synchronous function. Run the loop with asyncio.run(main(), debug=True) during development: the debug mode logs a warning for any callback that occupies the loop for more than 100 ms, which finds these without a bisect.

3. Building the entire task list before starting

gather over 50,000 coroutines creates 50,000 task objects at once. The semaphore correctly limits how many are calling the network, but every task and every prompt is resident, and if each task holds a row of a dataframe you have now copied the dataframe. For large inputs, process in chunks:

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

Progress and partial failure

gather gives you nothing until everything finishes, which is unpleasant on a job that runs for an hour. asyncio.as_completed yields awaitables in completion order and lets you write each result out as it lands:

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)

Appending one JSON object per line as results arrive means a crash costs you the in-flight calls and nothing else. That is the same discipline classifying 50,000 rows builds into a resumable job.

Deadlines and cancellation

An httpx read timeout bounds the gap between bytes, not the call. A model that streams steadily for ten minutes never trips a ninety-second read timeout, so a fan-out with no other bound can run far past any deadline the caller had in mind.

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")

Note that asyncio.timeout and wait_for both work by cancelling the task, and cancellation in asyncio is cooperative: it raises CancelledError at the next await. A task stuck in a blocking call — the second stall above — cannot be cancelled at all, and the timeout will simply not fire. If a deadline is not being honoured, look for blocking code before suspecting the timeout.

Two rules follow. Never swallow CancelledError with a bare except Exception — in Python 3.8 and later it inherits from BaseException precisely so that it does not, but code that catches BaseException for cleanup must re-raise it. And decide deliberately which side of the semaphore the deadline sits on: the wrappers above bound the queue wait as well as the request, which is right when the caller has an overall deadline, and wrong when you meant “45 seconds of provider” — for that, move the asyncio.timeout inside one_call, after async with sem, or a task that queued for three minutes fails on time it never spent talking to anybody.

When asyncio is the wrong tool

  • When the limit is theirs, not yours. Concurrency cannot exceed the rate limit for long. Above it you are converting throughput into 429s, and retries make the burst worse. Pair this with a client-side token bucket so the concurrency setting is a floor rather than a hope.
  • When the work is genuinely CPU-bound. Embedding 100,000 vectors locally, parsing 5,000 PDFs, or tokenising a corpus is not I/O wait. Those want ProcessPoolExecutor, and async buys nothing.
  • When a provider offers a batch endpoint. Many charge substantially less for work submitted asynchronously with a long completion window. If the job can wait hours, that is a bigger win than any concurrency setting — batch inference APIs covers the trade.
  • When the pipeline must survive a restart. An asyncio fan-out lives and dies with the process. Once the job is long enough that a deploy will interrupt it, it wants a queue — background jobs for long AI tasks.

Related