How to Monitor and Manage AI Models Deployed via Generative-AI-GCP

Use the GoogleCloudPlatform/generative-ai repository's vapo_lib.py for training job health checks, GcpEvaluation for batch inference tracking, and Vertex AI Model Monitoring with Cloud Logging to observe production endpoints in real time.

The GoogleCloudPlatform/generative-ai repository provides production-ready patterns to monitor and manage AI models deployed via generative-ai-gcp. Whether you are fine-tuning Gemini models through Vertex AI Custom Jobs or serving LLM predictions via managed endpoints, the repository's utility modules automate health checks, metric collection, and continuous evaluation workflows.

Deployment Lifecycle Overview

The repository implements a four-stage lifecycle for generative AI workloads on Google Cloud:

1. Training / Fine-tuning – Executed as a Vertex AI Custom Job, monitored via the monitor_progress helper in vapo_lib.py.
2. Model Registration – The trained artifact is uploaded to the Vertex AI Model Registry using aiplatform.Model.upload().
3. Endpoint Provisioning – A Vertex AI Endpoint is created and the model version is deployed via model.deploy().
4. Observability – Cloud Logging, Cloud Monitoring, and Vertex AI Model Monitoring capture metrics, logs, and drift alerts.

Monitoring Training Jobs with vapo_lib.py

The monitor_progress Implementation

The gemini/prompts/prompt_optimizer/vapo_lib.py file contains a robust polling mechanism for Custom Job health tracking. The monitor_progress method (line 419) streams state transitions to Cloud Logging and blocks until the job succeeds or fails.

# gemini/prompts/prompt_optimizer/vapo_lib.py – simplified excerpt

import time
import logging
from google.cloud import aiplatform
from google.cloud.aiplatform.gapic.schema import job_state

class VapoLib:
    def monitor_progress(self, job: aiplatform.CustomJob) -> bool:
        """Poll job state and log transitions until completion."""
        while True:
            state = job.state
            logging.info(f"[MONITOR] Job {job.resource_name} state: {state.name}")
            
            if state == job_state.JobState.JOB_STATE_SUCCEEDED:
                return True
            if state in (
                job_state.JobState.JOB_STATE_FAILED,
                job_state.JobState.JOB_STATE_CANCELLED,
            ):
                raise RuntimeError(f"Job ended with state: {state.name}")
            
            time.sleep(30)

This pattern guarantees that downstream deployment steps only trigger after a successful training run, with full audit trails available in Cloud Logging for compliance and debugging.

Model Registry and Endpoint Management

Uploading and Deploying Models

Once training completes, the model artifact is registered and exposed via a managed endpoint. The following pattern, derived from the repository's sample applications, demonstrates the SDK calls that automatically enable Cloud Logging sinks for request/response payloads.

from google.cloud import aiplatform

def register_and_deploy(
    model_dir: str, 
    display_name: str, 
    endpoint_name: str
) -> tuple[aiplatform.Model, aiplatform.Endpoint]:
    """Upload model to registry and deploy to a new endpoint."""
    aiplatform.init(project="YOUR_PROJECT", location="us-central1")
    
    # Register model artifact

    model = aiplatform.Model.upload(
        display_name=display_name,
        artifact_uri=model_dir,
        serving_container_image_uri="us-docker.pkg.dev/vertex-ai/prediction/gemini-2-0-flash:latest",
    )
    
    # Provision endpoint

    endpoint = aiplatform.Endpoint.create(
        display_name=endpoint_name,
        machine_type="n1-standard-4",
    )
    
    # Deploy with 100% traffic allocation

    model.deploy(
        endpoint=endpoint,
        traffic_percentage=100,
        machine_type="n1-standard-4",
    )
    
    return model, endpoint

Runtime Observability and Evaluation

Batch Inference with GcpEvaluation

The tools/llmevalkit/src/gcp_evaluation.py module provides the GcpEvaluation class for systematic batch testing. It executes prompts against a deployed endpoint, streams logs to Cloud Logging, and persists results to BigQuery for longitudinal analysis.

# tools/llmevalkit/src/gcp_evaluation.py – key implementation pattern

import pandas as pd
import logging
from google.cloud import aiplatform

class GcpEvaluation:
    def __init__(self, endpoint_id: str, project: str, location: str):
        self.client = aiplatform.Endpoint(
            endpoint_name=endpoint_id, 
            project=project, 
            location=location
        )
        
    def run(self, prompts: list[str]) -> pd.DataFrame:
        """Run batch inference and store results."""
        results = []
        for prompt in prompts:
            resp = self.client.predict(instances=[{"prompt": prompt}])
            logging.info(f"[EVAL] Prompt: {prompt!r} → Response: {resp.predictions}")
            results.append({
                "prompt": prompt, 
                "response": resp.predictions,
                "timestamp": pd.Timestamp.now()
            })
            
        df = pd.DataFrame(results)
        return df

Continuous Monitoring Configuration

To enable automated alerting on latency spikes or prediction drift, configure Vertex AI Model Monitoring. The following pattern aligns with the observability stack used in the Quickbot sample applications.

def enable_model_monitoring(endpoint: aiplatform.Endpoint):
    """Activate drift detection and latency alerts."""
    monitoring = endpoint.get_model_monitoring()
    monitoring.update(
        sampling_rate=0.05,  # Monitor 5% of traffic

        alert_config=aiplatform.gapic.AlertConfig(
            email_alerts=["ml-ops@example.com"],
            sms_alerts=["+1-555-1234"],
        ),
    )
    print("🔔 Model monitoring enabled – alerts configured for latency and drift")

Metrics are automatically exported to Cloud Monitoring under the vertex_ai namespace (e.g., vertex_ai/prediction/latency, vertex_ai/prediction/error_count).

End-to-End Observability Checklist

Concern	Recommended Implementation	Source Reference
Job health	Poll CustomJob state with monitor_progress	gemini/prompts/prompt_optimizer/vapo_lib.py
Model registry audit	Persist metadata to BigQuery via GcpDataset	tools/llmevalkit/src/gcp_dataset.py
Endpoint latency & errors	Enable Vertex AI Model Monitoring + Cloud Monitoring dashboards	gemini/sample-apps/quickbot/**/service/vertex_ai.py
Prompt drift	Periodic batch evaluation with GcpEvaluation	tools/llmevalkit/src/gcp_evaluation.py
Alerting	Cloud Logging-based metrics → Alerting policies (email/SMS)	Quickbot README monitoring section
Access control	IAM roles Vertex AI Administrator, Logging Admin, Monitoring Viewer	GCP best practices

Summary

• Training pipeline health is managed via the monitor_progress method in vapo_lib.py, which polls Vertex AI Custom Jobs and streams state transitions to Cloud Logging.
• Model lifecycle management relies on the Vertex AI SDK (Model.upload and deploy) to register artifacts and provision endpoints, automatically enabling request/response logging.
• Runtime observability is implemented through the GcpEvaluation class in tools/llmevalkit/src/gcp_evaluation.py, which executes batch inference, writes structured logs, and persists results to BigQuery for drift analysis.
• Alerting and monitoring leverage Vertex AI Model Monitoring and Cloud Monitoring to track latency, error rates, and prediction drift, with notification channels configured via the SDK or Cloud Console.

Frequently Asked Questions

How do I check if my Vertex AI Custom Job finished successfully?

Use the monitor_progress method from gemini/prompts/prompt_optimizer/vapo_lib.py. This helper polls the job state every 30 seconds, logs transitions to Cloud Logging, and returns True only when the job reaches SUCCEEDED status. If the job fails or is cancelled, it raises a RuntimeError with the terminal state.

What is the best way to log predictions from a deployed model?

The GcpEvaluation class in tools/llmevalkit/src/gcp_evaluation.py demonstrates the recommended pattern. It calls endpoint.predict() for each input, writes the prompt and response to Cloud Logging via Python's standard logging module, and appends the results to a BigQuery table. This creates an immutable audit trail suitable for debugging and drift detection.

How do I enable automated alerts for model latency or drift?

Configure Vertex AI Model Monitoring using the endpoint.get_model_monitoring().update() method. Set a sampling_rate (e.g., 0.05 for 5% of traffic) and provide an AlertConfig with email or SMS recipients. This automatically publishes metrics to Cloud Monitoring under the vertex_ai namespace and triggers alerts when latency thresholds or prediction drift are detected.

Can I reuse these monitoring utilities for non-Gemini models?

Yes. The utilities in tools/llmevalkit and gemini/prompts/prompt_optimizer/vapo_lib.py are container-agnostic. They interact with the Vertex AI SDK (aiplatform.Model, aiplatform.Endpoint, aiplatform.CustomJob), which supports any model artifact compatible with Vertex AI Prediction, including custom containers, TensorFlow, PyTorch, and Hugging Face models.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}