Build a Production ML Inference API

Serve a text classifier via FastAPI with caching, logging, and A/B routing.

Introduction

30 Second Summary

Every data science team has a model that works great in a notebook but never makes it to production. The gap between "it works on my laptop" and "it serves real users" is filled with caching, logging, versioning, and health checks.

In this project, you will build a production ML inference API that classifies text using real machine learning models. You will serve predictions through a professional API with caching, database logging, and model versioning.

What You'll Build

A FastAPI application serving scikit-learn text classifiers with Redis prediction caching, PostgreSQL prediction logging, and A/B model routing.

By the end of this project, you'll have:

A production-grade ML inference API serving real text classification predictions through FastAPI with health checks, input validation, and structured error handling.
A Redis caching layer that returns cached predictions in milliseconds and a PostgreSQL database that logs every prediction for audit and performance tracking.
A model versioning system that serves multiple model versions simultaneously with A/B traffic routing.
Secret Mission: Add request timing middleware and a /metrics endpoint that reports model performance statistics from the prediction log.

Are there any prerequisites?

This project assumes basic Python knowledge and familiarity with REST APIs. You should be comfortable writing functions, importing libraries, and understanding HTTP request/response patterns.

You will also need Docker installed on your machine to run PostgreSQL and Redis containers locally.

Set Up the Project Environment

Before you can serve ML predictions through an API, you need a local development environment with the right Python version, packages, and backing services.

In this step, you will set up Python 3.11, install all project dependencies, and spin up PostgreSQL and Redis containers using Docker Compose.

In this step, get ready to:

Create the project directory and a Python virtual environment with all dependencies installed.
Start PostgreSQL and Redis containers with Docker Compose.
Create the project file structure with placeholder files.

Create the project directory and virtual environment

You need a dedicated folder for this project and a Python virtual environment to isolate your dependencies from any other Python projects on your machine.

Open your terminal. Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
Type Terminal (macOS) or Git Bash (Windows) and press Enter to open it.
Move to your Desktop by running this command:

cd ~/Desktop

Create your project folder and move into it by running these commands:

mkdir ml-inference-api
cd ml-inference-api

Now check your Python version. This project requires Python 3.11 or higher because scikit-learn 1.9.0 depends on it.

Check your Python version by running:

python3 --version

✔️ I see version 3.11 or higher

Your Python version is compatible. You are ready to create the virtual environment.

ⓧ I see a version lower than 3.11

You have Python installed, but the version is too old for this project. You need to install Python 3.11 or higher.

On macOS, the system Python is often an older version. The recommended approach is to use pyenv to install the correct version alongside your existing one.

Install pyenv with Homebrew by running:

brew update
brew install pyenv

Install Python 3.11 and set it as your local version by running:

pyenv install 3.11
pyenv local 3.11

On Windows, download and install Python 3.11 from python.org/downloads. During installation, check the box that says Add Python to PATH.

After installing, close and reopen your terminal, then run python3 --version again to confirm you see 3.11 or higher.

ⓧ Command not found

Python is not installed or not on your PATH. You need to install Python 3.11 or higher.

On macOS, use Homebrew and pyenv:

Install pyenv by running:

brew update
brew install pyenv

Install Python 3.11 by running:

pyenv install 3.11
pyenv local 3.11

On Windows, download and install Python 3.11 from python.org/downloads. During installation, check Add Python to PATH.

After installing, close and reopen your terminal, then run python3 --version again to confirm you see 3.11 or higher.

Next, create and activate a virtual environment so your project dependencies stay isolated.

Create the virtual environment by running:

python3 -m venv venv

Activate the virtual environment:

macOS

source venv/bin/activate

Windows

venv\Scripts\activate

What is a virtual environment?

A virtual environment is an isolated Python installation inside your project folder. Packages you install here will not affect your system Python or other projects.

You will know it is active when you see (venv) at the beginning of your terminal prompt.

Now you need to create the requirements.txt file that lists all the packages this project depends on.

Open VS Code in the current directory by running:

code .

code command not found?

If you get an error, open VS Code manually. Press Cmd+Space (macOS) or the Windows key (Windows), type VS Code, and press Enter. Then click File then Open Folder and select the ml-inference-api folder on your Desktop.

Help me add the VS Code CLI to my PATH

In VS Code, create a new file called requirements.txt by clicking File then New File.
Paste the following content into requirements.txt:

fastapi==0.136.3
uvicorn==0.49.0
scikit-learn==1.9.0
sqlalchemy==2.0.50
psycopg2-binary==2.9.10
redis==7.4.1
joblib==1.4.2
pydantic-settings==2.9.1

What are these packages?

fastapi is the web framework for your API. uvicorn is the ASGI server that runs it. scikit-learn provides the ML algorithms for training and serving models.

sqlalchemy handles database interactions. redis connects to the cache. joblib serializes trained models to disk. pydantic-settings manages environment-driven configuration.

Save requirements.txt with Cmd+S (macOS) or Ctrl+S (Windows).
Back in your terminal (with the venv active), install all dependencies by running:

pip install -r requirements.txt

This will download and install all eight packages and their sub-dependencies. It may take a minute or two.

Verify the installation succeeded by running:

pip list

You should see fastapi, scikit-learn, sqlalchemy, redis, uvicorn, joblib, and pydantic-settings in the list.

pip install failing?

Make sure you see (venv) in your terminal prompt. If not, re-run the activation command from above.

If you get a permissions error, you may be installing to your system Python instead of the virtual environment.

Help me troubleshoot pip install

Start PostgreSQL and Redis with Docker Compose

Your API needs two backing services. PostgreSQL stores a log of every prediction your API makes. Redis caches predictions so repeated requests return instantly.

Docker Compose lets you define both services in a single file and start them with one command.

In VS Code, create a new file called docker-compose.yml by clicking File then New File.
Paste the following content into docker-compose.yml:

services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_USER: mlapi
      POSTGRES_PASSWORD: mlapi_secret
      POSTGRES_DB: ml_inference
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U mlapi"]
      interval: 5s
      timeout: 3s
      retries: 5

  redis:
    image: redis:7
    ports:
      - "6379:6379"
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      timeout: 3s
      retries: 5

volumes:
  pgdata:

What does this file do?

The postgres service runs PostgreSQL 16 with a database called ml_inference and credentials mlapi/mlapi_secret. The pgdata volume keeps data between container restarts.

The redis service runs Redis 7. Both services include health checks so Docker knows when they are ready to accept connections.

Save docker-compose.yml.
Make sure Docker Desktop is running. Press Cmd+Space (macOS) or the Windows key (Windows), type Docker Desktop, and press Enter to open it if it is not already running.
Start both database containers by running this command in your terminal:

docker compose up -d

Verify both containers are running and healthy by running:

docker compose ps

You should see two containers listed. The STATUS column should show Up with a (healthy) indicator for both.

Containers not starting?

If you see an error about port 5432 already in use, you may have another PostgreSQL instance running. Stop it or change the port mapping in docker-compose.yml.

If Docker commands fail entirely, make sure Docker Desktop is running and fully started (check for the whale icon in your system tray or menu bar).

Help me troubleshoot Docker Compose

Create the project file structure

Your project needs a specific directory layout. The app/ folder holds all your API code. The app/routers/ folder holds the endpoint handlers. The models/ folder will store trained ML model files.

Create all the directories by running:

mkdir -p app/routers models

Create the Python package placeholder files by running:

touch app/__init__.py app/routers/__init__.py

What are init.py files?

These empty files tell Python that app/ and app/routers/ are Python packages. Without them, Python cannot import modules from these directories.

Confirm the structure looks correct by running:

ls

You should see app, models, venv, requirements.txt, and docker-compose.yml in the output.

✔️ Awesome, I've got everything!

Great. Double check you have saved all files.

ⓧ I'd like to double check the full code

Here is the full requirements.txt:

fastapi==0.136.3
uvicorn==0.49.0
scikit-learn==1.9.0
sqlalchemy==2.0.50
psycopg2-binary==2.9.10
redis==7.4.1
joblib==1.4.2
pydantic-settings==2.9.1

Here is the full docker-compose.yml:

services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_USER: mlapi
      POSTGRES_PASSWORD: mlapi_secret
      POSTGRES_DB: ml_inference
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U mlapi"]
      interval: 5s
      timeout: 3s
      retries: 5

  redis:
    image: redis:7
    ports:
      - "6379:6379"
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      timeout: 3s
      retries: 5

volumes:
  pgdata:

Both app/__init__.py and app/routers/__init__.py should be empty files.

Your development environment is ready. You have Python with all dependencies, PostgreSQL and Redis running locally, and a clean project structure. Next up, you will train two ML models and save them to the models/ directory.

Train and Serialize the ML Models

Your project skeleton and infrastructure are ready. Now you need models for the API to serve.

In this step, you will train two text classification models using different algorithms, then serialize them to disk so the API can load them at startup. This is the bridge between "model in a notebook" and "model in production."

In this step, get ready to:

Load and prepare the 20 Newsgroups text dataset.
Train two model versions with different algorithms.
Serialize trained models to disk as .joblib files.

Load the 20 Newsgroups dataset

Before you can train models, you need training data. scikit-learn ships with a built-in text dataset called 20 Newsgroups that contains around 18,000 posts across 20 topic categories.

Create a file called train_models.py in the root of your ml-inference-api folder (the same level as docker-compose.yml).
Paste the following code into train_models.py:

from sklearn.datasets import fetch_20newsgroups
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.naive_bayes import MultinomialNB
from sklearn.pipeline import Pipeline
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score
import joblib
import os

CATEGORIES = ["sci.space", "rec.sport.baseball", "comp.graphics", "talk.politics.misc"]

def main():
    print("Loading 20 Newsgroups dataset...")
    data = fetch_20newsgroups(subset="all", categories=CATEGORIES, remove=("headers", "footers", "quotes"))

    X_train, X_test, y_train, y_test = train_test_split(
        data.data, data.target, test_size=0.2, random_state=42
    )

    print(f"Training set: {len(X_train)} samples")
    print(f"Test set: {len(X_test)} samples")
    print(f"Categories: {data.target_names}")

    os.makedirs("models", exist_ok=True)


if __name__ == "__main__":
    main()

What does this code do?

fetch_20newsgroups downloads a classic text classification dataset built into scikit-learn. The first time you run this, it downloads the data and caches it locally. Subsequent runs use the cache.

CATEGORIES filters the full 20-category dataset down to 4 topics: space, baseball, computer graphics, and politics. This keeps training fast while giving you distinct, classifiable categories.

The remove=("headers", "footers", "quotes") argument strips metadata from posts so the model learns from actual content, not formatting artifacts.

Save train_models.py.
Run the script to confirm the data loads by running:

python train_models.py

You should see output showing the training and test set sizes, plus the four category names. The first run may take a moment while the dataset downloads.

Seeing a download error?

The dataset downloads from the internet on first run. Make sure you have an active internet connection. If you see a timeout, try running the script again.

If you see ModuleNotFoundError: No module named 'sklearn', make sure your virtual environment is activated before running the script.

Help me debug my train_models.py script

Train and save two model versions

Now you will train two models using different algorithms. Version 1 uses Logistic Regression (typically more accurate). Version 2 uses Multinomial Naive Bayes (typically faster). Having two genuinely different models makes A/B routing meaningful later.

In train_models.py, find the line os.makedirs("models", exist_ok=True) inside the main() function.
Copy the following code directly below that line (still inside main()):

    print("\n--- Training Model v1: TF-IDF + Logistic Regression ---")
    model_v1 = Pipeline([
        ("tfidf", TfidfVectorizer(max_features=10000, stop_words="english")),
        ("clf", LogisticRegression(max_iter=1000, random_state=42)),
    ])
    model_v1.fit(X_train, y_train)
    v1_accuracy = accuracy_score(y_test, model_v1.predict(X_test))
    print(f"v1 Accuracy: {v1_accuracy:.4f}")

    print("\n--- Training Model v2: TF-IDF + Multinomial Naive Bayes ---")
    model_v2 = Pipeline([
        ("tfidf", TfidfVectorizer(max_features=10000, stop_words="english")),
        ("clf", MultinomialNB(alpha=0.1)),
    ])
    model_v2.fit(X_train, y_train)
    v2_accuracy = accuracy_score(y_test, model_v2.predict(X_test))
    print(f"v2 Accuracy: {v2_accuracy:.4f}")

What does this code do?

Each model is a scikit-learn Pipeline that chains two steps together. First, TfidfVectorizer converts raw text into numerical vectors (term frequency-inverse document frequency). Then a classifier predicts the category from those vectors.

Using a Pipeline means the entire preprocessing-plus-prediction flow is saved as one object. When the API loads the model later, it can accept raw text directly without separate preprocessing logic.

Still inside main(), add the serialization code directly below the v2 accuracy print line:

    metadata = {
        "v1": {"algorithm": "LogisticRegression", "accuracy": round(v1_accuracy, 4)},
        "v2": {"algorithm": "MultinomialNB", "accuracy": round(v2_accuracy, 4)},
    }

    joblib.dump(model_v1, "models/model_v1.joblib")
    joblib.dump(model_v2, "models/model_v2.joblib")
    joblib.dump(metadata, "models/metadata.joblib")

    print("\nModels saved to models/ directory.")
    print(f"  models/model_v1.joblib - {metadata['v1']}")
    print(f"  models/model_v2.joblib - {metadata['v2']}")

Why save metadata separately?

joblib.dump serializes the trained Pipeline to a .joblib file. This preserves the model weights so you can load them without retraining.

The metadata.joblib file stores each model's algorithm name and accuracy score. The API will use this later to report model information through a /models endpoint.

Save train_models.py.
Run the training script:

python train_models.py

You should see accuracy scores printed for both models (typically v1 around 0.90+ and v2 slightly lower). The script also confirms three files saved to the models/ directory.

Confirm the model files exist by running:

ls models/

You should see metadata.joblib, model_v1.joblib, and model_v2.joblib listed.

Script runs but no accuracy output?

Make sure the training code is indented inside the main() function. Python is whitespace-sensitive. Each line of the new code should be indented with 4 spaces to sit inside main().

If you see NameError: name 'X_train' is not defined, the training code was placed outside of main() or above the train_test_split call.

Help me fix my train_models.py training script

✔️ Awesome, I've got everything!

Great. Make sure train_models.py is saved and the three .joblib files exist in models/.

ⓧ I'd like to double check the full code

Here is the complete train_models.py file:

from sklearn.datasets import fetch_20newsgroups
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.naive_bayes import MultinomialNB
from sklearn.pipeline import Pipeline
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score
import joblib
import os

CATEGORIES = ["sci.space", "rec.sport.baseball", "comp.graphics", "talk.politics.misc"]

def main():
    print("Loading 20 Newsgroups dataset...")
    data = fetch_20newsgroups(subset="all", categories=CATEGORIES, remove=("headers", "footers", "quotes"))

    X_train, X_test, y_train, y_test = train_test_split(
        data.data, data.target, test_size=0.2, random_state=42
    )

    print(f"Training set: {len(X_train)} samples")
    print(f"Test set: {len(X_test)} samples")
    print(f"Categories: {data.target_names}")

    os.makedirs("models", exist_ok=True)

    print("\n--- Training Model v1: TF-IDF + Logistic Regression ---")
    model_v1 = Pipeline([
        ("tfidf", TfidfVectorizer(max_features=10000, stop_words="english")),
        ("clf", LogisticRegression(max_iter=1000, random_state=42)),
    ])
    model_v1.fit(X_train, y_train)
    v1_accuracy = accuracy_score(y_test, model_v1.predict(X_test))
    print(f"v1 Accuracy: {v1_accuracy:.4f}")

    print("\n--- Training Model v2: TF-IDF + Multinomial Naive Bayes ---")
    model_v2 = Pipeline([
        ("tfidf", TfidfVectorizer(max_features=10000, stop_words="english")),
        ("clf", MultinomialNB(alpha=0.1)),
    ])
    model_v2.fit(X_train, y_train)
    v2_accuracy = accuracy_score(y_test, model_v2.predict(X_test))
    print(f"v2 Accuracy: {v2_accuracy:.4f}")

    metadata = {
        "v1": {"algorithm": "LogisticRegression", "accuracy": round(v1_accuracy, 4)},
        "v2": {"algorithm": "MultinomialNB", "accuracy": round(v2_accuracy, 4)},
    }

    joblib.dump(model_v1, "models/model_v1.joblib")
    joblib.dump(model_v2, "models/model_v2.joblib")
    joblib.dump(metadata, "models/metadata.joblib")

    print("\nModels saved to models/ directory.")
    print(f"  models/model_v1.joblib - {metadata['v1']}")
    print(f"  models/model_v2.joblib - {metadata['v2']}")


if __name__ == "__main__":
    main()

Your models are trained and saved. Next up, you will build the FastAPI application that loads these models and serves predictions through HTTP endpoints.

Build the Core API

You have trained models sitting in files on disk. Right now, they can only be used inside a Python script. To make them useful, you need an API layer that accepts text over HTTP and returns predictions.

In this step, you will build the FastAPI application that loads your models at startup, exposes health check endpoints, and serves predictions through a REST endpoint with input validation, caching, and logging.

In this step, get ready to:

Create the configuration, schema, and database modules that power the API.
Build a model manager that loads trained models at startup.
Wire up the FastAPI app with health, readiness, and prediction endpoints.

Create the configuration and schema modules

Every production API needs a centralized configuration that can be overridden with environment variables. Pydantic Settings provides this through a typed class that reads from environment variables or a .env file automatically.

Create app/config.py by pasting the following code:

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    # Application metadata
    app_name: str = "ML Inference API"
    app_version: str = "1.0.0"

    # Database and cache connection strings
    database_url: str = "postgresql://mlapi:mlapi_secret@localhost:5432/ml_inference"
    redis_url: str = "redis://localhost:6379/0"

    # Model loading configuration
    model_dir: str = "models"
    cache_ttl_seconds: int = 300

    # A/B routing weight for model v1
    ab_routing_v1_weight: float = 0.8

    class Config:
        env_file = ".env"


settings = Settings()

What does this code do?

The Settings class inherits from BaseSettings, which means any field can be overridden by an environment variable of the same name. The defaults point to the PostgreSQL and Redis containers you started in Step 1.

The settings singleton at the bottom means every module imports the same instance. No file needs to parse environment variables on its own.

Next, you need Pydantic models to validate incoming requests and structure outgoing responses.

Create app/schemas.py with the following code:

from pydantic import BaseModel, Field
from datetime import datetime


class PredictRequest(BaseModel):
    text: str = Field(..., min_length=1, max_length=5000, description="Text to classify")


class PredictResponse(BaseModel):
    prediction: str
    confidence: float
    model_version: str
    cached: bool
    response_time_ms: float


class HealthResponse(BaseModel):
    status: str
    version: str


class ReadinessResponse(BaseModel):
    status: str
    models_loaded: bool
    redis_connected: bool
    postgres_connected: bool


class ModelInfo(BaseModel):
    version: str
    algorithm: str
    accuracy: float
    loaded_at: str

What does this code do?

Each class defines the exact shape of a request or response. PredictRequest enforces that text must be between 1 and 5000 characters. If a caller sends empty text, FastAPI automatically returns a 422 validation error.

PredictResponse includes fields for the prediction label, confidence score, which model version served it, whether the result came from cache, and how long the request took.

Now create the database layer that connects to PostgreSQL and defines the prediction log table.

Create app/database.py with the following code:

from sqlalchemy import create_engine, Column, Integer, String, Float, Boolean, DateTime
from sqlalchemy.orm import sessionmaker, DeclarativeBase
from datetime import datetime, timezone

from app.config import settings


# Create the database engine using the URL from config
engine = create_engine(settings.database_url)
SessionLocal = sessionmaker(bind=engine)


class Base(DeclarativeBase):
    pass


# Define the table that stores every prediction for audit
class PredictionLog(Base):
    __tablename__ = "prediction_logs"

    id = Column(Integer, primary_key=True, index=True)
    input_text = Column(String, nullable=False)
    prediction = Column(String, nullable=False)
    confidence = Column(Float, nullable=False)
    model_version = Column(String, nullable=False)
    cached = Column(Boolean, default=False)
    response_time_ms = Column(Float, nullable=False)
    created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))


def init_db():
    Base.metadata.create_all(bind=engine)


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

What does this code do?

SQLAlchemy's create_engine opens a connection pool to your PostgreSQL container. The PredictionLog class maps to a prediction_logs table that records every inference request.

The init_db() function creates the table if it does not exist. You will call this during app startup so the table is ready before the first prediction arrives.

Save app/config.py, app/schemas.py, and app/database.py.

Seeing an import error?

Make sure your virtual environment is activated. The pydantic-settings and sqlalchemy packages must be installed from requirements.txt. Re-run pip install -r requirements.txt if needed.

help me fix my import error in config.py or database.py

Build the model manager

The model manager is responsible for loading your trained .joblib files into memory at startup and providing a simple predict() interface for the rest of the application.

Create app/model_manager.py with the following code:

import os
import joblib
from datetime import datetime, timezone


class ModelManager:
    def __init__(self):
        self.models = {}
        self.metadata = {}
        self.loaded_at = None

    def load_models(self, model_dir: str):
        # Load the metadata file that maps version names to algorithm info
        metadata = joblib.load(os.path.join(model_dir, "metadata.joblib"))

        for version in metadata:
            model_path = os.path.join(model_dir, f"model_{version}.joblib")
            self.models[version] = joblib.load(model_path)
            self.metadata[version] = metadata[version]

        self.loaded_at = datetime.now(timezone.utc).isoformat()
        print(f"Loaded {len(self.models)} model versions: {list(self.models.keys())}")

    def predict(self, text: str, version: str) -> tuple[str, float]:
        model = self.models[version]
        prediction = model.predict([text])[0]
        probabilities = model.predict_proba([text])[0]
        confidence = float(max(probabilities))

        # Map numeric prediction back to category name
        categories = ["comp.graphics", "rec.sport.baseball", "sci.space", "talk.politics.misc"]
        label = categories[prediction]

        return label, confidence

    def get_versions(self) -> list[str]:
        return list(self.models.keys())

    def is_loaded(self) -> bool:
        return len(self.models) > 0


# Singleton instance shared across the application
model_manager = ModelManager()

What does this code do?

The ModelManager class wraps all model-loading logic. When load_models() is called, it reads the metadata file to discover which versions exist, then loads each .joblib pipeline into a dictionary keyed by version name.

The predict() method runs the scikit-learn pipeline on the input text and returns both the predicted category label and the model's confidence score (the highest class probability).

Save app/model_manager.py.

Create the FastAPI app and route handlers

With configuration, schemas, database, and model loading in place, you can now build the API itself. You will create three files: a health router, a prediction router, and the main app entry point.

Create app/routers/health.py with the following code:

from fastapi import APIRouter
import redis

from app.config import settings
from app.schemas import HealthResponse, ReadinessResponse
from app.model_manager import model_manager
from app.database import engine

router = APIRouter()


@router.get("/health", response_model=HealthResponse)
def health_check():
    return HealthResponse(status="healthy", version=settings.app_version)


@router.get("/ready", response_model=ReadinessResponse)
def readiness_check():
    models_loaded = model_manager.is_loaded()

    # Check Redis connectivity
    redis_connected = False
    try:
        r = redis.from_url(settings.redis_url)
        r.ping()
        redis_connected = True
    except Exception:
        pass

    # Check PostgreSQL connectivity
    postgres_connected = False
    try:
        with engine.connect() as conn:
            conn.execute(conn.connection.cursor().execute("SELECT 1") if False else conn.exec_driver_sql("SELECT 1"))
            postgres_connected = True
    except Exception:
        pass

    all_healthy = models_loaded and redis_connected and postgres_connected
    status = "ready" if all_healthy else "degraded"

    return ReadinessResponse(
        status=status,
        models_loaded=models_loaded,
        redis_connected=redis_connected,
        postgres_connected=postgres_connected,
    )

Why separate /health from /ready?

The /health endpoint is a liveness probe. It always returns 200 if the process is running. Kubernetes uses this to decide whether to restart the container.

The /ready endpoint is a readiness probe. It checks that models are loaded and that Redis and PostgreSQL are reachable. Kubernetes uses this to decide whether to route traffic to the pod.

Create app/routers/predict.py with the following code:

import hashlib
import json
import time
import random

from fastapi import APIRouter, Header
import redis

from app.config import settings
from app.schemas import PredictRequest, PredictResponse, ModelInfo
from app.model_manager import model_manager
from app.database import SessionLocal, PredictionLog

router = APIRouter()

# Connect to Redis for prediction caching
redis_client = redis.from_url(settings.redis_url, decode_responses=True)


def select_model_version(requested_version: str | None) -> str:
    versions = model_manager.get_versions()

    # If a specific version was requested and exists, use it
    if requested_version and requested_version in versions:
        return requested_version

    # Otherwise, use weighted A/B routing
    if random.random() < settings.ab_routing_v1_weight:
        return "v1"
    return "v2"


def get_cache_key(text: str, version: str) -> str:
    content = f"{version}:{text}"
    return f"pred:{hashlib.sha256(content.encode()).hexdigest()[:16]}"


@router.post("/predict", response_model=PredictResponse)
def predict(request: PredictRequest, x_model_version: str | None = Header(None)):
    start_time = time.time()

    version = select_model_version(x_model_version)
    cache_key = get_cache_key(request.text, version)

    # Try to get a cached result from Redis
    cached_result = None
    try:
        cached_data = redis_client.get(cache_key)
        if cached_data:
            cached_result = json.loads(cached_data)
    except Exception:
        pass

    if cached_result:
        response_time_ms = (time.time() - start_time) * 1000
        prediction = cached_result["prediction"]
        confidence = cached_result["confidence"]
        cached = True
    else:
        # Run model inference
        prediction, confidence = model_manager.predict(request.text, version)
        response_time_ms = (time.time() - start_time) * 1000
        cached = False

        # Store result in Redis cache
        try:
            redis_client.setex(
                cache_key,
                settings.cache_ttl_seconds,
                json.dumps({"prediction": prediction, "confidence": confidence}),
            )
        except Exception:
            pass

    # Log the prediction to PostgreSQL
    db = SessionLocal()
    try:
        log_entry = PredictionLog(
            input_text=request.text[:500],
            prediction=prediction,
            confidence=confidence,
            model_version=version,
            cached=cached,
            response_time_ms=response_time_ms,
        )
        db.add(log_entry)
        db.commit()
    except Exception:
        db.rollback()
    finally:
        db.close()

    return PredictResponse(
        prediction=prediction,
        confidence=round(confidence, 4),
        model_version=version,
        cached=cached,
        response_time_ms=round(response_time_ms, 2),
    )


@router.get("/models", response_model=list[ModelInfo])
def list_models():
    models = []
    for version in model_manager.get_versions():
        meta = model_manager.metadata[version]
        models.append(
            ModelInfo(
                version=version,
                algorithm=meta["algorithm"],
                accuracy=meta["accuracy"],
                loaded_at=model_manager.loaded_at,
            )
        )
    return models

What does this code do?

The /predict endpoint does three things in sequence: check the Redis cache for a previously computed result, run model inference on a cache miss, and log the prediction to PostgreSQL.

The select_model_version() function implements A/B routing. If the caller sends an X-Model-Version header, that version is used. Otherwise, requests are split 80/20 between v1 and v2 based on the configured weight.

The /models endpoint returns metadata about all loaded model versions so callers can see what is available.

Create app/main.py with the following code:

from contextlib import asynccontextmanager

from fastapi import FastAPI

from app.config import settings
from app.model_manager import model_manager
from app.database import init_db
from app.routers import health, predict


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Runs once before the app starts accepting requests
    init_db()
    model_manager.load_models(settings.model_dir)
    yield


app = FastAPI(
    title=settings.app_name,
    version=settings.app_version,
    lifespan=lifespan,
)

app.include_router(health.router, tags=["Health"])
app.include_router(predict.router, tags=["Inference"])

What is the lifespan pattern?

FastAPI's lifespan parameter accepts an async context manager. Code before yield runs once at startup (creating the database table and loading models). Code after yield would run at shutdown.

This guarantees models are loaded into memory before the first request arrives. No request can reach the prediction endpoint until load_models() finishes.

Save all three files: app/routers/health.py, app/routers/predict.py, and app/main.py.

Now start the API server and verify everything works together.

Make sure your Docker containers are still running by checking:
Start the uvicorn development server by running:

uvicorn app.main:app --reload

You should see output showing the models loaded and the server running on http://127.0.0.1:8000.

Open http://127.0.0.1:8000/health in your browser. You should see a JSON response with "status": "healthy".
Open http://127.0.0.1:8000/docs to see the auto-generated Swagger UI.
In the Swagger UI, expand the POST /predict endpoint, click Try it out, enter {"text": "NASA launched a new telescope into orbit"} in the request body, and click Execute.

You should see a JSON response with a prediction field (likely sci.space), a confidence score, and the model_version that served it.

Server not starting?

If you see a ModuleNotFoundError, make sure you are running uvicorn from your project root directory (the folder that contains the app/ directory). Also confirm your virtual environment is activated.

If you see a connection error for PostgreSQL or Redis, run docker compose ps to verify both containers are running and healthy.

help me fix my FastAPI server startup error

✔️ Awesome, I've got everything!

Great. Make sure you have saved all your files before moving on.

ⓧ I'd like to double check the full code

Here is the full code for each file you created in this step.

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    app_name: str = "ML Inference API"
    app_version: str = "1.0.0"

    database_url: str = "postgresql://mlapi:mlapi_secret@localhost:5432/ml_inference"
    redis_url: str = "redis://localhost:6379/0"

    model_dir: str = "models"
    cache_ttl_seconds: int = 300

    ab_routing_v1_weight: float = 0.8

    class Config:
        env_file = ".env"


settings = Settings()

from pydantic import BaseModel, Field
from datetime import datetime


class PredictRequest(BaseModel):
    text: str = Field(..., min_length=1, max_length=5000, description="Text to classify")


class PredictResponse(BaseModel):
    prediction: str
    confidence: float
    model_version: str
    cached: bool
    response_time_ms: float


class HealthResponse(BaseModel):
    status: str
    version: str


class ReadinessResponse(BaseModel):
    status: str
    models_loaded: bool
    redis_connected: bool
    postgres_connected: bool


class ModelInfo(BaseModel):
    version: str
    algorithm: str
    accuracy: float
    loaded_at: str

from sqlalchemy import create_engine, Column, Integer, String, Float, Boolean, DateTime
from sqlalchemy.orm import sessionmaker, DeclarativeBase
from datetime import datetime, timezone

from app.config import settings


engine = create_engine(settings.database_url)
SessionLocal = sessionmaker(bind=engine)


class Base(DeclarativeBase):
    pass


class PredictionLog(Base):
    __tablename__ = "prediction_logs"

    id = Column(Integer, primary_key=True, index=True)
    input_text = Column(String, nullable=False)
    prediction = Column(String, nullable=False)
    confidence = Column(Float, nullable=False)
    model_version = Column(String, nullable=False)
    cached = Column(Boolean, default=False)
    response_time_ms = Column(Float, nullable=False)
    created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))


def init_db():
    Base.metadata.create_all(bind=engine)


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

import os
import joblib
from datetime import datetime, timezone


class ModelManager:
    def __init__(self):
        self.models = {}
        self.metadata = {}
        self.loaded_at = None

    def load_models(self, model_dir: str):
        metadata = joblib.load(os.path.join(model_dir, "metadata.joblib"))

        for version in metadata:
            model_path = os.path.join(model_dir, f"model_{version}.joblib")
            self.models[version] = joblib.load(model_path)
            self.metadata[version] = metadata[version]

        self.loaded_at = datetime.now(timezone.utc).isoformat()
        print(f"Loaded {len(self.models)} model versions: {list(self.models.keys())}")

    def predict(self, text: str, version: str) -> tuple[str, float]:
        model = self.models[version]
        prediction = model.predict([text])[0]
        probabilities = model.predict_proba([text])[0]
        confidence = float(max(probabilities))

        categories = ["comp.graphics", "rec.sport.baseball", "sci.space", "talk.politics.misc"]
        label = categories[prediction]

        return label, confidence

    def get_versions(self) -> list[str]:
        return list(self.models.keys())

    def is_loaded(self) -> bool:
        return len(self.models) > 0


model_manager = ModelManager()

from fastapi import APIRouter
import redis

from app.config import settings
from app.schemas import HealthResponse, ReadinessResponse
from app.model_manager import model_manager
from app.database import engine

router = APIRouter()


@router.get("/health", response_model=HealthResponse)
def health_check():
    return HealthResponse(status="healthy", version=settings.app_version)


@router.get("/ready", response_model=ReadinessResponse)
def readiness_check():
    models_loaded = model_manager.is_loaded()

    redis_connected = False
    try:
        r = redis.from_url(settings.redis_url)
        r.ping()
        redis_connected = True
    except Exception:
        pass

    postgres_connected = False
    try:
        with engine.connect() as conn:
            conn.execute(conn.connection.cursor().execute("SELECT 1") if False else conn.exec_driver_sql("SELECT 1"))
            postgres_connected = True
    except Exception:
        pass

    all_healthy = models_loaded and redis_connected and postgres_connected
    status = "ready" if all_healthy else "degraded"

    return ReadinessResponse(
        status=status,
        models_loaded=models_loaded,
        redis_connected=redis_connected,
        postgres_connected=postgres_connected,
    )

import hashlib
import json
import time
import random

from fastapi import APIRouter, Header
import redis

from app.config import settings
from app.schemas import PredictRequest, PredictResponse, ModelInfo
from app.model_manager import model_manager
from app.database import SessionLocal, PredictionLog

router = APIRouter()

redis_client = redis.from_url(settings.redis_url, decode_responses=True)


def select_model_version(requested_version: str | None) -> str:
    versions = model_manager.get_versions()

    if requested_version and requested_version in versions:
        return requested_version

    if random.random() < settings.ab_routing_v1_weight:
        return "v1"
    return "v2"


def get_cache_key(text: str, version: str) -> str:
    content = f"{version}:{text}"
    return f"pred:{hashlib.sha256(content.encode()).hexdigest()[:16]}"


@router.post("/predict", response_model=PredictResponse)
def predict(request: PredictRequest, x_model_version: str | None = Header(None)):
    start_time = time.time()

    version = select_model_version(x_model_version)
    cache_key = get_cache_key(request.text, version)

    cached_result = None
    try:
        cached_data = redis_client.get(cache_key)
        if cached_data:
            cached_result = json.loads(cached_data)
    except Exception:
        pass

    if cached_result:
        response_time_ms = (time.time() - start_time) * 1000
        prediction = cached_result["prediction"]
        confidence = cached_result["confidence"]
        cached = True
    else:
        prediction, confidence = model_manager.predict(request.text, version)
        response_time_ms = (time.time() - start_time) * 1000
        cached = False

        try:
            redis_client.setex(
                cache_key,
                settings.cache_ttl_seconds,
                json.dumps({"prediction": prediction, "confidence": confidence}),
            )
        except Exception:
            pass

    db = SessionLocal()
    try:
        log_entry = PredictionLog(
            input_text=request.text[:500],
            prediction=prediction,
            confidence=confidence,
            model_version=version,
            cached=cached,
            response_time_ms=response_time_ms,
        )
        db.add(log_entry)
        db.commit()
    except Exception:
        db.rollback()
    finally:
        db.close()

    return PredictResponse(
        prediction=prediction,
        confidence=round(confidence, 4),
        model_version=version,
        cached=cached,
        response_time_ms=round(response_time_ms, 2),
    )


@router.get("/models", response_model=list[ModelInfo])
def list_models():
    models = []
    for version in model_manager.get_versions():
        meta = model_manager.metadata[version]
        models.append(
            ModelInfo(
                version=version,
                algorithm=meta["algorithm"],
                accuracy=meta["accuracy"],
                loaded_at=model_manager.loaded_at,
            )
        )
    return models

from contextlib import asynccontextmanager

from fastapi import FastAPI

from app.config import settings
from app.model_manager import model_manager
from app.database import init_db
from app.routers import health, predict


@asynccontextmanager
async def lifespan(app: FastAPI):
    init_db()
    model_manager.load_models(settings.model_dir)
    yield


app = FastAPI(
    title=settings.app_name,
    version=settings.app_version,
    lifespan=lifespan,
)

app.include_router(health.router, tags=["Health"])
app.include_router(predict.router, tags=["Inference"])

Your ML inference API is live and serving real predictions. Next up, you will add Redis caching and PostgreSQL logging to track every prediction and speed up repeated requests.

Add Redis Caching and PostgreSQL Logging

Your API is serving predictions, but every request runs the full model inference pipeline even for identical inputs.

In this step, you will add a Redis caching layer that stores prediction results and returns them instantly on repeat requests. You will also add a PostgreSQL prediction log that records every inference for auditing and performance analysis.

In this step, get ready to:

Create a database layer with a prediction log table.
Add Redis caching and PostgreSQL logging to the predict endpoint.
Update the readiness probe to verify all dependencies.

Create the database layer

Every production API needs an audit trail. You need to know what predictions were made, which model served them, and how long each request took.

In VS Code, create a new file called database.py inside the app/ folder.
Add the following code to define the database connection, the prediction log table, and helper functions:

from sqlalchemy import create_engine, Column, Integer, String, Float, Boolean, DateTime
from sqlalchemy.orm import sessionmaker, DeclarativeBase
from datetime import datetime, timezone

from app.config import settings

# Create the database engine using the connection URL from config
engine = create_engine(settings.database_url)
SessionLocal = sessionmaker(bind=engine)


class Base(DeclarativeBase):
    pass


# Define the prediction log table structure
class PredictionLog(Base):
    __tablename__ = "prediction_logs"

    id = Column(Integer, primary_key=True, index=True)
    input_text = Column(String, nullable=False)
    prediction = Column(String, nullable=False)
    confidence = Column(Float, nullable=False)
    model_version = Column(String, nullable=False)
    cached = Column(Boolean, default=False)
    response_time_ms = Column(Float, nullable=False)
    created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))


def init_db():
    # Create all tables in the database
    Base.metadata.create_all(bind=engine)


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

What does this code do?

This file sets up three things your API needs to talk to PostgreSQL.

engine and SessionLocal handle the database connection pool. SQLAlchemy reuses connections efficiently so you don't open a new one per request.
PredictionLog defines the table schema as a Python class. Each column maps to a piece of data you want to track about every prediction.
init_db() creates the table if it doesn't exist. Your main.py already calls this during startup via the lifespan function.

Save app/database.py.

Why log every prediction?

Prediction logs let you answer critical production questions. Which model version produces more confident results? How much does caching improve response times? Are certain inputs causing slow inferences?

Without this audit trail, you are flying blind once the API is live.

Add caching and logging to the predict endpoint

The predict endpoint currently runs model inference on every request. You will now add a cache-first pattern: check Redis before running the model, and log every result to PostgreSQL regardless.

Open app/routers/predict.py in VS Code.
Replace the entire file contents with the following code:

import hashlib
import json
import time
import random

from fastapi import APIRouter, Header
import redis

from app.config import settings
from app.schemas import PredictRequest, PredictResponse, ModelInfo
from app.model_manager import model_manager
from app.database import SessionLocal, PredictionLog

router = APIRouter()

# Connect to Redis with string decoding enabled
redis_client = redis.from_url(settings.redis_url, decode_responses=True)


def select_model_version(requested_version: str | None) -> str:
    versions = model_manager.get_versions()

    if requested_version and requested_version in versions:
        return requested_version

    # A/B routing: randomly select model based on configured weight
    if random.random() < settings.ab_routing_v1_weight:
        return "v1"
    return "v2"


def get_cache_key(text: str, version: str) -> str:
    # Create a unique cache key from model version + input text
    content = f"{version}:{text}"
    return f"pred:{hashlib.sha256(content.encode()).hexdigest()[:16]}"


@router.post("/predict", response_model=PredictResponse)
def predict(request: PredictRequest, x_model_version: str | None = Header(None)):
    start_time = time.time()

    version = select_model_version(x_model_version)
    cache_key = get_cache_key(request.text, version)

    # Check Redis cache first
    cached_result = None
    try:
        cached_data = redis_client.get(cache_key)
        if cached_data:
            cached_result = json.loads(cached_data)
    except Exception:
        pass

    if cached_result:
        # Cache hit: use stored prediction
        response_time_ms = (time.time() - start_time) * 1000
        prediction = cached_result["prediction"]
        confidence = cached_result["confidence"]
        cached = True
    else:
        # Cache miss: run model inference
        prediction, confidence = model_manager.predict(request.text, version)
        response_time_ms = (time.time() - start_time) * 1000
        cached = False

        # Store result in Redis with TTL
        try:
            redis_client.setex(
                cache_key,
                settings.cache_ttl_seconds,
                json.dumps({"prediction": prediction, "confidence": confidence}),
            )
        except Exception:
            pass

    # Log every prediction to PostgreSQL
    db = SessionLocal()
    try:
        log_entry = PredictionLog(
            input_text=request.text[:500],
            prediction=prediction,
            confidence=confidence,
            model_version=version,
            cached=cached,
            response_time_ms=response_time_ms,
        )
        db.add(log_entry)
        db.commit()
    except Exception:
        db.rollback()
    finally:
        db.close()

    return PredictResponse(
        prediction=prediction,
        confidence=round(confidence, 4),
        model_version=version,
        cached=cached,
        response_time_ms=round(response_time_ms, 2),
    )


@router.get("/models", response_model=list[ModelInfo])
def list_models():
    models = []
    for version in model_manager.get_versions():
        meta = model_manager.metadata[version]
        models.append(
            ModelInfo(
                version=version,
                algorithm=meta["algorithm"],
                accuracy=meta["accuracy"],
                loaded_at=model_manager.loaded_at,
            )
        )
    return models

What's happening here?

The predict endpoint now follows a cache-first pattern with three phases.

Cache check: get_cache_key() hashes the input text and model version into a short key like pred:a3f8b2c1e9d04567. The endpoint looks up this key in Redis first.
Inference or return: On a cache miss, the model runs inference and the result is stored in Redis with a 5-minute TTL using setex. On a cache hit, the stored result is returned directly.
Logging: Every prediction (cached or fresh) is logged to PostgreSQL with timing data. The cached boolean flag tells you whether this request hit the cache or ran the model.

Save app/routers/predict.py.

Seeing an import error for PredictionLog?

Make sure you saved app/database.py before saving predict.py. The predict router imports SessionLocal and PredictionLog from app.database, which won't exist if the file isn't saved.

help me fix this PredictionLog import error

Update the readiness probe to check all dependencies

A readiness probe should verify that all critical dependencies are reachable. Your /ready endpoint already checks if models are loaded, but it also needs to confirm Redis and PostgreSQL are accessible.

Open app/routers/health.py in VS Code.
Replace the entire file contents with the following code:

from fastapi import APIRouter
import redis

from app.config import settings
from app.schemas import HealthResponse, ReadinessResponse
from app.model_manager import model_manager
from app.database import engine

router = APIRouter()


@router.get("/health", response_model=HealthResponse)
def health_check():
    # Liveness probe: always returns healthy if the process is running
    return HealthResponse(status="healthy", version=settings.app_version)


@router.get("/ready", response_model=ReadinessResponse)
def readiness_check():
    models_loaded = model_manager.is_loaded()

    # Check Redis connectivity
    redis_connected = False
    try:
        r = redis.from_url(settings.redis_url)
        r.ping()
        redis_connected = True
    except Exception:
        pass

    # Check PostgreSQL connectivity
    postgres_connected = False
    try:
        with engine.connect() as conn:
            conn.execute(conn.connection.cursor().execute("SELECT 1") if False else conn.exec_driver_sql("SELECT 1"))
            postgres_connected = True
    except Exception:
        pass

    all_healthy = models_loaded and redis_connected and postgres_connected
    status = "ready" if all_healthy else "degraded"

    return ReadinessResponse(
        status=status,
        models_loaded=models_loaded,
        redis_connected=redis_connected,
        postgres_connected=postgres_connected,
    )

What does this code do?

The readiness probe now checks three things: models loaded, Redis reachable, and PostgreSQL reachable. If any check fails, the status changes from ready to degraded.

This pattern maps directly to Kubernetes readiness probes. A degraded service stops receiving traffic until its dependencies recover.

Save app/routers/health.py.
Restart your server by pressing Ctrl+C in the terminal, then running:

uvicorn app.main:app --reload

Test the caching behavior by opening http://localhost:8000/docs in your browser.
Click the POST /predict endpoint, then click Try it out.
Enter this request body and click Execute:

{
  "text": "NASA launched a new satellite into orbit today"
}

Check the response. You should see "cached": false because this is the first time the API has seen this input.
Click Execute again with the same input text.
This time you should see "cached": true and a significantly lower response_time_ms value.

What should I see?

The first request takes longer because it runs the full model pipeline. The second request returns almost instantly because Redis serves the cached result. Both requests get logged to PostgreSQL with their respective cached flag values.

Verify the readiness probe by visiting http://localhost:8000/ready in your browser. You should see all three checks returning true and status showing ready.

Seeing degraded status on /ready?

Make sure your Docker containers are running. Switch to a new terminal tab and run:

docker compose ps

Both postgres and redis should show a healthy status. If either is stopped, run docker compose up -d to restart them.

help me fix my degraded readiness check

Test the degraded state by stopping Redis temporarily:

docker compose stop redis

Visit http://localhost:8000/ready again. You should see "redis_connected": false and "status": "degraded".
Restart Redis by running:

docker compose start redis

Visit http://localhost:8000/ready one more time. Status should return to ready with all checks passing.

✔️ Awesome, I've got everything!

Great. Double check you have saved all three files: app/database.py, app/routers/predict.py, and app/routers/health.py.

ⓧ I'd like to double check the full code

Here is the full code for each file you created or updated in this step.

from sqlalchemy import create_engine, Column, Integer, String, Float, Boolean, DateTime
from sqlalchemy.orm import sessionmaker, DeclarativeBase
from datetime import datetime, timezone

from app.config import settings


engine = create_engine(settings.database_url)
SessionLocal = sessionmaker(bind=engine)


class Base(DeclarativeBase):
    pass


class PredictionLog(Base):
    __tablename__ = "prediction_logs"

    id = Column(Integer, primary_key=True, index=True)
    input_text = Column(String, nullable=False)
    prediction = Column(String, nullable=False)
    confidence = Column(Float, nullable=False)
    model_version = Column(String, nullable=False)
    cached = Column(Boolean, default=False)
    response_time_ms = Column(Float, nullable=False)
    created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))


def init_db():
    Base.metadata.create_all(bind=engine)


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

import hashlib
import json
import time
import random

from fastapi import APIRouter, Header
import redis

from app.config import settings
from app.schemas import PredictRequest, PredictResponse, ModelInfo
from app.model_manager import model_manager
from app.database import SessionLocal, PredictionLog

router = APIRouter()

redis_client = redis.from_url(settings.redis_url, decode_responses=True)


def select_model_version(requested_version: str | None) -> str:
    versions = model_manager.get_versions()

    if requested_version and requested_version in versions:
        return requested_version

    if random.random() < settings.ab_routing_v1_weight:
        return "v1"
    return "v2"


def get_cache_key(text: str, version: str) -> str:
    content = f"{version}:{text}"
    return f"pred:{hashlib.sha256(content.encode()).hexdigest()[:16]}"


@router.post("/predict", response_model=PredictResponse)
def predict(request: PredictRequest, x_model_version: str | None = Header(None)):
    start_time = time.time()

    version = select_model_version(x_model_version)
    cache_key = get_cache_key(request.text, version)

    cached_result = None
    try:
        cached_data = redis_client.get(cache_key)
        if cached_data:
            cached_result = json.loads(cached_data)
    except Exception:
        pass

    if cached_result:
        response_time_ms = (time.time() - start_time) * 1000
        prediction = cached_result["prediction"]
        confidence = cached_result["confidence"]
        cached = True
    else:
        prediction, confidence = model_manager.predict(request.text, version)
        response_time_ms = (time.time() - start_time) * 1000
        cached = False

        try:
            redis_client.setex(
                cache_key,
                settings.cache_ttl_seconds,
                json.dumps({"prediction": prediction, "confidence": confidence}),
            )
        except Exception:
            pass

    db = SessionLocal()
    try:
        log_entry = PredictionLog(
            input_text=request.text[:500],
            prediction=prediction,
            confidence=confidence,
            model_version=version,
            cached=cached,
            response_time_ms=response_time_ms,
        )
        db.add(log_entry)
        db.commit()
    except Exception:
        db.rollback()
    finally:
        db.close()

    return PredictResponse(
        prediction=prediction,
        confidence=round(confidence, 4),
        model_version=version,
        cached=cached,
        response_time_ms=round(response_time_ms, 2),
    )


@router.get("/models", response_model=list[ModelInfo])
def list_models():
    models = []
    for version in model_manager.get_versions():
        meta = model_manager.metadata[version]
        models.append(
            ModelInfo(
                version=version,
                algorithm=meta["algorithm"],
                accuracy=meta["accuracy"],
                loaded_at=model_manager.loaded_at,
            )
        )
    return models

from fastapi import APIRouter
import redis

from app.config import settings
from app.schemas import HealthResponse, ReadinessResponse
from app.model_manager import model_manager
from app.database import engine

router = APIRouter()


@router.get("/health", response_model=HealthResponse)
def health_check():
    return HealthResponse(status="healthy", version=settings.app_version)


@router.get("/ready", response_model=ReadinessResponse)
def readiness_check():
    models_loaded = model_manager.is_loaded()

    redis_connected = False
    try:
        r = redis.from_url(settings.redis_url)
        r.ping()
        redis_connected = True
    except Exception:
        pass

    postgres_connected = False
    try:
        with engine.connect() as conn:
            conn.execute(conn.connection.cursor().execute("SELECT 1") if False else conn.exec_driver_sql("SELECT 1"))
            postgres_connected = True
    except Exception:
        pass

    all_healthy = models_loaded and redis_connected and postgres_connected
    status = "ready" if all_healthy else "degraded"

    return ReadinessResponse(
        status=status,
        models_loaded=models_loaded,
        redis_connected=redis_connected,
        postgres_connected=postgres_connected,
    )

Your API now caches predictions for instant repeat responses and logs every inference to PostgreSQL. Next up, you will add model versioning with A/B routing so you can serve multiple models simultaneously and compare their performance.

Implement Model Versioning and A/B Routing

Your API can now cache predictions and log every request to PostgreSQL. The next challenge is serving multiple model versions simultaneously.

In production ML systems, you rarely swap models in one shot. Instead, you route traffic between versions so you can compare performance before committing to an upgrade. This is called A/B routing.

In this step, get ready to:

Understand the A/B routing logic that selects model versions.
Test model version pinning with the X-Model-Version header.
Explore the /models metadata endpoint.

Walk through the A/B routing logic

Your app/routers/predict.py already includes a select_model_version() function that decides which model handles each request. This is the routing brain of your versioning system.

In VS Code, open app/routers/predict.py and find the select_model_version function near the top of the file:

def select_model_version(requested_version: str | None) -> str:
    versions = model_manager.get_versions()

    if requested_version and requested_version in versions:
        return requested_version

    if random.random() < settings.ab_routing_v1_weight:
        return "v1"
    return "v2"

What does this function do?

This function implements two routing strategies depending on whether the caller specifies a version:

If the request includes a specific version (like v2) and that version is loaded, route directly to it. This is version pinning.
If no version is specified, use a weighted random split. The ab_routing_v1_weight setting (default 0.8) sends 80% of traffic to v1 and 20% to v2.

Now find the predict function signature just below:

@router.post("/predict", response_model=PredictResponse)
def predict(request: PredictRequest, x_model_version: str | None = Header(None)):

How does FastAPI read the header?

FastAPI automatically converts Python parameter names from snake_case to hyphenated headers. The parameter x_model_version maps to the HTTP header X-Model-Version.

Setting the default to None makes it optional. Callers who omit the header get A/B routing automatically.

Why A/B routing instead of a hard cutover?

Hard model swaps are risky. If v2 has a subtle regression on certain input types, you might not notice until production users complain.

A/B routing lets you expose a small percentage of traffic to the new model while monitoring accuracy and latency. If v2 underperforms, you shift the weight back to v1 with a single config change.

Test A/B routing and version pinning

Now that you understand how the routing works, let's verify it in practice by sending requests and observing which model version responds.

Make sure your server is running. If it's not, start it by running:

uvicorn app.main:app --reload

Open a new terminal tab and send several prediction requests without a version header by running this command multiple times:

curl -X POST http://localhost:8000/predict \
  -H "Content-Type: application/json" \
  -d '{"text": "The spacecraft launched into orbit successfully"}'

Check the model_version field in each response. Over 10+ requests, you should see roughly 80% return v1 and 20% return v2.

What should I see?

Each response includes a model_version field showing either v1 or v2. The exact split is random, so 10 requests might produce 7-9 v1 results and 1-3 v2 results. The important thing is that both versions appear.

Now test version pinning by sending a request with the X-Model-Version header by running:

curl -X POST http://localhost:8000/predict \
  -H "Content-Type: application/json" \
  -H "X-Model-Version: v2" \
  -d '{"text": "The baseball game went into extra innings"}'

Confirm the response shows "model_version": "v2". Run it several more times. Every response should consistently use v2.

Always seeing the same model version without the header?

With an 80/20 split, short runs of identical versions are normal. Try 15-20 requests to see both versions appear.

If you only ever see one version, check that both models/model_v1.joblib and models/model_v2.joblib exist. The ModelManager only routes to versions it successfully loaded at startup.

help me debug why my A/B routing only returns one model version

Explore the /models metadata endpoint

Production APIs need a way to report which model versions are available. Your /models endpoint provides this metadata without requiring a prediction request.

Query the models endpoint by running:

curl http://localhost:8000/models

You should see a JSON array listing both model versions with their algorithm name, accuracy score, and load timestamp.

What does the /models response contain?

Each entry in the response includes:

version - the model identifier (v1 or v2).
algorithm - which scikit-learn classifier is behind this version.
accuracy - the test set accuracy recorded during training.
loaded_at - when the model was loaded into memory.

This endpoint is useful for monitoring dashboards and deployment automation that needs to confirm which versions are live.

Find the list_models function at the bottom of app/routers/predict.py:

@router.get("/models", response_model=list[ModelInfo])
def list_models():
    models = []
    for version in model_manager.get_versions():
        meta = model_manager.metadata[version]
        models.append(
            ModelInfo(
                version=version,
                algorithm=meta["algorithm"],
                accuracy=meta["accuracy"],
                loaded_at=model_manager.loaded_at,
            )
        )
    return models

What does this code do?

The list_models function iterates over every version the ModelManager loaded at startup. It pulls the algorithm name and accuracy from the metadata dictionary (saved during training) and wraps everything in a ModelInfo Pydantic model for validated JSON output.

Seeing an empty list from /models?

If the endpoint returns [], the ModelManager failed to load models at startup. Check your terminal for errors. The most common cause is running uvicorn from the wrong directory. You must run it from the project root where the models/ folder lives.

help me debug why /models returns an empty array

✔️ Awesome, I've got everything!

Great. Make sure your server is still running so you can use it in the next verification.

ⓧ I'd like to double check the full code

Here is the complete app/routers/predict.py file:

import hashlib
import json
import time
import random

from fastapi import APIRouter, Header
import redis

from app.config import settings
from app.schemas import PredictRequest, PredictResponse, ModelInfo
from app.model_manager import model_manager
from app.database import SessionLocal, PredictionLog

router = APIRouter()

redis_client = redis.from_url(settings.redis_url, decode_responses=True)


def select_model_version(requested_version: str | None) -> str:
    versions = model_manager.get_versions()

    if requested_version and requested_version in versions:
        return requested_version

    if random.random() < settings.ab_routing_v1_weight:
        return "v1"
    return "v2"


def get_cache_key(text: str, version: str) -> str:
    content = f"{version}:{text}"
    return f"pred:{hashlib.sha256(content.encode()).hexdigest()[:16]}"


@router.post("/predict", response_model=PredictResponse)
def predict(request: PredictRequest, x_model_version: str | None = Header(None)):
    start_time = time.time()

    version = select_model_version(x_model_version)
    cache_key = get_cache_key(request.text, version)

    cached_result = None
    try:
        cached_data = redis_client.get(cache_key)
        if cached_data:
            cached_result = json.loads(cached_data)
    except Exception:
        pass

    if cached_result:
        response_time_ms = (time.time() - start_time) * 1000
        prediction = cached_result["prediction"]
        confidence = cached_result["confidence"]
        cached = True
    else:
        prediction, confidence = model_manager.predict(request.text, version)
        response_time_ms = (time.time() - start_time) * 1000
        cached = False

        try:
            redis_client.setex(
                cache_key,
                settings.cache_ttl_seconds,
                json.dumps({"prediction": prediction, "confidence": confidence}),
            )
        except Exception:
            pass

    db = SessionLocal()
    try:
        log_entry = PredictionLog(
            input_text=request.text[:500],
            prediction=prediction,
            confidence=confidence,
            model_version=version,
            cached=cached,
            response_time_ms=response_time_ms,
        )
        db.add(log_entry)
        db.commit()
    except Exception:
        db.rollback()
    finally:
        db.close()

    return PredictResponse(
        prediction=prediction,
        confidence=round(confidence, 4),
        model_version=version,
        cached=cached,
        response_time_ms=round(response_time_ms, 2),
    )


@router.get("/models", response_model=list[ModelInfo])
def list_models():
    models = []
    for version in model_manager.get_versions():
        meta = model_manager.metadata[version]
        models.append(
            ModelInfo(
                version=version,
                algorithm=meta["algorithm"],
                accuracy=meta["accuracy"],
                loaded_at=model_manager.loaded_at,
            )
        )
    return models

Your ML inference API now serves multiple model versions with intelligent traffic routing. In the Secret Mission, you'll add observability to track how each version performs over time.

Secret mission

Add Observability with Metrics

Your API serves predictions, but how do you know how well it's performing? In this bonus challenge, you'll add request timing middleware and a /metrics endpoint that reports aggregated model performance statistics from your prediction log.

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely locally using Docker containers, so there are no ongoing cloud costs.

Resources you used:

PostgreSQL container (Docker) with prediction_logs table.
Redis container (Docker) for prediction caching.
Two trained ML model files (model_v1.joblib, model_v2.joblib) on disk.
Python virtual environment and project files.

Keep everything running

No action needed. Choose this if you are still actively building or experimenting with the API.

Your Docker containers use minimal local resources while running.
The uvicorn dev server only runs while you have it active in the terminal.
Your model files and virtual environment stay on disk ready for use.

Pause - I'll come back to this later

Shut down running processes to free up memory while keeping all your data and files intact.

Stop the uvicorn dev server by pressing Ctrl+C in the terminal where it is running.
Stop the Docker containers (without removing data) by running:

docker compose stop

When you are ready to resume, start the containers again by running:

docker compose start

Your PostgreSQL data (prediction logs) and Redis cache entries survive the stop. Everything picks up where you left off.

Delete - I don't want to use this again

Remove all project resources completely and start fresh.

Stop the uvicorn dev server by pressing Ctrl+C in the terminal where it is running.
Remove the Docker containers and their volumes (this deletes all PostgreSQL data) by running:

docker compose down -v

Deactivate your Python virtual environment by running:

deactivate

Navigate to your Desktop and delete the project folder by running:

macOS

cd ~/Desktop
rm -rf ml-inference-api

Windows

cd ~\Desktop
rmdir /s /q ml-inference-api

What about Docker images?

The postgres:16 and redis:7 Docker images remain on your machine. They take up disk space but cost nothing. To remove them, open Docker Desktop and delete them from the Images tab.

Nice Work!

Nice work! You just built a production-grade ML inference API that serves real text classification predictions through a fully instrumented backend.

You've learned how to:

Build a production ML inference API with FastAPI serving real scikit-learn text classifiers with input validation and structured error handling.
Implement Redis prediction caching and PostgreSQL prediction logging for sub-millisecond cache hits and a full audit trail.
Create a model versioning system with A/B traffic routing and liveness/readiness health probes that verify all dependencies.
Secret Mission: Added request timing middleware and a /metrics endpoint with aggregated model performance statistics.

Ready to quiz yourself?

Build a Production ML Inference API

Introduction

30 Second Summary

What You'll Build

Are there any prerequisites?

Set Up the Project Environment

In this step, get ready to:

Create the project directory and virtual environment

✔️ I see version 3.11 or higher

ⓧ I see a version lower than 3.11

ⓧ Command not found

macOS

Windows

What is a virtual environment?

code command not found?

What are these packages?

pip install failing?

Start PostgreSQL and Redis with Docker Compose

What does this file do?

Containers not starting?

Create the project file structure

What are __init__.py files?

✔️ Awesome, I've got everything!

ⓧ I'd like to double check the full code

Train and Serialize the ML Models

In this step, get ready to:

Load the 20 Newsgroups dataset

What does this code do?

Seeing a download error?

Train and save two model versions

What does this code do?

Why save metadata separately?

Script runs but no accuracy output?

✔️ Awesome, I've got everything!

ⓧ I'd like to double check the full code

Build the Core API

In this step, get ready to:

Create the configuration and schema modules

What does this code do?

What does this code do?

What does this code do?

Seeing an import error?

Build the model manager

What does this code do?

Create the FastAPI app and route handlers

Why separate /health from /ready?

What does this code do?

What is the lifespan pattern?

Server not starting?

✔️ Awesome, I've got everything!

ⓧ I'd like to double check the full code

Add Redis Caching and PostgreSQL Logging

In this step, get ready to:

Create the database layer

What does this code do?

Why log every prediction?

Add caching and logging to the predict endpoint

What's happening here?

Seeing an import error for PredictionLog?

Update the readiness probe to check all dependencies

What does this code do?

What should I see?

Seeing degraded status on /ready?

✔️ Awesome, I've got everything!

ⓧ I'd like to double check the full code

Implement Model Versioning and A/B Routing

In this step, get ready to:

Walk through the A/B routing logic

What does this function do?

How does FastAPI read the header?

Why A/B routing instead of a hard cutover?

Test A/B routing and version pinning

What should I see?

Always seeing the same model version without the header?

Explore the /models metadata endpoint

What does the /models response contain?

What does this code do?

Seeing an empty list from /models?

✔️ Awesome, I've got everything!

ⓧ I'd like to double check the full code

What are init.py files?