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

> Effectively monitor and manage generative AI models deployed with generative-ai-gcp. Learn to use vapo_lib.py, GcpEvaluation, and Vertex AI for robust AI model oversight.

- Repository: [Google Cloud Platform/generative-ai](https://github.com/GoogleCloudPlatform/generative-ai)
- Tags: how-to-guide
- Published: 2026-03-09

---

**Use the `GoogleCloudPlatform/generative-ai` repository's [`vapo_lib.py`](https://github.com/GoogleCloudPlatform/generative-ai/blob/main/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`](https://github.com/GoogleCloudPlatform/generative-ai/blob/main/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`](https://github.com/GoogleCloudPlatform/generative-ai/blob/main/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.

```python

# 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.

```python
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`](https://github.com/GoogleCloudPlatform/generative-ai/blob/main/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.

```python

# 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.

```python
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`](https://github.com/GoogleCloudPlatform/generative-ai/blob/main/gemini/prompts/prompt_optimizer/vapo_lib.py) |
| **Model registry audit** | Persist metadata to BigQuery via `GcpDataset` | [`tools/llmevalkit/src/gcp_dataset.py`](https://github.com/GoogleCloudPlatform/generative-ai/blob/main/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`](https://github.com/GoogleCloudPlatform/generative-ai/blob/main/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`](https://github.com/GoogleCloudPlatform/generative-ai/blob/main/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`](https://github.com/GoogleCloudPlatform/generative-ai/blob/main/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`](https://github.com/GoogleCloudPlatform/generative-ai/blob/main/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`](https://github.com/GoogleCloudPlatform/generative-ai/blob/main/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`](https://github.com/GoogleCloudPlatform/generative-ai/blob/main/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.