Free lesson · AI Engineering
Async Python
Async Python
Why Async Is Not Optional for AI Applications
At Meta, the first version of our AI documentation assistant processed requests sequentially. A user query triggered this sequence: embed the query (200ms API call to embedding service), search the vector database (150ms), retrieve the full documents (100ms), call the LLM (3,000ms), and apply guardrails (80ms). Total: approximately 3,530ms per request. One request at a time. If ten users submitted queries simultaneously, the tenth user waited 35 seconds.
We rewrote the pipeline with async Python. The embed and search calls - which were independent - ran in parallel instead of sequentially. The LLM call streamed tokens while guardrails ran asynchronously on completed output segments. The per-request latency dropped from 3,530ms to 3,250ms (modest improvement - the LLM call dominates). But the throughput exploded: the server now handled 50 concurrent requests instead of 1, because while one request was waiting for the LLM, the server processed embedding and search calls for other requests.
This is why async matters for AI engineering. AI applications spend most of their time waiting - waiting for API responses, waiting for database queries, waiting for model inference. While one request waits, the server can do work for other requests. Sequential code wastes 95% of the server's capacity on waiting. Async code uses that waiting time to serve other users.
Understanding Async: The Restaurant Analogy
In a synchronous restaurant, the waiter takes your order, walks to the kitchen, stands there watching the chef cook your meal, carries the meal back to you, and only then goes to the next table. The waiter serves one table at a time. If the kitchen takes 20 minutes, the waiter is idle for 20 minutes.
In an async restaurant, the waiter takes your order, gives it to the kitchen, and immediately goes to the next table. When the kitchen signals that your meal is ready, the waiter picks it up and delivers it. The waiter serves many tables concurrently because they do not wait for each meal to be prepared.
Python's asyncio library implements the async restaurant. An async function is like a waiter who can leave and come back. The await keyword is where the waiter says "I'll come back when this is ready" and goes to serve another table. The event loop is the restaurant manager who keeps track of which meals are ready and tells the waiter where to go next.
Critically important: Async is concurrency, not parallelism. One waiter (one thread) serves many tables (many requests) by interleaving work. The waiter does not clone themselves. They simply do not waste time standing idle. For AI applications, this is exactly the right model - the bottleneck is not CPU computation (which would need parallelism via multiprocessing), it is I/O waiting (network calls to APIs, databases, and model providers), which async handles perfectly. If you need CPU parallelism (heavy numerical computation, image processing), use multiprocessing or concurrent.futures.ProcessPoolExecutor instead of or alongside asyncio.
The httpx alternative to aiohttp: While aiohttp is the established async HTTP library, httpx has gained significant adoption because it provides both sync and async interfaces with the same API. The OpenAI and Anthropic Python SDKs use httpx internally and provide native async clients (AsyncOpenAI, AsyncAnthropic), so for LLM API calls specifically, you rarely need to write raw HTTP calls - the SDK handles async for you.
The Async Basics: async/await
import asyncio
import time
# A synchronous function - blocks while waiting
def sync_embed(text: str) -> list[float]:
time.sleep(0.2) # Simulate 200ms API call
return [0.1, 0.8, 0.3]
# An async function - yields control while waiting
async def async_embed(text: str) -> list[float]:
await asyncio.sleep(0.2) # Non-blocking 200ms wait
return [0.1, 0.8, 0.3]
# Sequential: embed 5 texts one at a time
async def sequential_embed(texts: list[str]):
start = time.time()
results = []
for text in texts:
result = await async_embed(text)
results.append(result)
elapsed = time.time() - start
print(f"Sequential: {elapsed:.2f}s for {len(texts)} embeddings")
return results
# Concurrent: embed 5 texts simultaneously
async def concurrent_embed(texts: list[str]):
start = time.time()
tasks = [async_embed(text) for text in texts]
results = await asyncio.gather(*tasks)
elapsed = time.time() - start
print(f"Concurrent: {elapsed:.2f}s for {len(texts)} embeddings")
return results
# Run both
async def main():
texts = ["text1", "text2", "text3", "text4", "text5"]
await sequential_embed(texts) # ~1.0s (5 × 0.2s)
await concurrent_embed(texts) # ~0.2s (all 5 run simultaneously)
asyncio.run(main())
The key insight: Sequential embedding of 5 texts takes 1.0 seconds (5 × 0.2s). Concurrent embedding takes 0.2 seconds (all 5 calls run simultaneously, total time equals the slowest single call). The difference grows with scale: 100 sequential embeddings take 20 seconds; 100 concurrent embeddings take 0.2 seconds. For AI pipelines that make dozens or hundreds of API calls per request, async is the difference between seconds and minutes.
asyncio.gather: Running Multiple Async Tasks
asyncio.gather() is the primary tool for running multiple async operations concurrently. It takes multiple coroutines (async function calls), runs them all simultaneously, and returns the results in the same order the coroutines were provided.
import asyncio
import time
async def call_embedding_api(text: str) -> dict:
"""Simulate an embedding API call."""
await asyncio.sleep(0.15)
return {"text": text, "embedding": [0.1, 0.2, 0.3]}
async def call_search_api(embedding: list[float]) -> dict:
"""Simulate a vector search API call."""
await asyncio.sleep(0.1)
return {"results": [{"text": "result", "score": 0.94}]}
async def call_llm_api(prompt: str) -> dict:
"""Simulate an LLM API call."""
await asyncio.sleep(0.5)
return {"response": f"Answer to: {prompt[:30]}..."}
# Pattern 1: Independent calls in parallel
async def parallel_independent():
"""Run independent operations simultaneously."""
start = time.time()
# These three calls have no dependencies on each other
embedding, search, llm = await asyncio.gather(
call_embedding_api("query text"),
call_search_api([0.1, 0.2]),
call_llm_api("What is RAG?")
)
elapsed = time.time() - start
print(f"Parallel independent: {elapsed:.2f}s (max of individual calls, not sum)")
return embedding, search, llm
# Pattern 2: Some calls depend on others (sequential + parallel mix)
async def pipeline_with_parallelism():
"""A RAG pipeline with parallel steps where possible."""
start = time.time()
# Step 1: Embed the query (must happen first)
embedding_result = await call_embedding_api("What is RAG?")
# Step 2: Search + prepare prompt can happen in parallel
search_task = call_search_api(embedding_result["embedding"])
# (In a real pipeline, prompt preparation might involve other async work)
search_result = await search_task
# Step 3: Call LLM with search results (must happen after search)
context = search_result["results"][0]["text"]
llm_result = await call_llm_api(f"Context: {context}\nQuestion: What is RAG?")
elapsed = time.time() - start
print(f"Pipeline: {elapsed:.2f}s")
return llm_result
asyncio.run(parallel_independent())
asyncio.run(pipeline_with_parallelism())
aiohttp: Async HTTP Requests
The standard requests library is synchronous - it blocks while waiting for HTTP responses. For async AI applications, use aiohttp (or httpx with async support) to make non-blocking HTTP requests.
import aiohttp
import asyncio
import json
async def call_openai_async(session: aiohttp.ClientSession,
prompt: str,
api_key: str) -> dict:
"""Make an async call to the OpenAI API."""
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "gpt-4o-mini",
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 200
}
async with session.post(
"https://api.openai.com/v1/chat/completions",
headers=headers,
json=payload
) as response:
result = await response.json()
return result
async def batch_llm_calls(prompts: list[str], api_key: str) -> list[dict]:
"""Process multiple prompts concurrently."""
async with aiohttp.ClientSession() as session:
tasks = [
call_openai_async(session, prompt, api_key)
for prompt in prompts
]
results = await asyncio.gather(*tasks, return_exceptions=True)
# Handle errors gracefully
processed = []
for i, result in enumerate(results):
if isinstance(result, Exception):
print(f" Prompt {i} failed: {result}")
processed.append({"error": str(result)})
else:
processed.append(result)
return processed
The return_exceptions=True pattern: By default, asyncio.gather raises the first exception and cancels the remaining tasks. With return_exceptions=True, exceptions are returned as values in the results list rather than raised. This is critical for batch LLM calls - if one call out of fifty fails (rate limit, timeout), you want the other 49 results, not a crash. Check each result for isinstance(result, Exception) to identify and handle failures.
Concurrency Control: Semaphores
Making 1,000 concurrent API calls will trigger rate limits immediately. Semaphores control the number of concurrent operations, keeping you within provider rate limits.
import asyncio
import time
async def rate_limited_api_call(semaphore: asyncio.Semaphore,
call_id: int) -> dict:
"""API call with concurrency control."""
async with semaphore: # Only N calls proceed simultaneously
await asyncio.sleep(0.1) # Simulate API call
return {"id": call_id, "status": "success"}
async def batch_with_rate_limit():
"""Process 20 calls with max 5 concurrent."""
semaphore = asyncio.Semaphore(5) # Max 5 concurrent calls
start = time.time()
tasks = [rate_limited_api_call(semaphore, i) for i in range(20)]
results = await asyncio.gather(*tasks)
elapsed = time.time() - start
print(f"20 calls with max 5 concurrent: {elapsed:.2f}s")
print(f" (Without limit: ~0.1s, Sequential: ~2.0s)")
print(f" Successful: {sum(1 for r in results if r['status'] == 'success')}")
asyncio.run(batch_with_rate_limit())
Setting the semaphore value: Match it to the provider's rate limit. If the provider allows 60 requests per minute (1 per second), set the semaphore to the number of concurrent requests that stays within that rate. For OpenAI with a 500 RPM limit, a semaphore of 8-10 keeps you well within the limit while providing good throughput. For self-hosted models with no rate limit, set the semaphore to the number of concurrent requests the model can handle without degrading latency (typically the GPU batch size).
Async Patterns for AI Pipelines
Pattern 1: Fan-Out / Fan-In
Process multiple items in parallel (fan-out), then combine results (fan-in). This is the pattern for batch embedding, batch classification, and any operation applied to multiple independent inputs.
import asyncio
import time
async def embed_chunk(chunk: str, chunk_id: int) -> dict:
"""Embed a single chunk (simulated)."""
await asyncio.sleep(0.1)
return {"chunk_id": chunk_id, "embedding": [0.1 * chunk_id]}
async def fan_out_fan_in():
"""Embed 10 chunks in parallel, collect all results."""
chunks = [f"chunk_{i}" for i in range(10)]
start = time.time()
# Fan-out: create all tasks
tasks = [embed_chunk(chunk, i) for i, chunk in enumerate(chunks)]
# Fan-in: wait for all to complete
results = await asyncio.gather(*tasks)
elapsed = time.time() - start
print(f"Fan-out/fan-in: {len(results)} embeddings in {elapsed:.2f}s")
print(f" (Sequential would take: {len(results) * 0.1:.1f}s)")
asyncio.run(fan_out_fan_in())
Pattern 2: Pipeline with Async Stages
Each stage of the pipeline is async, enabling concurrent request handling across the full pipeline.
async def rag_pipeline(query: str) -> str:
"""Complete RAG pipeline with async stages."""
# Stage 1: Embed query
query_embedding = await embed_async(query)
# Stage 2: Search (depends on stage 1)
search_results = await search_async(query_embedding, top_k=5)
# Stage 3: Assemble context (CPU-bound, but fast)
context = assemble_context(search_results)
# Stage 4: Generate response (depends on stage 3)
response = await llm_generate_async(query, context)
# Stage 5: Apply guardrails (depends on stage 4)
safe_response = await guardrails_async(response)
return safe_response
The pipeline itself is sequential (each stage depends on the previous). But because each stage is async, the server handles other requests while any stage is waiting. While request A waits for the LLM (3 seconds), requests B through F can be embedding, searching, and assembling context.
graph TD
subgraph Sync["Synchronous Server"]
S1["Request 1: embed → search → LLM → respond\n(3.5 seconds)"]
S2["Request 2: waiting...\n(starts after request 1 finishes)"]
S3["Request 3: waiting...\n(starts after request 2 finishes)"]
end
subgraph Async["Async Server"]
A1["Request 1: embed → search → LLM (waiting)"]
A2["Request 2: embed → search (while R1 waits for LLM)"]
A3["Request 3: embed (while R1 and R2 wait)"]
A4["All 3 complete in ~3.5s total\n(not 10.5s)"]
end
style Sync fill:#991b1b,stroke:#fca5a5,stroke-width:2px
style Async fill:#166534,stroke:#86efac,stroke-width:2px
This diagram contrasts synchronous and asynchronous server behaviour for AI requests. A synchronous server processes one request at a time: Request 1 takes 3.5 seconds, Request 2 waits until Request 1 finishes, and Request 3 waits until Request 2 finishes - totalling 10.5 seconds for three requests. An async server interleaves work: while Request 1 waits for the LLM, Request 2 performs embedding and search, and Request 3 starts embedding. All three requests complete in approximately 3.5 seconds total because the waiting time is used productively.
graph LR
subgraph Patterns["Async Patterns for AI"]
P1["Fan-Out / Fan-In\nBatch embed 100 chunks\n→ asyncio.gather()"]
P2["Pipeline Stages\nEmbed → Search → Generate\n→ each stage async"]
P3["Rate-Limited Batch\n1000 API calls at 10 concurrent\n→ asyncio.Semaphore"]
P4["Streaming\nYield tokens as generated\n→ async generator"]
end
style Patterns fill:#312e81,stroke:#a5b4fc,stroke-width:2px
This diagram lists four async patterns for AI applications. Fan-out/fan-in uses asyncio.gather to batch-embed 100 chunks in parallel. Pipeline stages make each step (embed, search, generate) async so the server handles other requests during waits. Rate-limited batching uses asyncio.Semaphore to process 1,000 API calls with a maximum of 10 concurrent. Streaming uses async generators to yield tokens as they are generated from the LLM.
Interview: AI engineering interview connection: Async Python appears in AI engineering interviews when candidates are asked to design systems that handle concurrent requests. "How would your RAG pipeline handle 100 concurrent users?" The expected answer involves async: the server processes multiple requests concurrently because each request spends most of its time waiting for I/O (API calls, database queries). Candidates who describe sequential request processing are flagged as lacking production experience.
What to demonstrate: Show that you understand the difference between concurrency (async - one thread interleaving I/O waits) and parallelism (multiprocessing - multiple CPU cores). Show that you know when to use
asyncio.gather(independent tasks), when to use semaphores (rate-limited APIs), and when to keep steps sequential (dependent stages in a pipeline).
Exercise: Starter: Write an async function that takes a list of 10 text strings and embeds them concurrently (simulate each embedding call with
await asyncio.sleep(0.2)). Compare the total time against sequential embedding. How much faster is the concurrent version?Intermediate: Write an async RAG pipeline with three stages: embed (0.2s), search (0.15s), generate (1.5s). The embed and search stages are independent for different requests. Design the pipeline so the server can handle 10 concurrent requests, measuring total time to complete all 10. What is the theoretical minimum time?
Advanced: Write a batch processing system that sends 500 prompts to an LLM API with a rate limit of 50 requests per minute. Use semaphores for concurrency control and implement retry logic for failed requests (with exponential backoff). Track progress (completed/total), handle partial failures (return results for successful calls even if some fail), and report the total processing time and success rate.
Key Takeaways
Async Python is not optional for production AI applications. AI systems spend 90-95% of their time waiting for I/O - API calls, database queries, model inference. Synchronous code wastes this waiting time. Async code uses it to serve other requests, increasing throughput from 1 concurrent request to 50+ without additional hardware.
Three async primitives cover 95% of AI engineering needs: async/await for non-blocking function calls, asyncio.gather() for running multiple independent calls concurrently, and asyncio.Semaphore for controlling concurrency to stay within rate limits. Master these three and you can build async AI pipelines for any use case.
The fan-out/fan-in pattern (batch multiple independent calls with asyncio.gather) and the async pipeline pattern (each stage is async, enabling concurrent request handling across the full pipeline) are the two architectural patterns that appear in every production AI application. Fan-out/fan-in handles batch processing. Async pipelines handle concurrent user requests. Together, they transform a single-request-at-a-time prototype into a production service that handles real traffic.
The return_exceptions=True parameter on asyncio.gather is essential for production code. Without it, one failed API call out of fifty crashes the entire batch. With it, failures are captured as values, the other 49 calls succeed, and your error handling processes the failures gracefully.