Advanced22 min

Async Python and concurrency

Run many slow I/O calls at once with async/await, and know when threads or processes are the right tool instead.

Why this matters in AI / ML / GenAI

An LLM call takes one to five seconds, and almost all of it is waiting on the network. Sequential calls make a batch of 50 documents take minutes. Async turns that into seconds. FastAPI endpoints that call models are async for exactly this reason.

Why async exists

Python's GIL means threads do not speed up CPU-bound work. But most AI application code is I/O-bound: waiting on an HTTP response, a vector database, or a disk read. While waiting, the CPU is idle.

async def defines a coroutine. await says "pause here, let other work run, resume when this finishes". asyncio.gather(...) runs many coroutines concurrently.

The rule of thumb:

Work type	Right tool
Many API calls, DB reads	`asyncio`
Blocking libraries you cannot change	threads (`ThreadPoolExecutor`)
Heavy CPU: tokenizing millions of docs	processes (`ProcessPoolExecutor`)
GPU training	the framework's own loaders

Rules that trip people up

You can only await inside an async def. Calling a coroutine without awaiting it returns a coroutine object and runs nothing — a silent no-op bug.

Never call blocking code inside async. One time.sleep(2) or a synchronous requests.get freezes the entire event loop, including every other in-flight request on your server. Use asyncio.sleep and an async HTTP client (httpx.AsyncClient), or push blocking work to asyncio.to_thread.

Bound your concurrency with asyncio.Semaphore. Firing 500 simultaneous requests at an LLM provider earns you 429 rate limits, not speed. Twenty concurrent calls with retry is the practical shape.

Locally you start the loop with asyncio.run(main()). The compiler on this page already runs inside an event loop, so the examples below use top-level await directly.

Copy-paste examples

Copy into your own editor, or load one into the compiler below and press Run.

Sequential vs concurrent

Five calls of 0.3s each: 1.5s sequential, about 0.3s concurrent.

import asyncio
import time

async def fake_llm_call(i):
    await asyncio.sleep(0.3)
    return f"answer-{i}"

start = time.perf_counter()
sequential = []
for i in range(5):
    sequential.append(await fake_llm_call(i))
seq_ms = (time.perf_counter() - start) * 1000

start = time.perf_counter()
concurrent = await asyncio.gather(*(fake_llm_call(i) for i in range(5)))
conc_ms = (time.perf_counter() - start) * 1000

print("sequential:", sequential, f"{seq_ms:.0f} ms")
print("concurrent:", concurrent, f"{conc_ms:.0f} ms")
print(f"speedup: {seq_ms / conc_ms:.1f}x")

Limit concurrency with a semaphore

Protects you from provider rate limits. Twenty is a sane default.

import asyncio

limit = asyncio.Semaphore(3)
in_flight = {"now": 0, "peak": 0}

async def guarded_call(i):
    async with limit:
        in_flight["now"] += 1
        in_flight["peak"] = max(in_flight["peak"], in_flight["now"])
        await asyncio.sleep(0.1)
        in_flight["now"] -= 1
        return i

results = await asyncio.gather(*(guarded_call(i) for i in range(12)))
print("completed:", len(results))
print("peak concurrent calls:", in_flight["peak"])

Handle partial failures

return_exceptions=True keeps one bad call from killing the whole batch.

import asyncio

async def maybe_fail(i):
    await asyncio.sleep(0.05)
    if i % 3 == 0:
        raise TimeoutError(f"call {i} timed out")
    return f"ok-{i}"

results = await asyncio.gather(*(maybe_fail(i) for i in range(6)), return_exceptions=True)

succeeded = [r for r in results if not isinstance(r, Exception)]
failed = [r for r in results if isinstance(r, Exception)]
print("succeeded:", succeeded)
print("failed:", [str(e) for e in failed])
print(f"success rate: {len(succeeded) / len(results):.0%}")

Async LLM batch (run locally)

Real shape with httpx. Copy into your project after pip install httpx.

# pip install httpx
import asyncio
import httpx

API_URL = "https://api.openai.com/v1/chat/completions"

async def ask(client, semaphore, question, api_key):
    async with semaphore:
        response = await client.post(
            API_URL,
            headers={"Authorization": f"Bearer {api_key}"},
            json={
                "model": "gpt-4.1-mini",
                "messages": [{"role": "user", "content": question}],
                "temperature": 0,
            },
            timeout=30.0,
        )
        response.raise_for_status()
        return response.json()["choices"][0]["message"]["content"]

async def main(questions, api_key):
    semaphore = asyncio.Semaphore(20)
    async with httpx.AsyncClient() as client:
        tasks = [ask(client, semaphore, q, api_key) for q in questions]
        return await asyncio.gather(*tasks, return_exceptions=True)

# asyncio.run(main(["What is RAG?", "What is MLOps?"], "sk-..."))

Batch 10 calls with bounded concurrency

Try it — in-browser Python

Change the semaphore limit to 1 and compare the elapsed time.

Output

Python runs in your browser. First run downloads the runtime.

Press Run (or Ctrl+Enter) to execute.

CPython in WebAssembly. Stdlib works. NumPy and pandas load on demand. No input(), no GPU, no network installs.

Takeaways

async/await wins on I/O-bound work: LLM calls, HTTP, database reads.
One blocking call inside async freezes the whole event loop.
Bound concurrency with a Semaphore and use return_exceptions for partial failures.