Chapter 4 · AI Services and Security
Serving an ML Model
- Page 14 of 22
- 20 min read
A model in a notebook is not a product. Users, other services and your own frontend can only reach it through an API, and the moment you do that, new questions appear that accuracy scores never asked: how do you load the model once, not on every request? What stops a request with a negative petal length from reaching it? Which version produced this answer? What happens when the model is slow, and how would you even know? This page builds a small, honest prediction service with FastAPI and scikit-learn that answers all of them, using the lifespan, settings and thread-pool ideas from Dependencies, Lifespan, Middleware and Settings.
The model is deliberately tiny (iris flowers, four numbers in, a species out) so you can run everything in seconds. The service around it is the part that carries over unchanged to a fraud model, a recommender or a text classifier.
What you will learn
- The difference between training and inference, and why a service only does the second.
- Train a preprocessing-plus-model pipeline, save it with
joblibas a versioned artifact, and load every version once inlifespan, with a warm-up. - A prediction endpoint with Pydantic feature validation, single and batch requests, and the model version in every response.
- Timeouts, a concurrency limit and thread offloading so a slow model never freezes the event loop; CPU versus GPU notes.
- Prediction logging, latency metrics, error handling, and tests with a tiny model.
Training versus inference
| Training | Inference (serving) | |
|---|---|---|
| Job | Learn parameters from labelled data | Apply the learned parameters to new inputs |
| Runs | Occasionally, offline, minutes to days | Continuously, per request, milliseconds |
| Cares about | Accuracy, experiments, big data, GPUs | Latency, throughput, correctness, uptime |
| Failure looks like | A bad metric you can re-run | A user-facing error or a silently wrong answer |
| Lives in | A script or notebook (train.py) | The API process (app/) |
The bridge between them is an artifact: a file the training job writes and the service reads. Everything else on this page is about handling that file responsibly.
Set up the project
mkdir -p ~/projects/iris-api
cd ~/projects/iris-api
uv init --bare -q
uv add -q "fastapi[standard]" pydantic-settings scikit-learn joblib
uv add -q --dev pytest httpx2
# with pip: python -m venv .venv && source .venv/bin/activate && pip install -q "fastapi[standard]" pydantic-settings scikit-learn joblib pytest httpx2
mkdir -p app tests
touch app/__init__.py
cat >> pyproject.toml <<'EOF'
[tool.fastapi]
entrypoint = "app.main:app"
[tool.pytest.ini_options]
pythonpath = ["."]
EOFTrain and save a model
Training code stays out of the app/ package: the service must never depend on it. The script trains two candidates on the classic iris dataset (bundled with scikit-learn, so no download) and saves each as a versioned file. Every random choice has a fixed random_state, so you get the same numbers every time.
# file: train.py
"""Train two small classifiers and save them as versioned artifacts."""
import sys
from pathlib import Path
import joblib
import sklearn
from sklearn.datasets import load_iris
from sklearn.ensemble import RandomForestClassifier
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import StandardScaler
FEATURES = ["sepal_length", "sepal_width", "petal_length", "petal_width"]
CLASSES = ["setosa", "versicolor", "virginica"]
def build_bundle(version, estimator, X, y):
"""Fit scaler + estimator as ONE pipeline, so preprocessing can never be forgotten."""
pipeline = make_pipeline(StandardScaler(), estimator)
pipeline.fit(X, y)
return {"version": version, "features": FEATURES, "classes": CLASSES,
"sklearn": sklearn.__version__, "pipeline": pipeline}
def main(out_dir="models"):
iris = load_iris()
X_train, X_test, y_train, y_test = train_test_split(
iris.data, iris.target, test_size=0.25, random_state=42, stratify=iris.target)
candidates = {
"v1": LogisticRegression(max_iter=200),
"v2": RandomForestClassifier(n_estimators=50, random_state=42),
}
Path(out_dir).mkdir(exist_ok=True)
for version, estimator in candidates.items():
bundle = build_bundle(version, estimator, X_train, y_train)
accuracy = bundle["pipeline"].score(X_test, y_test)
joblib.dump(bundle, Path(out_dir) / f"iris-{version}.joblib")
print(f"{version}: {type(estimator).__name__}, test accuracy {accuracy:.3f}")
if __name__ == "__main__":
main(*sys.argv[1:])python train.py
ls modelsv1: LogisticRegression, test accuracy 0.921
v2: RandomForestClassifier, test accuracy 0.921
iris-v1.joblib
iris-v2.joblib- The pipeline holds
StandardScaler(the preprocessing) and the estimator together. The classic bug is training on scaled data and serving unscaled data; with one saved object that cannot happen. The service never re-implements preprocessing. - The artifact is a dictionary, not a bare model: it carries the
version, the exact feature order the model expects, the class names and the scikit-learn version it was trained with. A model file without that context is a mystery in six months. - The test accuracy was measured on rows the model never saw. Report it with the artifact; "v2 is better" must be a number, not a feeling. Here both models score the same on this small split, which is itself a lesson: choose by more than one number (latency, size, behaviour on edge cases).
joblib.loaduses pickle, which can run arbitrary code while loading. Only load artifacts your own pipeline produced, never a file uploaded by a user or downloaded from an unknown source. Also load with the same scikit-learn version that saved the file; a mismatch can warn or break. Safer formats (ONNX,skops) exist when files cross trust boundaries.
The service pieces
Settings come from the environment, as on the previous page:
# file: app/settings.py
from pathlib import Path
from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_prefix="IRIS_", extra="ignore")
model_dir: Path = Path("models")
default_version: str = "v1"
predict_timeout_s: float = Field(default=2.0, gt=0) # give up on one prediction after this
max_concurrent: int = Field(default=4, ge=1) # predictions running at onceThe schemas are the contract with callers. Pydantic is the guard at the door: the model only ever sees four finite, plausible numbers.
# file: app/schemas.py
from typing import Annotated
from pydantic import BaseModel, ConfigDict, Field
MAX_BATCH = 100
Centimetres = Annotated[float, Field(gt=0, le=20, allow_inf_nan=False)]
class Flower(BaseModel):
model_config = ConfigDict(extra="forbid") # a misspelt feature is an error, not ignored
sepal_length: Centimetres
sepal_width: Centimetres
petal_length: Centimetres
petal_width: Centimetres
class BatchIn(BaseModel):
items: list[Flower] = Field(min_length=1, max_length=MAX_BATCH)
class Result(BaseModel):
label: str
probabilities: dict[str, float]
class Prediction(Result):
model_version: str
class BatchOut(BaseModel):
model_version: str
results: list[Result]Now the registry: it finds every artifact, loads it, checks it against the API contract, and warms it up. The first prediction of a real model often pays one-time costs (lazy imports, memory allocation, GPU kernel compilation); doing one throwaway prediction at startup moves that cost out of the first user's request.
# file: app/registry.py
from pathlib import Path
import joblib
import numpy as np
from app.schemas import Flower
WARMUP_ROW = [[5.1, 3.5, 1.4, 0.2]]
def predict_rows(bundle: dict, rows: list[list[float]]) -> list[dict]:
"""Plain blocking function: one call for any number of rows."""
proba = bundle["pipeline"].predict_proba(np.array(rows))
results = []
for row in proba:
best = int(row.argmax())
results.append({
"label": bundle["classes"][best],
"probabilities": {name: round(float(p), 3) for name, p in zip(bundle["classes"], row)},
})
return results
def load_registry(model_dir: Path) -> dict[str, dict]:
"""Load every iris-*.joblib file, keyed by version. Fail loudly, at startup."""
registry = {}
for path in sorted(model_dir.glob("iris-*.joblib")):
bundle = joblib.load(path)
if bundle["features"] != list(Flower.model_fields):
raise RuntimeError(f"{path.name}: features {bundle['features']} do not match the API schema")
predict_rows(bundle, WARMUP_ROW) # warm-up: result thrown away
registry[bundle["version"]] = bundle
if not registry:
raise RuntimeError(f"No model files found in {model_dir}")
return registryFinally the app. Read it top to bottom: the lifespan loads everything once, a dependency picks the model version, and infer is the single place where a prediction is run, timed, limited, logged and counted.
# file: app/main.py
import asyncio
import json
import logging
import time
from contextlib import asynccontextmanager
from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException, Query, Request
from app.registry import load_registry, predict_rows
from app.schemas import BatchIn, BatchOut, Flower, Prediction, Result
from app.settings import Settings
log = logging.getLogger("iris_api.predictions")
class Metrics:
"""In-memory counters. In production export these to Prometheus, Datadog or your platform."""
def __init__(self) -> None:
self.requests = self.errors = self.slow = 0
self.total_ms = self.max_ms = 0.0
def record(self, ms: float, *, ok: bool) -> None:
self.requests += 1
self.errors += 0 if ok else 1
self.slow += 1 if ms > 1000 else 0
self.total_ms += ms
self.max_ms = max(self.max_ms, ms)
def snapshot(self) -> dict:
average = self.total_ms / self.requests if self.requests else 0.0
return {"requests": self.requests, "errors": self.errors, "slow_over_1s": self.slow,
"avg_ms": round(average, 2), "max_ms": round(self.max_ms, 2)}
def create_app(settings: Settings | None = None) -> FastAPI:
settings = settings or Settings()
@asynccontextmanager
async def lifespan(app: FastAPI):
logging.basicConfig(level="INFO", format="%(asctime)s %(levelname)s %(name)s %(message)s")
app.state.settings = settings
app.state.registry = load_registry(settings.model_dir) # blocks startup, not requests
if settings.default_version not in app.state.registry:
raise RuntimeError(f"Default model {settings.default_version!r} was not loaded")
app.state.predict = predict_rows
app.state.metrics = Metrics()
app.state.slots = asyncio.Semaphore(settings.max_concurrent)
yield
app.state.registry.clear() # release the memory
app = FastAPI(title="BitByte Iris API", lifespan=lifespan)
def get_bundle(request: Request, version: Annotated[str | None, Query(pattern=r"^[a-z0-9.-]{1,20}$")] = None) -> dict:
registry = request.app.state.registry
wanted = version or request.app.state.settings.default_version
if wanted not in registry:
raise HTTPException(404, f"Unknown model version {wanted!r}; available: {sorted(registry)}")
return registry[wanted]
BundleDep = Annotated[dict, Depends(get_bundle)]
async def infer(request: Request, bundle: dict, flowers: list[Flower]) -> list[dict]:
state = request.app.state
rows = [[getattr(flower, name) for name in bundle["features"]] for flower in flowers]
started = time.perf_counter()
ok = False
try:
async with state.slots: # at most N predictions at once
work = asyncio.to_thread(state.predict, bundle, rows) # off the event loop
results = await asyncio.wait_for(work, state.settings.predict_timeout_s)
ok = True
except TimeoutError:
raise HTTPException(504, "The model took too long to answer") from None
except Exception:
log.exception("inference failed") # details stay in the logs
raise HTTPException(500, "Prediction failed") from None
finally:
duration_ms = (time.perf_counter() - started) * 1000
state.metrics.record(duration_ms, ok=ok)
fields = {"model_version": bundle["version"], "n_items": len(rows), "ok": ok,
"labels": [r["label"] for r in results], "duration_ms": round(duration_ms, 1)}
log.info("prediction %s", json.dumps(fields))
return results
@app.get("/health")
def health() -> dict:
return {"status": "ok", "versions": sorted(app.state.registry)}
@app.get("/v1/model")
def model_info(bundle: BundleDep) -> dict:
return {"version": bundle["version"], "features": bundle["features"],
"classes": bundle["classes"], "estimator": type(bundle["pipeline"][-1]).__name__}
@app.post("/v1/predict")
async def predict(flower: Flower, request: Request, bundle: BundleDep) -> Prediction:
(result,) = await infer(request, bundle, [flower])
return Prediction(model_version=bundle["version"], **result)
@app.post("/v1/predict/batch")
async def predict_batch(data: BatchIn, request: Request, bundle: BundleDep) -> BatchOut:
results = await infer(request, bundle, data.items)
return BatchOut(model_version=bundle["version"], results=[Result(**r) for r in results])
@app.get("/v1/metrics")
def metrics() -> dict:
return app.state.metrics.snapshot()
return app
app = create_app()Run it and call it
fastapi run --port 8000 > server.log 2>&1 &
export SERVER_PID=$!
export BASE=http://localhost:8000
curl -s --retry 20 --retry-connrefused --retry-delay 0 $BASE/health
echo
curl -s $BASE/v1/model{"status":"ok","versions":["v1","v2"]}
{"version":"v1","features":["sepal_length","sepal_width","petal_length","petal_width"],"classes":["setosa","versicolor","virginica"],"estimator":"LogisticRegression"}The server started only after both models were loaded and warmed up, so the first request is not special. Now a prediction, then the same flower with the other model version:
curl -s $BASE/v1/predict -H "Content-Type: application/json" \
-d '{"sepal_length": 6.3, "sepal_width": 2.9, "petal_length": 5.6, "petal_width": 1.8}'
echo
curl -s "$BASE/v1/predict?version=v2" -H "Content-Type: application/json" \
-d '{"sepal_length": 6.3, "sepal_width": 2.9, "petal_length": 5.6, "petal_width": 1.8}'
echo
curl -s "$BASE/v1/predict?version=v9" -H "Content-Type: application/json" \
-d '{"sepal_length": 6.3, "sepal_width": 2.9, "petal_length": 5.6, "petal_width": 1.8}'{"label":"virginica","probabilities":{"setosa":0.001,"versicolor":0.195,"virginica":0.804},"model_version":"v1"}
{"label":"virginica","probabilities":{"setosa":0.0,"versicolor":0.0,"virginica":1.0},"model_version":"v2"}
{"detail":"Unknown model version 'v9'; available: ['v1', 'v2']"}Every response names the model_version that produced it. When a user complains about an answer next month, that field tells you which artifact to reproduce it with, and it makes a gradual rollout (10% of callers on v2) measurable. A model that serves answers without saying who produced them cannot be debugged.
Validation protects the model
A model never complains: given a negative petal it happily returns a species. Garbage in, confident garbage out. Pydantic rejects bad input before it gets near the model:
import os
from fastapi.testclient import TestClient
from app.main import create_app
from app.settings import Settings
app = create_app(Settings())
client = TestClient(app).__enter__() # runs the lifespan, like a real startup
good = {"sepal_length": 6.3, "sepal_width": 2.9, "petal_length": 5.6, "petal_width": 1.8}
bad_inputs = {
"negative": {**good, "petal_length": -1},
"missing": {k: v for k, v in good.items() if k != "petal_width"},
"text": {**good, "sepal_width": "wide"},
"misspelt": {**good, "petal_lenght": 5.6},
"huge": {**good, "sepal_length": 9999},
}
for name, body in bad_inputs.items():
response = client.post("/v1/predict", json=body)
problem = response.json()["detail"][0]
print(f"{name:9} {response.status_code} {problem['loc'][-1]}: {problem['msg']}")negative 422 petal_length: Input should be greater than 0
missing 422 petal_width: Field required
text 422 sepal_width: Input should be a valid number, unable to parse string as a number
misspelt 422 petal_lenght: Extra inputs are not permitted
huge 422 sepal_length: Input should be less than or equal to 20The model is never called for any of these. Choose bounds from your training data and your domain (a petal is never 9999 cm), not from what Pydantic can express; and remember a value can be valid yet far outside anything the model saw. Detecting that "drift" belongs in monitoring (below), not in a validator.
Batch versus single
Models are vectorised: predicting 100 rows in one call costs little more than predicting one, while 100 separate calls pay the per-call overhead 100 times (and 100 HTTP round trips, if each is a request). A batch endpoint is the right tool for callers that have many items; it also needs a cap (MAX_BATCH = 100 here) so one request cannot occupy a worker for minutes.
import time
import numpy as np
from app.registry import predict_rows
bundle = app.state.registry["v1"]
rows = np.tile([[6.3, 2.9, 5.6, 1.8]], (100, 1)).tolist()
start = time.perf_counter()
for row in rows:
predict_rows(bundle, [row])
one_by_one = time.perf_counter() - start
start = time.perf_counter()
results = predict_rows(bundle, rows)
as_batch = time.perf_counter() - start
print(len(results), "results; batch faster than 100 single calls:", as_batch < one_by_one)
response = client.post("/v1/predict/batch", json={"items": [good, {**good, "petal_length": 1.4, "petal_width": 0.2}]})
print(response.json())
print(client.post("/v1/predict/batch", json={"items": []}).status_code,
client.post("/v1/predict/batch", json={"items": [good] * 101}).status_code)100 results; batch faster than 100 single calls: True
{'model_version': 'v1', 'results': [{'label': 'virginica', 'probabilities': {'setosa': 0.001, 'versicolor': 0.195, 'virginica': 0.804}}, {'label': 'versicolor', 'probabilities': {'setosa': 0.373, 'versicolor': 0.627, 'virginica': 0.0}}]}
422 422Do not block the event loop
The previous page showed that blocking code inside async def freezes every request. A model call is exactly that: CPU-bound and synchronous. The infer function therefore does three things around it:
- Moves it to a thread with
asyncio.to_thread, so the event loop keeps answering/healthand other requests while the model computes. (Many numeric libraries release Python's GIL while they compute, so threads genuinely overlap. For pure-Python models they would not; then use more worker processes.) - Limits concurrency with a semaphore: ten simultaneous predictions on four CPU cores are slower for everyone than four at a time. Excess requests wait their turn instead of thrashing.
- Applies a timeout with
asyncio.wait_for, so a stuck or pathological input becomes a clean 504 instead of a hung connection.
Here is the timeout in action, using a deliberately slow replacement of the prediction function (the design with app.state.predict exists precisely so tests can do this):
def slow_predict(bundle, rows):
time.sleep(0.5)
return predict_rows(bundle, rows)
tight = create_app(Settings(predict_timeout_s=0.1))
with TestClient(tight) as tight_client:
tight.state.predict = slow_predict
response = tight_client.post("/v1/predict", json=good)
print(response.status_code, response.json())
tight.state.predict = predict_rows
print(tight_client.post("/v1/predict", json=good).status_code)
print(tight_client.get("/v1/metrics").json()["errors"], "error counted")504 {'detail': 'The model took too long to answer'}
200
1 error countedBe honest about what a timeout does: it frees the caller, but the worker thread cannot be killed and finishes its computation anyway. That is why the semaphore matters, and why truly unbounded work (huge inputs, runaway generation) needs hard limits on input size and a separate worker process that can be terminated.
CPU versus GPU
| CPU serving | GPU serving | |
|---|---|---|
| Good for | Classical ML (scikit-learn, XGBoost), small neural nets, low traffic | Large neural nets, LLMs, images, audio |
| Scaling | More worker processes (fastapi run --workers N); each loads its own copy of the model, so memory is N times the model size | Usually one process per GPU; the GPU, not Python, is the bottleneck |
| Batching | Helps a little | Essential: a GPU is efficient only when fed batches, so servers often collect requests for a few milliseconds ("dynamic batching") |
| Startup | Seconds | Loading weights and warming up can take minutes: warm-up and readiness checks matter most here |
| Cost | Cheap and everywhere | Expensive; do not leave idle |
Do not reach for a GPU because it sounds serious: a logistic regression on a GPU is slower than on a CPU. In PyTorch you would additionally call model.eval() and run under torch.inference_mode(), and move inputs to the model's device; the lifespan, validation, timeout and versioning pattern on this page stays the same. Dedicated servers (vLLM, Triton, BentoML, TorchServe) implement dynamic batching for you when you outgrow a hand-written endpoint.
Prediction logging and latency monitoring
Two different questions need two different tools. Logs answer "what happened on this request?"; metrics answer "how is the service doing overall?". The infer function emits both: one log line per prediction (model version, number of items, predicted labels, duration, success) and counters (requests, errors, slow requests, average and maximum latency).
import json
import logging
lines = []
class Capture(logging.Handler):
def emit(self, record):
lines.append(record.getMessage())
logging.getLogger("iris_api").addHandler(Capture())
logging.getLogger("iris_api").setLevel(logging.INFO)
client.post("/v1/predict?version=v2", json=good)
entry = json.loads(lines[-1].removeprefix("prediction "))
print({key: value for key, value in entry.items() if key != "duration_ms"})
print("duration under 1 s:", entry["duration_ms"] < 1000)
stats = client.get("/v1/metrics").json()
print(stats["requests"] > 0, stats["errors"], stats["slow_over_1s"], "average under 1 s:", stats["avg_ms"] < 1000){'model_version': 'v2', 'n_items': 1, 'ok': True, 'labels': ['virginica']}
duration under 1 s: True
True 0 0 average under 1 s: True- What to log: version, sizes, outputs, duration, request ID. Log raw inputs only when you have a reason and the right to: features may be personal data. Common compromises are sampling, hashing, or logging only aggregates. Without logged predictions you cannot investigate a complaint, build a dataset for the next training run, or notice drift (the live inputs slowly stop looking like the training data).
- What to alert on: error rate, and a high percentile of latency (p95 or p99), not only the average, which hides the slow tail. The in-memory counters above are a teaching device: they reset on restart and are per process. Real services export histograms to a monitoring system.
- Health:
/healthreports loaded versions; a service whose model failed to load should not start, which is whyload_registryraises at startup instead of failing on the first request.
Error handling
| Situation | Response | Where handled |
|---|---|---|
| Invalid or out-of-range features | 422 with field-level messages | Pydantic, before the model |
| Unknown model version | 404 listing available versions | get_bundle |
| Model too slow | 504 | wait_for in infer |
| Model raises an exception | 500, generic message; stack trace only in logs | infer |
| Artifact missing or incompatible | The service refuses to start | load_registry |
Never return exception text from inside the model to the caller: it can reveal paths, library versions and data. In a real project, route all of these through the single error format from FastAPI Architecture: Routers, Schemas and Error Handling. Stop the demo server before the tests:
kill $SERVER_PID
wait $SERVER_PID 2>/dev/null
echo stoppedstoppedTest with a tiny model
Tests must not depend on the models you trained last Tuesday. They build a tiny model in a temporary folder (a few rows, seconds to fit, deterministic) and point the app at it. They also reuse the real build_bundle, so training and serving stay consistent.
# file: tests/conftest.py
import joblib
import pytest
from fastapi.testclient import TestClient
from sklearn.datasets import load_iris
from sklearn.linear_model import LogisticRegression
from app.main import create_app
from app.settings import Settings
from train import build_bundle
GOOD = {"sepal_length": 5.1, "sepal_width": 3.5, "petal_length": 1.4, "petal_width": 0.2} # a setosa
@pytest.fixture
def client(tmp_path):
iris = load_iris()
X, y = iris.data[::5], iris.target[::5] # 30 rows: a tiny model
joblib.dump(build_bundle("test", LogisticRegression(max_iter=200), X, y), tmp_path / "iris-test.joblib")
app = create_app(Settings(model_dir=tmp_path, default_version="test", predict_timeout_s=0.2))
with TestClient(app) as test_client:
yield test_client# file: tests/test_api.py
import time
import pytest
from app.registry import predict_rows
from tests.conftest import GOOD
def test_prediction_shape_and_version(client):
body = client.post("/v1/predict", json=GOOD).json()
assert body["model_version"] == "test"
assert body["label"] == "setosa"
assert abs(sum(body["probabilities"].values()) - 1) < 0.01
def test_batch_matches_single(client):
single = client.post("/v1/predict", json=GOOD).json()["label"]
batch = client.post("/v1/predict/batch", json={"items": [GOOD, GOOD]}).json()
assert [r["label"] for r in batch["results"]] == [single, single]
@pytest.mark.parametrize("field,value", [("petal_length", -1), ("sepal_width", "wide"), ("sepal_length", 9999)])
def test_invalid_features_are_rejected(client, field, value):
assert client.post("/v1/predict", json={**GOOD, field: value}).status_code == 422
def test_unknown_version_is_404(client):
assert client.post("/v1/predict?version=nope", json=GOOD).status_code == 404
def test_slow_model_times_out_and_service_recovers(client):
def slow(bundle, rows):
time.sleep(0.5)
return predict_rows(bundle, rows)
client.app.state.predict = slow
assert client.post("/v1/predict", json=GOOD).status_code == 504
client.app.state.predict = predict_rows
assert client.post("/v1/predict", json=GOOD).status_code == 200
def test_model_crash_is_a_generic_500(client):
def broken(bundle, rows):
raise ValueError("secret internal detail")
client.app.state.predict = broken
response = client.post("/v1/predict", json=GOOD)
assert response.status_code == 500
assert "secret" not in response.textpytest -q........ [100%]
8 passed in 0.97sThe tests check behaviour (shape, version, errors, timeout), not the model's quality: model quality is measured by evaluation on held-out data in the training pipeline, and by monitoring in production. Keep those two concerns separate.
Real-world use
- Fraud, churn, recommendation and ranking models are served exactly like this: a validated feature payload in, a score and a model version out, with latency budgets of tens of milliseconds.
- Teams keep artifacts in a model registry (MLflow, Hugging Face Hub, an S3 bucket with version folders) instead of a local
models/directory; the lifespan then downloads a pinned version. See Git for AI Projects: Data, Models, LFS, DVC and Leaked Secrets for why you do not commit large files to Git. - Rollouts use the version field: serve
v2to a small share of callers, compare error rates and business metrics, then switch the default. - LLM backends add streaming and provider calls on top of the same skeleton; that is the next page.
Common mistakes
- Loading the model inside the endpoint. Every request pays seconds of disk and deserialisation. Load once in
lifespan. - Preprocessing outside the saved pipeline. Scaling or encoding re-implemented in the service drifts from training and silently degrades answers. Save the whole pipeline.
- Calling the model directly in
async def. One slow prediction stalls the whole server. Use a thread (or a plaindefroute), a concurrency limit and a timeout. - No validation and no version in the response. You serve nonsense confidently and cannot tell which model said it. Validate every feature; return
model_version. - Unseeded, unreproducible training and untrusted pickles. Fix
random_state, record library versions, and onlyjoblib.loadfiles you produced.
Try it yourself
- Easy: load
models/iris-v1.joblibin Python and print the predicted label and probabilities for the flower[5.1, 3.5, 1.4, 0.2]. - Medium: write a test that sends 100 rows to
/v1/predict/batchand checks that 100 results come back, then check the endpoint refuses 101. - Hard: add a field
confidence(the highest probability) toResult, and a 422-style guard: if the model's confidence is below 0.6, the single endpoint should still answer 200 but add"low_confidence": true. Show it with a flower that sits between two species.
Answers
# 1. Easy
import joblib
from app.registry import predict_rows
loaded = joblib.load("models/iris-v1.joblib")
print(predict_rows(loaded, [[5.1, 3.5, 1.4, 0.2]]))
# 2. Medium (run it as a pytest test: add it to tests/test_api.py)
def test_batch_limit(client):
from tests.conftest import GOOD
assert len(client.post("/v1/predict/batch", json={"items": [GOOD] * 100}).json()["results"]) == 100
assert client.post("/v1/predict/batch", json={"items": [GOOD] * 101}).status_code == 422
# 3. Hard: the check can live in one small function that works on a result
def with_confidence(result: dict) -> dict:
confidence = max(result["probabilities"].values())
return {**result, "confidence": confidence, "low_confidence": confidence < 0.6}
between = predict_rows(loaded, [[6.0, 2.7, 4.9, 1.7]])[0]
print(between["probabilities"])
print(with_confidence(between)["low_confidence"], with_confidence(predict_rows(loaded, [[5.1, 3.5, 1.4, 0.2]])[0])["low_confidence"])[{'label': 'setosa', 'probabilities': {'setosa': 0.981, 'versicolor': 0.019, 'virginica': 0.0}}]
{'setosa': 0.003, 'versicolor': 0.455, 'virginica': 0.542}
True FalseSummary
- Training is offline and produces an artifact; serving loads it and only predicts. Save preprocessing and model as one pipeline, with version, feature order and library versions.
- Load all model versions once in
lifespan, check them against the API schema, warm them up, and refuse to start if anything is wrong. Treatjoblibfiles as code: load only trusted ones. - Validate features with Pydantic before they reach the model, support single and capped batch requests, and return the
model_versionin every response. - Keep the event loop free: run the model in a thread, limit concurrency, and apply a timeout; a timeout frees the caller but not the thread. Use GPUs only for models that need them, and batch when you do.
- Log each prediction without leaking data, track error rate and tail latency, and test the service with a tiny model built in a temporary folder.
Next: Building an LLM Backend with Streaming keeps this skeleton (settings, lifespan, validation, timeouts, logging) and swaps the local model for a language-model provider, streaming the answer to the client as it is generated.