CI/CD for Machine Learning — AI Tutorials

Why CI/CD for ML Is Different

Software CI/CD validates code correctness. ML CI/CD must validate three things simultaneously:

Layer	What Can Break	How to Test
Code	Training script bugs, dependency conflicts	Unit tests, linting, type checking
Data	Schema changes, distribution shift, missing columns	Data validation tests
Model	Accuracy regression, latency increase, fairness degradation	Model evaluation gates

A broken deployment in software: wrong output. A broken deployment in ML: wrong predictions at scale, often without an obvious error.

The ML CI/CD Pipeline

A complete ML CI/CD pipeline has four phases:

Continuous Integration — validate code and data on every push
Continuous Training — retrain when data or code changes
Continuous Evaluation — test the new model against the baseline
Continuous Deployment — promote to production if evaluation passes

Model Testing

Model testing goes beyond unit tests. There are four categories:

1. Data Tests

Validate that input data meets expectations before training or inference.

Data Validation with Great Expectationspython

import great_expectations as ge

df = ge.read_csv("data/prepared/train.csv")

# Schema tests
df.expect_column_to_exist("user_id")
df.expect_column_values_to_be_of_type("amount", "float")

# Distribution tests
df.expect_column_values_to_be_between("amount", min_value=0, max_value=100_000)
df.expect_column_values_to_not_be_null("is_fraud")
df.expect_column_mean_to_be_between("amount", min_value=50, max_value=500)

# Cardinality
df.expect_column_unique_value_count_to_be_between("user_id", 10_000, 1_000_000)

results = df.validate()
assert results.success, f"Data validation failed: {results}"

Model Performance and Behavioral Testspython

import pytest
import joblib
import numpy as np
from sklearn.metrics import f1_score, roc_auc_score

@pytest.fixture(scope="module")
def model():
    return joblib.load("models/fraud_detector.pkl")

@pytest.fixture(scope="module")
def test_data():
    import pandas as pd
    df = pd.read_parquet("data/prepared/test.parquet")
    return df.drop("is_fraud", axis=1), df["is_fraud"]

# Performance gate: new model must beat baseline
def test_f1_exceeds_baseline(model, test_data):
    X_test, y_test = test_data
    f1 = f1_score(y_test, model.predict(X_test))
    assert f1 >= 0.82, f"F1 {f1:.3f} is below 0.82 baseline"

def test_auc_exceeds_baseline(model, test_data):
    X_test, y_test = test_data
    auc = roc_auc_score(y_test, model.predict_proba(X_test)[:, 1])
    assert auc >= 0.91, f"AUC {auc:.3f} is below 0.91 baseline"

# Latency gate
def test_inference_latency(model, test_data):
    import time
    X_test, _ = test_data
    batch = X_test.head(100).values
    start = time.time()
    model.predict(batch)
    elapsed_ms = (time.time() - start) * 1000
    assert elapsed_ms < 50, f"Inference {elapsed_ms:.1f}ms exceeds 50ms limit"

# Behavioral invariant: model should predict more fraud for high amounts
def test_amount_invariant(model):
    low_amount = np.array([[100, 1, 0, 0, 0, 0, 0, 0]])   # low amount
    high_amount = np.array([[9999, 1, 0, 0, 0, 0, 0, 0]])  # high amount, same other features
    p_low = model.predict_proba(low_amount)[0, 1]
    p_high = model.predict_proba(high_amount)[0, 1]
    assert p_high > p_low, "Model should assign higher fraud prob to high amounts"

GitHub Actions ML Pipeline

GitHub Actions is a natural fit for ML CI/CD: it's free for public repos, has GPU runners, and integrates with DVC and MLflow.

.github/workflows/ml-pipeline.ymlyaml

name: ML Pipeline

on:
  push:
    branches: [main]
    paths:
      - 'src/**'
      - 'params.yaml'
      - 'data/**/*.dvc'
  schedule:
    - cron: '0 2 * * *'  # Nightly retraining

env:
  MLFLOW_TRACKING_URI: ${{ secrets.MLFLOW_URI }}

jobs:
  validate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: '3.11' }
      - run: pip install -r requirements.txt
      - run: ruff check src/
      - run: mypy src/
      - run: pytest tests/unit/ -v

  data-validation:
    needs: validate
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: iterative/setup-dvc@v1
      - run: dvc pull data/prepared/
        env:
          AWS_ACCESS_KEY_ID: ${{ secrets.AWS_KEY }}
          AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET }}
      - run: python src/validate_data.py  # runs great_expectations suite

  train-and-evaluate:
    needs: data-validation
    runs-on: ubuntu-latest
    outputs:
      model_version: ${{ steps.register.outputs.version }}
      passed_gates: ${{ steps.evaluate.outputs.passed }}
    steps:
      - uses: actions/checkout@v4
      - uses: iterative/setup-dvc@v1
      - run: dvc pull
        env:
          AWS_ACCESS_KEY_ID: ${{ secrets.AWS_KEY }}
          AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET }}
      - run: dvc repro  # runs full pipeline
      - id: evaluate
        run: |
          python src/evaluate.py --threshold 0.82
          echo "passed=true" >> $GITHUB_OUTPUT
      - id: register
        if: steps.evaluate.outputs.passed == 'true'
        run: |
          VERSION=$(python src/register_model.py)
          echo "version=$VERSION" >> $GITHUB_OUTPUT

  deploy:
    needs: train-and-evaluate
    if: needs.train-and-evaluate.outputs.passed_gates == 'true'
    runs-on: ubuntu-latest
    environment: production  # requires manual approval
    steps:
      - run: |
          python src/deploy.py \
            --model-version ${{ needs.train-and-evaluate.outputs.model_version }} \
            --environment production

Automated Retraining Triggers

Retraining should not happen on a fixed schedule alone. Trigger retraining when:

Trigger Type	Signal	Example
Performance degradation	F1 drops below threshold	Monitoring alert → retrain
Data drift	PSI or KS-test exceeds threshold	Distribution shift detected
New labeled data	Sufficient new labels collected	> 10,000 new labeled examples
Calendar	Scheduled cadence	Weekly, nightly
Data pipeline completion	Upstream data available	Airflow task success event

Retraining Trigger via GitHub Actions APIpython

import requests
import os

def trigger_retraining(reason: str, model_name: str):
    """Trigger the ML training pipeline via GitHub Actions repository dispatch."""
    token = os.environ["GITHUB_TOKEN"]
    repo = "my-org/ml-platform"

    response = requests.post(
        f"https://api.github.com/repos/{repo}/dispatches",
        headers={
            "Authorization": f"Bearer {token}",
            "Accept": "application/vnd.github+json",
        },
        json={
            "event_type": "retrain",
            "client_payload": {
                "model": model_name,
                "reason": reason,
                "triggered_by": "monitoring",
            },
        },
    )
    response.raise_for_status()
    print(f"Retraining triggered: {reason}")

# Called from your monitoring service when drift is detected
if psi_score > 0.2:
    trigger_retraining(
        reason=f"PSI score {psi_score:.3f} exceeds threshold 0.2",
        model_name="fraud-detector",
    )

Knowledge check

What is a behavioral (metamorphic) test in the context of ML model testing?

Summary

ML CI/CD must validate three layers: code (unit tests), data (schema/distribution tests), and model (performance gates)
Great Expectations provides declarative data validation with JSON result reporting
Model tests cover performance gates, latency SLAs, and behavioral invariants
GitHub Actions can orchestrate the full train → evaluate → deploy pipeline with DVC and MLflow integration
Automated retraining triggers should fire on drift detection, performance degradation, or new labeled data — not just on a fixed schedule

Next: production monitoring with data drift detection, Evidently, Arize, and Fiddler.

← MLOps & AI Engineering