Why this matters

AI workloads are overwhelmingly I/O bound. A single LLM call takes 300ms to 3 seconds. If you process 50 documents serially, you are waiting for minutes of cumulative network time. Async Python lets you turn that serial wait into concurrent requests — often achieving 10x throughput improvement with minimal code change.

But async also has sharp edges. Unconstrained concurrency hammers provider rate limits. Blocking calls inside async code stall the entire event loop. Missing timeouts turn degraded providers into hung services. This lesson covers the patterns that pay off and the mistakes that create new problems.

Core concepts

asyncio.gather for concurrent LLM calls

The most common async pattern in AI work is gathering a batch of independent calls:

import asyncio
from typing import Any


async def batch_generate(prompts: list[str], client) -> list[dict]:
    tasks = [client.generate(prompt) for prompt in prompts]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return [r for r in results if not isinstance(r, Exception)]

return_exceptions=True is important: without it, a single failure cancels all pending tasks. With it, you get exceptions back as values and can handle them per-item.

Rate limiting with semaphores

Calling 50 LLM endpoints simultaneously will saturate provider rate limits. A Semaphore limits concurrency to a safe window:

import asyncio


async def rate_limited_generate(
    prompts: list[str],
    client,
    max_concurrent: int = 5,
) -> list[dict | Exception]:
    semaphore = asyncio.Semaphore(max_concurrent)

    async def _one(prompt: str) -> dict:
        async with semaphore:
            return await client.generate(prompt)

    return await asyncio.gather(*[_one(p) for p in prompts], return_exceptions=True)

Choose max_concurrent based on the provider's documented rate limit. For OpenAI's API tier, 5-10 concurrent requests is a safe starting point.

Timeout handling

Every network call needs an explicit timeout. Without one, a slow or hung provider blocks your task indefinitely:

import asyncio


async def generate_with_timeout(client, prompt: str, timeout_s: float = 30.0) -> dict:
    try:
        return await asyncio.wait_for(client.generate(prompt), timeout=timeout_s)
    except asyncio.TimeoutError:
        raise TimeoutError(f"Provider call timed out after {timeout_s}s")

In practice, set timeouts at two levels: per-request (30–60s for LLM calls) and per-batch (total wall clock limit for the whole gather).

Connection pooling

For high-throughput scenarios, creating a new HTTP connection per request is expensive. Use a shared httpx.AsyncClient or aiohttp.ClientSession across the lifetime of the service:

import httpx
from contextlib import asynccontextmanager


class ProviderClient:
    def __init__(self, base_url: str, api_key: str) -> None:
        self._client = httpx.AsyncClient(
            base_url=base_url,
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=httpx.Timeout(30.0),
            limits=httpx.Limits(max_connections=20, max_keepalive_connections=10),
        )

    async def generate(self, payload: dict) -> dict:
        response = await self._client.post("/v1/generate", json=payload)
        response.raise_for_status()
        return response.json()

    async def close(self) -> None:
        await self._client.aclose()

The limits parameter controls the connection pool size. A pool of 10-20 connections is usually the right balance between throughput and resource consumption.

Retry with exponential backoff

Provider calls fail transiently. A retry wrapper with bounded backoff handles most cases:

import asyncio
import random


async def with_retry(
    coro_fn,
    *args,
    max_attempts: int = 3,
    base_delay: float = 1.0,
    **kwargs,
):
    last_exc = None
    for attempt in range(max_attempts):
        try:
            return await coro_fn(*args, **kwargs)
        except Exception as exc:
            last_exc = exc
            if attempt < max_attempts - 1:
                delay = base_delay * (2 ** attempt) + random.uniform(0, 0.5)
                await asyncio.sleep(delay)
    raise last_exc

The jitter (random.uniform) prevents thundering herd problems when many retries fire simultaneously.

Working example

A complete async batch processor that combines all the patterns:

import asyncio
import logging
import time
from typing import Any

logger = logging.getLogger(__name__)


async def process_batch(
    items: list[dict],
    client: ProviderClient,
    max_concurrent: int = 5,
    timeout_s: float = 30.0,
) -> tuple[list[dict], list[dict]]:
    semaphore = asyncio.Semaphore(max_concurrent)
    successes, failures = [], []

    async def _process_one(item: dict) -> dict:
        async with semaphore:
            start = time.monotonic()
            result = await asyncio.wait_for(
                with_retry(client.generate, item["prompt"]),
                timeout=timeout_s,
            )
            logger.info(
                "item_processed",
                extra={"id": item["id"], "latency_ms": int((time.monotonic() - start) * 1000)},
            )
            return {"id": item["id"], "result": result}

    tasks = [_process_one(item) for item in items]
    raw_results = await asyncio.gather(*tasks, return_exceptions=True)

    for item, result in zip(items, raw_results):
        if isinstance(result, Exception):
            failures.append({"id": item["id"], "error": str(result)})
        else:
            successes.append(result)

    return successes, failures

Common mistakes

Blocking calls inside async functions. time.sleep(), requests.get(), or any synchronous I/O inside an async def blocks the entire event loop. Use asyncio.sleep(), httpx.AsyncClient, or asyncio.to_thread() to run blocking code without stalling other coroutines.

Forgetting return_exceptions=True in gather. Without it, one failure cancels all pending tasks. In a batch of 50 documents, one bad payload kills the other 49.

No timeout on provider calls. A provider that starts hanging instead of returning errors will stall your event loop indefinitely. Always wrap with asyncio.wait_for.

Creating a new HTTP session per request. Session creation involves TCP handshakes and TLS negotiation. Reuse a single AsyncClient for the lifetime of the process or request context.

Treating semaphore count as a performance dial. Max concurrent is an operational decision based on rate limits, not a number to maximize. Setting it too high causes 429 errors; too low wastes throughput. Start conservative and measure.

Try it yourself

1. Take a script that processes documents serially with a synchronous HTTP client. Rewrite it to use asyncio.gather with a semaphore of 5. Measure the wall-clock time difference on 20 documents.
2. Add retry logic to a provider call that currently has none. Verify it handles a transient 500 error without propagating the exception.
3. Profile a batch job: use time.monotonic() around the gather call and log the total latency along with the per-item breakdown. Find which items took longest.