Neural Networks & Deep Learning

Chapter 21: MLOps — From Jupyter Notebook to Production

Bridging the Last Mile Between Model Training and Real-World Impact

⏱️ Reading Time: ~3 hours  |  📖 Part VI: MLOps & Deployment  |  🚀 Engineering + Code Chapter

📋 Prerequisites: Chapters 12–17 (CNN, Transfer Learning), Basic Python, Familiarity with REST APIs

21.1 The ML Lifecycle — Why Notebooks Are Not Products

The journey from data to value follows a structured lifecycle. Most university courses cover only the first three stages. Production ML requires mastering all six:

The "Hidden Technical Debt" Problem

A landmark Google paper (Sculley et al., 2015) revealed that in real-world ML systems, the actual model training code is just a tiny fraction of the total codebase. The vast majority is infrastructure: data pipelines, feature stores, serving infrastructure, monitoring, and configuration management.

21.2 Model Serialization — Saving Models for Production

Your trained model exists as weights in GPU memory. To deploy it, you must serialize it — convert it to a portable, self-contained format that any server can load without your original training code.

TensorFlow SavedModel

# Save a Keras model as TensorFlow SavedModel
import tensorflow as tf

# After training your model (e.g., plant disease classifier from Ch17)
model = tf.keras.models.load_model('plant_disease_model.h5')

# Save as SavedModel (directory format)
tf.saved_model.save(model, './saved_model/plant_disease/1')

# The '1' is the version number — critical for model versioning!
# Directory structure:
# saved_model/plant_disease/1/
#   ├── saved_model.pb        ← computation graph
#   ├── fingerprint.pb
#   └── variables/
#       ├── variables.data-00000-of-00001  ← weights
#       └── variables.index

# Load it back
loaded = tf.saved_model.load('./saved_model/plant_disease/1')
print(loaded.signatures)  # Shows input/output specsPython

ONNX Export from PyTorch

import torch
import torch.onnx

# Assume `model` is a trained PyTorch model
model.eval()

# Create dummy input matching your model's expected shape
dummy_input = torch.randn(1, 3, 224, 224)  # Batch=1, 3 channels, 224×224

# Export to ONNX
torch.onnx.export(
    model,
    dummy_input,
    "plant_disease.onnx",
    input_names=["image"],
    output_names=["prediction"],
    dynamic_axes={"image": {0: "batch_size"}},  # Allow variable batch
    opset_version=13
)

# Verify with ONNX Runtime
import onnxruntime as ort
session = ort.InferenceSession("plant_disease.onnx")
result = session.run(None, {"image": dummy_input.numpy()})
print(f"Prediction shape: {result[0].shape}")Python

TorchScript

# Method 1: Tracing (works for models without control flow)
traced_model = torch.jit.trace(model, dummy_input)
traced_model.save("plant_disease_traced.pt")

# Method 2: Scripting (works with if/else, loops)
scripted_model = torch.jit.script(model)
scripted_model.save("plant_disease_scripted.pt")

# Load without needing the model class definition!
loaded = torch.jit.load("plant_disease_traced.pt")
output = loaded(dummy_input)Python

21.3 REST API with FastAPI — The /predict Endpoint

FastAPI is the modern Python web framework of choice for ML serving. It's built on Starlette (async ASGI) and Pydantic (data validation), offering automatic OpenAPI docs, type validation, and async request handling out of the box.

Why FastAPI Over Flask for ML?

Complete FastAPI ML Server

# app.py — Production FastAPI ML serving application
from fastapi import FastAPI, File, UploadFile, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from typing import List, Optional
import tensorflow as tf
import numpy as np
from PIL import Image
import io
import time
import logging

# ─── Setup ────────────────────────────────────────
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI(
    title="Plant Disease Classifier API",
    description="Predicts plant disease from leaf images (Ch17 model)",
    version="1.0.0"
)

# CORS — allow Streamlit frontend to call this API
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

# ─── Load model at startup (not per-request!) ────
CLASS_NAMES = [
    "Healthy", "Early Blight", "Late Blight",
    "Bacterial Spot", "Yellow Leaf Curl"
]
model = None

@app.on_event("startup")
async def load_model():
    global model
    logger.info("Loading plant disease model...")
    model = tf.saved_model.load("./saved_model/plant_disease/1")
    logger.info("✅ Model loaded successfully!")

# ─── Pydantic Response Schemas ────────────────────
class PredictionResult(BaseModel):
    predicted_class: str = Field(..., example="Early Blight")
    confidence: float = Field(..., ge=0, le=1, example=0.94)
    all_probabilities: dict = Field(...)
    inference_time_ms: float = Field(..., example=45.2)

class HealthResponse(BaseModel):
    status: str = "healthy"
    model_loaded: bool = True
    version: str = "1.0.0"

# ─── Helper: preprocess image ─────────────────────
def preprocess_image(image_bytes: bytes) -> np.ndarray:
    """Resize to 224x224, normalize to [0,1], add batch dim."""
    img = Image.open(io.BytesIO(image_bytes)).convert("RGB")
    img = img.resize((224, 224))
    arr = np.array(img, dtype=np.float32) / 255.0
    return np.expand_dims(arr, axis=0)  # (1, 224, 224, 3)

# ─── Endpoints ────────────────────────────────────
@app.get("/health", response_model=HealthResponse)
async def health_check():
    return HealthResponse(model_loaded=model is not None)

@app.post("/predict", response_model=PredictionResult)
async def predict(file: UploadFile = File(...)):
    """Upload a leaf image and get disease prediction."""
    # Validate file type
    if file.content_type not in ["image/jpeg", "image/png"]:
        raise HTTPException(
            status_code=400,
            detail="Only JPEG/PNG images accepted"
        )

    # Read & preprocess
    image_bytes = await file.read()
    input_tensor = preprocess_image(image_bytes)

    # Inference with timing
    start = time.perf_counter()
    infer = model.signatures["serving_default"]
    predictions = infer(tf.constant(input_tensor))
    output_key = list(predictions.keys())[0]
    probs = tf.nn.softmax(predictions[output_key]).numpy()[0]
    elapsed_ms = (time.perf_counter() - start) * 1000

    # Build response
    predicted_idx = int(np.argmax(probs))
    return PredictionResult(
        predicted_class=CLASS_NAMES[predicted_idx],
        confidence=round(float(probs[predicted_idx]), 4),
        all_probabilities={
            name: round(float(p), 4)
            for name, p in zip(CLASS_NAMES, probs)
        },
        inference_time_ms=round(elapsed_ms, 2)
    )Python

Run & Test

# Terminal: Start the server
$ uvicorn app:app --host 0.0.0.0 --port 8000 --workers 4

# Another terminal: Test with curl
$ curl -X POST "http://localhost:8000/predict" \
    -F "file=@tomato_leaf.jpg"

# Or open browser → http://localhost:8000/docs
# FastAPI auto-generates interactive Swagger UI!Bash

21.4 Docker Containerization for ML

Docker solves the "it works on my machine" problem. Your container bundles the OS, Python version, libraries, model weights, and application code into a single portable image.

Dockerfile for ML Serving (Multi-Stage Build)

# Dockerfile — Optimized for ML serving
# Stage 1: Builder (install heavy dependencies)
FROM python:3.11-slim AS builder

WORKDIR /build
COPY requirements.txt .
RUN pip install --no-cache-dir --prefix=/install -r requirements.txt

# Stage 2: Runtime (minimal image)
FROM python:3.11-slim

# Metadata
LABEL maintainer="mlops-team@startup.in"
LABEL version="1.0"

# Security: non-root user
RUN groupadd -r mluser && useradd -r -g mluser mluser

WORKDIR /app

# Copy installed packages from builder
COPY --from=builder /install /usr/local

# Copy application code
COPY app.py .
COPY saved_model/ ./saved_model/

# Switch to non-root user
USER mluser

# Expose port
EXPOSE 8000

# Health check
HEALTHCHECK --interval=30s --timeout=5s --retries=3 \
    CMD curl -f http://localhost:8000/health || exit 1

# Start server
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]Dockerfile

requirements.txt

fastapi==0.104.1
uvicorn[standard]==0.24.0
tensorflow-cpu==2.15.0     # Use CPU version to keep image small
numpy==1.26.2
Pillow==10.1.0
python-multipart==0.0.6    # Required for file uploadsText

Build & Run

# Build the image
$ docker build -t plant-disease-api:v1 .

# Check image size
$ docker images plant-disease-api
# REPOSITORY          TAG    SIZE
# plant-disease-api   v1     1.2GB  (with TF-CPU)

# Run the container
$ docker run -d \
    --name plant-api \
    -p 8000:8000 \
    --memory=2g \
    --cpus=2 \
    plant-disease-api:v1

# Test it
$ curl -X POST http://localhost:8000/predict -F "file=@leaf.jpg"

# View logs
$ docker logs -f plant-apiBash

.dockerignore

# Don't copy training artifacts into production image
__pycache__/
*.pyc
.git/
.gitignore
data/
notebooks/
*.ipynb
checkpoints/
wandb/
mlruns/
*.h5
README.mdText

21.5 TensorFlow Serving — Industrial-Grade Model Serving

While FastAPI is great for lightweight deployment, TensorFlow Serving is purpose-built for serving TF models at scale. It handles model versioning, batching, and hardware acceleration natively.

Deploy with TensorFlow Serving (Docker)

# Directory structure expected by TF Serving:
# models/
#   └── plant_disease/
#       ├── 1/               ← Version 1
#       │   └── saved_model.pb + variables/
#       └── 2/               ← Version 2 (auto-detected!)
#           └── saved_model.pb + variables/

# Run TensorFlow Serving container
$ docker run -d \
    --name tf-serving \
    -p 8501:8501 \
    -p 8500:8500 \
    -v "$(pwd)/models:/models" \
    -e MODEL_NAME=plant_disease \
    tensorflow/serving:latest

# REST API prediction (port 8501)
$ curl -X POST http://localhost:8501/v1/models/plant_disease:predict \
    -H "Content-Type: application/json" \
    -d '{"instances": [[[0.1, 0.2, 0.3], ...]]}'

# Check model status
$ curl http://localhost:8501/v1/models/plant_diseaseBash

TF Serving Configuration for A/B Testing

# model_config.config — Serve multiple versions simultaneously
model_config_list {
  config {
    name: "plant_disease"
    base_path: "/models/plant_disease"
    model_platform: "tensorflow"
    model_version_policy {
      specific {
        versions: 1    # Champion model (90% traffic)
        versions: 2    # Challenger model (10% traffic)
      }
    }
  }
}

# Start TF Serving with config file
$ docker run -d \
    -p 8501:8501 \
    -v "$(pwd)/models:/models" \
    -v "$(pwd)/model_config.config:/config" \
    tensorflow/serving \
    --model_config_file=/configConfig

A/B Traffic Splitting with Python

import random
import requests

def predict_with_ab_test(image_data, champion_pct=0.9):
    """Route traffic: 90% to v1 (champion), 10% to v2 (challenger)."""
    base_url = "http://localhost:8501/v1/models/plant_disease"

    if random.random() < champion_pct:
        version = 1  # Champion
    else:
        version = 2  # Challenger

    url = f"{base_url}/versions/{version}:predict"
    response = requests.post(url, json={"instances": [image_data]})

    result = response.json()
    result["model_version"] = version
    return resultPython

21.6 Model Versioning & Experiment Tracking

In production, you don't have one model — you have dozens of experiments, each with different hyperparameters, data versions, and feature sets. Model versioning answers: "Which model is running? What data was it trained on? Who approved it?"

MLflow Experiment Tracking

import mlflow
import mlflow.tensorflow
import tensorflow as tf

# Set tracking URI (local or remote server)
mlflow.set_tracking_uri("http://mlflow-server.internal:5000")
mlflow.set_experiment("plant-disease-classifier")

# Start a training run
with mlflow.start_run(run_name="resnet50-augmented-v3"):

    # Log hyperparameters
    mlflow.log_params({
        "model_arch": "ResNet50",
        "learning_rate": 0.001,
        "batch_size": 32,
        "epochs": 50,
        "augmentation": "random_flip+rotate+zoom",
        "dataset_version": "plantvillage-v3-india-augmented",
        "optimizer": "Adam",
    })

    # Train the model
    model = build_resnet50_classifier()
    history = model.fit(train_ds, validation_data=val_ds, epochs=50)

    # Log metrics
    mlflow.log_metrics({
        "val_accuracy": history.history["val_accuracy"][-1],
        "val_loss": history.history["val_loss"][-1],
        "train_accuracy": history.history["accuracy"][-1],
        "model_size_mb": 98.5,
        "inference_latency_ms": 42.0,
    })

    # Log the model artifact
    mlflow.tensorflow.log_model(
        model,
        artifact_path="model",
        registered_model_name="PlantDiseaseClassifier"
    )

    # Log training curves as artifact
    import matplotlib.pyplot as plt
    fig, ax = plt.subplots()
    ax.plot(history.history["val_accuracy"])
    ax.set_title("Validation Accuracy")
    fig.savefig("training_curve.png")
    mlflow.log_artifact("training_curve.png")

print("Run logged! Check MLflow UI at http://localhost:5000")Python

Model Registry — Promoting to Production

from mlflow.tracking import MlflowClient

client = MlflowClient()

# Promote model version 3 to Production
client.transition_model_version_stage(
    name="PlantDiseaseClassifier",
    version=3,
    stage="Production"
)

# Archive the old version
client.transition_model_version_stage(
    name="PlantDiseaseClassifier",
    version=2,
    stage="Archived"
)

# Load the production model for serving
model_uri = "models:/PlantDiseaseClassifier/Production"
prod_model = mlflow.tensorflow.load_model(model_uri)
print("✅ Production model v3 loaded!")Python

21.7 Monitoring — Data Drift, Prediction Drift, Concept Drift

A model deployed without monitoring is a ticking time bomb. Real-world data changes over time — user behavior shifts, seasonal patterns emerge, economic conditions fluctuate. This phenomenon is called drift.

Drift Detection Implementation

import numpy as np
from scipy import stats
from datetime import datetime
import json
import logging

logger = logging.getLogger("drift_monitor")

class DriftMonitor:
    """Monitor data drift, prediction drift, and concept drift."""

    def __init__(self, reference_data: np.ndarray,
                 reference_predictions: np.ndarray,
                 feature_names: list,
                 ks_threshold: float = 0.1,
                 psi_threshold: float = 0.2):
        self.reference_data = reference_data
        self.reference_preds = reference_predictions
        self.feature_names = feature_names
        self.ks_threshold = ks_threshold
        self.psi_threshold = psi_threshold
        self.alerts = []

    def _calculate_psi(self, expected: np.ndarray,
                       actual: np.ndarray,
                       bins: int = 10) -> float:
        """Population Stability Index — measures distribution shift."""
        expected_pct, bin_edges = np.histogram(expected, bins=bins)
        actual_pct, _ = np.histogram(actual, bins=bin_edges)

        # Normalize and add small epsilon to avoid division by zero
        expected_pct = expected_pct / len(expected) + 1e-6
        actual_pct = actual_pct / len(actual) + 1e-6

        psi = np.sum(
            (actual_pct - expected_pct) * np.log(actual_pct / expected_pct)
        )
        return float(psi)

    def check_data_drift(self, production_data: np.ndarray) -> dict:
        """KS test per feature: training vs production distribution."""
        drift_report = {"timestamp": datetime.now().isoformat(),
                        "drifted_features": [], "details": {}}

        for i, feature in enumerate(self.feature_names):
            ks_stat, p_value = stats.ks_2samp(
                self.reference_data[:, i],
                production_data[:, i]
            )
            is_drifted = ks_stat > self.ks_threshold

            drift_report["details"][feature] = {
                "ks_statistic": round(ks_stat, 4),
                "p_value": round(p_value, 6),
                "is_drifted": is_drifted
            }

            if is_drifted:
                drift_report["drifted_features"].append(feature)
                logger.warning(f"🚨 DRIFT DETECTED: {feature} "
                              f"(KS={ks_stat:.4f}, p={p_value:.6f})")

        return drift_report

    def check_prediction_drift(self, prod_preds: np.ndarray) -> dict:
        """PSI on prediction distribution."""
        psi = self._calculate_psi(self.reference_preds, prod_preds)
        is_drifted = psi > self.psi_threshold

        if is_drifted:
            logger.warning(f"🚨 PREDICTION DRIFT: PSI={psi:.4f}")

        return {
            "psi": round(psi, 4),
            "is_drifted": is_drifted,
            "threshold": self.psi_threshold,
            "reference_mean": float(np.mean(self.reference_preds)),
            "production_mean": float(np.mean(prod_preds)),
        }

# ─── Usage Example ────────────────────────────────
# Suppose we have reference (training) and production data
np.random.seed(42)
ref_data = np.random.randn(1000, 5)             # Training data
prod_data = np.random.randn(500, 5) + 0.3      # Shifted production data
ref_preds = np.random.uniform(0, 1, 1000)       # Training predictions

monitor = DriftMonitor(
    reference_data=ref_data,
    reference_predictions=ref_preds,
    feature_names=["income", "age", "credit_score",
                   "loan_amount", "tenure"],
)

report = monitor.check_data_drift(prod_data)
print(json.dumps(report, indent=2))Python

21.8 Edge Deployment — TFLite for Android in India

India has 750+ million smartphone users, but large portions still rely on 2G/3G connectivity. Sending images to a cloud API requires stable bandwidth and adds latency. Edge deployment puts the model on the phone itself — predictions happen offline.

Convert to TFLite with Quantization

import tensorflow as tf
import numpy as np

# Load the trained Keras model
model = tf.keras.models.load_model("plant_disease_model.h5")

# ─── Method 1: Dynamic Range Quantization (simplest) ──
converter = tf.lite.TFLiteConverter.from_keras_model(model)
converter.optimizations = [tf.lite.Optimize.DEFAULT]
tflite_model = converter.convert()

with open("plant_disease_dynamic.tflite", "wb") as f:
    f.write(tflite_model)
print(f"Dynamic quant: {len(tflite_model) / 1e6:.1f} MB")

# ─── Method 2: Full Integer Quantization (smallest) ───
def representative_dataset():
    """Provide ~100 samples for calibration."""
    for _ in range(100):
        sample = np.random.rand(1, 224, 224, 3).astype(np.float32)
        yield [sample]

converter2 = tf.lite.TFLiteConverter.from_keras_model(model)
converter2.optimizations = [tf.lite.Optimize.DEFAULT]
converter2.representative_dataset = representative_dataset
converter2.target_spec.supported_ops = [
    tf.lite.OpsSet.TFLITE_BUILTINS_INT8
]
converter2.inference_input_type = tf.uint8
converter2.inference_output_type = tf.uint8

tflite_int8 = converter2.convert()
with open("plant_disease_int8.tflite", "wb") as f:
    f.write(tflite_int8)
print(f"Int8 quant: {len(tflite_int8) / 1e6:.1f} MB")Python

Run TFLite Inference (Python — mimics Android behavior)

# TFLite inference — same API available in Android Java/Kotlin
interpreter = tf.lite.Interpreter(model_path="plant_disease_int8.tflite")
interpreter.allocate_tensors()

input_details = interpreter.get_input_details()
output_details = interpreter.get_output_details()

# Prepare input
img = tf.io.read_file("test_leaf.jpg")
img = tf.image.decode_jpeg(img, channels=3)
img = tf.image.resize(img, [224, 224])
img = tf.cast(img, tf.uint8)  # Int8 model expects uint8 input
input_data = tf.expand_dims(img, 0)

# Run inference
interpreter.set_tensor(input_details[0]["index"], input_data.numpy())
interpreter.invoke()
output = interpreter.get_tensor(output_details[0]["index"])

CLASS_NAMES = ["Healthy", "Early Blight", "Late Blight",
               "Bacterial Spot", "Yellow Leaf Curl"]
predicted = CLASS_NAMES[np.argmax(output)]
print(f"Prediction: {predicted}")Python

Section A — Multiple-Choice Questions (10)

Section B — Short Answer Questions (5)

Section C — Long Answer Questions (3)

Section D — Programming Exercises (2)

Section E — Mini-Project