Building a Production-Grade End-to-End MLOps Pipeline from Scratch

Most machine learning tutorials end at model.fit(). But in the real world, training a model is barely 10% of the work. The remaining 90% is everything that makes ML actually work in production — data versioning, validation, experiment tracking, API serving, automated testing, containerization, CI/CD, monitoring, and drift detection.

In this walkthrough, I'll take you through building a complete, production-grade MLOps pipeline using the classic Titanic survival prediction problem. The focus isn't on building a complex model — it's on implementing every production practice that separates a Jupyter notebook experiment from a deployable ML system.

GitHub Repository: End-to-end-MLOPS-Pipeline

Here's the complete system we're building:

Data (Titanic CSV)
 │
 ▼
DVC (Data Versioning)
 │
 ▼
Data Validation (Schema & Constraint Checks)
 │
 ▼
Scikit-learn Training Pipeline
 │
 ▼
MLflow (Experiment Tracking & Model Registry)
 │
 ▼
Pytest (Automated Testing)
 │
 ▼
Docker (Containerization)
 │
 ▼
GitHub Actions (CI/CD)
 │
 ▼
FastAPI (Inference Service)
 │
 ├── Prometheus (Monitoring)
 │
 └── Evidently AI (Drift Detection)

Component	Tool
Version Control	Git, GitHub
Data Versioning	DVC
Model Training	Scikit-learn
Experiment Tracking	MLflow
API Framework	FastAPI
Testing	Pytest
Containerization	Docker
CI/CD	GitHub Actions
Monitoring	Prometheus
Drift Detection	Evidently AI

End-to-end-MLOPS-Pipeline/
│
├── data/
│   └── raw/
│       └── titanic.csv
│
├── models/
│   └── model.joblib
│
├── src/
│   ├── data_ingestion.py
│   ├── data_validation.py
│   ├── data_preprocess.py
│   ├── train.py
│   └── drift_detection.py
│
├── api/
│   ├── __init__.py
│   └── main.py
│
├── tests/
│   ├── __init__.py
│   ├── test_api.py
│   └── test_model.py
│
├── monitoring/
│   ├── prometheus.yml
│   └── drift_report.html
│
├── .github/
│   └── workflows/
│       └── ci_cd.yml
│
├── Dockerfile
├── docker-compose.yml
├── requirements.txt
├── .gitignore
├── .dockerignore
└── .dvc/

Start by initializing Git and creating a virtual environment:

git init
python -m venv venv

.\venv\Scripts\Activate.ps1
source venv/bin/activate

.gitignore Create a .gitignore that keeps our repository clean:

venv/
env/
.env

__pycache__/
*.pyc

data/raw/*
!data/raw/*.dvc
!data/raw/*.csv
data/processed/

models/
mlruns/

requirements.txt

pandas>=2.0.0
numpy>=1.24.0
scikit-learn>=1.3.0
joblib>=1.3.0

dvc>=3.0.0
mlflow>=2.10.0

fastapi>=0.100.0
uvicorn[standard]>=0.22.0
pydantic>=2.0.0

pytest>=7.4.0
httpx>=0.24.0
prometheus-client>=0.17.0
evidently

Install everything:

pip install -r requirements.txt

Git was designed for source code — small text files. Datasets can be hundreds of megabytes or gigabytes. Committing them directly to Git makes your repository bloated, slow to clone, and impossible to manage at scale.

DVC (Data Version Control) solves this by:

.dvc metadata files in Git (a few bytes).git checkout. Download the Titanic dataset and place it in data/raw/titanic.csv.

dvc init

dvc add data/raw/titanic.csv

git add .
git commit -m "Initialize project and track dataset with DVC"

DVC creates a data/raw/titanic.csv.dvc file — a tiny metadata pointer that Git tracks instead of the actual CSV.

src/data_ingestion.py) The data ingestion module is responsible for raw data from disk with defensive checks — ensuring the file exists and isn't empty before passing it downstream.

import os
import logging
import pandas as pd

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)

def load_data(file_path: str) -> pd.DataFrame:
    """
    Load raw CSV data and validate it.
    """
    if not os.path.exists(file_path):
        logging.error(f"File not found at path: {file_path}")
        raise FileNotFoundError(f"File does not exist: {file_path}")

    logging.info(f" data from {file_path}")
    df = pd.read_csv(file_path)

    if df.empty:
        logging.error("Loaded dataframe is empty")
        raise ValueError("Dataframe is empty")

    logging.info(f"Successfully loaded dataset with shape: {df.shape}")
    return df

if __name__ == "__main__":
    raw_data_path = os.path.join("data", "raw", "titanic.csv")
    df = load_data(raw_data_path)
    print(df.head())

logging instead of print()? In production MLOps, print() statements disappear into the void. logging gives you:

INFO, WARNING, ERROR) to filter noise. src/data_validation.py) Before touching any ML model, we validate that the incoming data matches our expectations. This is the first line of defense against silent model failures caused by upstream data changes.

import os
import sys
import logging
import pandas as pd

sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from src.data_ingestion import load_data

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")

REQUIRED_COLUMNS = [
    "Survived", "Pclass", "Name", "Sex", "Age", 
    "SibSp", "Parch", "Ticket", "Fare", "Embarked"
]
VALID_PCLASS = {1, 2, 3}
VALID_SEX = {"male", "female"}

def validate_data(df: pd.DataFrame) -> bool:
    """
    Validates dataset integrity, schema, and column value constraints.
    """
    logging.info("Starting data validation checks...")

    missing_cols = [col for col in REQUIRED_COLUMNS if col not in df.columns]
    if missing_cols:
        logging.error(f"Validation failed: Missing columns: {missing_cols}")
        raise ValueError(f"Missing required columns: {missing_cols}")

    invalid_pclass = set(df["Pclass"].dropna().unique()) - VALID_PCLASS
    if invalid_pclass:
        logging.error(f"Validation failed: Unexpected Pclass values: {invalid_pclass}")
        raise ValueError(f"Invalid values in Pclass: {invalid_pclass}")

    invalid_sex = set(df["Sex"].dropna().unique()) - VALID_SEX
    if invalid_sex:
        logging.error(f"Validation failed: Unexpected Sex values: {invalid_sex}")
        raise ValueError(f"Invalid values in Sex: {invalid_sex}")

    if df["Survived"].isnull().any():
        logging.error("Validation failed: Target column 'Survived' contains null values.")
        raise ValueError("Target column 'Survived' cannot contain null values.")

    if (df["Fare"] < 0).any():
        logging.error("Validation failed: Found negative values in 'Fare'.")
        raise ValueError("Fare values cannot be negative.")

    logging.info("All data validation checks passed successfully!")
    return True

if __name__ == "__main__":
    raw_data_path = os.path.join("data", "raw", "titanic.csv")
    df = load_data(raw_data_path)
    validate_data(df)

Check	Why It Matters
Required columns exist	If someone upstream renames Sex toGender , your pipeline failshere — not deep inside model training with a cryptic error.
Pclass ∈ {1, 2, 3}	Unknown categories would crash one-hot encoding or produce garbage features.
Survived has no nulls	Your target variable for supervised learning must be complete.
Fare >= 0	Negative fares are logically impossible and indicate data corruption.

Run it:

python src/data_validation.py

src/data_preprocess.py)

import os
import sys
import logging
import pandas as pd
from sklearn.model_selection import train_test_split

sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from src.data_ingestion import load_data
from src.data_validation import validate_data

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")

def preprocess_data(df: pd.DataFrame) -> pd.DataFrame:
    """
    Cleans raw titanic data using simple pandas operations:
    1. Fills missing values (Age with median, Embarked with mode).
    2. Drops columns that aren't useful (PassengerId, Name, Ticket, Cabin).
    3. Converts Sex and Embarked to numbers using one-hot encoding (get_dummies).
    """
    df = df.copy()

    df["Age"] = df["Age"].fillna(df["Age"].median())
    df["Fare"] = df["Fare"].fillna(df["Fare"].median())
    df["Embarked"] = df["Embarked"].fillna(df["Embarked"].mode()[0])

    drop_cols = ["PassengerId", "Name", "Ticket", "Cabin"]
    df = df.drop(columns=[col for col in drop_cols if col in df.columns])

    df = pd.get_dummies(df, columns=["Sex", "Embarked"], drop_first=True, dtype=int)

    return df

def split_data(df: pd.DataFrame, test_size: float = 0.2, random_state: int = 42):
    """
    Splits the cleaned dataframe into train and test sets.
    """
    X = df.drop(columns=["Survived"])
    y = df["Survived"]

    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=test_size, random_state=random_state, stratify=y
    )

    return X_train, X_test, y_train, y_test

if __name__ == "__main__":
    raw_data_path = os.path.join("data", "raw", "titanic.csv")
    df = load_data(raw_data_path)
    validate_data(df)

    cleaned_df = preprocess_data(df)
    logging.info(f"Cleaned data shape: {cleaned_df.shape}")
    print("\nCleaned Data Preview:")
    print(cleaned_df.head())

    X_train, X_test, y_train, y_test = split_data(cleaned_df)
    logging.info(f"Train samples: {len(X_train)}, Test samples: {len(X_test)}")

fillna with median/modePassengerId is a unique identifier (no predictive value). Name and Ticket are high-cardinality text. Cabin is >70% null.pd.get_dummies(drop_first=True)`` Sex (male/female) → Sex_male (1/0) and Embarked (C/Q/S) → Embarked_Q, Embarked_S. The drop_first avoids the stratify=y

python src/data_preprocess.py

src/train.py) This is where most tutorials stop. But we're not just training a model — we're logging every experiment so we can compare runs, reproduce results, and promote the best model to production.

import os
import sys
import logging
import joblib
import mlflow
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import (
    accuracy_score,
    precision_score,
    recall_score,
    f1_score
)

sys.path.append(
    os.path.abspath(
        os.path.join(os.path.dirname(__file__), "..")
    )
)
from src.data_ingestion import load_data
from src.data_preprocess import preprocess_data, split_data

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)

def train_model(
    n_estimators: int = 100,
    max_depth: int = 5,
    random_state: int = 42
):
    """
    Train a Random Forest model and log results to MLflow.
    """
    raw_data_path = os.path.join("data", "raw", "titanic.csv")
    df = load_data(raw_data_path)

    cleaned_df = preprocess_data(df)

    X_train, X_test, y_train, y_test = split_data(
        cleaned_df,
        random_state=random_state
    )

    mlflow.set_tracking_uri("sqlite:///mlflow.db")
    mlflow.set_experiment("Titanic-Survival-Prediction")

    with mlflow.start_run():
        logging.info("Training Random Forest Classifier...")

        model = RandomForestClassifier(
            n_estimators=n_estimators,
            max_depth=max_depth,
            random_state=random_state
        )
        model.fit(X_train, y_train)

        y_pred = model.predict(X_test)

        acc = accuracy_score(y_test, y_pred)
        prec = precision_score(y_test, y_pred)
        rec = recall_score(y_test, y_pred)
        f1 = f1_score(y_test, y_pred)

        logging.info(
            f"Accuracy={acc:.4f}, "
            f"Precision={prec:.4f}, "
            f"Recall={rec:.4f}, "
            f"F1={f1:.4f}"
        )

        mlflow.log_param("n_estimators", n_estimators)
        mlflow.log_param("max_depth", max_depth)
        mlflow.log_param("random_state", random_state)

        mlflow.log_metric("accuracy", acc)
        mlflow.log_metric("precision", prec)
        mlflow.log_metric("recall", rec)
        mlflow.log_metric("f1_score", f1)

        os.makedirs("models", exist_ok=True)
        model_path = os.path.join("models", "model.joblib")
        joblib.dump(model, model_path)
        logging.info(f"Model saved successfully at: {model_path}")

        mlflow.log_artifact(model_path, artifact_path="model")
        logging.info("Experiment successfully logged to MLflow.")

    return model, acc

if __name__ == "__main__":
    train_model(n_estimators=100, max_depth=5)

Every time train_model() runs, MLflow creates a unique Run ID and stores:

What	Why
Parameters (n_estimators ,max_depth )	So you can compare what configuration produced each result.
Metrics (accuracy ,f1_score )	Track model performance across experiments.
Artifacts (model.joblib )	The actual trained binary, versioned and downloadable.

python src/train.py

python -m mlflow ui

You can now visually compare multiple training runs, their hyperparameters, and metrics side-by-side in the MLflow UI.

api/main.py) End-users don't run Python scripts. They send HTTP requests. Our FastAPI service:

import os
import time
import joblib
import pandas as pd
from fastapi import FastAPI, HTTPException, Response
from pydantic import BaseModel, Field
from prometheus_client import Counter, Histogram, generate_latest, CONTENT_TYPE_LATEST

app = FastAPI(
    title="Titanic Survival Prediction API",
    description="Production-ready inference service with Prometheus monitoring.",
    version="1.0.0"
)

REQUEST_COUNT = Counter(
    "api_request_count_total", 
    "Total HTTP requests received", 
    ["method", "endpoint", "status"]
)
REQUEST_LATENCY = Histogram(
    "api_request_latency_seconds", 
    "Histogram of request latencies in seconds", 
    ["endpoint"]
)
PREDICTION_COUNT = Counter(
    "model_prediction_count_total", 
    "Total predictions generated by the model", 
    ["prediction"]
)

MODEL_PATH = os.path.join("models", "model.joblib")
if not os.path.exists(MODEL_PATH):
    raise FileNotFoundError(f"Trained model not found at {MODEL_PATH}!")

model = joblib.load(MODEL_PATH)

class PassengerInput(BaseModel):
    Pclass: int = Field(..., ge=1, le=3, description="Ticket class (1 = 1st, 2 = 2nd, 3 = 3rd)")
    Sex: str = Field(..., description="Gender: 'male' or 'female'")
    Age: float = Field(..., ge=0, le=120, description="Age in years")
    SibSp: int = Field(..., ge=0, description="Number of siblings/spouses aboard")
    Parch: int = Field(..., ge=0, description="Number of parents/children aboard")
    Fare: float = Field(..., ge=0.0, description="Passenger fare")
    Embarked: str = Field(..., description="Port of Embarkation: 'C', 'Q', or 'S'")

class PredictionResponse(BaseModel):
    survived: bool
    survival_probability: float

@app.get("/metrics")
def get_metrics():
    """Exposes Prometheus application metrics."""
    return Response(content=generate_latest(), media_type=CONTENT_TYPE_LATEST)

@app.get("/")
def health_check():
    return {"status": "healthy", "service": "titanic-survival-prediction"}

@app.post("/predict", response_model=PredictionResponse)
def predict(passenger: PassengerInput):
    start_time = time.time()
    try:
        input_data = pd.DataFrame([{
            "Pclass": passenger.Pclass,
            "Age": passenger.Age,
            "SibSp": passenger.SibSp,
            "Parch": passenger.Parch,
            "Fare": passenger.Fare,
            "Sex_male": 1 if passenger.Sex.lower() == "male" else 0,
            "Embarked_Q": 1 if passenger.Embarked.upper() == "Q" else 0,
            "Embarked_S": 1 if passenger.Embarked.upper() == "S" else 0,
        }])

        prediction = model.predict(input_data)[0]
        probability = model.predict_proba(input_data)[0][1]

        PREDICTION_COUNT.labels(prediction=str(int(prediction))).inc()
        REQUEST_COUNT.labels(method="POST", endpoint="/predict", status="200").inc()
        REQUEST_LATENCY.labels(endpoint="/predict").observe(time.time() - start_time)

        return PredictionResponse(
            survived=bool(prediction == 1),
            survival_probability=round(float(probability), 4)
        )

    except Exception as e:
        REQUEST_COUNT.labels(method="POST", endpoint="/predict", status="500").inc()
        raise HTTPException(status_code=500, detail=str(e))

Notice the manual construction of Sex_male, Embarked_Q, and Embarked_S. These must exactly match the dummy columns created during training by pd.get_dummies(..., drop_first=True). If you forget Embarked_Q or add Sex_female instead of Sex_male, the model will silently produce garbage predictions — no errors, just wrong numbers.

Request:

{
  "Pclass": 1,
  "Sex": "female",
  "Age": 29.0,
  "SibSp": 0,
  "Parch": 0,
  "Fare": 100.0,
  "Embarked": "S"
}

Response:

{
  "survived": true,
  "survival_probability": 0.9200
}
uvicorn api.main:app --reload --port 8000

Tests are CI/CD guardrails. If any test fails, deployment stops automatically.

tests/test_model.py — Model Integrity Tests

import os
import joblib
import pandas as pd

def test_model_file_exists():
    """Verify the trained model artifact exists on disk."""
    model_path = os.path.join("models", "model.joblib")
    assert os.path.exists(model_path), "Model artifact 'model.joblib' is missing!"

def test_model_prediction_output():
    """Verify model can take a sample input row and output 0 or 1."""
    model_path = os.path.join("models", "model.joblib")
    model = joblib.load(model_path)

    sample = pd.DataFrame([{
        "Pclass": 3,
        "Age": 22.0,
        "SibSp": 1,
        "Parch": 0,
        "Fare": 7.25,
        "Sex_male": 1,
        "Embarked_Q": 0,
        "Embarked_S": 1
    }])

    pred = model.predict(sample)
    assert pred[0] in [0, 1], f"Unexpected prediction value: {pred[0]}"

tests/test_api.py — API Contract Tests

from fastapi.testclient import TestClient
from api.main import app

client = TestClient(app)

def test_health_check():
    """Test GET / returns healthy status."""
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"status": "healthy", "service": "titanic-survival-prediction"}

def test_predict_endpoint_valid_input():
    """Test POST /predict with valid passenger JSON."""
    payload = {
        "Pclass": 1,
        "Sex": "female",
        "Age": 29.0,
        "SibSp": 0,
        "Parch": 0,
        "Fare": 100.0,
        "Embarked": "S"
    }
    response = client.post("/predict", json=payload)
    assert response.status_code == 200

    data = response.json()
    assert "survived" in data
    assert "survival_probability" in data
    assert isinstance(data["survived"], bool)
    assert 0.0 <= data["survival_probability"] <= 1.0

def test_predict_endpoint_invalid_input():
    """Test POST /predict fails when invalid data is passed."""
    invalid_payload = {
        "Pclass": 1,
        "Sex": "female",
        "Age": -5.0,  # Negative age should be rejected by Pydantic ge=0
        "SibSp": 0,
        "Parch": 0,
        "Fare": 50.0,
        "Embarked": "S"
    }
    response = client.post("/predict", json=invalid_payload)
    assert response.status_code == 422  # Unprocessable Entity

Run:

pytest tests/ -v

Your model works on your machine with Python 3.13 on Windows. But it might fail on an Ubuntu cloud server because of missing C++ build tools, conflicting package versions, or different file paths. Docker packages your Python version, dependencies, model artifact, and FastAPI server into an isolated, reproducible Linux container.

Dockerfile

FROM python:3.11-slim

ENV PYTHONDONTWRITEBYTECODE=1 \
    PYTHONUNBUFFERED=1

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY api/ ./api/
COPY src/ ./src/
COPY models/ ./models/

EXPOSE 8000

CMD ["uvicorn", "api.main:app", "--host", "0.0.0.0", "--port", "8000"]

Notice that COPY requirements.txt and RUN pip install come before COPY api/ and COPY src/. Docker caches layers. If you only change your source code (not dependencies), Docker skips the entire pip install step and rebuilds in ~2 seconds instead of 2 minutes.

.dockerignore

venv/
.env
__pycache__/
*.pyc
.git/
.dvc/
data/raw/
mlruns/
mlflow.db
tests/
notebooks/

docker-compose.yml

services:
  titanic-api:
    image: titanic-mlops
    container_name: titanic-mlops-api
    ports:
      - "8000:8000"
    restart: unless-stopped

  prometheus:
    image: prom/prometheus:latest
    container_name: titanic-prometheus
    ports:
      - "9090:9090"
    volumes:
      - ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml
    restart: unless-stopped
docker build -t titanic-mlops .

docker compose up -d

Now visit:

Every push to main triggers an automated pipeline that validates data, trains the model, runs tests, and builds the Docker image — all on GitHub's free cloud runners.

.github/workflows/ci_cd.yml

name: Titanic MLOps Pipeline

on:
  push:
    branches: [ "main", "master" ]
  pull_request:
    branches: [ "main", "master" ]

jobs:
  mlops_pipeline:
    runs-on: ubuntu-latest

    steps:
    - name: Checkout Code
      uses: actions/checkout@v4

    - name: Set up Python 3.11
      uses: actions/setup-python@v5
      with:
        python-version: "3.11"
        cache: 'pip'

    - name: Install Dependencies
      run: |
        python -m pip install --upgrade pip
        pip install -r requirements.txt

    - name: Run Data Validation
      run: |
        python src/data_validation.py

    - name: Train Model
      run: |
        python src/train.py

    - name: Run Pytest Suite
      env:
        PYTHONPATH: .
      run: |
        pytest tests/ -v

    - name: Build Docker Image
      run: |
        docker build -t titanic-mlops-api:latest .

PYTHONPATH: . Fix On Linux CI runners, the project root isn't automatically in Python's import search path. Without PYTHONPATH: ., pytest can't find the api module and fails with ModuleNotFoundError: No module named 'api'.

Push and watch it run:

git add .
git commit -m "Add CI/CD pipeline"
git push origin main

Check the Actions tab on your GitHub repository to see the pipeline execute live.

In our api/main.py, we defined three metric types:

Metric	Type	What It Measures
api_request_count_total	Counter	Total HTTP requests by endpoint and status code
api_request_latency_seconds	Histogram	Inference response time in seconds
model_prediction_count_total	Counter	How many survived (1) vs died (0) predictions

monitoring/prometheus.yml

global:
  scrape_interval: 5s

scrape_configs:
  - job_name: "titanic-api"
    metrics_path: "/metrics"
    static_configs:
      - targets: ["titanic-api:8000"]

Prometheus pulls from the FastAPI /metrics endpoint every 5 seconds and stores time-series data.

Imagine your model is deployed and serving 10,000 requests per day. Prometheus helps you detect:

survived=0 when it's normally 38% → something is wrong with the incoming data. After running some predictions through the API, visit http://localhost:9090 and query:

api_request_count_total
model_prediction_count_total
rate(api_request_latency_seconds_sum[5m]) / rate(api_request_latency_seconds_count[5m])

A model trained on 1912 Titanic passengers assumes certain demographic distributions — median age ~28, mostly 3rd class, mostly male. If the model is deployed live and incoming traffic suddenly changes (e.g., all passengers are age 60+ with fares >$300), the model won't crash — it will silently produce unreliable, inaccurate predictions.

This is called Data Drift (or Covariate Shift), and it's the #1 silent killer of production ML systems.

src/drift_detection.py

import os
import sys
import logging
import pandas as pd

try:
    from evidently.report import Report
    from evidently.metric_preset import DataDriftPreset, DataQualityPreset
except ModuleNotFoundError:
    from evidently.legacy.report import Report
    from evidently.legacy.metric_preset import DataDriftPreset, DataQualityPreset

sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from src.data_ingestion import load_data
from src.data_preprocess import preprocess_data

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")

def generate_drift_report(
    reference_data: pd.DataFrame, 
    current_data: pd.DataFrame, 
    output_html_path: str = "monitoring/drift_report.html"
):
    """
    Compares reference (training) data with current (production) data
    and generates an interactive HTML drift dashboard.
    """
    logging.info("Building Evidently Data Drift and Data Quality Report...")

    features = [col for col in reference_data.columns if col != "Survived"]
    ref_df = reference_data[features]
    curr_df = current_data[features]

    report = Report(metrics=[
        DataDriftPreset(),
        DataQualityPreset()
    ])

    report.run(reference_data=ref_df, current_data=curr_df)

    os.makedirs(os.path.dirname(output_html_path), exist_ok=True)
    report.save_html(output_html_path)

    logging.info(f"Drift report successfully generated at: {output_html_path}")
    return output_html_path

if __name__ == "__main__":
    raw_path = os.path.join("data", "raw", "titanic.csv")
    df = load_data(raw_path)
    reference_df = preprocess_data(df)

    current_df = reference_df.copy()
    current_df["Age"] = current_df["Age"] + 25.0  # Simulated drift in Age
    current_df["Fare"] = current_df["Fare"] * 3.5  # Simulated drift in Fare

    report_file = generate_drift_report(reference_df, current_df)
    print(f"\nReport ready! Open in browser: {os.path.abspath(report_file)}")

Evidently runs statistical tests on every feature:

Feature Type	Statistical Test	What It Measures
Numerical ( Age ,Fare )	Kolmogorov-Smirnov test	Whether two distributions are from the same population
Categorical ( Pclass ,Sex_male )	Chi-Square test	Whether category proportions have changed

If the p-value falls below the threshold (default 0.05), the feature is flagged as drifted.

python src/drift_detection.py

Open monitoring/drift_report.html in your browser to see the interactive Evidently dashboard with per-feature distribution charts, p-values, and drift flags.

Let's step back and appreciate what we've built:

┌──────────────────────────────────────────────────────┐
│                  Developer Pushes Code               │
└────────────────────────┬─────────────────────────────┘
                         │
                         ▼
┌──────────────────────────────────────────────────────┐
│              GitHub Actions CI/CD Pipeline            │
│                                                      │
│  ✅ Install Dependencies                             │
│  ✅ Validate Data Schema                             │
│  ✅ Train Model + Log to MLflow                      │
│  ✅ Run Pytest Suite (Model + API tests)             │
│  ✅ Build Docker Image                               │
└────────────────────────┬─────────────────────────────┘
                         │
                         ▼
┌──────────────────────────────────────────────────────┐
│            Docker Container (Production)             │
│                                                      │
│  FastAPI (:8000)                                     │
│  ├── POST /predict  → Model Inference                │
│  ├── GET  /metrics  → Prometheus Scrape Endpoint     │
│  └── GET  /docs     → Interactive Swagger UI         │
└────────────────────────┬─────────────────────────────┘
                         │
              ┌──────────┴──────────┐
              ▼                     ▼
┌────────────────────┐  ┌────────────────────┐
│  Prometheus (:9090) │  │  Evidently AI       │
│                    │  │                    │
│  Request Counts    │  │  Data Drift        │
│  Latency Tracking  │  │  Feature Drift     │
│  Prediction Dist.  │  │  Quality Metrics   │
└────────────────────┘  └────────────────────┘

Data Versioning (DVC): Never commit raw data to Git. Track it with DVC for reproducibility and rollback.

Validate Before You Train: Schema checks, constraint checks, and missing-value guardrails catch data problems before they silently corrupt your model.

Track Every Experiment (MLflow): Every training run should log parameters, metrics, and artifacts. "I think I got 82% accuracy last Tuesday" is not acceptable in production.

Serve via API, Not Scripts: End-users send HTTP requests. FastAPI with Pydantic validation gives you type-safe, self-documenting endpoints.

Test Everything Automatically: Model integrity tests + API contract tests in Pytest, executed automatically in CI.

Containerize for Reproducibility: Docker eliminates "works on my machine" problems entirely.

Automate with CI/CD: GitHub Actions ensures every push is validated, tested, trained, and container-built before reaching production.

Monitor in Production: Prometheus gives you real-time visibility into traffic, latency, and prediction behavior. Without it, you're flying blind.

Detect Drift: Evidently AI catches the silent killer of ML systems — when production data distribution shifts away from what the model was trained on.

If you found this walkthrough helpful, consider giving the GitHub repo a ⭐ and sharing this post with someone learning MLOps!

source & further reading

dev.to — original article Routing OpenCode Tasks with Jev Airport: Manage Multiple AI Agent Terminals Happily ✈️ Why AI Agents Need an Execution Boundary