vertex-agent-builder

name	vertex-agent-builder
description	Build and deploy production-ready generative AI agents using Vertex AI, Gemini models, and Google Cloud infrastructure with RAG, function calling, and multi-modal capabilities. Use when appropriate context detected. Trigger with relevant phrases based on skill purpose.
allowed-tools	Read, Write, Edit, Grep, Bash(cmd:*)
version	1.0.0
author	Jeremy Longshore <[e-mail]>
license	MIT

Vertex AI Agent Builder Skill

This skill provides automated assistance for vertex agent builder tasks.

Prerequisites

• Appropriate file access permissions
• Required dependencies installed

Instructions

1. Invoke this skill when the trigger conditions are met
2. Provide necessary context and parameters
3. Review the generated output
4. Apply modifications as needed

Output

The skill produces structured output relevant to the task.

Error Handling

• Invalid input: Prompts for correction
• Missing dependencies: Lists required components
• Permission errors: Suggests remediation steps

Overview

This skill provides automated assistance for vertex agent builder tasks. This skill provides production-ready scaffolding and deployment for generative AI agents using Google Cloud's Vertex AI platform. Built on actual Google Cloud source code from GoogleCloudPlatform/generative-ai and agent-starter-pack repositories, it offers:

• Gemini Model Integration (1.5 Pro, 1.5 Flash, experimental models)
• Multi-modal Capabilities (text, image, video, audio processing)
• RAG Implementation (Retrieval Augmented Generation with Vector Search)
• Function Calling (Tool use with Gemini models)
• Production Deployment (Cloud Run, Vertex AI Endpoints)
• Evaluation Framework (AutoSxS, ROUGE, custom metrics)
• Observability (Cloud Logging, Monitoring, Trace)

Installation

# Install Google Cloud SDK
curl https://sdk.cloud.google.com | bash
exec -l $SHELL
gcloud init

# Install Python SDK
pip install google-cloud-aiplatform>=1.38.0
pip install vertexai>=1.46.0
pip install google-generativeai>=0.3.2

# Clone source repositories
git clone https://github.com/GoogleCloudPlatform/generative-ai.git
git clone https://github.com/GoogleCloudPlatform/agent-starter-pack.git
git clone https://github.com/GoogleCloudPlatform/vertex-ai-samples.git

# Install this plugin
/plugin install jeremy-vertex-ai@jeremylongshore

Quick Start (5 Minutes)

Create Your First Vertex AI Agent

import vertexai
from vertexai.generative_models import GenerativeModel, ChatSession
from vertexai.preview.generative_models import grounding

# Initialize Vertex AI
vertexai.init(project="your-project-id", location="us-central1")

# Create Gemini model
model = GenerativeModel(
    "gemini-1.5-pro-002",
    system_instruction="""You are a helpful AI assistant that can:
    - Search the web for information
    - Analyze documents and images
    - Execute Python code
    - Call external APIs
    """
)

# Start chat session
chat = model.start_chat()

# Send message with grounding
response = chat.send_message(
    "What are the latest developments in quantum computing?",
    generation_config={
        "temperature": 0.3,
        "max_output_tokens": 2048,
        "top_p": 0.95,
    },
    safety_settings={
        "HARM_CATEGORY_HATE_SPEECH": "BLOCK_MEDIUM_AND_ABOVE",
        "HARM_CATEGORY_DANGEROUS_CONTENT": "BLOCK_MEDIUM_AND_ABOVE",
    },
    tools=[grounding.Tool(google_search_retrieval=grounding.GoogleSearchRetrieval())]
)

print(response.text)

Agent Patterns

1. Production Agent with Agent Builder

Based on GoogleCloudPlatform/agent-starter-pack:

from vertexai.preview import agents
from vertexai.preview.generative_models import Tool, FunctionDeclaration
import functions_framework

class ProductionAgent:
    """
    Production-ready agent using Vertex AI Agent Builder
    """

    def __init__(self, project_id: str, location: str = "us-central1"):
        self.project_id = project_id
        self.location = location

        # Initialize agent
        self.agent = agents.Agent.create(
            project=project_id,
            location=location,
            display_name="production-agent",
            description="Production-ready agent with tools and RAG",
            model="gemini-1.5-pro-002",
            tools=self._create_tools(),
            system_instruction=self._get_system_instruction()
        )

    def _create_tools(self):
        """Create function calling tools"""
        return [
            Tool(function_declarations=[
                FunctionDeclaration(
                    name="search_knowledge_base",
                    description="Search internal knowledge base",
                    parameters={
                        "type": "object",
                        "properties": {
                            "query": {"type": "string"},
                            "top_k": {"type": , : }
                        }
                    }
                ),
                FunctionDeclaration(
                    name=,
                    description=,
                    parameters={
                        : ,
                        : {
                            : {: },
                            : {: }
                        }
                    }
                ),
                FunctionDeclaration(
                    name=,
                    description=,
                    parameters={
                        : ,
                        : {
                            : {: },
                            : {: },
                            : {: }
                        }
                    }
                )
            ])
        ]

     ():
        
         

      ():
        
        session = .agent.start_session()
        response =  session.send_message(user_input)

        
         response.function_calls:
             func_call  response.function_calls:
                result =  ._execute_function(
                    func_call.name,
                    func_call.args
                )
                
                response =  session.send_message(
                    content=,
                    function_responses=[{
                        : func_call.name,
                        : result
                    }]
                )

         response.text

      ():
        
         name == :
              ._search_kb(args[], args.get(, ))
         name == :
              ._execute_bigquery(args[], args[])
         name == :
              ._send_email(args[], args[], args[])



 ():
    
    agent = ProductionAgent(
        project_id=,
        location=
    )

    request_json = request.get_json()
    response = agent.process_request(request_json[])

     {: response}

2. RAG-Enhanced Agent

Based on GoogleCloudPlatform/generative-ai/gemini/use-cases/retrieval-augmented-generation:

from vertexai.language_models import TextEmbeddingModel
from vertexai.preview import rag
from google.cloud import aiplatform

class RAGAgent:
    """
    Agent with Retrieval Augmented Generation using Vertex AI Search
    """

    def __init__(self, project_id: str, corpus_name: str):
        self.project_id = project_id

        # Create RAG corpus
        self.corpus = rag.create_corpus(
            display_name=corpus_name,
            description="Knowledge base for RAG"
        )

        # Initialize embedding model
        self.embedding_model = TextEmbeddingModel.from_pretrained(
            "text-embedding-004"
        )

        # Initialize Gemini model
        self.model = GenerativeModel("gemini-1.5-pro-002")

    async def import_documents(self, gcs_uris: list):
        """Import documents into RAG corpus"""
        import_job = rag.import_files(
            corpus_name=self.corpus.name,
            gcs_uris=gcs_uris,
            chunk_size=512,
            chunk_overlap=100
        )

        # Wait for import to complete
        import_job.wait()

        return import_job

    async def query_with_rag(self, query: str, similarity_top_k:  = ):
        

        
        retrieval_response = rag.retrieval_query(
            rag_resources=[
                rag.RagResource(
                    rag_corpus=.corpus.name,
                    similarity_top_k=similarity_top_k
                )
            ],
            query=query
        )

        
        context = .join([
            
             chunk  retrieval_response.contexts
        ])

        
        prompt = 

        response = .model.generate_content(
            prompt,
            generation_config={
                : ,
                : ,
            }
        )

         {
            : response.text,
            : [chunk.source  chunk  retrieval_response.contexts],
            : retrieval_response.attribution_score
        }

3. Multi-Modal Agent

Based on GoogleCloudPlatform/generative-ai/gemini/multimodality:

from vertexai.generative_models import GenerativeModel, Part
import vertexai.preview.generative_models as generative_models

class MultiModalAgent:
    """
    Agent that processes text, images, video, and audio
    """

    def __init__(self):
        self.model = GenerativeModel("gemini-1.5-flash-002")

    async def analyze_image(self, image_path: str, prompt: str):
        """Analyze image with text prompt"""
        image = Part.from_uri(
            uri=f"gs://your-bucket/{image_path}",
            mime_type="image/jpeg"
        )

        response = self.model.generate_content([prompt, image])
        return response.text

    async def analyze_video(self, video_path: str, prompt: str):
        """Analyze video content"""
        video = Part.from_uri(
            uri=f"gs://your-bucket/{video_path}",
            mime_type="video/mp4"
        )

        response = self.model.generate_content(
            [prompt, video],
            generation_config={
                "temperature": 0.4,
                "max_output_tokens": 2048,
            }
        )
        return response.text

    async def ():
        
        document = Part.from_uri(
            uri=,
            mime_type=
        )

        prompt = 

        response = .model.generate_content([prompt, document])
         response.text

      ():
        
        prompt = 

        response = .model.generate_content(
            prompt,
            generation_config={
                : ,
                : ,
            }
        )
         response.text

Deployment Patterns

1. Cloud Run Deployment

# cloudbuild.yaml
steps:
  # Build container
  - name: 'gcr.io/cloud-builders/docker'
    args: ['build', '-t', 'gcr.io/$PROJECT_ID/vertex-agent:$COMMIT_SHA', '.']

  # Push to Container Registry
  - name: 'gcr.io/cloud-builders/docker'
    args: ['push', 'gcr.io/$PROJECT_ID/vertex-agent:$COMMIT_SHA']

  # Deploy to Cloud Run
  - name: 'gcr.io/google.com/cloudsdktool/cloud-sdk'
    entrypoint: gcloud
    args:
      - 'run'
      - 'deploy'
      - 'vertex-agent'
      - '--image'
      - 'gcr.io/$PROJECT_ID/vertex-agent:$COMMIT_SHA'
      - '--region'
      - 'us-central1'
      - '--platform'
      - 'managed'
      - '--allow-unauthenticated'
      - '--set-env-vars'
      - 'GCP_PROJECT=$PROJECT_ID'

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application
COPY . .

# Run as non-root user
RUN useradd -m -u 1000 agent && chown -R agent:agent /app
USER agent

EXPOSE 8080

CMD ["gunicorn", "--bind", ":8080", "--workers", "1", "--threads", "8", "--timeout", "0", "main:app"]

2. Vertex AI Endpoint Deployment

from google.cloud import aiplatform

def deploy_to_vertex_endpoint():
    """Deploy model to Vertex AI Endpoint"""

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

    # Upload model
    model = aiplatform.Model.upload(
        display_name="vertex-agent-model",
        artifact_uri="gs://your-bucket/model",
        serving_container_image_uri="us-docker.pkg.dev/vertex-ai/prediction/sklearn-cpu.1-3:latest"
    )

    # Create endpoint
    endpoint = aiplatform.Endpoint.create(
        display_name="vertex-agent-endpoint",
    )

    # Deploy model to endpoint
    endpoint.deploy(
        model=model,
        deployed_model_display_name="vertex-agent-v1",
        machine_type="n1-standard-4",
        min_replica_count=1,
        max_replica_count=5,
        accelerator_type="NVIDIA_TESLA_T4",
        accelerator_count=1,
    )

    return endpoint

Evaluation Framework

Based on GoogleCloudPlatform/generative-ai/gemini/evaluation:

from vertexai.preview.evaluation import EvalTask, MetricPromptTemplate
import pandas as pd

class AgentEvaluator:
    """
    Evaluate agent performance using Vertex AI evaluation framework
    """

    def __init__(self):
        self.eval_dataset = self._load_eval_dataset()

    def _load_eval_dataset(self):
        """Load evaluation dataset"""
        return pd.DataFrame([
            {
                "prompt": "What is the capital of France?",
                "reference": "The capital of France is Paris.",
                "context": "France is a country in Western Europe."
            },
            # Add more test cases
        ])

    async def evaluate_accuracy(self, agent):
        """Evaluate agent accuracy"""
        eval_task = EvalTask(
            dataset=self.eval_dataset,
            metrics=[
                "rouge_l_sum",
                "bleu",
                "exact_match",
                MetricPromptTemplate(
                    metric="custom_factuality",
                    metric_prompt_template="""Rate the factual accuracy of the response.
                    Reference: {reference}
                    Response: {response}
                    Score (0-1):"""
                )
            ]
        )

        # Run evaluation
        results = []
        for _, row in .eval_dataset.iterrows():
            response =  agent.process_request(row[])
            results.append({
                : row[],
                : row[],
                : response
            })

        eval_result = eval_task.evaluate(
            model=agent.model,
            response_column=
        )

         eval_result.summary_metrics

Monitoring & Observability

from google.cloud import monitoring_v3
from google.cloud import logging
import time

class AgentMonitor:
    """
    Monitor agent performance and usage
    """

    def __init__(self, project_id: str):
        self.project_id = project_id
        self.monitoring_client = monitoring_v3.MetricServiceClient()
        self.logging_client = logging.Client()
        self.logger = self.logging_client.logger("vertex-agent")

    def log_request(self, request_id: str, user_input: str, response: str, latency: float):
        """Log agent request"""
        self.logger.log_struct({
            "request_id": request_id,
            "timestamp": time.time(),
            "user_input": user_input,
            "response": response[:500],  # Truncate long responses
            "latency_ms": latency * 1000,
            "model": "gemini-1.5-pro-002",
            "success": True
        })

    def create_custom_metric(self, metric_name: str, value: float):
        """Create custom metric in Cloud Monitoring"""
        project_name = 

        series = monitoring_v3.TimeSeries()
        series.metric. = 

        now = time.time()
        seconds = (now)
        nanos = ((now - seconds) *  ** )
        interval = monitoring_v3.TimeInterval(
            {: {: seconds, : nanos}}
        )
        point = monitoring_v3.Point({
            : interval,
            : {: value}
        })
        series.points = [point]

        .monitoring_client.create_time_series(
            name=project_name,
            time_series=[series]
        )

     ():
        
        
        dashboard_config = {
            : ,
            : [
                {
                    : ,
                    : {
                        : [{
                            : {
                                : {
                                    : 
                                }
                            }
                        }]
                    }
                },
                {
                    : ,
                    : {
                        : [{
                            : {
                                : {
                                    : 
                                }
                            }
                        }]
                    }
                }
            ]
        }
         dashboard_config

Cost Optimization

class CostOptimizedAgent:
    """
    Cost-optimized agent configuration
    """

    def __init__(self):
        # Model selection based on task complexity
        self.models = {
            "simple": "gemini-1.5-flash-002",  # Fastest and cheapest
            "standard": "gemini-1.5-pro-002",   # Balanced
            "complex": "gemini-1.5-pro-002",    # Most capable
        }

        # Caching for repeated queries
        self.response_cache = {}

    def select_model(self, query: str) -> str:
        """Select appropriate model based on query complexity"""
        # Simple heuristic based on query length and keywords
        complexity_score = len(query) / 100

        if "analyze" in query or "explain" in query:
            complexity_score += 0.3
        if "code" in query or "implement" in query:
            complexity_score += 0.4

        if complexity_score < 0.3:
            return self.models[]
         complexity_score < :
             .models[]
        :
             .models[]

      ():
        
        
        cache_key = hashlib.md5(query.encode()).hexdigest()
         cache_key  .response_cache:
             .response_cache[cache_key]

        
        model_name = .select_model(query)
        model = GenerativeModel(model_name)

        
        response = model.generate_content(
            query,
            generation_config={
                : ,  
                : ,  
            }
        )

        
        .response_cache[cache_key] = response.text

         response.text

Integration with Google Cloud Services

BigQuery Integration

from google.cloud import bigquery

class BigQueryAgent:
    """Agent with BigQuery data access"""

    def __init__(self, project_id: str):
        self.bq_client = bigquery.Client(project=project_id)
        self.model = GenerativeModel("gemini-1.5-pro-002")

    async def nl2sql(self, natural_language_query: str, dataset: str):
        """Convert natural language to SQL"""

        # Get table schemas
        tables = self.bq_client.list_tables(dataset)
        schema_info = []
        for table in tables:
            table_ref = self.bq_client.get_table(table.reference)
            schema_info.append(f"Table: {table.table_id}\nSchema: {table_ref.schema}")

        prompt = f"""Convert this natural language query to SQL:
        Query: {natural_language_query}

        Available tables and schemas:
        {chr(10).join(schema_info)}

        Return only the SQL query, no explanation.
        """

        response = self.model.generate_content(prompt)
        sql_query = response.text.strip()

        # Execute query
        query_job = self.bq_client.query(sql_query)
        results = query_job.result()

        return {
            "sql": sql_query,
            "results": [dict(row) for row  results],
            : results.total_rows
        }

Cloud Storage Integration

from google.cloud import storage

class StorageAgent:
    """Agent with Cloud Storage access"""

    def __init__(self, bucket_name: str):
        self.storage_client = storage.Client()
        self.bucket = self.storage_client.bucket(bucket_name)

    async def process_documents(self, prefix: str):
        """Process all documents in a GCS prefix"""
        blobs = self.bucket.list_blobs(prefix=prefix)

        results = []
        for blob in blobs:
            if blob.name.endswith(('.pdf', '.txt', '.docx')):
                # Download and process
                content = blob.download_as_text()
                analysis = await self.analyze_document(content)
                results.append({
                    "file": blob.name,
                    "analysis": analysis
                })

        return results

Best Practices

1. Security

from google.cloud import secretmanager

class SecureAgent:
    """Agent with security best practices"""

    def __init__(self, project_id: str):
        self.secret_client = secretmanager.SecretManagerServiceClient()
        self.project_id = project_id

    def get_secret(self, secret_id: str) -> str:
        """Get secret from Secret Manager"""
        name = f"projects/{self.project_id}/secrets/{secret_id}/versions/latest"
        response = self.secret_client.access_secret_version(request={"name": name})
        return response.payload.data.decode("UTF-8")

    def sanitize_input(self, user_input: str) -> str:
        """Sanitize user input"""
        # Remove potential injection attempts
        sanitized = user_input.replace("```", "")
        sanitized = sanitized.replace("<script>", "")
        # Add more sanitization as needed
        return sanitized

2. Error Handling

from tenacity import retry, stop_after_attempt, wait_exponential
import logging

class ResilientAgent:
    """Agent with robust error handling"""

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=4, max=60)
    )
    async def process_with_retry(self, request):
        """Process with automatic retry"""
        try:
            response = await self._process(request)
            return response
        except Exception as e:
            logging.error(f"Processing failed: {e}")
            raise

Examples Repository

Complete working examples from Google Cloud repositories:

1. Text Generation - /gemini/function-calling/
2. RAG Implementation - /gemini/use-cases/retrieval-augmented-generation/
3. Multi-modal Processing - /gemini/multimodality/
4. Agent Builder - /agent-builder/
5. Production Templates - From agent-starter-pack

Resources

• Vertex AI Documentation
• Gemini API Reference
• Agent Builder Guide
• GitHub: GoogleCloudPlatform/generative-ai
• GitHub: GoogleCloudPlatform/agent-starter-pack

---

Version: 1.0.0 Last Updated: October 2025 Author: Jeremy Longshore Based on: Official Google Cloud repositories License: MIT