Your personal operating system for career transition.

The async patterns that actually matter for AI workloads

Async Python is not automatically faster. The patterns that produce real throughput gains for AI workloads are specific, and the patterns that introduce new failure modes are equally specific. This article covers both.

Why async matters for AI workloads

LLM calls take 300ms to 3 seconds. Embedding 1000 documents at 100ms each takes 100 seconds serially. With async and a semaphore of 10, it takes 10 seconds. The math is straightforward: async concurrency is the single highest-leverage optimization for I/O-bound AI work.

CPU-bound work (document parsing, tokenization, scoring) does not benefit from async. Use ProcessPoolExecutor for CPU-bound parallelism.

Pattern 1: asyncio.gather with return_exceptions=True

async def batch_process(items, fn, max_concurrent=5):
    semaphore = asyncio.Semaphore(max_concurrent)

    async def _one(item):
        async with semaphore:
            return await fn(item)

    raw = await asyncio.gather(*[_one(i) for i in items], return_exceptions=True)
    successes = [(i, r) for i, r in zip(items, raw) if not isinstance(r, Exception)]
    failures = [(i, r) for i, r in zip(items, raw) if isinstance(r, Exception)]
    return successes, failures

return_exceptions=True is the most important parameter. Without it, one failure cancels all pending tasks. With it, you handle each item's result independently.

Pattern 2: asyncio.wait_for for per-call timeouts

async def call_with_timeout(client, prompt, timeout_s=30.0):
    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")

Every provider call needs an explicit timeout. Without one, a hung provider blocks your task indefinitely. The right timeout is your P95 call latency × 2, not an arbitrary constant.

Pattern 3: Shared AsyncClient lifecycle

import httpx

class ProviderClient:
    def __init__(self, base_url, api_key):
        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):
        resp = await self._client.post("/v1/generate", json=payload)
        resp.raise_for_status()
        return resp.json()

    async def close(self):
        await self._client.aclose()

A shared AsyncClient reuses TCP connections. Creating one client per request means a new TCP handshake and TLS negotiation for every call. At 10 requests/second, that overhead is measurable.

Common mistakes that introduce failure modes

Blocking calls inside async functions. time.sleep(), requests.get(), synchronous file I/O, or any other blocking call inside async def stalls the entire event loop. Use asyncio.sleep(), httpx.AsyncClient, and asyncio.to_thread() for blocking code.

# WRONG: blocks the event loop
async def bad_read(path):
    return open(path).read()  # blocking I/O

# RIGHT: runs blocking I/O in a thread pool
async def good_read(path):
    return await asyncio.to_thread(Path(path).read_text)

Missing timeout on gather. Per-item timeouts catch slow individual calls. But if you need a hard wall clock limit for the entire batch, add a timeout on the gather itself:

try:
    results = await asyncio.wait_for(
        asyncio.gather(*tasks, return_exceptions=True),
        timeout=total_budget_s,
    )
except asyncio.TimeoutError:
    # Some tasks are still running — they are now abandoned
    pass

Treating semaphore count as a performance maximizer. max_concurrent is an operational constraint (provider rate limit, downstream capacity), not a number to maximize. Start with 5, measure, and increase carefully.

When to use asyncio vs ProcessPoolExecutor

Work type	Use
API calls, network I/O	asyncio.gather
Database queries	asyncio with an async driver
CPU-intensive parsing	ProcessPoolExecutor
CPU + I/O mix	asyncio for I/O, to_thread for CPU stages

Practical starting point

For any AI batch job:

1. Start with asyncio.gather + Semaphore(5) + wait_for(timeout=30)
2. Measure actual throughput and provider 429 rate
3. Adjust max_concurrent based on measurements, not intuition