Serving models with FastAPI
Turn a model or RAG pipeline into a validated, documented, containerized HTTP service.
Why this matters in AI / ML / GenAI
A model in a notebook has no business value. FastAPI is the standard way Python teams expose inference — request validation from type hints, automatic OpenAPI docs, and async support for concurrent LLM calls.
Why FastAPI
You write a pydantic model for the request; FastAPI validates it and returns a clear 422 on bad input before your code runs. It generates interactive docs at /docs for free, and async def endpoints handle many concurrent LLM calls on one worker.
The critical pattern: load the model once at startup, not per request. Loading a transformer inside the handler adds seconds to every call and exhausts memory. Use the lifespan context manager and keep the model in application state.
Endpoints every service needs
GET /health— liveness. Returns 200 if the process is up. Kubernetes restarts the pod when this fails.GET /ready— readiness. Returns 200 only when the model is loaded. Keeps traffic away until you can actually serve.POST /predictor/chat— the real work.GET /metrics— Prometheus scrape endpoint for latency, throughput, and errors.
Return proper status codes: 400 for bad input, 404 for missing resources, 429 when rate limited, 503 when a dependency is down. Never return 200 with an error message inside — monitoring cannot see it.
Deployment shape
Run with uvicorn app:app --host 0.0.0.0 --port 8000 --workers 4. Rough sizing: workers equal to CPU cores for CPU inference; one worker per GPU for GPU inference, since each worker loads its own copy of the model.
Containerize with a slim base image, install pinned requirements, copy the code, expose the port, and add a HEALTHCHECK. Run as a non-root user.
Set request timeouts, cap the maximum request body size, and add a rate limit. Log a request id, model version, latency, and token counts on every call — that is what you will need when someone reports the service was slow last Tuesday.
Copy-paste examples
Copy into your own editor, or load one into the compiler below and press Run.
A complete FastAPI service (run locally)
pip install fastapi uvicorn pydantic, then: uvicorn app:app --reload
# app.py — pip install fastapi uvicorn pydantic
import logging
import time
import uuid
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException, Request
from pydantic import BaseModel, Field
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("service")
STATE = {"model": None, "ready": False}
def load_model():
time.sleep(0.5) # stand-in for real model loading
return lambda text: {"label": "positive" if "good" in text.lower() else "neutral"}
@asynccontextmanager
async def lifespan(app: FastAPI):
logger.info("loading model...")
STATE["model"] = load_model()
STATE["ready"] = True
yield
STATE["model"] = None
STATE["ready"] = False
app = FastAPI(title="Inference API", version="1.0.0", lifespan=lifespan)
class PredictRequest(BaseModel):
text: str = Field(min_length=1, max_length=5000)
threshold: float = Field(default=0.5, ge=0.0, le=1.0)
class PredictResponse(BaseModel):
request_id: str
label: str
latency_ms: float
@app.get("/health")
def health():
return {"status": "ok"}
@app.get("/ready")
def ready():
if not STATE["ready"]:
raise HTTPException(status_code=503, detail="model not loaded")
return {"status": "ready"}
@app.post("/predict", response_model=PredictResponse)
async def predict(payload: PredictRequest, request: Request):
if not STATE["ready"]:
raise HTTPException(status_code=503, detail="model not loaded")
request_id = request.headers.get("x-request-id", str(uuid.uuid4()))
start = time.perf_counter()
result = STATE["model"](payload.text)
latency_ms = (time.perf_counter() - start) * 1000
logger.info("predict request_id=%s chars=%d latency_ms=%.1f",
request_id, len(payload.text), latency_ms)
return PredictResponse(request_id=request_id, label=result["label"], latency_ms=latency_ms)Dockerfile
Slim base, pinned deps, non-root user, healthcheck.
# Dockerfile
FROM python:3.12-slim
ENV PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1
WORKDIR /app
RUN apt-get update && apt-get install -y --no-install-recommends curl \
&& rm -rf /var/lib/apt/lists/*
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
RUN useradd --create-home appuser && chown -R appuser:appuser /app
USER appuser
EXPOSE 8000
HEALTHCHECK --interval=30s --timeout=5s --retries=3 \
CMD curl -fsS http://localhost:8000/health || exit 1
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "2"]Request validation logic (runs here)
The same rules pydantic enforces, written by hand so you can see the checks.
def validate_request(payload):
errors = []
text = payload.get("text")
if not isinstance(text, str) or not text.strip():
errors.append("text: must be a non-empty string")
elif len(text) > 5000:
errors.append("text: exceeds 5000 characters")
threshold = payload.get("threshold", 0.5)
try:
threshold = float(threshold)
if not 0.0 <= threshold <= 1.0:
errors.append("threshold: must be between 0 and 1")
except (TypeError, ValueError):
errors.append("threshold: must be a number")
if errors:
return {"status": 422, "errors": errors}
return {"status": 200, "text": text.strip(), "threshold": threshold}
for payload in [
{"text": "This service is good", "threshold": 0.7},
{"text": " "},
{"text": "ok", "threshold": 5},
]:
print(payload, "->", validate_request(payload))Client with retry (run locally)
How another service should call yours: timeout, retry on 5xx, never on 4xx.
# pip install requests
import requests
import time
def call_predict(text, url="http://localhost:8000/predict", attempts=3):
for attempt in range(attempts):
try:
response = requests.post(url, json={"text": text}, timeout=10)
if response.status_code >= 500:
raise requests.HTTPError(f"server error {response.status_code}")
response.raise_for_status()
return response.json()
except (requests.Timeout, requests.HTTPError) as err:
if attempt == attempts - 1:
raise
wait = 2 ** attempt
print(f"retry in {wait}s after: {err}")
time.sleep(wait)
# print(call_predict("this product is good"))Simulate the request lifecycle
Try it — in-browser Python
Add a request that exceeds the length limit and confirm it returns 422.
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
- Load the model once at startup with lifespan — never inside the request handler.
- Ship /health and /ready separately so orchestrators route traffic correctly.
- Validate with pydantic, return real status codes, and log request id plus latency.