Tutorial Lengkap Vertex AI Model Monitoring: Monitoring ML Berkelanjutan

Vertex AI Model Monitoring secara otomatis mendeteksi data drift, prediction drift, dan perubahan feature attribution pada model yang dideploy. Membantu menjaga performa dan reliabilitas model di production.

Mengapa Model Monitoring?

Manfaat Utama:

• Deteksi drift: Identifikasi data dan concept drift
• Alert otomatis: Dapat notifikasi masalah
• Feature attribution: Lacak perubahan importance
• Monitoring berkelanjutan: Pengawasan model 24/7
• Integrasi: Bekerja dengan Vertex AI endpoints

Prerequisites

pip install google-cloud-aiplatform
gcloud auth login

Setup Monitoring

1. Aktifkan Monitoring pada Endpoint

from google.cloud import aiplatform

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

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

Buat monitoring job
monitoringjob = aiplatform.ModelDeploymentMonitoringJob.create(

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

    },
    scheduleconfig={"monitorinterval": {"seconds": 3600}},  # Per jam

    driftthresholds={
        "numericalfeatures": 0.3,

        "categoricalfeatures": 0.3
    }
)

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

2. Konfigurasi Deteksi Drift

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

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


Konfigurasi 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

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

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

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

        "defaultdriftthreshold": 0.2

    }
)

Tipe Monitoring

1. Training-Serving Skew

# Deteksi perbedaan antara training dan 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 perubahan distribusi prediksi
driftconfig = {

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

    }
}

3. Feature Attribution Drift

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

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

Lihat Hasil Monitoring

1. Dapatkan Statistik Monitoring

# Dapatkan statistik monitoring
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

# Dapatkan anomali yang terdeteksi
anomalies = monitoringjob.listmodeldeploymentmonitoringanomalies()

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

    print(f"Tipe Anomali: {anomaly.anomalytype}")
    print(f"Score: {anomaly.anomalyscore}")

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

3. Akses 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):

    """Hitung 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)


    # Hindari pembagian dengan nol
    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

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

if psiscore > 0.2:
    print("Drift signifikan terdeteksi!")

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 atau retraining
                print(f"Drift terdeteksi di {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
    )

Kelola Monitoring Jobs

1. Update Monitoring Job

# Update konfigurasi monitoring
monitoringjob.update(

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

2. Pause dan Resume

# Pause monitoring
monitoringjob.pause()


Resume monitoring
monitoringjob.resume()

3. Hapus Monitoring Job

# Hapus monitoring job
monitoringjob.delete()

Best Practices

1. Set Threshold yang Sesuai

# Mulai dengan threshold konservatif
thresholds = {
    "highimportancefeatures": 0.15,  # Lebih ketat

    "mediumimportancefeatures": 0.25,

    "lowimportancefeatures": 0.35  # Lebih longgar

}

2. Konfigurasi Alerts

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

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

Kesimpulan

Vertex AI Model Monitoring menyediakan:

Deteksi drift: Training-serving skew dan prediction drift

Alert otomatis: Email dan notification channels

Feature attribution: Lacak perubahan importance

Integrasi: Monitoring endpoint native

Insights: Export BigQuery untuk analisis

Key takeaways:

• Aktifkan monitoring pada semua production endpoints
• Set threshold berdasarkan dampak bisnis
• Konfigurasi alerts untuk respons tepat waktu
• Review stats monitoring secara berkala
• Trigger retraining ketika drift melebihi threshold