Mastering asyncio for High-Throughput AI Apps
If you're running evaluation sweeps, batch-processing documents through an LLM, or building a chatbot that handles concurrent users, sequential API calls will kill your performance. A request to Claude takes 1–5 seconds. Process 100 documents sequentially and you're looking at 2–8 minutes. Process them concurrently and you're down to under 30 seconds — bounded only by your rate limits.
Python's asyncio is the answer. As a C# developer, you already understand async/await — Python's version is conceptually identical but with important runtime differences.
async/await: The Conceptual Mapping
// Async method returns Task<T>
public async Task<string> CallClaudeAsync(
string prompt)
{
var response = await client
.Messages.CreateAsync(request);
return response.Content[0].Text;
}
// Run concurrent tasks
var tasks = prompts
.Select(p => CallClaudeAsync(p));
var results = await Task.WhenAll(tasks);
# async def returns a coroutine
async def call_claude(prompt: str) -> str:
response = await client.chat.completions.create(
model="gpt-4o",
max_tokens=512,
messages=[{"role": "user", "content": prompt}],
)
return response.choices[0].message.content
# Run concurrent coroutines
results = await asyncio.gather(
*[call_claude(p) for p in prompts]
)
The Critical Difference: The Event Loop
In C#, Task.WhenAll uses the thread pool. Python's asyncio.gather runs in a single thread on an event loop. This means:
- asyncio helps I/O-bound work — while one coroutine waits for an API response, the event loop runs other coroutines. No thread overhead.
-
asyncio does NOT help CPU-bound work — if you need to run heavy computation (tokenization, embedding post-processing), asyncio won't parallelize it. Use
multiprocessinginstead (covered in Lesson 1.4). - LLM API calls are always I/O-bound — asyncio is exactly right for concurrent API requests.
The Async OpenAI Client
The OpenAI SDK ships both sync and async clients. Use AsyncOpenAI for concurrent workloads:
import asyncio
from openai import OpenAI
# Use AsyncOpenAI for async code
async_client = AsyncOpenAI(api_key="...")
async def summarize(text: str, client: AsyncOpenAI) -> str:
"""Async version — awaitable."""
response = await client.chat.completions.create(
model="gpt-4o",
max_tokens=256,
messages=[{
"role": "user",
"content": f"Summarize in 2 sentences:\n{text}",
}],
)
return response.choices[0].message.content
async def main():
async with AsyncOpenAI() as client:
result = await summarize("Python is a programming language...", client)
print(result)
# Entry point: run the event loop
asyncio.run(main())
asyncio.gather: Concurrent Requests
asyncio.gather() is the equivalent of Task.WhenAll(). It runs multiple coroutines concurrently and collects their results in order:
import asyncio
import time
from openai import OpenAI
async def process_document(
doc_id: str,
text: str,
client: AsyncOpenAI,
) -> dict:
"""Process a single document through the LLM."""
try:
response = await client.chat.completions.create(
model="gpt-4o",
max_tokens=512,
messages=[{
"role": "user",
"content": f"Extract key topics from this text:\n\n{text}",
}],
)
return {
"id": doc_id,
"topics": response.choices[0].message.content,
"tokens": response.usage.completion_tokens,
"status": "success",
}
except Exception as e:
return {"id": doc_id, "status": "error", "error": str(e)}
async def process_batch(documents: list[dict]) -> list[dict]:
"""Process all documents concurrently."""
start = time.monotonic()
async with AsyncOpenAI() as client:
# Create all coroutines — they don't run yet
tasks = [
process_document(doc["id"], doc["text"], client)
for doc in documents
]
# Run all concurrently, collect results in order
results = await asyncio.gather(*tasks)
elapsed = time.monotonic() - start
print(f"Processed {len(documents)} documents in {elapsed:.1f}s")
return list(results)
# Demo: 10 documents processed concurrently
documents = [{"id": f"doc_{i}", "text": f"Document {i} content..."} for i in range(10)]
results = asyncio.run(process_batch(documents))
Performance Comparison
This is the payoff that makes asyncio worth the mental overhead. When you fire ten LLM calls sequentially, total time is the sum of all ten; when you run them concurrently with asyncio.gather, total time is roughly the slowest single call, because the waiting overlaps. Since AI apps spend almost all their time waiting on network I/O, this is often a 5–10x throughput win for the same hardware.
import asyncio
import time
from openai import OpenAI
PROMPTS = [f"What is {topic}?" for topic in [
"machine learning", "neural networks", "transformers", "embeddings",
"RAG", "fine-tuning", "RLHF", "attention mechanism",
]]
# Sequential — naive approach
def run_sequential(client: OpenAI) -> list[str]:
return [
client.chat.completions.create(
model="gpt-4o-mini-20240307", # fast model for demo
max_tokens=64,
messages=[{"role": "user", "content": p}],
).choices[0].message.content
for p in PROMPTS
]
# Concurrent with asyncio
async def run_concurrent(client: AsyncOpenAI) -> list[str]:
async def one(prompt: str) -> str:
r = await client.chat.completions.create(
model="gpt-4o-mini-20240307",
max_tokens=64,
messages=[{"role": "user", "content": prompt}],
)
return r.choices[0].message.content
return await asyncio.gather(*[one(p) for p in PROMPTS])
# Sequential: ~16s (8 × ~2s per request)
# Concurrent: ~3s (all 8 in parallel, limited by slowest)
# Speedup: ~5x — limited by rate limits and the slowest single request
Controlling Concurrency with Semaphores
Launching 1000 concurrent requests will hit rate limits instantly. A asyncio.Semaphore limits how many coroutines run at once — like a thread pool size in .NET:
import asyncio
from openai import OpenAI
async def process_with_semaphore(
documents: list[dict],
max_concurrent: int = 10, # match your API rate limit tier
) -> list[dict]:
semaphore = asyncio.Semaphore(max_concurrent)
async def process_one(doc: dict, client: AsyncOpenAI) -> dict:
async with semaphore: # blocks if max_concurrent already running
return await process_document(doc["id"], doc["text"], client)
async with AsyncOpenAI() as client:
tasks = [process_one(doc, client) for doc in documents]
results = await asyncio.gather(*tasks, return_exceptions=True)
# Filter out exceptions from results
successes = [r for r in results if not isinstance(r, Exception)]
errors = [r for r in results if isinstance(r, Exception)]
print(f"Completed: {len(successes)}, Errors: {len(errors)}")
return successes
Async Context Managers and Generators
Two async variants of familiar constructs matter for AI work. An async context manager (async with) cleanly opens and closes resources like an HTTP client or connection pool — the await using of the Python world. An async generator (async for over yield) is how you consume a streaming response token-by-token without buffering the whole thing, the equivalent of an IAsyncEnumerable.
import asyncio
from openai import OpenAI
# Async context manager — same as with statement, but for async resources
async def stream_response(prompt: str) -> None:
"""Stream a response token by token."""
async with AsyncOpenAI() as client:
async with client.chat.completions.stream(
model="gpt-4o",
max_tokens=512,
messages=[{"role": "user", "content": prompt}],
) as stream:
async for text in stream.text_stream:
print(text, end="", flush=True)
print() # newline at end
# Async generator for processing a stream
async def process_stream_chunks(prompt: str):
"""Yield chunks as they arrive."""
async with AsyncOpenAI() as client:
async with client.chat.completions.stream(
model="gpt-4o",
max_tokens=512,
messages=[{"role": "user", "content": prompt}],
) as stream:
async for text in stream.text_stream:
yield text # async generator
async def main():
await stream_response("Explain Python asyncio in one paragraph.")
asyncio.run(main())
asyncio Error Handling
Concurrency complicates error handling: when you run many calls together, one failure shouldn't necessarily sink the rest. The key tool is asyncio.gather(..., return_exceptions=True), which collects exceptions alongside successful results instead of aborting the whole batch — essential when you're processing a thousand documents and a handful hit rate limits. The examples show handling per-task failures and enforcing timeouts.
import asyncio
async def safe_process(doc: dict, client) -> dict:
"""Process one document, handling errors gracefully."""
try:
return await process_document(doc["id"], doc["text"], client)
except openai.RateLimitError:
# Simple per-task backoff
await asyncio.sleep(5)
return await process_document(doc["id"], doc["text"], client)
except Exception as e:
return {"id": doc["id"], "status": "error", "error": str(e)}
async def gather_with_errors(documents: list[dict]) -> tuple[list, list]:
"""Gather results, separating successes from failures."""
async with AsyncOpenAI() as client:
# return_exceptions=True prevents one failure from canceling all
results = await asyncio.gather(
*[safe_process(doc, client) for doc in documents],
return_exceptions=True,
)
successes = [r for r in results if isinstance(r, dict)]
failures = [r for r in results if isinstance(r, Exception)]
return successes, failures
Never call a blocking sync function (like time.sleep() or the sync OpenAI client) inside an async function. It blocks the entire event loop — all other coroutines pause. Use await asyncio.sleep() for delays, and the AsyncOpenAI client for API calls. For truly blocking operations, use loop.run_in_executor() to offload to a thread.
Key Takeaways
- asyncio is single-threaded cooperative concurrency — it helps I/O-bound work (API calls, network, disk) but NOT CPU-bound computation
- Use
AsyncOpenAI()for async LLM calls;await client.chat.completions.create()yields control to the event loop while waiting -
asyncio.gather(*coroutines)runs multiple coroutines concurrently — equivalent to C#'sTask.WhenAll() -
asyncio.Semaphore(n)limits concurrency tonsimultaneous operations — essential for staying within API rate limits - Use
return_exceptions=Trueingather()so a single failure doesn't cancel all other concurrent tasks - Entry point:
asyncio.run(main())starts the event loop; only call it once at the top level
Comments
Loading comments…