MLOps Fundamentals: Experiment Tracking, Model Registry, Serving, and Drift Monitoring

March 26, 2026 17 min read AI & Machine Learning

mlopsmlflowmodel-servingdriftbentomltritonmachine-learning

Contents

A model that performs well in a notebook is a science project. A model that performs well in production six months after deployment is engineering. The gap between the two is MLOps: the practices, tooling, and culture that make machine learning reproducible, deployable, and maintainable at scale.

This guide covers the core MLOps pillars — experiment tracking so you can reproduce any result, a model registry so you know what’s in production and why, model serving infrastructure that handles real traffic, and drift monitoring so you know when your model stops working before your users do.

Why MLOps Matters

The failure modes of unmanaged ML are specific and painful:

Can’t reproduce results: a model trained six months ago performed better, but nobody remembers the exact hyperparameters or data version
Deployment is manual and fragile: models get moved to production via “copy the pickle file” or “ask the data scientist to push it”
No visibility into production performance: the model degraded gradually over three months; nobody noticed until accuracy dropped 20%
Training/serving skew: the preprocessing pipeline in training differs subtly from the one in production — the model sees different data than it was trained on

MLOps addresses each of these systematically.

Experiment Tracking with MLflow

MLflow is the de facto open-source standard for experiment tracking. It logs parameters, metrics, artifacts, and model code for every training run, giving you a complete audit trail.

Running MLflow Tracking Server

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53


# Run the tracking server with PostgreSQL backend and S3 artifact store
pip install mlflow psycopg2-binary boto3

mlflow server \
  --backend-store-uri postgresql://mlflow:password@postgres:5432/mlflow \
  --artifacts-destination s3://mlflow-artifacts/experiments \
  --host 0.0.0.0 \
  --port 5000

# Or with Docker Compose
cat > docker-compose.yml <<'EOF'
version: '3.8'
services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_USER: mlflow
      POSTGRES_PASSWORD: password
      POSTGRES_DB: mlflow
    volumes:
      - pgdata:/var/lib/postgresql/data

  mlflow:
    image: ghcr.io/mlflow/mlflow:latest
    ports:
      - "5000:5000"
    environment:
      MLFLOW_S3_ENDPOINT_URL: http://minio:9000
      AWS_ACCESS_KEY_ID: minioadmin
      AWS_SECRET_ACCESS_KEY: minioadmin
    command: >
      mlflow server
        --backend-store-uri postgresql://mlflow:password@postgres:5432/mlflow
        --artifacts-destination s3://mlflow-artifacts
        --host 0.0.0.0
    depends_on:
      - postgres
      - minio

  minio:
    image: quay.io/minio/minio:latest
    ports:
      - "9000:9000"
    command: server /data --console-address :9001
    volumes:
      - minio_data:/data

volumes:
  pgdata:
  minio_data:
EOF

docker compose up -d

Tracking Experiments

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78


import mlflow
import mlflow.sklearn
import mlflow.pytorch
from mlflow.models import infer_signature
import numpy as np
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.metrics import accuracy_score, roc_auc_score, f1_score
from sklearn.model_selection import train_test_split

# Configure tracking server
mlflow.set_tracking_uri("http://localhost:5000")
mlflow.set_experiment("customer-churn-prediction")

def train_and_log(
    n_estimators: int = 100,
    max_depth: int = 3,
    learning_rate: float = 0.1,
    subsample: float = 0.8,
):
    with mlflow.start_run(run_name=f"gbt-n{n_estimators}-d{max_depth}") as run:
        # Log all parameters upfront
        mlflow.log_params({
            "n_estimators": n_estimators,
            "max_depth": max_depth,
            "learning_rate": learning_rate,
            "subsample": subsample,
            "model_type": "GradientBoostingClassifier",
            "data_version": "v2.3",   # Tag your data version!
            "feature_set": "base_v1",
        })

        # Train
        model = GradientBoostingClassifier(
            n_estimators=n_estimators,
            max_depth=max_depth,
            learning_rate=learning_rate,
            subsample=subsample,
        )
        model.fit(X_train, y_train)

        # Evaluate and log metrics
        for split_name, X, y in [("train", X_train, y_train), ("val", X_val, y_val)]:
            preds = model.predict(X)
            probs = model.predict_proba(X)[:, 1]
            mlflow.log_metrics({
                f"{split_name}_accuracy": accuracy_score(y, preds),
                f"{split_name}_auc": roc_auc_score(y, probs),
                f"{split_name}_f1": f1_score(y, preds),
            })

        # Log feature importances as a figure
        import matplotlib.pyplot as plt
        fig, ax = plt.subplots(figsize=(10, 6))
        feature_imp = sorted(zip(feature_names, model.feature_importances_), key=lambda x: -x[1])
        ax.barh([f[0] for f in feature_imp[:20]], [f[1] for f in feature_imp[:20]])
        ax.set_title("Top 20 Feature Importances")
        mlflow.log_figure(fig, "feature_importances.png")
        plt.close()

        # Log the model with its input/output signature
        signature = infer_signature(X_val, model.predict(X_val))
        input_example = X_val[:5]

        mlflow.sklearn.log_model(
            model,
            artifact_path="model",
            signature=signature,
            input_example=input_example,
            registered_model_name="customer-churn-gbt",   # Register in model registry
        )

        print(f"Run ID: {run.info.run_id}")
        return run.info.run_id

# Run a hyperparameter sweep
for lr in [0.05, 0.1, 0.2]:
    for depth in [3, 4, 5]:
        train_and_log(learning_rate=lr, max_depth=depth)

Logging During Training (Step Metrics)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29


import mlflow
import torch

with mlflow.start_run():
    mlflow.log_params({"epochs": 50, "lr": 3e-4, "batch_size": 32})

    for epoch in range(50):
        train_loss = train_one_epoch(model, train_loader, optimizer)
        val_loss, val_acc = evaluate(model, val_loader)

        # Log step metrics for learning curves
        mlflow.log_metrics({
            "train_loss": train_loss,
            "val_loss": val_loss,
            "val_accuracy": val_acc,
        }, step=epoch)

        # Log model checkpoint at best validation loss
        if val_loss < best_val_loss:
            best_val_loss = val_loss
            mlflow.pytorch.log_model(
                model,
                artifact_path="best_model",
                registered_model_name="my-transformer",
            )

    # Log final artifacts
    mlflow.log_artifact("confusion_matrix.png")
    mlflow.log_artifact("data/preprocessing_config.json")  # CRITICAL: log config!

Querying the Experiment Store

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22


import mlflow

client = mlflow.MlflowClient()

# Find the best run by validation AUC
experiment = client.get_experiment_by_name("customer-churn-prediction")
runs = client.search_runs(
    experiment_ids=[experiment.experiment_id],
    filter_string="metrics.val_auc > 0.85",
    order_by=["metrics.val_auc DESC"],
    max_results=10,
)

best_run = runs[0]
print(f"Best run: {best_run.info.run_id}")
print(f"Params: {best_run.data.params}")
print(f"Val AUC: {best_run.data.metrics['val_auc']:.4f}")

# Load a specific model from a run
model = mlflow.sklearn.load_model(
    f"runs:/{best_run.info.run_id}/model"
)

Model Registry: Lifecycle Management

The model registry is where models graduate from experiment results to managed artifacts with defined lifecycle stages.

Stages: None → Staging → Production → Archived

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29


client = mlflow.MlflowClient()

# Transition a model version to Staging for testing
client.transition_model_version_stage(
    name="customer-churn-gbt",
    version=12,
    stage="Staging",
    archive_existing_versions=False,  # Keep other staging versions
)

# After validation, promote to Production
client.transition_model_version_stage(
    name="customer-churn-gbt",
    version=12,
    stage="Production",
    archive_existing_versions=True,  # Auto-archive previous production version
)

# Add approval notes
client.update_model_version(
    name="customer-churn-gbt",
    version=12,
    description="Approved by Jane. Val AUC 0.912. Validated on holdout 2026-Q1 data.",
)

# Load the current production model by alias (not version number)
production_model = mlflow.sklearn.load_model(
    "models:/customer-churn-gbt/Production"
)

Model Cards and Tags

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18


# Tag models with metadata for governance
client.set_model_version_tag(
    name="customer-churn-gbt",
    version=12,
    key="data_version",
    value="v2.3-2026Q1",
)
client.set_model_version_tag(name="customer-churn-gbt", version=12,
    key="validated_by", value="jane.smith@company.com")
client.set_model_version_tag(name="customer-churn-gbt", version=12,
    key="fairness_review", value="passed")
client.set_model_version_tag(name="customer-churn-gbt", version=12,
    key="pii_in_features", value="false")

# Search for models with specific tags
versions = client.search_model_versions(
    "name='customer-churn-gbt' and tags.fairness_review='passed'"
)

Model Serving

Getting a model into a REST API that handles production traffic requires solving: containerization, batching, hardware acceleration, versioning, and graceful deployment.

BentoML: Pythonic Model Serving

BentoML is the most developer-friendly serving framework. You define your service in Python, and BentoML handles containerization, batching, and deployment.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58


# service.py
import numpy as np
import bentoml
from bentoml.io import NumpyNdarray, JSON
from pydantic import BaseModel

# Save a model to the BentoML store
import mlflow.sklearn
model = mlflow.sklearn.load_model("models:/customer-churn-gbt/Production")
bento_model = bentoml.sklearn.save_model(
    "customer-churn-gbt",
    model,
    metadata={"mlflow_run_id": "abc123", "val_auc": 0.912},
)

# Define the service
svc = bentoml.Service("churn-prediction", runners=[
    bentoml.sklearn.get("customer-churn-gbt:latest").to_runner()
])

class PredictionRequest(BaseModel):
    customer_id: str
    tenure_months: int
    monthly_charges: float
    total_charges: float
    contract_type: str
    payment_method: str
    num_support_tickets: int

class PredictionResponse(BaseModel):
    customer_id: str
    churn_probability: float
    will_churn: bool
    model_version: str

@svc.api(input=JSON(pydantic_model=PredictionRequest),
         output=JSON(pydantic_model=PredictionResponse))
async def predict(request: PredictionRequest) -> PredictionResponse:
    # Feature engineering matches training pipeline EXACTLY
    features = np.array([[
        request.tenure_months,
        request.monthly_charges,
        request.total_charges,
        1 if request.contract_type == "Month-to-month" else 0,
        1 if request.payment_method == "Electronic check" else 0,
        request.num_support_tickets,
    ]])

    runner = svc.runners[0]
    probability = await runner.predict_proba.async_run(features)
    churn_prob = float(probability[0][1])

    return PredictionResponse(
        customer_id=request.customer_id,
        churn_probability=churn_prob,
        will_churn=churn_prob > 0.5,
        model_version=bentoml.sklearn.get("customer-churn-gbt:latest").tag.version,
    )

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19


# bentofile.yaml — defines the Bento (deployable artifact)
service: "service:svc"
labels:
  team: data-science
  model: customer-churn-gbt
  stage: production
include:
  - "service.py"
  - "preprocessing.py"
python:
  packages:
    - scikit-learn==1.4.0
    - numpy==1.26.0
    - pydantic==2.6.0
    - mlflow==2.11.0
docker:
  base_image: python:3.11-slim
  system_packages:
    - libgomp1

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22


# Build and serve
bentoml build
bentoml serve service:svc --port 3000

# Test
curl -X POST http://localhost:3000/predict \
  -H "Content-Type: application/json" \
  -d '{
    "customer_id": "cust-001",
    "tenure_months": 12,
    "monthly_charges": 65.50,
    "total_charges": 786.0,
    "contract_type": "Month-to-month",
    "payment_method": "Electronic check",
    "num_support_tickets": 3
  }'

# Build container image
bentoml containerize churn-prediction:latest

# Push to registry and deploy to Kubernetes
docker push myorg/churn-prediction:$(bentoml get churn-prediction:latest -o json | jq -r '.tag')

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37


# Kubernetes deployment
apiVersion: apps/v1
kind: Deployment
metadata:
  name: churn-prediction
  namespace: ml-serving
spec:
  replicas: 3
  selector:
    matchLabels:
      app: churn-prediction
  template:
    metadata:
      labels:
        app: churn-prediction
    spec:
      containers:
        - name: churn-prediction
          image: myorg/churn-prediction:v1.2.0
          ports:
            - containerPort: 3000
          resources:
            requests:
              cpu: 500m
              memory: 512Mi
            limits:
              cpu: 2
              memory: 2Gi
          livenessProbe:
            httpGet:
              path: /healthz
              port: 3000
            initialDelaySeconds: 30
          readinessProbe:
            httpGet:
              path: /readyz
              port: 3000

NVIDIA Triton Inference Server: High-Throughput GPU Serving

Triton is NVIDIA’s production inference server. It handles TensorFlow, PyTorch, ONNX, TensorRT, Python, and more — with dynamic batching, concurrent model execution, and GPU utilization optimization.

# Model repository structure
model_repository/
├── churn_model/
│   ├── config.pbtxt          ← Model configuration
│   └── 1/                    ← Version 1
│       └── model.onnx
├── embedding_model/
│   ├── config.pbtxt
│   └── 1/
│       └── model.pt
└── ensemble_pipeline/
    ├── config.pbtxt          ← Ensemble: embedding → ranking
    └── 1/
        └── (no files — ensemble)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30


# config.pbtxt — Triton model configuration
name: "churn_model"
platform: "onnxruntime_onnx"
max_batch_size: 256

input [
  {
    name: "features"
    data_type: TYPE_FP32
    dims: [6]    # 6 input features
  }
]
output [
  {
    name: "probabilities"
    data_type: TYPE_FP32
    dims: [2]    # Binary classification probabilities
  }
]

# Dynamic batching: accumulate requests for up to 5ms, batch up to 256
dynamic_batching {
  preferred_batch_size: [32, 64, 128]
  max_queue_delay_microseconds: 5000
}

# Run 2 model instances in parallel on the same GPU
instance_group [
  { count: 2, kind: KIND_GPU }
]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20


# Client using Triton's HTTP API
import tritonclient.http as httpclient
import numpy as np

client = httpclient.InferenceServerClient(url="localhost:8000")

# Check server and model are ready
assert client.is_server_ready()
assert client.is_model_ready("churn_model")

# Prepare inputs
features = np.array([[12, 65.5, 786.0, 1, 1, 3]], dtype=np.float32)
input_tensor = httpclient.InferInput("features", features.shape, "FP32")
input_tensor.set_data_from_numpy(features)

# Run inference
outputs = [httpclient.InferRequestedOutput("probabilities")]
response = client.infer("churn_model", inputs=[input_tensor], outputs=outputs)
probabilities = response.as_numpy("probabilities")
print(f"Churn probability: {probabilities[0][1]:.3f}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22


# Export PyTorch model to ONNX for Triton
import torch
model = torch.load("churn_model.pt")
dummy_input = torch.randn(1, 6)
torch.onnx.export(
    model,
    dummy_input,
    "model_repository/churn_model/1/model.onnx",
    input_names=["features"],
    output_names=["probabilities"],
    dynamic_axes={"features": {0: "batch_size"}, "probabilities": {0: "batch_size"}},
    opset_version=17,
)

# Run Triton
docker run --gpus all --rm -p 8000:8000 -p 8001:8001 -p 8002:8002 \
  -v $(pwd)/model_repository:/models \
  nvcr.io/nvidia/tritonserver:24.01-py3 \
  tritonserver --model-repository=/models

# Monitor with Prometheus
# Triton exports metrics at :8002/metrics

Drift Detection and Model Monitoring

A model that was accurate in January degrades silently by June. Two types of drift cause this:

Data drift (covariate shift): the distribution of input features changes. The population of users sending requests differs from the training distribution. The model was never trained on this kind of data.

Concept drift (label shift): the relationship between features and the target changes. What predicted churn six months ago no longer predicts it today — customer behavior evolved.

Detecting Data Drift with Evidently

Evidently is the most practical open-source drift detection library:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62


import pandas as pd
from evidently import ColumnMapping
from evidently.report import Report
from evidently.metrics import (
    DataDriftTable,
    DatasetDriftMetric,
    ColumnDriftMetric,
    DatasetMissingValuesMetric,
)
from evidently.test_suite import TestSuite
from evidently.tests import (
    TestNumberOfDriftedColumns,
    TestShareOfDriftedColumns,
)

# Load reference data (training distribution) and current data (production)
reference_data = pd.read_parquet("data/training_features_2025Q4.parquet")
current_data = pd.read_parquet("data/production_features_2026Q1.parquet")

column_mapping = ColumnMapping(
    target="churn",
    prediction="churn_probability",
    numerical_features=["tenure_months", "monthly_charges", "total_charges", "num_support_tickets"],
    categorical_features=["contract_type", "payment_method"],
)

# Data drift report
report = Report(metrics=[
    DatasetDriftMetric(),      # Overall dataset drift
    DataDriftTable(),           # Per-column drift table
    ColumnDriftMetric(column_name="monthly_charges"),  # Individual column
    DatasetMissingValuesMetric(),
])

report.run(
    reference_data=reference_data,
    current_data=current_data,
    column_mapping=column_mapping,
)

report.save_html("drift_report.html")
results = report.as_dict()

drift_detected = results["metrics"][0]["result"]["dataset_drift"]
drift_share = results["metrics"][0]["result"]["share_of_drifted_columns"]
print(f"Dataset drift detected: {drift_detected}")
print(f"Share of drifted columns: {drift_share:.1%}")

# Automated drift test suite (for CI/monitoring pipelines)
tests = TestSuite(tests=[
    TestNumberOfDriftedColumns(lt=3),     # Fail if more than 2 columns drift
    TestShareOfDriftedColumns(lt=0.3),    # Fail if more than 30% of columns drift
])

tests.run(
    reference_data=reference_data,
    current_data=current_data,
    column_mapping=column_mapping,
)

if not tests.as_dict()["summary"]["all_passed"]:
    print("ALERT: Drift tests failed! Review before next deployment.")

Statistical Drift Tests

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41


from scipy import stats
import numpy as np

def detect_drift(reference: np.ndarray, current: np.ndarray,
                 feature_name: str, threshold: float = 0.05) -> dict:
    """Run multiple drift tests and return results."""

    # Kolmogorov-Smirnov test (continuous features)
    ks_stat, ks_p = stats.ks_2samp(reference, current)

    # Population Stability Index (PSI) — common in finance/credit
    def psi(expected, actual, n_bins=10):
        breakpoints = np.percentile(expected, np.linspace(0, 100, n_bins + 1))
        expected_percents = np.histogram(expected, bins=breakpoints)[0] / len(expected)
        actual_percents = np.histogram(actual, bins=breakpoints)[0] / len(actual)
        # Avoid division by zero
        expected_percents = np.where(expected_percents == 0, 0.0001, expected_percents)
        actual_percents = np.where(actual_percents == 0, 0.0001, actual_percents)
        psi_value = np.sum((actual_percents - expected_percents) *
                          np.log(actual_percents / expected_percents))
        return psi_value

    psi_value = psi(reference, current)

    result = {
        "feature": feature_name,
        "ks_statistic": ks_stat,
        "ks_p_value": ks_p,
        "ks_drift_detected": ks_p < threshold,
        "psi": psi_value,
        # PSI: < 0.1 = no significant change, 0.1-0.2 = moderate, > 0.2 = significant
        "psi_drift_detected": psi_value > 0.2,
        "reference_mean": reference.mean(),
        "current_mean": current.mean(),
        "mean_shift_pct": (current.mean() - reference.mean()) / reference.mean() * 100,
    }

    if result["ks_drift_detected"] or result["psi_drift_detected"]:
        print(f"⚠️  DRIFT: {feature_name} — KS p={ks_p:.4f}, PSI={psi_value:.3f}")

    return result

Monitoring Prediction Distribution (Output Drift)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58


# Track model output distribution over time
# Even without labels, output distribution shift signals potential issues

import mlflow
from collections import defaultdict
import time

class ModelMonitor:
    def __init__(self, model_name: str, window_days: int = 7):
        self.model_name = model_name
        self.window_days = window_days
        self.predictions_buffer = []

    def log_prediction(self, features: dict, prediction: float, timestamp: float = None):
        """Log a production prediction for monitoring."""
        self.predictions_buffer.append({
            "timestamp": timestamp or time.time(),
            "prediction": prediction,
            **features,
        })

        # Flush to storage periodically
        if len(self.predictions_buffer) >= 1000:
            self._flush()

    def _flush(self):
        df = pd.DataFrame(self.predictions_buffer)
        df.to_parquet(
            f"s3://monitoring/predictions/{self.model_name}/{int(time.time())}.parquet"
        )
        self.predictions_buffer = []

    def compute_daily_stats(self, df: pd.DataFrame) -> dict:
        """Compute daily prediction statistics for alerting."""
        return {
            "date": df["timestamp"].dt.date.max(),
            "prediction_count": len(df),
            "mean_prediction": df["prediction"].mean(),
            "std_prediction": df["prediction"].std(),
            "p5": df["prediction"].quantile(0.05),
            "p95": df["prediction"].quantile(0.95),
            "high_risk_rate": (df["prediction"] > 0.7).mean(),  # > 70% churn prob
        }

    def detect_output_drift(self, reference_stats: dict, current_stats: dict) -> list:
        """Compare current prediction stats to reference (training) distribution."""
        alerts = []

        # Mean shift > 20%
        mean_shift = abs(current_stats["mean_prediction"] - reference_stats["mean_prediction"])
        if mean_shift / reference_stats["mean_prediction"] > 0.2:
            alerts.append(f"Mean prediction shifted {mean_shift:.3f} from reference")

        # High-risk rate doubled
        if current_stats["high_risk_rate"] > reference_stats["high_risk_rate"] * 2:
            alerts.append(f"High-risk rate doubled: {current_stats['high_risk_rate']:.1%}")

        return alerts

A/B Testing New Model Versions

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35


# traffic_router.py — gradual rollout with shadow mode and canary
import random
import mlflow

class ModelRouter:
    def __init__(self):
        self.production_model = mlflow.sklearn.load_model(
            "models:/customer-churn-gbt/Production"
        )
        self.challenger_model = mlflow.sklearn.load_model(
            "models:/customer-churn-gbt-v2/Staging"
        )
        self.canary_traffic_pct = 0.10   # 10% to challenger

    def predict(self, features, request_id: str) -> dict:
        production_pred = self._predict_model(self.production_model, features, "production")

        # Shadow mode: run challenger but don't use its result
        challenger_pred = self._predict_model(self.challenger_model, features, "challenger")

        # Log both predictions for offline comparison
        self._log_comparison(request_id, production_pred, challenger_pred)

        # Canary: use challenger result for canary_traffic_pct of traffic
        if random.random() < self.canary_traffic_pct:
            return challenger_pred
        return production_pred

    def _predict_model(self, model, features, model_name: str) -> dict:
        prob = model.predict_proba(features)[0][1]
        return {"probability": float(prob), "model": model_name}

    def _log_comparison(self, request_id, prod, challenger):
        # Store to time-series DB for analysis
        pass  # Write to InfluxDB/Prometheus/etc.

Prometheus Metrics for Model Serving

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49


# metrics.py — expose model metrics to Prometheus
from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time

# Counters
prediction_requests = Counter(
    "model_prediction_requests_total",
    "Total prediction requests",
    ["model_name", "model_version", "status"]
)

# Latency histogram
prediction_latency = Histogram(
    "model_prediction_latency_seconds",
    "Prediction latency in seconds",
    ["model_name"],
    buckets=[0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.5, 1.0]
)

# Prediction distribution
prediction_score = Histogram(
    "model_prediction_score",
    "Distribution of model prediction scores",
    ["model_name"],
    buckets=[0.0, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0]
)

# Data quality
missing_feature_rate = Gauge(
    "model_missing_feature_rate",
    "Rate of requests with missing features",
    ["model_name", "feature_name"]
)

def predict_with_metrics(features, model_name: str, model_version: str):
    start = time.time()
    try:
        result = model.predict(features)
        prediction_requests.labels(model_name, model_version, "success").inc()
        prediction_score.labels(model_name).observe(result)
        return result
    except Exception as e:
        prediction_requests.labels(model_name, model_version, "error").inc()
        raise
    finally:
        prediction_latency.labels(model_name).observe(time.time() - start)

# Start metrics server
start_http_server(8080)

Putting It All Together: The MLOps Pipeline

Data Versioning (DVC/Delta Lake)
    │
    ▼
Feature Store (Feast/Hopsworks)
    │
    ▼
Experiment Tracking (MLflow)
    │  ← Log params, metrics, artifacts for every run
    ▼
Model Registry (MLflow Registry)
    │  ← None → Staging → Production → Archived
    ▼
CI/CD Pipeline (GitHub Actions / Argo Workflows)
    │  ← Automated tests: data validation, model evaluation, drift checks
    ▼
Model Serving (BentoML / Triton)
    │  ← Canary deployment, shadow mode
    ▼
Monitoring (Prometheus + Grafana + Evidently)
    │  ← Latency, error rate, prediction distribution, feature drift
    ▼
Alert → Trigger Retraining Pipeline

Quick Reference

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33


# MLflow
mlflow server --backend-store-uri sqlite:///mlflow.db --default-artifact-root ./artifacts
export MLFLOW_TRACKING_URI=http://localhost:5000

# Python tracking
import mlflow
with mlflow.start_run():
    mlflow.log_param("lr", 0.01)
    mlflow.log_metric("val_auc", 0.91, step=10)
    mlflow.log_artifact("model.pkl")
    mlflow.sklearn.log_model(model, "model", registered_model_name="my-model")

# Model registry
client = mlflow.MlflowClient()
client.transition_model_version_stage("my-model", version=3, stage="Production")
model = mlflow.sklearn.load_model("models:/my-model/Production")

# BentoML
bentoml build
bentoml serve service:svc --port 3000
bentoml containerize churn-prediction:latest

# Triton
docker run --gpus all -p 8000:8000 -v ./models:/models \
  nvcr.io/nvidia/tritonserver:24.01-py3 tritonserver --model-repository=/models
# Metrics at :8002/metrics

# Evidently drift report
from evidently.report import Report
from evidently.metrics import DatasetDriftMetric
report = Report(metrics=[DatasetDriftMetric()])
report.run(reference_data=ref_df, current_data=curr_df)
report.save_html("report.html")

The maturity of the MLOps ecosystem means there’s no longer an excuse for “we can’t reproduce that result” or “we don’t know if the model is still working.” The tooling exists. MLflow tracks everything automatically. BentoML removes the “how do I deploy a Python model” problem. Evidently makes drift detection a five-line script. The investment in setting this up once pays back every time you retrain, redeploy, or investigate a production incident.