Accelerating AI Agent Development with Gemma 4

import vertexai
from vertexai.preview.generative_models import (
    GenerativeModel, FunctionDeclaration, Tool, Part
)
 
vertexai.init(project="YOUR_PROJECT_ID", location="us-central1")
 
# Define the tool set
web_search = FunctionDeclaration(
    name="web_search",
    description="Search the internet for current information on a topic",
    parameters={
        "type": "object",
        "properties": {
            "query": {"type": "string", "description": "The search query"}
        },
        "required": ["query"]
    }
)
 
get_stock_price = FunctionDeclaration(
    name="get_stock_price",
    description="Retrieve the current price of a publicly traded stock",
    parameters={
        "type": "object",
        "properties": {
            "ticker": {
                "type": "string",
                "description": "Stock ticker symbol (e.g., GOOG, AAPL)"
            }
        },
        "required": ["ticker"]
    }
)
 
calculate = FunctionDeclaration(
    name="calculate",
    description="Evaluate a mathematical expression and return the result",
    parameters={
        "type": "object",
        "properties": {
            "expression": {
                "type": "string",
                "description": "Mathematical expression as a string (e.g., '185.42 * 100')"
            }
        },
        "required": ["expression"]
    }
)
 
tools = [Tool(function_declarations=[web_search, get_stock_price, calculate])]
 
model = GenerativeModel(
    "google/gemma-4-31b-it",
    tools=tools,
    system_instruction="""You are a financial research assistant.
    Use available tools proactively to provide accurate, current information.
    Always verify numeric data with the appropriate tool before reporting it."""
)
 
def execute_tool(name: str, args: dict) -> dict:
    """Route tool calls to their implementations."""
    if name == "get_stock_price":
        # In production, call a real stock data API here
        return {"ticker": args["ticker"], "price": 185.42, "currency": "USD"}
    elif name == "calculate":
        # Safe evaluation - restrict to numeric operations only
        allowed_names = {"__builtins__": {}}
        result = eval(args["expression"], allowed_names)
        return {"result": result}
    elif name == "web_search":
        return {"results": [f"Search result for: {args['query']}"]}
    return {"error": "Unknown tool"}
 
def run_agent(user_message: str, max_iterations: int = 5) -> str:
    """Execute a ReAct agent loop until completion or max iterations."""
    chat = model.start_chat()
    response = chat.send_message(user_message)
    
    for _ in range(max_iterations):
        parts = response.candidates[0].content.parts
        function_calls = [p for p in parts if hasattr(p, 'function_call') and p.function_call.name]
        
        if not function_calls:
            return response.text  # No more tool calls — final answer
        
        # Execute all requested tools and collect responses
        function_responses = []
        for part in function_calls:
            fc = part.function_call
            result = execute_tool(fc.name, dict(fc.args))
            function_responses.append(
                Part.from_function_response(name=fc.name, response={"result": result})
            )
        
        response = chat.send_message(function_responses)
    
    return response.text
 
# Example usage
result = run_agent(
    "What is Apple's current stock price, and what would 50 shares cost at that price?"
)
print(result)

from vertexai.preview.generative_models import GenerativeModel, Part, Tool, FunctionDeclaration
 
create_work_order = FunctionDeclaration(
    name="create_work_order",
    description="Create a maintenance work order when a defect is detected",
    parameters={
        "type": "object",
        "properties": {
            "severity": {
                "type": "string",
                "enum": ["low", "medium", "high", "critical"]
            },
            "defect_type": {"type": "string"},
            "location": {"type": "string"},
            "recommended_action": {"type": "string"}
        },
        "required": ["severity", "defect_type", "location", "recommended_action"]
    }
)
 
alert_supervisor = FunctionDeclaration(
    name="alert_supervisor",
    description="Send an alert to the floor supervisor",
    parameters={
        "type": "object",
        "properties": {
            "message": {"type": "string"},
            "urgency": {"type": "string", "enum": ["normal", "urgent", "emergency"]}
        },
        "required": ["message", "urgency"]
    }
)
 
# E4B works well for real-time edge inference in manufacturing
inspection_model = GenerativeModel(
    "google/gemma-4-e4b",
    tools=[Tool(function_declarations=[create_work_order, alert_supervisor])],
    system_instruction=(
        "You are a quality control inspector. Analyze product images and "
        "take action immediately if you detect any defects or anomalies. "
        "Use create_work_order for defects, alert_supervisor for critical issues."
    )
)
 
def inspect_product_image(image_path: str):
    with open(image_path, "rb") as f:
        image_bytes = f.read()
    
    chat = inspection_model.start_chat()
    response = chat.send_message([
        Part.from_data(mime_type="image/jpeg", data=image_bytes),
        "Inspect this product and take appropriate action if you detect any issues."
    ])
    return response
 
# inspect_product_image("line_capture_001.jpg")

import asyncio
from vertexai.preview.generative_models import GenerativeModel
 
async def run_market_research_agent(companies: list[str]) -> str:
    """
    Fetch data for multiple companies in parallel, then synthesize.
    Sequential execution: N * API_latency
    Parallel execution: API_latency (for any N)
    """
    # Fetch all market data concurrently
    async def fetch_company_data(ticker: str) -> dict:
        await asyncio.sleep(0.2)  # Simulates API call latency
        return {"ticker": ticker, "price": 185.0, "market_cap": "2.8T", "pe_ratio": 28.5}
    
    results = await asyncio.gather(*[fetch_company_data(c) for c in companies])
    
    # Now synthesize with a single model call using all collected data
    model = GenerativeModel("google/gemma-4-31b-it")
    prompt = "Compare these companies and identify which appears strongest:\n"
    for r in results:
        prompt += f"- {r['ticker']}: Price ${r['price']}, Market Cap {r['market_cap']}, P/E {r['pe_ratio']}\n"
    
    response = model.generate_content(prompt + "\nProvide a concise analysis in 3 paragraphs.")
    return response.text
 
result = asyncio.run(run_market_research_agent(["GOOG", "AAPL", "MSFT", "META"]))
print(result)

import json
from vertexai.preview.generative_models import GenerativeModel, GenerationConfig
 
model = GenerativeModel("google/gemma-4-31b-it")
 
def classify_and_route_request(user_message: str) -> dict:
    """Classify an incoming request and route it to the appropriate handler."""
    response = model.generate_content(
        f"""Classify this customer request and determine routing:
 
        Request: "{user_message}"
 
        Return ONLY valid JSON matching this exact schema:
        {{
          "intent": "billing|technical_support|feature_request|complaint|general",
          "urgency": "low|medium|high|critical",
          "sentiment": "positive|neutral|negative",
          "key_entities": ["array of important nouns from the request"],
          "suggested_department": "string",
          "estimated_resolution_time": "string (e.g., '2 hours', '1-2 business days')"
        }}""",
        generation_config=GenerationConfig(
            response_mime_type="application/json",
            temperature=0.0
        )
    )
    return json.loads(response.text)
 
# Example
classification = classify_and_route_request(
    "My payment failed three times and I urgently need this resolved for a client demo tomorrow."
)
print(f"Intent: {classification['intent']}")
print(f"Urgency: {classification['urgency']}")
print(f"Route to: {classification['suggested_department']}")
print(f"ETA: {classification['estimated_resolution_time']}")