Complete Vertex AI Model Monitoring Tutorial: Continuous ML Monitoring

Vertex AI Model Monitoring automatically detects data drift, prediction drift, and feature attribution changes in deployed models. It helps maintain model performance and reliability in production.

Why Model Monitoring?

Key Benefits:

• Drift detection: Identify data and concept drift
• Automatic alerts: Get notified of issues
• Feature attribution: Track feature importance changes
• Continuous monitoring: 24/7 model oversight
• Integration: Works with Vertex AI endpoints

Prerequisites

pip install google-cloud-aiplatform
gcloud auth login

Setup Monitoring

1. Enable Monitoring on Endpoint

from google.cloud import aiplatform

aiplatform.init(project="your-project", location="us-central1")

Get endpoint
endpoint = aiplatform.Endpoint("projects/123/locations/us-central1/endpoints/456")

Create monitoring job
monitoringjob = aiplatform.ModelDeploymentMonitoringJob.create(

    displayname="model-monitoring-job",
    endpoint=endpoint,
    loggingsamplingstrategy={
        "randomsampleconfig": {"samplerate": 0.8}

    },
    scheduleconfig={"monitorinterval": {"seconds": 3600}},  # Hourly

    driftthresholds={
        "numericalfeatures": 0.3,

        "categoricalfeatures": 0.3
    }
)

print(f"Monitoring job created: {monitoringjob.resourcename}")

2. Configure Drift Detection

from google.cloud.aiplatformv1 import (
    ModelDeploymentMonitoringJob,
    ModelDeploymentMonitoringObjectiveConfig,
    ModelDeploymentMonitoringScheduleConfig,
    SamplingStrategy
)

Training dataset for baseline
trainingdataset = "bq://project.dataset.trainingdata"


Configure monitoring objectives
objectiveconfig = ModelDeploymentMonitoringObjectiveConfig(
    deployedmodelid=deployedmodelid,
    objectiveconfig={

        "trainingdataset": trainingdataset,

        "trainingpredictionskewdetectionconfig": {

            "skewthresholds": {
                "age": {"value": 0.3},
                "monthlycharges": {"value": 0.3}

            }
        },
        "predictiondriftdetectionconfig": {
            "driftthresholds": {

                "prediction": {"value": 0.2}
            }
        }
    }
)

3. Set Alert Thresholds

# Configure email alerts
monitoringjob = aiplatform.ModelDeploymentMonitoringJob.create(
    displayname="monitoring-with-alerts",

    endpoint=endpoint,
    alertconfig={
        "emailalertconfig": {
            "useremails": ["team@company.com"]

        },
        "enablelogging": True
    },
    driftthresholds={

        "defaultdriftthreshold": 0.2

    }
)

Monitoring Types

1. Training-Serving Skew

# Detect differences between training and serving data
skewconfig = {
    "skewthresholds": {

        "age": {"value": 0.3},
        "tenuremonths": {"value": 0.3},
        "monthlycharges": {"value": 0.25}

    },
    "attributionscoreskewthresholds": {
        "age": {"value": 0.2}
    }
}

2. Prediction Drift

# Monitor prediction distribution changes
driftconfig = {

    "driftthresholds": {
        "predictionscore": {"value": 0.15}

    }
}

3. Feature Attribution Drift

# Monitor feature importance changes
attributionconfig = {
    "attributionscoredriftthresholds": {

        "age": {"value": 0.2},
        "tenuremonths": {"value": 0.2}
    }
}

View Monitoring Results

1. Get Monitoring Stats

# Get monitoring statistics
stats = monitoringjob.getmodeldeploymentmonitoringstats()


for stat in stats:
    print(f"Feature: {stat.featurename}")
    print(f"Drift Score: {stat.driftscore}")

    print(f"Baseline Mean: {stat.baselinestats.mean}")
    print(f"Current Mean: {stat.currentstats.mean}")

2. Query Anomalies

# Get detected anomalies
anomalies = monitoringjob.listmodeldeploymentmonitoringanomalies()

for anomaly in anomalies:
    print(f"Feature: {anomaly.featurename}")

    print(f"Anomaly Type: {anomaly.anomalytype}")
    print(f"Score: {anomaly.anomalyscore}")

    print(f"Time: {anomaly.detectiontime}")

3. Access via BigQuery

from google.cloud import bigquery

client = bigquery.Client()

query = """
SELECT
    featurename,

    driftscore,
    detectiontime

FROM project.dataset.monitoringstats
WHERE driftscore > 0.2

ORDER BY detectiontime DESC
"""

results = client.query(query)
for row in results:
    print(f"{row.featurename}: {row.driftscore}")

Custom Monitoring

1. Custom Drift Detection

import pandas as pd
from scipy import stats

def calculatepsi(expected, actual, buckets=10):

    """Calculate Population Stability Index."""
    def scalerange(x, minval, maxval):
        return (x - minval) / (maxval - minval)


    breakpoints = np.linspace(0, 1, buckets + 1)

    expectedscaled = scalerange(expected, expected.min(), expected.max())

    actualscaled = scalerange(actual, actual.min(), actual.max())


    expectedcounts = np.histogram(expectedscaled, breakpoints)[0] / len(expected)

    actualcounts = np.histogram(actualscaled, breakpoints)[0] / len(actual)


    # Avoid division by zero
    expectedcounts = np.where(expectedcounts == 0, 0.0001, expectedcounts)
    actualcounts = np.where(actualcounts == 0, 0.0001, actualcounts)


    psi = np.sum((actualcounts - expectedcounts) * np.log(actualcounts / expectedcounts))

    return psi

Calculate PSI
psiscore = calculatepsi(trainingdata["feature"], servingdata["feature"])

if psiscore > 0.2:
    print("Significant drift detected!")

2. Scheduled Monitoring Job

from google.cloud import aiplatform
from kfp import dsl

@dsl.component
def checkmodeldrift(endpointname: str, threshold: float):

    from google.cloud import aiplatform

    endpoint = aiplatform.Endpoint(endpointname)
    monitoringjobs = aiplatform.ModelDeploymentMonitoringJob.list(

        filter=f'endpoint="{endpointname}"'
    )

    for job in monitoringjobs:

        stats = job.getmodeldeploymentmonitoringstats()

        for stat in stats:
            if stat.driftscore > threshold:
                # Trigger alert or retraining
                print(f"Drift detected in {stat.featurename}")

                return True
    return False

@dsl.pipeline(name="model-monitoring-pipeline")
def monitoringpipeline():
    driftcheck = checkmodeldrift(

        endpointname="projects/123/locations/us-central1/endpoints/456",
        threshold=0.2
    )

Manage Monitoring Jobs

1. Update Monitoring Job

# Update monitoring configuration
monitoringjob.update(

    driftthresholds={"defaultdriftthreshold": 0.15},
    scheduleconfig={"monitorinterval": {"seconds": 1800}}  # 30 minutes
)

2. Pause and Resume

# Pause monitoring
monitoringjob.pause()


Resume monitoring
monitoringjob.resume()

3. Delete Monitoring Job

# Delete monitoring job
monitoringjob.delete()

Best Practices

1. Set Appropriate Thresholds

# Start with conservative thresholds
thresholds = {
    "highimportancefeatures": 0.15,  # Stricter

    "mediumimportancefeatures": 0.25,

    "lowimportancefeatures": 0.35  # More lenient

}

2. Configure Alerts

# Set up comprehensive alerting
alertconfig = {
    "emailalertconfig": {
        "useremails": ["ml-team@company.com", "oncall@company.com"]

    },
    "notificationchannels": ["projects/123/notificationChannels/456"]
}

Conclusion

Vertex AI Model Monitoring provides:

Drift detection: Training-serving skew and prediction drift

Automatic alerts: Email and notification channels

Feature attribution: Track importance changes

Integration: Native endpoint monitoring

Insights: BigQuery export for analysis

Key takeaways:

• Enable monitoring on all production endpoints
• Set thresholds based on business impact
• Configure alerts for timely response
• Review monitoring stats regularly
• Trigger retraining when drift exceeds thresholds