Python Async Programming for AI

AI workloads are I/O monsters. You're waiting on OpenAI's API, streaming tokens from Claude, fetching embeddings from a vector DB, or pulling training data from S3. Standard synchronous Python processes these one by one. Async lets you orchestrate thousands of these waits simultaneously without spawning thousands of threads.

Why Async Matters for AI

Most AI bottlenecks aren't compute — they're waiting:

Async shines on the first three. It does nothing for the last one.

The 30-Second Primer

import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI()

async def generate(prompt: str) -> str:
    response = await client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

async def main():
    prompts = ["Summarize AI", "Explain async", "Write a haiku"]
    # Run all three concurrently
    results = await asyncio.gather(*(generate(p) for p in prompts))
    print(results)

asyncio.run(main())

async def declares a coroutine. await yields control back to the event loop while waiting. asyncio.gather() runs them concurrently. Three API calls that would take ~9 seconds synchronously finish in ~3.

Real-World AI Patterns

1. Bulk Embedding Generation

async def embed_batch(texts: list[str]) -> list[list[float]]:
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_embedding(session, t) for t in texts]
        return await asyncio.gather(*tasks)

Fire 100 embedding requests at once. The event loop juggles them as responses arrive.

2. Streaming with Backpressure

async def stream_tokens(prompt: str):
    stream = await client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}],
        stream=True
    )
    async for chunk in stream:
        yield chunk.choices[0].delta.content

Process tokens as they arrive — no buffering the full response in memory.

3. Semaphore-Limited Concurrency

Unbounded concurrency gets you rate-limited. Use a semaphore:

semaphore = asyncio.Semaphore(10)

async def safe_generate(prompt: str):
    async with semaphore:
        return await generate(prompt)

The Brutal Truth: When Async Fails

Async is not parallelism. It runs on a single thread. If you drop a CPU-heavy task into the event loop, everything freezes.

# DON'T: Blocks the entire event loop
async def bad_idea():
    result = heavy_model_inference(tensor)  # CPU-bound, 5s
    return result

Fix it with run_in_executor:

from concurrent.futures import ProcessPoolExecutor

executor = ProcessPoolExecutor()

async def good_idea(tensor):
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(executor, heavy_model_inference, tensor)

This offloads CPU work to a separate process, freeing the event loop to keep handling I/O.

Quick Decision Matrix

The Checklist

• Use AsyncOpenAI, AsyncAnthropic, etc. — not the sync clients
• Limit concurrency with asyncio.Semaphore to respect rate limits
• Offload CPU work to ProcessPoolExecutor
• Use asyncio.gather() for batch operations, not loops
• Profile with asyncio.run() — don't call async functions from sync code haphazardly

Bottom Line

Async Python won't make your model train faster. But it will 10x your throughput on API-driven AI pipelines. Master async/await, respect the event loop, and keep CPU work out of it. The result: fewer servers, lower latency, and no more watching one API call finish while ninety-nine others sit in a queue.