Evidently AI: ML Model Monitoring, Drift Detection & Reports

Generating Data Drift Reports

Install with pip install evidently. Compare a reference dataset against current production data:

import pandas as pd
from evidently.report import Report
from evidently.metric_preset import DataDriftPreset, DataQualityPreset
from evidently.metric_preset import ClassificationPreset, RegressionPreset

# Load reference (training) and current (production) data
reference_data = pd.read_csv("train_data.csv")
current_data = pd.read_csv("prod_data_last_24h.csv")

# Data drift report — detects feature distribution changes
drift_report = Report(metrics=[
    DataDriftPreset(drift_share_threshold=0.2),  # fail if >20% features drift
    DataQualityPreset(),                           # missing values, outliers, etc.
])

drift_report.run(
    reference_data=reference_data,
    current_data=current_data,
    column_mapping=None   # or specify target/prediction/datetime columns
)

# Save interactive HTML report
drift_report.save_html("drift_report.html")

# Extract metrics programmatically
result = drift_report.as_dict()
drift_share = result["metrics"][0]["result"]["share_of_drifted_columns"]
print(f"Drifted features: {drift_share * 100:.1f}%")

# Model quality report (requires ground truth labels)
quality_report = Report(metrics=[ClassificationPreset()])
quality_report.run(
    reference_data=reference_data,
    current_data=current_data.assign(prediction=model.predict(current_data)),
)
quality_report.save_html("quality_report.html")

Test Suites for CI/CD Monitoring Gates

Use Test Suites instead of Reports when you need pass/fail decisions in pipelines:

from evidently.test_suite import TestSuite
from evidently.tests import (
    TestShareOfDriftedColumns,
    TestNumberOfMissingValues,
    TestColumnDrift,
    TestMeanInNSigmas,
    TestShareOfOutRangeValues,
)

suite = TestSuite(tests=[
    # Fail if more than 20% of features drift
    TestShareOfDriftedColumns(lt=0.2),

    # Fail if missing values exceed 5%
    TestNumberOfMissingValues(lt=0.05),

    # Monitor specific high-impact features individually
    TestColumnDrift(column_name="user_age"),
    TestColumnDrift(column_name="purchase_value"),

    # Fail if mean prediction outside 3 sigma of reference
    TestMeanInNSigmas(column_name="prediction", n=3),

    # Fail if out-of-range values exceed 1%
    TestShareOfOutRangeValues(column_name="feature_score", lt=0.01),
])

suite.run(reference_data=reference_data, current_data=current_data)
suite.save_html("test_results.html")

# CI/CD gate: exit non-zero if any test fails
result = suite.as_dict()
all_passed = all(t["status"] == "SUCCESS" for t in result["tests"])
if not all_passed:
    failed = [t["name"] for t in result["tests"] if t["status"] != "SUCCESS"]
    print(f"FAILED tests: {failed}")
    raise SystemExit(1)

Batch Pipeline Integration (Airflow / Prefect)

Schedule Evidently monitoring as a daily batch job:

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
import pandas as pd
from evidently.report import Report
from evidently.metric_preset import DataDriftPreset

def run_monitoring(**context):
    ds = context["ds"]  # execution date

    # Load reference data from feature store
    reference = pd.read_parquet("s3://ml-bucket/reference/train_features.parquet")

    # Load yesterday's production data
    current = pd.read_parquet(f"s3://ml-bucket/prod/{ds}/features.parquet")

    report = Report(metrics=[DataDriftPreset(drift_share_threshold=0.15)])
    report.run(reference_data=reference, current_data=current)

    # Save report and metrics to S3
    report.save_html(f"/tmp/drift_report_{ds}.html")
    metrics = report.as_dict()

    drift_share = metrics["metrics"][0]["result"]["share_of_drifted_columns"]
    if drift_share > 0.15:
        send_slack_alert(f"[ALERT] Data drift detected: {drift_share:.0%} features drifted on {ds}")

with DAG("ml_monitoring", schedule_interval="0 6 * * *", start_date=datetime(2024, 1, 1)) as dag:
    monitor = PythonOperator(task_id="run_drift_monitoring", python_callable=run_monitoring)

Evidently vs WhyLogs vs NannyML vs Arize