Async batch client with asyncio

Yesterday I built an SSE streaming client that reads tokens as they arrive from the server. That works well for a single interactive session. But when I need to evaluate a model across 5,000 prompts, or run a throughput benchmark, I cannot afford to wait for each request to finish before sending the next one. I need a batch client that fires requests concurrently, handles backpressure, and collects results cleanly.

Python's asyncio event loop paired with aiohttp makes this surprisingly ergonomic. The pattern is not complicated, but there are a few pitfalls that cost me hours the first time I wrote one. This post walks through the client I actually use for production load testing and batch evaluation.

Why not threads?

The work here is almost entirely I/O-bound: we send an HTTP request, wait for the response, and move on. Threads would work, but they come with overhead that matters at scale:

Memory: each thread allocates a stack (typically 8 MB on Linux). With 500 concurrent requests, that is 4 GB of stack space alone.
Context switching: the OS scheduler has to juggle hundreds of kernel threads, and the GIL serializes Python code anyway.
Coordination: collecting results from threads requires explicit synchronization primitives.

Asyncio coroutines are cooperative. They share a single thread, yield control at every await, and consume kilobytes of memory each. For I/O-bound fan-out, they are the right tool.

The basic structure

The pattern has three layers: a single-request coroutine, a concurrency limiter, and a gather-and-collect driver. Here is the skeleton:

import asyncio
import aiohttp
from typing import Any

API_URL = "http://localhost:8000/v1/completions"

async def send_one(
    session: aiohttp.ClientSession,
    semaphore: asyncio.Semaphore,
    prompt: str,
    idx: int,
) -> dict[str, Any]:
    """Send a single completion request, respecting the semaphore."""
    payload = {
        "model": "meta-llama/Llama-3.1-8B",
        "prompt": prompt,
        "max_tokens": 256,
        "temperature": 0.0,
    }
    async with semaphore:
        async with session.post(API_URL, json=payload) as resp:
            data = await resp.json()
            return {"idx": idx, "text": data["choices"][0]["text"]}


async def run_batch(
    prompts: list[str],
    concurrency: int = 64,
) -> list[dict]:
    """Fire all prompts with bounded concurrency."""
    sem = asyncio.Semaphore(concurrency)
    connector = aiohttp.TCPConnector(limit=concurrency)
    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = [
            send_one(session, sem, p, i)
            for i, p in enumerate(prompts)
        ]
        results = await asyncio.gather(*tasks)
    return sorted(results, key=lambda r: r["idx"])

The asyncio.Semaphore is the key piece. Without it, gather would try to open all 5,000 connections at once, which would overwhelm the server and likely trigger connection resets or OOM on the client side. The semaphore gates entry so that at most concurrency requests are in flight at any moment.

Matching the TCPConnector limit

A subtle bug I hit early: I set the semaphore to 64 but left the aiohttp.TCPConnector at its default limit of 100. That works, but if you flip it (semaphore at 200, connector at 100), the semaphore lets coroutines through but they block on the connector's internal pool, defeating the purpose. The rule is simple: set the connector limit equal to or greater than the semaphore value. I usually set them to the same number.

Handling errors without losing the batch

In a real benchmark, some requests will fail. The server might return a 503 under load, or a request might time out. If any task in asyncio.gather raises, the default behavior is to cancel everything and propagate the exception. That is not what you want for batch evaluation.

async def send_one_safe(
    session: aiohttp.ClientSession,
    semaphore: asyncio.Semaphore,
    prompt: str,
    idx: int,
    max_retries: int = 3,
) -> dict[str, Any]:
    """Retry with exponential backoff; never crash the batch."""
    for attempt in range(max_retries):
        try:
            async with semaphore:
                timeout = aiohttp.ClientTimeout(total=120)
                async with session.post(
                    API_URL,
                    json={"model": "meta-llama/Llama-3.1-8B",
                          "prompt": prompt,
                          "max_tokens": 256},
                    timeout=timeout,
                ) as resp:
                    if resp.status == 429:
                        wait = 2 ** attempt
                        await asyncio.sleep(wait)
                        continue
                    resp.raise_for_status()
                    data = await resp.json()
                    return {"idx": idx, "text": data["choices"][0]["text"]}
        except (aiohttp.ClientError, asyncio.TimeoutError):
            if attempt < max_retries - 1:
                await asyncio.sleep(2 ** attempt)
    return {"idx": idx, "text": None, "error": "max retries exceeded"}

Two things to note. First, the retry backoff happens inside the semaphore context on 429s but outside on exceptions, because a failed connection has already released the socket. Second, the function always returns a dict, never raises. This lets asyncio.gather complete cleanly, and I can filter for errors afterward.

Tracking progress

When you have 5,000 prompts in flight, you want a progress bar. The trick is to wrap results in an asyncio.as_completed loop instead of using gather:

from tqdm.asyncio import tqdm_asyncio

async def run_batch_with_progress(prompts, concurrency=64):
    sem = asyncio.Semaphore(concurrency)
    connector = aiohttp.TCPConnector(limit=concurrency)
    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = [
            send_one_safe(session, sem, p, i)
            for i, p in enumerate(prompts)
        ]
        results = await tqdm_asyncio.gather(*tasks)
    return sorted(results, key=lambda r: r["idx"])

The tqdm.asyncio integration updates the bar as each coroutine completes. It is a small thing, but staring at a silent terminal for ten minutes wondering if the script is stuck is not a good use of engineering time.

Choosing the right concurrency

This is the part that requires measurement rather than intuition. Too low and you leave server capacity on the table. Too high and you saturate the inference engine's batch queue, causing timeouts and wasted GPU cycles.

The right number depends on:

Server batch size: if vLLM or SGLang is configured with max_num_seqs=256, sending 512 concurrent requests means half are just queued.
Request duration: longer generations (high max_tokens) mean each slot is occupied longer, so you need fewer concurrent requests to keep the server busy.
Network latency: over a WAN, you need more in-flight requests to keep the pipe full than over localhost.

My rule of thumb: start at 2x the server's max batch size for short generations, 1x for long ones. Then look at the server's num_requests_waiting metric. If the queue is always empty, increase concurrency. If it is growing unboundedly, back off.

Practical tip

When benchmarking throughput, I log timestamps for each request start and end, then compute the actual achieved concurrency as (requests in flight) averaged over the run. This is more useful than the semaphore value because it accounts for server-side queuing delays.

Streaming responses in a batch client

Sometimes you want batch evaluation and streaming, for example to measure time-to-first-token across thousands of prompts. The trick is to use the SSE approach from day 68 inside each coroutine, reading lines from the response stream until you see the first token, then recording the timestamp:

async def measure_ttft(session, semaphore, prompt, idx):
    t0 = asyncio.get_event_loop().time()
    async with semaphore:
        async with session.post(
            API_URL.replace("completions", "completions"),
            json={"prompt": prompt, "max_tokens": 64,
                  "stream": True},
        ) as resp:
            async for line in resp.content:
                text = line.decode().strip()
                if text.startswith("data:") and "choices" in text:
                    ttft = asyncio.get_event_loop().time() - t0
                    return {"idx": idx, "ttft_ms": ttft * 1000}
    return {"idx": idx, "ttft_ms": None}

This is how I collected the TTFT distributions in day 52. The async client fires requests at the target concurrency, each one measures its own first-token latency, and I get a full histogram at the end.

Putting it together

The final script I use for batch evaluation looks like this at the top level:

import asyncio
import json

async def main():
    with open("prompts.jsonl") as f:
        prompts = [json.loads(line)["prompt"] for line in f]

    results = await run_batch_with_progress(prompts, concurrency=128)

    succeeded = [r for r in results if r.get("text")]
    failed = [r for r in results if r.get("error")]
    print(f"Done: {len(succeeded)} ok, {len(failed)} failed")

    with open("results.jsonl", "w") as f:
        for r in results:
            f.write(json.dumps(r) + "\n")

if __name__ == "__main__":
    asyncio.run(main())

The async batch client is not glamorous infrastructure. It is plumbing. But bad plumbing is how you spend three hours on a benchmark that should take ten minutes, or worse, collect results from a server you accidentally DDoS-ed into dropping requests.

Tomorrow I will use this client to talk to inference endpoints across multiple cloud regions and explore geo-aware routing.