The Concept
Why ML Serving Is Different from Web Dev
A typical web endpoint queries a database and returns HTML — the request takes 10-50ms. An ML endpoint runs a model inference or calls an LLM API — the request takes 500ms to 30 seconds. During those seconds, if your server is blocked waiting for one request to finish, it can't serve anyone else. A single slow LLM call would freeze the entire API.
This is why asynchronous programming is not optional for ML serving — it's the difference between serving 10 users and 10,000. The idea is simple: when your code is waiting for something (an LLM API response, a database query, a file read), it should yield control back to the system so other requests can be processed. When the wait is over, your code resumes. This is called cooperative multitasking, and in Python it's implemented with async/await.
The Event Loop: One Thread, Many Tasks
Python's asyncio runs a single-threaded event loop. The loop maintains a queue of tasks. When a task hits an await point (waiting for I/O), the loop suspends that task and runs the next one. When the I/O completes, the suspended task resumes. This lets a single thread handle thousands of concurrent I/O-bound operations — because at any given moment, most tasks are waiting, not computing.
The critical rule: never call blocking code inside an async function. A blocking call (like requests.post() or time.sleep()) freezes the entire event loop — no other task can run until it finishes. This is the #1 mistake in async ML serving. Use httpx.AsyncClient instead of requests, use asyncio.sleep() instead of time.sleep(), and use async database drivers instead of synchronous ones.
FastAPI: ML Model Serving
FastAPI is built on top of asyncio and automatically handles the event loop for you. You define endpoints with async def, and FastAPI runs them on the event loop. Pydantic models define the request and response contracts — the same validation you learned in the previous lesson, now applied to HTTP endpoints.
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import Literal
import time
app = FastAPI(title="Sentiment API", version="1.0.0")
# Pydantic models define the API contract
class SentimentRequest(BaseModel):
text: str = Field(..., min_length=1, max_length=10000)
model: Literal["distilbert", "roberta"] = "distilbert"
class SentimentResponse(BaseModel):
label: Literal["positive", "negative", "neutral"]
confidence: float = Field(ge=0.0, le=1.0)
latency_ms: float
You have an async endpoint that calls an LLM API with requests.post(). The endpoint is slow and blocks other requests. What's wrong and how do you fix it?
requests is synchronous — it blocks the entire event loop while waiting for the HTTP response. In async code, use httpx.AsyncClient (or aiohttp) so the event loop can process other requests while awaiting the LLM API response.
Load model once at startup (not per request)
model_cache: dict[str, any] = {}
@app.on_event("startup") async def load_models(): # In production, load from disk or model registry model_cache["distilbert"] = lambda text: ("positive", 0.92) model_cache["roberta"] = lambda text: ("positive", 0.95)
@app.post("/predict", response_model=SentimentResponse) async def predict(request: SentimentRequest): if request.model not in model_cache: raise HTTPException(status_code=400, detail=f"Model {request.model} not loaded")
start = time.perf_counter() model = model_cache[request.model] label, confidence = model(request.text) latency = (time.perf_counter() - start) * 1000
return SentimentResponse( label=label, confidence=confidence, latency_ms=round(latency, 2) )
@app.get("/health") async def health(): return {"status": "healthy", "models": list(model_cache.keys())}
### Async/Await: Concurrent API Calls
The key insight: **async is not about speed per call — it's about concurrency**. When your code waits for an API response, async lets other requests proceed instead of blocking.
import asyncio
import httpx
from typing import Any
# SLOW: Sequential calls (each waits for the previous)
async def generate_embeddings_sequential(texts: list[str]) -> list[dict]:
results = []
async with httpx.AsyncClient() as client:
for text in texts:
response = await client.post(
"https://api.openai.com/v1/embeddings",
json={"model": "text-embedding-3-small", "input": text},
headers={"Authorization": "Bearer sk-..."}
)
results.append(response.json())
return results # 10 texts × 200ms = 2 seconds total
# FAST: Concurrent calls (all fire at once)
async def generate_embeddings_concurrent(texts: list[str]) -> list[dict]:
async with httpx.AsyncClient() as client:
tasks = [
client.post(
"https://api.openai.com/v1/embeddings",
json={"model": "text-embedding-3-small", "input": text},
headers={"Authorization": "Bearer sk-..."}
)
for text in texts
]
responses = await asyncio.gather(*tasks)
return [r.json() for r in responses] # 10 texts × 200ms = ~200ms total
# With rate limiting: semaphore limits concurrent calls
async def generate_embeddings_rate_limited(
texts: list[str],
max_concurrent: int = 5
) -> list[dict]:
semaphore = asyncio.Semaphore(max_concurrent)
async def fetch_one(client: httpx.AsyncClient, text: str) -> dict:
async with semaphore:
response = await client.post(
"https://api.openai.com/v1/embeddings",
json={"model": "text-embedding-3-small", "input": text},
headers={"Authorization": "Bearer sk-..."}
)
return response.json()
async with httpx.AsyncClient() as client:
tasks = [fetch_one(client, text) for text in texts]
return await asyncio.gather(*tasks)
Retry with Exponential Backoff
import asyncio
import httpx
from typing import Any
async def call_with_retry(
client: httpx.AsyncClient,
url: str,
max_retries: int = 3,
**kwargs
) -> httpx.Response:
"""Call an API with exponential backoff retry."""
last_error = None
for attempt in range(max_retries):
try:
response = await client.post(url, **kwargs)
if response.status_code == 429: # rate limited
wait = 2 ** attempt
await asyncio.sleep(wait)
continue
response.raise_for_status()
return response
except (httpx.HTTPError, httpx.HTTPStatusError) as e:
last_error = e
wait = 2 ** attempt
await asyncio.sleep(wait)
raise last_error
CLI Tools with Typer
import typer
from typing import Optional
from pathlib import Path
app = typer.Typer(help="AI workflow CLI")
@app.command()
def embed(
input_file: Path = typer.Argument(..., help="Path to text file"),
output_file: Path = typer.Option("embeddings.npy", "--output", "-o"),
model: str = typer.Option("text-embedding-3-small", "--model", "-m"),
):
"""Generate embeddings for a text file."""
texts = input_file.read_text().splitlines()
typer.echo(f"Embedding {len(texts)} lines with {model}...")
# Run async code from sync CLI
import asyncio
results = asyncio.run(generate_embeddings_concurrent(texts))
import numpy as np
embeddings = np.array([r["data"][0]["embedding"] for r in results])
np.save(output_file, embeddings)
typer.echo(f"Saved {embeddings.shape} to {output_file}")
@app.command()
def batch_predict(
input_file: Path = typer.Argument(...),
endpoint: str = typer.Option("http://localhost:8000/predict", "--endpoint", "-e"),
):
"""Run batch predictions against a deployed model."""
import asyncio
import httpx
import json
async def run():
texts = input_file.read_text().splitlines()
async with httpx.AsyncClient() as client:
tasks = [
client.post(endpoint, json={"text": text})
for text in texts
]
responses = await asyncio.gather(*tasks)
for i, r in enumerate(responses):
result = r.json()
print(f"{i}: {result['label']} ({result['confidence']:.2f})")
asyncio.run(run())
if __name__ == "__main__":
app()