Async Python for AI Pipelines: Concurrency, Backpressure, and Not Melting Your API Bill

An LLM call is 500ms–30s of waiting on somebody else's GPU. A vector search is 20–200ms of waiting on somebody else's index. An embedding batch is a network round trip. If you write this code with threads you pay 8MB per idle worker; with processes you pay much more; with asyncio a single Python process happily juggles thousands of in-flight requests using kilobytes of memory each.

The catch is that async is unforgiving of sloppy patterns. This article is the small set of primitives — Semaphore, TaskGroup, wait_for, retries with jitter, streaming — that make async pipelines actually production-safe. It pairs with building production AI agents with n8n (when n8n isn't the right tool and you drop to Python) and designing REST APIs LLMs can use (the server side of the same story).

An asyncio program has one event loop per thread. Coroutines are cooperative — they voluntarily await at IO points, which lets the loop switch to another ready coroutine. If any coroutine does blocking work (a synchronous requests.get(), a CPU-heavy loop, a time.sleep()), it freezes the entire loop and every other in-flight request stalls.

The three rules that follow from this:

1. Use async-native libraries for all IO — httpx.AsyncClient, asyncpg, aiofiles, redis.asyncio. Never call the sync version.
2. Wrap unavoidably-blocking calls in asyncio.to_thread(...) so they run on the default thread pool and don't block the loop.
3. Debug loop stalls with PYTHONASYNCIODEBUG=1 — it will print warnings when any callback runs longer than 100ms.

The single most important pattern in async AI code is bounding concurrency. If you have 10,000 records and naïvely do await asyncio.gather(*[embed(r) for r in records]), you'll instantly hit the provider's rate limit, get a wave of 429s, and either lose half the batch or trigger an angry email from your account manager.

The correct pattern is a Semaphore sized to the concurrency budget the provider actually gives you:

import asyncio
from typing import Awaitable, Callable, TypeVar, Sequence

T = TypeVar("T")
R = TypeVar("R")

async def bounded_gather(
    items: Sequence[T],
    worker: Callable[[T], Awaitable[R]],
    concurrency: int,
) -> list[R]:
    sem = asyncio.Semaphore(concurrency)

    async def _run(item: T) -> R:
        async with sem:
            return await worker(item)

    return await asyncio.gather(*(_run(i) for i in items))

# usage
results = await bounded_gather(records, embed_one, concurrency=32)

Set concurrency based on the provider's documented request-per-second limit, divided by your average request latency. For OpenAI-tier-3 embeddings that's usually 30–100 concurrent requests; for chat completions it's 10–40. Start conservative, watch your 429 rate, and only raise it when you see the rate below 1%.

asyncio does not enforce timeouts for you. A hung TCP connection to a provider that suddenly stops responding will happily consume a semaphore slot forever, and after a few of those your whole pipeline is deadlocked with zero visible errors.

Two timeouts, always:

import httpx

# Per-request timeout on the client
client = httpx.AsyncClient(
    timeout=httpx.Timeout(
        connect=5.0,   # TCP handshake
        read=60.0,     # waiting for LLM to stream
        write=10.0,    # sending the request body
        pool=5.0,      # waiting for a connection from the pool
    ),
    limits=httpx.Limits(max_connections=100, max_keepalive_connections=20),
)

# Per-batch deadline
async with asyncio.timeout(600):  # 10-minute hard cap for the whole batch
    results = await bounded_gather(records, embed_one, concurrency=32)

The Python 3.11 asyncio.timeout() context manager is the modern replacement for asyncio.wait_for() — it composes with TaskGroup and reports cleaner tracebacks. Use it for anything with a natural deadline.

429s and 5xxs are normal at scale. Retrying naively is the fastest way to turn a temporary blip into an outage — every client comes back at the same moment and re-hammers the provider. The fix is exponential backoff with full jitter:

import asyncio, random
from httpx import HTTPStatusError, TimeoutException

RETRYABLE = {408, 429, 500, 502, 503, 504}

async def with_retry(fn, *, attempts: int = 5, base: float = 0.5, cap: float = 30.0):
    for attempt in range(attempts):
        try:
            return await fn()
        except HTTPStatusError as e:
            if e.response.status_code not in RETRYABLE or attempt == attempts - 1:
                raise
        except (TimeoutException, ConnectionError):
            if attempt == attempts - 1:
                raise
        # Full jitter: sleep is a uniform random in [0, min(cap, base * 2**attempt)]
        delay = random.uniform(0, min(cap, base * (2 ** attempt)))
        await asyncio.sleep(delay)

• Only retry idempotent failures — 400/401/403/422 mean the request itself is wrong; retrying won't fix it and might charge you.
• Honor Retry-After — when the provider tells you when to come back, obey; don't guess.
• Cap total wait — a request that would take 15 minutes across retries should just fail; something upstream is broken.
• Log each retry with a request id — silent retries hide problems until they compound.

asyncio.gather() has a fatal flaw: if one task fails and you didn't pass return_exceptions=True, siblings keep running in the background and their exceptions are silently dropped. TaskGroup (Python 3.11+) fixes this — one failure cancels the group, all tasks are awaited, and exceptions are collected into an ExceptionGroup.

async def process_document(doc):
    async with asyncio.TaskGroup() as tg:
        summary_task = tg.create_task(summarize(doc))
        entities_task = tg.create_task(extract_entities(doc))
        embed_task = tg.create_task(embed(doc))
    # All three completed successfully, or the whole block raised.
    return {
        "summary": summary_task.result(),
        "entities": entities_task.result(),
        "embedding": embed_task.result(),
    }

Use TaskGroup whenever tasks share a logical unit of work. Use plain gather(..., return_exceptions=True) only for genuinely independent fan-out where one failure shouldn't kill the batch.

LLM chat completions stream tokens for a reason — perceived latency drops from 8 seconds to 200ms when you show the first token immediately. But naïvely collecting the stream into a string defeats the point. The pattern:

async def stream_answer(prompt: str, on_token):
    async with client.stream(
        "POST",
        "https://ai.gateway.lovable.dev/v1/chat/completions",
        headers={"Lovable-API-Key": os.environ["LOVABLE_API_KEY"]},
        json={"model": "google/gemini-2.5-flash", "messages": [...], "stream": True},
    ) as resp:
        resp.raise_for_status()
        async for line in resp.aiter_lines():
            if not line.startswith("data: "):
                continue
            payload = line[6:]
            if payload == "[DONE]":
                break
            chunk = json.loads(payload)
            token = chunk["choices"][0]["delta"].get("content")
            if token:
                await on_token(token)

Use streaming when a human is waiting on the other end (chat UIs, live assistants). Buffer when a downstream service needs the whole response (a summarizer feeding a database, an agent evaluating a tool call). Mixing the two — streaming while also buffering server-side — gives you the best of both.

Creating a new HTTPS client per request adds a TLS handshake (~50–200ms) to every call. In async code this multiplies badly. The fix is one AsyncClient per process, held for the process lifetime:

# module-level singleton
_client: httpx.AsyncClient | None = None

def get_client() -> httpx.AsyncClient:
    global _client
    if _client is None:
        _client = httpx.AsyncClient(
            http2=True,
            timeout=httpx.Timeout(60.0, connect=5.0),
            limits=httpx.Limits(max_connections=200, max_keepalive_connections=50),
        )
    return _client

# on shutdown
async def close_client():
    if _client is not None:
        await _client.aclose()

For Postgres, use asyncpg's pool with 10–20 connections per worker; for Redis, redis.asyncio's connection pool defaults are sensible. Never open a fresh connection per query in an async handler.

A queue-fed async worker that pulls from Redis/Kafka/SQS faster than it can process silently accumulates in-memory tasks until it OOMs. Backpressure is the mechanism that stops that:

async def worker(queue_url: str, max_in_flight: int = 32):
    sem = asyncio.Semaphore(max_in_flight)
    async with asyncio.TaskGroup() as tg:
        while True:
            await sem.acquire()  # blocks if we're at capacity
            msg = await pull_one(queue_url)
            if msg is None:
                sem.release()
                await asyncio.sleep(1)
                continue

            async def _handle(m=msg):
                try:
                    await process(m)
                    await ack(m)
                finally:
                    sem.release()

            tg.create_task(_handle())

This pattern is the async equivalent of an SQS long-poll worker with a bounded prefetch. It composes cleanly with the retry helper above and produces steady memory usage instead of a sawtooth.

Async code hides bugs. A silently-swallowed exception, a task that never awakens, a semaphore leaked because of an early return — these are invisible without instrumentation. The minimum:

• Structured logs with a request id propagated via contextvars.ContextVar, not thread-locals.
• OpenTelemetry spans around each network call. Every provider SDK has an OTel integration in 2026.
• Metrics: in-flight count, queue depth, retry count, 429 rate, p50/p95/p99 latency per provider.
• PYTHONASYNCIODEBUG=1 in staging to catch slow callbacks and un-awaited coroutines.

For the full AI-specific observability story (prompts, tokens, drift), see observability for AI systems.

Pulling all the patterns together into something you can adapt:

import asyncio, os, random, json
import httpx

CONCURRENCY = 32
BATCH_SIZE = 100

client = httpx.AsyncClient(
    base_url="https://ai.gateway.lovable.dev/v1",
    headers={"Lovable-API-Key": os.environ["LOVABLE_API_KEY"]},
    timeout=httpx.Timeout(60.0, connect=5.0),
    limits=httpx.Limits(max_connections=200, max_keepalive_connections=50),
    http2=True,
)

async def embed_batch(texts: list[str]) -> list[list[float]]:
    async def _call():
        r = await client.post(
            "/embeddings",
            json={"model": "google/text-embedding-004", "input": texts},
        )
        r.raise_for_status()
        return [d["embedding"] for d in r.json()["data"]]
    return await with_retry(_call)

async def run(records: list[dict]):
    sem = asyncio.Semaphore(CONCURRENCY)
    batches = [records[i:i+BATCH_SIZE] for i in range(0, len(records), BATCH_SIZE)]

    async def _process(batch):
        async with sem:
            vecs = await embed_batch([r["text"] for r in batch])
            await save_batch(batch, vecs)

    async with asyncio.timeout(3600):  # 1-hour hard cap
        async with asyncio.TaskGroup() as tg:
            for b in batches:
                tg.create_task(_process(b))

On a modest VM this processes 100k documents in 8–15 minutes at a total cost of a few dollars, with steady memory around 200MB. The same shape works for chat completions (change the endpoint and payload).

• CPU-bound work — image resizing, PDF parsing, local model inference. Use processes or a job queue.
• Long-running batch jobs that outlive a single process. Use a real workflow engine (Airflow, Dagster) — see modern data stack 2026.
• Team unfamiliarity — async in the wrong hands produces bugs that are harder to reason about than the throughput was worth. Sync + threads is fine for 100 QPS.
• Multi-machine fan-out — asyncio doesn't cross process boundaries. Use Celery/Arq/Redis Streams.

The bar is: is my bottleneck concurrent IO in a single process? If yes, asyncio. If no, something else.

• bare asyncio.gather() on a list of 10k coroutines — always bound with a Semaphore.
• No timeouts — one stuck request eats the whole pipeline.
• Retries without jitter — thundering herd against the provider.
• Retries on 4xx — retrying validation errors just wastes money.
• A fresh httpx.AsyncClient per request — kills throughput via TLS handshakes.
• Sync IO inside a coroutine — requests.get(), time.sleep(), psycopg2. All freeze the loop.
• asyncio.run() called from inside an existing loop — raises at runtime; use await or get_event_loop() pattern.
• Global mutable state without a Lock — coroutines can race between awaits.