Why this matters

AI workloads are almost entirely network I/O: LLM calls, embedding requests, vector database queries, reranker inference. A single provider call takes 300ms to 5 seconds. If you run 20 provider calls serially, you wait up to 100 seconds. Run them concurrently with proper rate limiting and you finish in 5-10 seconds.

Async Python is the right tool — not threads, not multiprocessing. But async has specific failure modes in AI work: unconstrained concurrency triggers rate limit errors, missing timeouts turn degraded providers into hung services, and streaming responses require a different mental model than request-response.

Core concepts

asyncio.gather for parallel provider calls

The fundamental pattern: fire multiple coroutines simultaneously, collect results when all complete.

import asyncio


async def call_three_providers(prompt: str, claude_client, openai_client, gemini_client) -> dict:
    results = await asyncio.gather(
        claude_client.generate(prompt, model="claude-3-5-sonnet-20241022"),
        openai_client.generate(prompt, model="gpt-4o-mini"),
        gemini_client.generate(prompt, model="gemini-1.5-flash"),
        return_exceptions=True,
    )

    output = {}
    providers = ["claude", "openai", "gemini"]
    for provider, result in zip(providers, results):
        if isinstance(result, Exception):
            output[provider] = {"error": str(result), "success": False}
        else:
            output[provider] = {"content": result.content, "success": True}
    return output

return_exceptions=True is non-negotiable for multi-provider calls. Without it, one provider failure cancels all pending coroutines.

Semaphores for rate limiting concurrent requests

Calling 50 endpoints simultaneously saturates rate limits. A Semaphore constrains concurrency to a safe window:

import asyncio


async def batch_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)

The semaphore does not queue requests — it blocks coroutines until a slot opens. All 50 coroutines are created immediately; only 5 run at a time.

Timeouts at two levels

Every provider call needs a timeout. Set timeouts at two levels:

import asyncio


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


async def batch_with_total_timeout(
    prompts: list[str],
    client,
    max_concurrent: int = 5,
    per_request_s: float = 30.0,
    total_timeout_s: float = 120.0,
) -> list[dict | Exception]:
    semaphore = asyncio.Semaphore(max_concurrent)

    async def _one(prompt: str) -> dict:
        async with semaphore:
            return await generate_with_timeout(client, prompt, per_request_s)

    try:
        return await asyncio.wait_for(
            asyncio.gather(*[_one(p) for p in prompts], return_exceptions=True),
            timeout=total_timeout_s,
        )
    except asyncio.TimeoutError:
        raise TimeoutError(f"Batch timed out after {total_timeout_s}s total")

The per-request timeout catches slow individual calls; the total timeout catches a stalling batch.

Streaming response handling with async generators

Streaming LLM responses arrive as a sequence of tokens. An async generator forwards tokens as they arrive:

from collections.abc import AsyncIterator


async def stream_response(client, prompt: str) -> AsyncIterator[str]:
    async with client.stream(prompt) as response:
        async for chunk in response:
            token = chunk.choices[0].delta.content
            if token:
                yield token


async def accumulate_stream(client, prompt: str) -> str:
    parts = []
    async for token in stream_response(client, prompt):
        parts.append(token)
    return "".join(parts)


async def forward_to_websocket(client, prompt: str, websocket) -> str:
    full_text = []
    async for token in stream_response(client, prompt):
        full_text.append(token)
        await websocket.send_text(token)
    return "".join(full_text)

The generator pattern decouples streaming from consuming: the same stream_response generator works whether accumulating text, forwarding to a WebSocket, or displaying a progress indicator.

Working example

Three providers in parallel with semaphore rate limiting, per-request timeouts, and structured error handling:

import asyncio
import logging
import time
from dataclasses import dataclass

logger = logging.getLogger(__name__)


@dataclass
class ProviderResult:
    provider: str
    content: str | None
    error: str | None
    latency_ms: int
    success: bool


async def multi_provider_eval(
    prompt: str,
    clients: dict,
    max_concurrent: int = 3,
    timeout_s: float = 45.0,
) -> list[ProviderResult]:
    semaphore = asyncio.Semaphore(max_concurrent)

    async def _call_one(name: str, client) -> ProviderResult:
        async with semaphore:
            start = time.monotonic()
            try:
                response = await asyncio.wait_for(client.generate(prompt), timeout=timeout_s)
                latency = int((time.monotonic() - start) * 1000)
                logger.info("provider_ok", extra={"provider": name, "latency_ms": latency})
                return ProviderResult(provider=name, content=response.content, error=None, latency_ms=latency, success=True)
            except Exception as exc:
                latency = int((time.monotonic() - start) * 1000)
                logger.warning("provider_failed", extra={"provider": name, "error": str(exc), "latency_ms": latency})
                return ProviderResult(provider=name, content=None, error=str(exc), latency_ms=latency, success=False)

    tasks = [_call_one(name, client) for name, client in clients.items()]
    results = await asyncio.gather(*tasks)
    successful = [r for r in results if r.success]
    logger.info("eval_complete", extra={"total": len(results), "successful": len(successful)})
    return list(results)

Common mistakes

Blocking calls inside async functions. time.sleep(), requests.get(), synchronous file reads inside async def block the event loop entirely. Use asyncio.sleep(), httpx.AsyncClient, and asyncio.to_thread() for blocking work.

Skipping return_exceptions=True. One failure in asyncio.gather cancels all pending coroutines without it. Always use it for production batch calls.

No timeout on streaming responses. A streaming call that starts but stops mid-stream can hang indefinitely. Wrap streaming calls in asyncio.wait_for or set a read timeout on the HTTP client.

Using max_concurrent as a performance dial. It is a rate limit dial. Set it based on provider documentation. Start at 5, measure for rate limit errors, adjust.

Try it yourself

1. Write an async function that calls two different LLM clients in parallel using asyncio.gather. Add per-request timeouts and return both results (or the error) in a structured dict.
2. Add a semaphore to a batch processing loop you already have. Compare wall-clock time for max_concurrent values of 1, 5, and 10 on a batch of 20 requests.
3. Implement an async generator that streams tokens from a provider. Write an assertion that the joined tokens equal the content a non-streaming call would return.