Shipping a TensorFlow Model to Production with FastAPI, Docker, and Model Monitoring

Table of Contents#

Prerequisites
Step 1: Prepare Your Trained TensorFlow Model for Serving
Step 2: Build a Production-Grade FastAPI Serving Layer
Step 3: Containerize Your Service with Docker for Reproducible Deployments
Step 4: Implement Model Monitoring for Drift and Performance Tracking
Production Best Practices for ML Serving
Common Pitfalls to Avoid
Real-World Use Cases
Conclusion
References

Prerequisites#

Before starting, ensure you have:

Python 3.11+ installed
Basic familiarity with TensorFlow 2.x and REST APIs
Docker Desktop installed on your local machine
A trained TensorFlow model (we use a sample text sentiment analysis model for examples)

Step 1: Prepare Your Trained TensorFlow Model for Serving#

TensorFlow supports two model storage formats, and choosing the right one is critical for seamless serving:

Format	Pros	Cons	Use Case
SavedModel (default)	Directory-based, supports custom layers (e.g. TextVectorization), optimized for serving	Multiple files	Recommended for all production use cases
HDF5 (.h5)	Single file, easy to share	Limited support for custom layers, requires explicit layer definition during loading	Experimental/development use only

Save Your Model in SavedModel Format#

Convert any existing HDF5 model to SavedModel, or save newly trained models directly to this format:

import tensorflow as tf
 
# Load your trained model (replace with your model path)
trained_model = tf.keras.models.load_model("trained_sentiment_model.h5")
 
# Save as SavedModel with versioning for easy rollbacks
trained_model.save("models/sentiment_v1/")

Always version your model directories (e.g. sentiment_v1, sentiment_v2) to enable seamless rollouts and A/B testing of new model versions.

Step 2: Build a Production-Grade FastAPI Serving Layer#

FastAPI is the leading choice for ML serving due to its built-in validation, async support, and automatic documentation, outperforming legacy frameworks like Flask for high-concurrency workloads.

Recommended Project Structure#

Follow this standardized src layout for maintainability and compliance with MLOps best practices:

sentiment-api/
├── src/
│   ├── api/               # FastAPI routes and endpoints
│   ├── core/              # Configuration, logging setup
│   ├── models/            # Model loading and prediction logic
│   └── utils/             # Helper functions
├── models/                # Saved TensorFlow models
├── configs/               # Reference data, drift thresholds
├── tests/                 # Unit and integration tests
├── requirements.txt
├── Dockerfile
├── .env
└── pyproject.toml

1. Configure Dependencies and Settings#

First, define your requirements in requirements.txt:

fastapi==0.109.0
uvicorn==0.27.0
gunicorn==21.2.0
pydantic-settings==2.1.0
tensorflow-cpu==2.15.0  # Use tensorflow-gpu for GPU deployments
evidently==0.4.10       # For drift detection
python-json-logger==2.0.7

Use Pydantic BaseSettings for environment-aware configuration, no hardcoded values:

# src/core/config.py
from pydantic_settings import BaseSettings
from typing import List
 
class Settings(BaseSettings):
    APP_NAME: str = "Sentiment Analysis API"
    API_VERSION: str = "1.0.0"
    MODEL_PATH: str = "models/sentiment_v1/"
    CORS_ORIGINS: List[str] = ["https://your-app.com"]  # Restrict in production
    PORT: int = 8000
    DRIFT_THRESHOLD: float = 0.2
 
    class Config:
        env_file = ".env"
 
settings = Settings()

2. Load Model Once at Startup#

Avoid the critical mistake of loading your model on every request (which adds 100ms+ latency per call). Initialize the model once during app startup:

# src/models/loader.py
import tensorflow as tf
from typing import List
from src.core.config import settings
 
class SentimentModel:
    def __init__(self, model_path: str):
        self.model = tf.keras.models.load_model(model_path)
    
    def predict(self, texts: List[str]) -> List[float]:
        predictions = self.model.predict(texts, verbose=0)
        return [float(pred[0]) for pred in predictions]
 
# Global model instance
model: SentimentModel | None = None
 
def init_model():
    global model
    model = SentimentModel(settings.MODEL_PATH)

3. Define API Endpoints#

Create routes with Pydantic validation for request/response schemas, plus health checks for orchestration tools like Kubernetes:

# src/api/routes.py
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel, Field
from typing import List
import json
from datetime import datetime, timezone
from src.models.loader import model
from src.core.config import settings
 
router = APIRouter()
 
# Request/Response schemas
class PredictionRequest(BaseModel):
    texts: List[str] = Field(..., min_length=1, description="List of text inputs to classify")
 
class PredictionResponse(BaseModel):
    predictions: List[float] = Field(..., description="Sentiment scores: 0 = negative, 1 = positive")
 
# Health checks for load balancers
@router.get("/health", tags=["Health"])
async def liveness_check():
    return {"status": "alive", "app_name": settings.APP_NAME, "version": settings.API_VERSION}
 
@router.get("/readiness", tags=["Health"])
async def readiness_check():
    if model is None:
        raise HTTPException(status_code=503, detail="Model not initialized")
    return {"status": "ready"}
 
# Prediction endpoint
@router.post("/api/v1/predict", response_model=PredictionResponse, tags=["Prediction"])
async def predict(request: PredictionRequest):
    try:
        start_time = datetime.now(timezone.utc)
        preds = model.predict(request.texts)
        latency = (datetime.now(timezone.utc) - start_time).total_seconds() * 1000
 
        # Log inputs and predictions for monitoring
        log_entry = json.dumps({
            "timestamp": start_time.isoformat(),
            "inputs": request.texts,
            "predictions": preds,
            "latency_ms": latency,
            "model_version": settings.API_VERSION
        })
        with open("/app/logs/predictions.jsonl", "a") as f:
            f.write(log_entry + "\n")
 
        return {"predictions": preds}
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Prediction failed: {str(e)}")

4. Initialize the FastAPI App#

Add CORS middleware and startup hooks to your main app file:

# src/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from src.api.routes import router
from src.core.config import settings
from src.models.loader import init_model
 
# Load model on app startup using lifespan (recommended over deprecated on_event)
@asynccontextmanager
async def lifespan(app: FastAPI):
    init_model()
    yield
 
app = FastAPI(title=settings.APP_NAME, version=settings.API_VERSION, lifespan=lifespan)
 
# CORS configuration
app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.CORS_ORIGINS,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
 
app.include_router(router)

Run the Production Server#

Use Gunicorn with Uvicorn workers for production (match worker count to your CPU core count):

gunicorn src.main:app --workers 4 --worker-class uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000

You can test the API and view auto-generated Swagger documentation at http://localhost:8000/docs.

Step 3: Containerize Your Service with Docker for Reproducible Deployments#

Docker eliminates "works on my machine" errors by packaging your model, code, dependencies, and runtime into a single portable image. Use this optimized Dockerfile to minimize image size and improve build speeds:

# Stage 1: Build dependencies (separate to reduce final image size)
FROM python:3.11-slim as builder
 
WORKDIR /app
ENV PYTHONDONTWRITEBYTECODE 1
ENV PYTHONUNBUFFERED 1
 
# Install system dependencies for TensorFlow
RUN apt-get update && apt-get install -y --no-install-recommends build-essential \
    && rm -rf /var/lib/apt/lists/*
 
# Install Python dependencies
COPY requirements.txt .
RUN pip install --user --no-cache-dir -r requirements.txt
 
# Stage 2: Final runtime image
FROM python:3.11-slim
 
WORKDIR /app
 
# Copy dependencies from builder stage
COPY --from=builder /root/.local/lib/python3.11/site-packages /usr/local/lib/python3.11/site-packages
COPY --from=builder /root/.local/bin /usr/local/bin
 
# Create logs directory
RUN mkdir -p /app/logs /app/reports
 
# Copy model first for Docker layer caching (avoids rebuilding model layer on code changes)
COPY models/ ./models/
# Copy app code
COPY src/ ./src/
COPY configs/ ./configs/
COPY .env .
 
# Expose serving port
EXPOSE 8000
 
# Run production server
CMD ["gunicorn", "src.main:app", "--workers", "4", "--worker-class", "uvicorn.workers.UvicornWorker", "--bind", "0.0.0.0:8000"]

Build and Run the Container#

# Build image
docker build -t sentiment-api:v1 .
 
# Run container
docker run -p 8000:8000 --name sentiment-api sentiment-api:v1

This image is fully reproducible and can be deployed to any cloud provider (AWS ECS, GCP Cloud Run, Kubernetes) without modification.

Step 4: Implement Model Monitoring for Drift and Performance Tracking#

Even the best-performing models degrade over time due to shifts in real-world data. Proactive monitoring tracks 4 key types of drift:

Data Drift: Changes in input feature distributions (e.g. new slang terms in text inputs)
Concept Drift: Changes in input-output relationships (e.g. what counts as "toxic" content evolves)
Prediction Drift: Changes in model output distributions
Training-Serving Skew: Mismatch between training data and initial production data

We use Evidently AI (open-source, 20M+ downloads) for drift detection, paired with Prometheus and Grafana for visualization.

Implement Drift Detection#

Create a scheduled job (e.g. using Airflow or cron) to run drift checks daily against your reference training data:

# src/utils/drift_check.py
import pandas as pd
from evidently.report import Report
from evidently.metrics import DataDriftTable, PredictionDriftMetric
from src.core.config import settings
import logging
 
def run_drift_check():
    # Load reference training data (store this in your configs folder)
    reference = pd.read_csv("configs/reference_sentiment_data.csv")
    # Load last 24h of production predictions
    current = pd.read_json("/app/logs/predictions.jsonl", lines=True)
 
    # Generate drift report
    report = Report(metrics=[DataDriftTable(), PredictionDriftMetric()])
    report.run(reference_data=reference, current_data=current)
 
    # Save report for debugging
    report.save_html("/app/reports/latest_drift_report.html")
 
    # Extract drift results
    report_dict = report.as_dict()
    dataset_drift = report_dict["metrics"][0]["result"]["dataset_drift"]
    if dataset_drift:
        drifted_columns = report_dict["metrics"][0]["result"]["drift_by_columns"]
        drifted_features = [col for col, val in drifted_columns.items() if val.get("drifted", False)]
        logging.critical(
            f"DRIFT ALERT: Dataset drift detected. Drifted features: {drifted_features}"
        )

Integrate with Prometheus + Grafana#

Expose custom ML metrics (prediction latency, request count, drift score) via the Prometheus FastAPI exporter to build real-time dashboards and set automated alerting rules for performance degradation:

# src/api/metrics.py
from prometheus_fastapi_instrumentator import Instrumentator
 
def setup_metrics(app):
    instrumentator = Instrumentator()
    instrumentator.instrument(app).expose(app, endpoint="/metrics")

Add prometheus-fastapi-instrumentator>=6.0.0 to your requirements and call setup_metrics(app) in your main app file. Then point Grafana at the /metrics endpoint to build dashboards tracking latency percentiles, error rates, and throughput.

Production Best Practices for ML Serving#

Security: Add JWT authentication for protected endpoints, rate limiting with SlowAPI, and security headers middleware to prevent attacks.
Model Versioning: Use MLflow to track model versions and enable seamless rollbacks if new models perform poorly.
Logging: Use structured JSON logging for all requests and predictions, and send logs to a central platform (Datadog, ELK Stack) for auditing.
Testing: Add unit tests for model predictions, integration tests for API endpoints, and load tests with Locust to ensure your service handles expected traffic.
Orchestration: For high-scale workloads, deploy the Docker container to Kubernetes with horizontal pod autoscaling based on request load.

Common Pitfalls to Avoid#

Loading the model per request: This adds massive latency and will crash your service under load. Always load the model once at startup.
Using the development Uvicorn server in production: The single-process Uvicorn server is not designed for production traffic. Use Gunicorn with Uvicorn workers.
Ignoring model monitoring: 60% of models degrade within 6 months of deployment. You will not catch drift unless you explicitly monitor for it.
Hardcoding configuration values: Use environment variables and Pydantic BaseSettings to avoid rebuilding images for minor config changes.
Oversized Docker images: Use multi-stage builds and tensorflow-cpu for CPU deployments to reduce image size by 70%+ and speed up deployment times.

Real-World Use Cases#

This pipeline is used across industries for production ML workloads:

Customer Support: Sentiment analysis API to prioritize negative customer feedback tickets
E-commerce: Image classification service to auto-tag product listings with 95% accuracy
Fintech: Real-time fraud detection service with drift alerts for evolving fraud patterns
Retail: Demand forecasting API that triggers retraining when customer purchasing patterns shift
Social Media: Content moderation API with model versioning to adapt to new types of toxic content

Conclusion#

Shipping a TensorFlow model to production does not need to be complicated. By following this pipeline:

Save your model in TensorFlow's SavedModel format for optimal serving
Build a validated, high-performance API with FastAPI
Containerize with Docker for consistent deployments across environments
Implement proactive monitoring to catch drift before it impacts users You can go from trained model to production-ready service in a few hours, with the scalability and reliability to support enterprise workloads.