MLOps, CI/CD, Testing, and Safe Releases for ML Systems

Software engineering has solved the problem of shipping code changes reliably: automated tests, CI/CD pipelines, staged rollouts. ML systems have the same problems plus additional ones: the data can be wrong, the model can degrade, and failures are often silent (wrong predictions, not crashes). This module covers how to apply engineering discipline to the ML lifecycle.

What Makes ML Testing Different

In software, a failing test raises an exception. In ML:

• A model can produce outputs with no error but wrong answers.
• A pipeline can succeed while silently processing bad data.
• A model can improve on validation but degrade in production.

ML testing requires three layers: data tests, model tests, and integration tests.

Data Validation

Test your data before training, not after your model fails:

pythonCopy
from dataclasses import dataclass
from typing import Any
import pandas as pd

@dataclass
class ColumnRule:
    name: str
    dtype: type
    null_rate_max: float = 0.05
    min_val: float = None
    max_val: float = None
    allowed_values: set = None
    unique_fraction_min: float = None

def validate_dataset(df: pd.DataFrame, rules: list[ColumnRule]) -> list[str]:
    errors = []
    for rule in rules:
        if rule.name not in df.columns:
            errors.append(f"Missing column: {rule.name}")
            continue
        col = df[rule.name]
        null_rate = col.isnull().mean()
        if null_rate > rule.null_rate_max:
            errors.append(f"{rule.name}: null rate {null_rate:.3f} > {rule.null_rate_max}")
        if rule.min_val is not None and (col.dropna() < rule.min_val).any():
            errors.append(f"{rule.name}: values below minimum {rule.min_val}")
        if rule.max_val is not None and (col.dropna() > rule.max_val).any():
            errors.append(f"{rule.name}: values above maximum {rule.max_val}")
        if rule.allowed_values:
            illegal = set(col.dropna().unique()) - rule.allowed_values
            if illegal:
                errors.append(f"{rule.name}: illegal values {illegal}")
    return errors

# Define rules
training_rules = [
    ColumnRule('user_id', str, null_rate_max=0.0),
    ColumnRule('age', float, min_val=18, max_val=120, null_rate_max=0.02),
    ColumnRule('label', int, allowed_values={0, 1}, null_rate_max=0.0),
    ColumnRule('purchase_count', float, min_val=0),
]

errors = validate_dataset(train_df, training_rules)
if errors:
    raise ValueError(f"Data validation failed:\n" + "\n".join(errors))

Model Tests

pythonCopy
import pytest
import numpy as np

class TestModelBehavior:
    def test_output_shape(self, trained_model, X_val):
        preds = trained_model.predict_proba(X_val)
        assert preds.shape == (len(X_val), 2), "Wrong output shape"

    def test_output_range(self, trained_model, X_val):
        probs = trained_model.predict_proba(X_val)[:, 1]
        assert probs.min() >= 0.0 and probs.max() <= 1.0, "Probabilities out of [0,1]"

    def test_no_nan_output(self, trained_model, X_val):
        probs = trained_model.predict_proba(X_val)[:, 1]
        assert not np.isnan(probs).any(), "NaN in model outputs"

    def test_performance_threshold(self, trained_model, X_val, y_val):
        from sklearn.metrics import roc_auc_score
        auc = roc_auc_score(y_val, trained_model.predict_proba(X_val)[:, 1])
        assert auc >= 0.75, f"Model AUC {auc:.4f} below minimum threshold 0.75"

    def test_not_trivially_predicting_one_class(self, trained_model, X_val):
        preds = trained_model.predict(X_val)
        unique_preds = len(set(preds))
        assert unique_preds > 1, "Model is predicting only one class"

The performance threshold test (test_performance_threshold) creates a gate: if a retrain produces a significantly worse model, CI fails and the model is not deployed.

Experiment Tracking and the Model Registry

Use MLflow or Weights & Biases to track every training run. The model registry stages models through a lifecycle:

Training → Staging → Production → Archived

pythonCopy
import mlflow
from mlflow.tracking import MlflowClient

client = MlflowClient()

# After training, register the model
run_id = mlflow.active_run().info.run_id
model_uri = f"runs:/{run_id}/model"
mv = mlflow.register_model(model_uri, "churn_classifier")

# Transition to staging after validation
client.transition_model_version_stage(
    name="churn_classifier",
    version=mv.version,
    stage="Staging",
    archive_existing_versions=False,
)

# Only after passing all tests, promote to production
client.transition_model_version_stage(
    name="churn_classifier",
    version=mv.version,
    stage="Production",
    archive_existing_versions=True,  # archive the current production model
)

# In serving, always load from "Production" stage
production_model = mlflow.sklearn.load_model(
    f"models:/churn_classifier/Production"
)

A Minimal ML CI/CD Pipeline

yamlCopy
# .github/workflows/ml-pipeline.yml
name: ML CI/CD

on:
  push:
    branches: [main]
  schedule:
    - cron: '0 2 * * *'  # Daily retrain at 2am UTC

jobs:
  validate-and-train:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - name: Set up Python
        uses: actions/setup-python@v4
        with: { python-version: '3.11' }
      - name: Install dependencies
        run: pip install -r requirements.txt

      - name: Validate training data
        run: python scripts/validate_data.py --date ${{ env.TRAIN_DATE }}

      - name: Run unit tests
        run: pytest tests/unit/ -v

      - name: Train model
        run: python scripts/train.py --config configs/production.yaml

      - name: Run model behavior tests
        run: pytest tests/model/ -v

      - name: Compare against production baseline
        run: python scripts/compare_models.py --threshold 0.005

      - name: Register model if all checks pass
        run: python scripts/register_model.py --stage staging

The compare_models.py step is critical: it loads the current production model and the new candidate, evaluates both on the same held-out test set, and fails CI if the new model is more than 0.005 AUC worse.

Rollback Strategy

Every deployment needs a rollback path. With a model registry:

pythonCopy
def rollback_to_previous(model_name: str):
    client = MlflowClient()
    # Find the latest archived version (previous production model)
    all_versions = client.search_model_versions(f"name='{model_name}'")
    archived = [v for v in all_versions if v.current_stage == 'Archived']
    if not archived:
        raise ValueError("No archived version to roll back to")
    latest_archived = max(archived, key=lambda v: int(v.version))

    # Promote it back to production
    client.transition_model_version_stage(
        name=model_name, version=latest_archived.version,
        stage="Production", archive_existing_versions=True
    )
    print(f"Rolled back to version {latest_archived.version}")

Common Mistakes and Bad Instincts

No data validation before training. Running a multi-hour training job on corrupted data is an expensive mistake. Validate data first, fail fast.

Testing the model on the same data it was trained on. Model tests must use a held-out set that was not used in any training decision. Otherwise your performance threshold test is measuring training set performance.

Manual model promotion. If promoting a model to production requires a human to run a command, it will be done inconsistently. Automate the promotion decision as part of CI.

Not archiving the previous production model. Before deploying a new model, archive the current one. This enables rollback without needing to retrain.

Where to Go Next

• Module 27 (Observability and Monitoring) covers what happens after deployment - detecting when a production model starts degrading.
• Module 24 (Data Pipelines) covers the upstream pipeline that feeds this CI/CD system.