Articles/Agents & Manager

◈ Agents & Manager/2026-04-19Advanced

Antigravity AI Agent Design: Multimodal Production Implementation Patterns

A complete guide to building multimodal AI agents with Google Antigravity (Gemma 4). Covers image+text integration, Function Calling, async batch processing, state management, error handling, and cost estimation — with production-ready code.

antigravity⁴²⁹ gemma-4¹⁹ agents¹²³ multimodal⁷ function-calling⁵ production⁷¹ python²⁵

✦ Premium Article

I still remember the day I first shipped an image-capable agent to production. What behaved predictably on text alone started misreading "what it was obviously looking at" the moment I handed it a single photo. That small unease — familiar to any indie developer shipping and supporting apps solo — is where my multimodal agent design really began. Multimodal agents — ones that can see images, reason about them, and take action — represent a qualitative jump from text-only agents. Antigravity AI, Google's Gemma 4-based agent infrastructure, is one of the more capable platforms for building them. The combination of image understanding, function calling, and generation quality opens up use cases that weren't tractable before: analyzing uploaded documents with embedded charts, checking product images against specifications, or processing screenshots to automate workflows.

The implementation complexity jumps alongside the capability. This article covers the full stack of building a production multimodal agent with the Antigravity API — from basic image+text integration through function calling, async batch processing, and the operational patterns that keep agents stable in production.

Multimodal Input: Images and Text Together

The Antigravity API accepts images and text in the same request. The image is passed as a PIL Image object alongside the text prompt:

import google.generativeai as genai
from PIL import Image
import requests
from io import BytesIO
 
genai.configure(api_key="YOUR_ANTIGRAVITY_API_KEY")
model = genai.GenerativeModel("gemini-2.5-pro-preview-05-06")
 
def load_image_from_url(url: str) -> Image.Image:
    response = requests.get(url, timeout=10)
    return Image.open(BytesIO(response.content))
 
def load_image_from_file(path: str) -> Image.Image:
    return Image.open(path)
 
def analyze_image(image: Image.Image, prompt: str) -> str:
    """Send an image + text prompt to the model"""
    response = model.generate_content([prompt, image])
    return response.text
 
# Example: generate a product description from an image
product_image = load_image_from_file("product.jpg")
description = analyze_image(
    product_image,
    "Describe this product in three bullet points suitable for an e-commerce listing. "
    "Focus on features a customer comparing products would care about."
)
print(description)

Multiple images can be passed in the same request by extending the list: [prompt, image1, image2, image3]. The model processes all images in context together, which is useful for comparison tasks or multi-page document analysis.

Function Calling: Tools the Agent Can Use

Function calling is what transforms a chat model into an agent. The model decides which tools to call, with what arguments, based on the user's request. Here's a complete function calling setup for an e-commerce assistant:

import json
from typing import Any, Callable
 
# Define the tools the agent has access to
tools = [
    {
        "function_declarations": [
            {
                "name": "search_product_database",
                "description": "Search the product catalog by keyword, category, and price",
                "parameters": {
                    "type": "OBJECT",
                    "properties": {
                        "query": {
                            "type": "STRING",
                            "description": "Search keyword"
                        },
                        "category": {
                            "type": "STRING",
                            "description": "Product category",
                            "enum": ["electronics", "clothing", "food", "all"]
                        },
                        "max_price": {
                            "type": "NUMBER",
                            "description": "Maximum price in USD"
                        }
                    },
                    "required": ["query"]
                }
            },
            {
                "name": "get_product_details",
                "description": "Get detailed information about a specific product by ID",
                "parameters": {
                    "type": "OBJECT",
                    "properties": {
                        "product_id": {"type": "STRING"}
                    },
                    "required": ["product_id"]
                }
            },
            {
                "name": "add_to_cart",
                "description": "Add a product to the shopping cart",
                "parameters": {
                    "type": "OBJECT",
                    "properties": {
                        "product_id": {"type": "STRING"},
                        "quantity": {
                            "type": "INTEGER",
                            "description": "Quantity to add (default: 1)"
                        }
                    },
                    "required": ["product_id"]
                }
            }
        ]
    }
]
 
# Actual tool implementations (mock data for illustration)
def search_product_database(query: str, category: str = "all", max_price: float = None) -> dict:
    products = [
        {"id": "P001", "name": "Wireless Earbuds Pro", "category": "electronics", "price": 89.99},
        {"id": "P002", "name": "Noise Canceling Headphones", "category": "electronics", "price": 159.99},
        {"id": "P003", "name": "Bluetooth Speaker", "category": "electronics", "price": 45.00},
    ]
    results = [
        p for p in products
        if query.lower() in p["name"].lower()
        and (category == "all" or p["category"] == category)
        and (max_price is None or p["price"] <= max_price)
    ]
    return {"products": results, "count": len(results)}
 
def get_product_details(product_id: str) -> dict:
    details = {
        "P001": {"id": "P001", "name": "Wireless Earbuds Pro", "stock": 23, "rating": 4.5},
        "P002": {"id": "P002", "name": "Noise Canceling Headphones", "stock": 8, "rating": 4.7},
    }
    return details.get(product_id, {"error": "Product not found"})
 
def add_to_cart(product_id: str, quantity: int = 1) -> dict:
    return {"success": True, "product_id": product_id, "quantity": quantity}
 
# Map tool names to their implementations
TOOL_IMPLEMENTATIONS: dict[str, Callable] = {
    "search_product_database": search_product_database,
    "get_product_details": get_product_details,
    "add_to_cart": add_to_cart,
}
 
def execute_tool(tool_name: str, args: dict) -> Any:
    """Execute a tool and return the result"""
    impl = TOOL_IMPLEMENTATIONS.get(tool_name)
    if not impl:
        return {"error": f"Tool not found: {tool_name}"}
    try:
        return impl(**args)
    except Exception as e:
        return {"error": str(e)}

✦

Thank you for reading this far.

Continue Reading

What follows includes implementation code, benchmarks, and practical content we hope you'll find useful. This site runs without ads — server and development costs are supported entirely by members like you. If it's been helpful, we'd be truly grateful for your support.

WHAT YOU'LL LEARN

✦Complete implementation code for a multimodal agent combining images with Function Calling

✦Permission design informed by the v2.2.1 unified permission model: deciding what to delegate and what to keep in human hands

✦Production judgment across cost estimation, observability logs, and staged rollout

Secure payment via Stripe · Cancel anytime

✦

Unlock This Article

Get full access to the rest of this article. Buy once, read anytime. This site is ad-free — your support goes directly toward keeping it running.

Unlock all articles with Membership →

The Agent Loop: Multimodal + Function Calling Together

The core of the agent is the loop that processes model responses, executes tool calls, and continues until the model produces a final answer:

from dataclasses import dataclass, field
from typing import Optional
import time
import uuid
 
@dataclass
class AgentState:
    """Track agent execution state and resource usage"""
    session_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
    tool_calls_count: int = 0
    total_input_tokens: int = 0
    total_output_tokens: int = 0
    start_time: float = field(default_factory=time.time)
    max_tool_calls: int = 30        # Prevent infinite loops
    max_duration_seconds: float = 300  # 5-minute hard timeout
    
    def is_budget_exceeded(self) -> bool:
        return (
            self.tool_calls_count >= self.max_tool_calls or
            time.time() - self.start_time > self.max_duration_seconds
        )
    
    def record_usage(self, input_tokens: int, output_tokens: int):
        self.total_input_tokens += input_tokens
        self.total_output_tokens += output_tokens
    
    def summary(self) -> dict:
        return {
            "session_id": self.session_id,
            "tool_calls": self.tool_calls_count,
            "total_tokens": self.total_input_tokens + self.total_output_tokens,
            "elapsed_seconds": round(time.time() - self.start_time, 1)
        }
 
class MultimodalAgent:
    """Multimodal agent with function calling support"""
    
    def __init__(self, model_name: str = "gemini-2.5-pro-preview-05-06"):
        self.model = genai.GenerativeModel(
            model_name=model_name,
            tools=tools,
            system_instruction=(
                "You are a helpful e-commerce assistant. Help users find products, "
                "check details, and manage their cart. When images are provided, "
                "analyze them carefully to understand what the user is looking for. "
                "Use tools efficiently — aim to accomplish tasks in the fewest calls possible."
            )
        )
    
    def run(
        self,
        user_message: str,
        images: list[Image.Image] = None,
        state: Optional[AgentState] = None
    ) -> dict:
        if state is None:
            state = AgentState()
        
        # Build initial content: text + optional images
        content_parts = [user_message]
        if images:
            content_parts.extend(images)
        
        chat = self.model.start_chat()
        current_content = content_parts
        
        while not state.is_budget_exceeded():
            response = chat.send_message(current_content)
            
            # Track token usage
            if response.usage_metadata:
                state.record_usage(
                    response.usage_metadata.prompt_token_count or 0,
                    response.usage_metadata.candidates_token_count or 0
                )
            
            # Check if the model wants to call tools
            candidate = response.candidates[0]
            has_function_call = any(
                hasattr(part, 'function_call') and part.function_call.name
                for part in candidate.content.parts
            )
            
            if not has_function_call:
                # Final answer — return it
                return {
                    "response": response.text,
                    "state": state.summary()
                }
            
            # Execute all tool calls in this response
            function_responses = []
            for part in candidate.content.parts:
                if hasattr(part, 'function_call') and part.function_call.name:
                    fc = part.function_call
                    state.tool_calls_count += 1
                    
                    result = execute_tool(fc.name, dict(fc.args))
                    function_responses.append(
                        genai.protos.Part(
                            function_response=genai.protos.FunctionResponse(
                                name=fc.name,
                                response={"result": json.dumps(result, ensure_ascii=False)}
                            )
                        )
                    )
            
            current_content = function_responses
        
        return {
            "response": "The agent reached its resource limit before completing the task.",
            "state": state.summary()
        }

The two hard limits — max_tool_calls and max_duration_seconds — are non-negotiable for production. Without them, a misbehaving orchestration loop will run indefinitely and consume tokens without bound. Set them conservatively and adjust based on observed behavior.

Async Batch Processing for High-Volume Image Tasks

When processing many images, sequential API calls create unnecessary latency. An async pattern with a concurrency limit handles this efficiently:

import asyncio
 
async def analyze_image_async(
    image: Image.Image,
    prompt: str,
    image_id: str
) -> dict:
    """Analyze a single image asynchronously"""
    loop = asyncio.get_event_loop()
    try:
        response = await loop.run_in_executor(
            None,
            lambda: model.generate_content([prompt, image])
        )
        return {"id": image_id, "result": response.text, "status": "success"}
    except Exception as e:
        return {"id": image_id, "error": str(e), "status": "failed"}
 
async def batch_analyze(
    items: list[dict],  # [{"id": str, "image": Image, "prompt": str}]
    max_concurrent: int = 5
) -> list[dict]:
    """Process multiple images with bounded concurrency"""
    semaphore = asyncio.Semaphore(max_concurrent)
    
    async def process_one(item: dict) -> dict:
        async with semaphore:
            return await analyze_image_async(
                item["image"], item["prompt"], item["id"]
            )
    
    tasks = [process_one(item) for item in items]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    return [
        r if isinstance(r, dict) else {"status": "failed", "error": str(r)}
        for r in results
    ]
 
# Usage
async def process_product_catalog(image_paths: list[str]):
    items = [
        {
            "id": f"product_{i}",
            "image": Image.open(path),
            "prompt": "Identify this product category and list three key features."
        }
        for i, path in enumerate(image_paths)
    ]
    
    results = await batch_analyze(items, max_concurrent=3)
    
    succeeded = [r for r in results if r.get("status") == "success"]
    failed = [r for r in results if r.get("status") == "failed"]
    print(f"Processed: {len(succeeded)} succeeded, {len(failed)} failed")
    return results

The max_concurrent=3 limit prevents hitting API rate limits during batch operations. Tune this based on your API tier's rate limits — start conservatively and increase if you're not seeing 429 responses.

Structured Output From the Agent

For downstream systems that need to parse agent responses programmatically, forcing structured JSON output prevents parsing failures:

def run_with_structured_output(
    user_message: str,
    output_schema: dict,
    images: list[Image.Image] = None
) -> dict:
    """
    Run the agent and enforce JSON output conforming to a schema.
    Useful when agent output feeds into another system.
    """
    schema_str = json.dumps(output_schema, indent=2)
    
    structured_model = genai.GenerativeModel(
        "gemini-2.5-pro-preview-05-06",
        tools=tools,
        system_instruction=(
            f"You are a data extraction assistant. After completing your analysis, "
            f"respond ONLY with a JSON object matching this schema:\n{schema_str}\n"
            f"No explanation, no markdown — pure JSON only."
        )
    )
    
    content = [user_message] + (images or [])
    response = structured_model.generate_content(content)
    
    raw = response.text.strip()
    if raw.startswith("```"):
        lines = raw.split("\n")
        raw = "\n".join(lines[1:-1])
    
    return json.loads(raw)
 
# Example: extract product attributes from an image
schema = {
    "type": "object",
    "properties": {
        "category": {"type": "string"},
        "features": {"type": "array", "items": {"type": "string"}},
        "estimated_price_range": {"type": "string"},
        "confidence": {"type": "number"}
    }
}
 
result = run_with_structured_output(
    "Analyze this product image and extract the requested attributes.",
    schema,
    images=[product_image]
)
print(f"Category: {result['category']}")
print(f"Confidence: {result['confidence']:.1%}")

Production Observability and Cost Tracking

Two things you need to add before shipping to production:

import logging
from datetime import datetime
 
logger = logging.getLogger("antigravity_agent")
 
@dataclass
class CostEstimator:
    """Estimate API costs based on token usage"""
    # Approximate Gemini 2.5 Pro pricing (USD per 1M tokens, as of Apr 2026)
    input_price_per_million: float = 1.25
    output_price_per_million: float = 5.00
    image_price_per_image: float = 0.001315
    
    def estimate(self, input_tokens: int, output_tokens: int, images: int = 0) -> dict:
        input_cost = input_tokens / 1_000_000 * self.input_price_per_million
        output_cost = output_tokens / 1_000_000 * self.output_price_per_million
        image_cost = images * self.image_price_per_image
        total = input_cost + output_cost + image_cost
        return {
            "input_usd": round(input_cost, 6),
            "output_usd": round(output_cost, 6),
            "image_usd": round(image_cost, 6),
            "total_usd": round(total, 6)
        }
 
class ProductionAgent(MultimodalAgent):
    """Agent wrapper with logging and cost tracking"""
    
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.cost_estimator = CostEstimator()
    
    def run(self, user_message: str, images: list = None, state=None) -> dict:
        request_id = f"req_{datetime.now().strftime('%Y%m%d_%H%M%S_%f')}"
        
        logger.info(json.dumps({
            "event": "request_start",
            "request_id": request_id,
            "has_images": bool(images),
            "message_chars": len(user_message)
        }))
        
        try:
            result = super().run(user_message, images, state)
            agent_state = result.get("state", {})
            
            cost = self.cost_estimator.estimate(
                agent_state.get("total_tokens", 0) // 2,  # rough split
                agent_state.get("total_tokens", 0) // 2,
                len(images) if images else 0
            )
            
            logger.info(json.dumps({
                "event": "request_complete",
                "request_id": request_id,
                **agent_state,
                "estimated_cost_usd": cost["total_usd"]
            }))
            
            result["request_id"] = request_id
            result["estimated_cost"] = cost
            return result
            
        except Exception as e:
            logger.error(json.dumps({
                "event": "request_error",
                "request_id": request_id,
                "error": str(e)
            }))
            raise

Key Lessons From Multimodal Agent Development

A few things that aren't obvious until you've built several multimodal agents:

Image resolution affects both quality and cost. Very high-resolution images (above 3000px on a side) don't proportionally improve analysis quality for most tasks, but they do consume more tokens. Resize images to a reasonable resolution (1024–2000px) before sending to the API unless you specifically need fine-grained detail.

The system instruction is the agent's personality and constraints. Everything about how the agent behaves — its communication style, what it will and won't do, how it prioritizes tools — flows from the system instruction. A vague system instruction produces an agent with inconsistent behavior. A specific, well-designed system instruction produces an agent that's predictable and auditable.

Tool call count is a leading indicator of quality. An agent that consistently takes more tool calls than expected to complete a task is a sign of an unclear system instruction (the model doesn't know what to prioritize) or a missing tool (the model is compensating for absent capabilities with extra steps). Monitor average tool calls per successful request and investigate outliers.

Test with adversarial inputs before going live. Multimodal agents are more susceptible to prompt injection via image content than text-only agents. Test with images that contain embedded text instructions, unusual content, or formats the model hasn't seen before. Ensure the agent's behavior remains within expected bounds.

The Antigravity API's combination of multimodal understanding and function calling creates agent capabilities that genuinely weren't possible with earlier systems. The patterns in this article — structured function declarations, the bounded agent loop, async batch processing, and production observability — are the foundation for making those capabilities reliable.

For foundational Antigravity API usage, the Antigravity articles provide the baseline. For broader agent design patterns, the Agents section covers orchestration, multi-agent coordination, and evaluation.

Advanced Tool Design: Error Contracts and Versioning

The function declarations shown so far handle the happy path. Production tools also need to communicate failure clearly — not just raise exceptions, but return structured error information that the agent can reason about and potentially recover from.

from enum import Enum
from typing import Union
 
class ToolErrorCode(str, Enum):
    NOT_FOUND = "not_found"
    PERMISSION_DENIED = "permission_denied"
    RATE_LIMITED = "rate_limited"
    INVALID_INPUT = "invalid_input"
    EXTERNAL_SERVICE_ERROR = "external_service_error"
    TIMEOUT = "timeout"
 
class ToolResult:
    """Typed wrapper for tool results that distinguishes success from failure."""
    
    def __init__(
        self,
        success: bool,
        data: dict | None = None,
        error_code: ToolErrorCode | None = None,
        error_message: str | None = None,
        retry_after_seconds: int | None = None
    ):
        self.success = success
        self.data = data or {}
        self.error_code = error_code
        self.error_message = error_message
        self.retry_after_seconds = retry_after_seconds
    
    def to_dict(self) -> dict:
        if self.success:
            return {"status": "success", **self.data}
        else:
            result = {
                "status": "error",
                "error_code": self.error_code.value if self.error_code else "unknown",
                "error_message": self.error_message or "An error occurred"
            }
            if self.retry_after_seconds:
                result["retry_after_seconds"] = self.retry_after_seconds
            return result
 
 
class InventoryTool:
    """Inventory lookup with structured error contracts."""
    
    TOOL_DECLARATION = {
        "name": "check_inventory",
        "description": (
            "Check current inventory levels for one or more product SKUs. "
            "Returns quantity on hand, reorder point, and supplier lead time. "
            "If a SKU is not found, returns error_code 'not_found' for that SKU. "
            "Rate limited to 100 requests per minute per agent instance."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "skus": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": "List of product SKUs to check (max 20 per request)",
                    "maxItems": 20
                }
            },
            "required": ["skus"]
        }
    }
    
    def __init__(self, db_client, rate_limiter):
        self.db = db_client
        self.limiter = rate_limiter
    
    def execute(self, skus: list[str]) -> dict:
        # Check rate limit first
        if not self.limiter.allow():
            return ToolResult(
                success=False,
                error_code=ToolErrorCode.RATE_LIMITED,
                error_message="Rate limit exceeded for inventory lookups",
                retry_after_seconds=self.limiter.reset_in_seconds()
            ).to_dict()
        
        results = {}
        for sku in skus[:20]:  # Enforce max
            try:
                record = self.db.get_inventory(sku)
                if record is None:
                    results[sku] = ToolResult(
                        success=False,
                        error_code=ToolErrorCode.NOT_FOUND,
                        error_message=f"SKU '{sku}' not found in inventory system"
                    ).to_dict()
                else:
                    results[sku] = ToolResult(
                        success=True,
                        data={
                            "quantity_on_hand": record.qty,
                            "reorder_point": record.reorder_point,
                            "supplier_lead_time_days": record.lead_time
                        }
                    ).to_dict()
            except TimeoutError:
                results[sku] = ToolResult(
                    success=False,
                    error_code=ToolErrorCode.TIMEOUT,
                    error_message="Database query timed out",
                    retry_after_seconds=5
                ).to_dict()
        
        return {"inventory": results}

The key design choice: when a tool fails, it returns a structured dict with status: "error" and an error_code, rather than raising an exception that bubbles up to the agent loop. This way, the model can reason about the failure — "inventory lookup returned not_found for SKU-789, I should inform the user and suggest alternatives" — rather than hitting a generic error handler.

Include the error contract in the tool's description field. The model reads this at each turn and knows what error codes to expect and what they mean.

Tool Versioning and Backward Compatibility

Tools evolve. A product you deploy today will have different tool signatures six months from now. Build versioning in from the start:

TOOL_REGISTRY = {
    "check_inventory_v1": InventoryTool_v1(),
    "check_inventory_v2": InventoryTool_v2(),  # Added reorder_point field
}
 
def get_tools_for_version(api_version: str) -> list[dict]:
    """Return the right set of tool declarations for a given API version."""
    if api_version >= "2025-01":
        return [
            TOOL_REGISTRY["check_inventory_v2"].TOOL_DECLARATION,
            # ...
        ]
    else:
        return [
            TOOL_REGISTRY["check_inventory_v1"].TOOL_DECLARATION,
            # ...
        ]
 
def dispatch_tool(tool_name: str, inputs: dict, api_version: str) -> dict:
    """Route tool calls to the correct implementation."""
    versioned_name = f"{tool_name}_v2" if api_version >= "2025-01" else f"{tool_name}_v1"
    tool = TOOL_REGISTRY.get(versioned_name)
    if tool is None:
        return {"status": "error", "error_code": "unknown_tool", "error_message": f"Tool '{tool_name}' not found"}
    return tool.execute(**inputs)

If you store agent conversation histories (for multi-turn or audit purposes), you also need to ensure that replaying an old conversation against a new tool schema doesn't produce confusing errors. Keep old tool versions available for at least as long as your retention policy requires.

Multi-Turn State: Persisting Agent Context Across Requests

The examples so far assume a single request-response cycle. Many real applications need the agent to remember context across multiple turns — a customer support bot that recalls an order number from two messages ago, or a code review assistant that maintains understanding of the codebase across a session.

The Antigravity API's messages array is stateless from the API's perspective — you pass the full conversation each time. The state management burden is on you.

import json
import time
from dataclasses import dataclass, field, asdict
 
@dataclass
class ConversationTurn:
    role: str  # "user" or "assistant"
    content: list[dict]  # API-format content blocks
    timestamp: float = field(default_factory=time.time)
    tool_calls: list[str] = field(default_factory=list)  # for auditing
 
@dataclass  
class SessionState:
    session_id: str
    user_id: str
    turns: list[ConversationTurn] = field(default_factory=list)
    metadata: dict = field(default_factory=dict)
    created_at: float = field(default_factory=time.time)
    last_active: float = field(default_factory=time.time)
    total_tokens_used: int = 0
 
class SessionManager:
    """
    Manages multi-turn conversation state with automatic compression.
    Stores sessions in Redis (or a similar key-value store) for persistence.
    """
    
    MAX_TURNS = 20          # Max turns before compression kicks in
    COMPRESSION_TARGET = 10  # Turns to keep after compression
    SESSION_TTL = 86400     # 24 hours in seconds
    
    def __init__(self, redis_client, anthropic_client):
        self.redis = redis_client
        self.client = anthropic_client
    
    def load_session(self, session_id: str) -> SessionState | None:
        raw = self.redis.get(f"session:{session_id}")
        if raw is None:
            return None
        data = json.loads(raw)
        session = SessionState(**{k: v for k, v in data.items() if k \!= "turns"})
        session.turns = [ConversationTurn(**t) for t in data["turns"]]
        return session
    
    def save_session(self, session: SessionState):
        session.last_active = time.time()
        self.redis.setex(
            f"session:{session.session_id}",
            self.SESSION_TTL,
            json.dumps(asdict(session))
        )
    
    def get_messages_for_api(self, session: SessionState) -> list[dict]:
        """Convert session turns to the API's messages format."""
        return [
            {"role": turn.role, "content": turn.content}
            for turn in session.turns
        ]
    
    async def compress_if_needed(self, session: SessionState):
        """
        If turn count exceeds MAX_TURNS, summarize the oldest turns
        using a fast model and replace them with a condensed system note.
        """
        if len(session.turns) <= self.MAX_TURNS:
            return
        
        # Turns to compress (oldest, keeping COMPRESSION_TARGET recent turns)
        compress_count = len(session.turns) - self.COMPRESSION_TARGET
        old_turns = session.turns[:compress_count]
        
        # Build a plain-text version for the summarizer
        history_text = "\n".join(
            f"{t.role.upper()}: {self._extract_text(t.content)}"
            for t in old_turns
        )
        
        summary_response = self.client.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=500,
            messages=[{
                "role": "user",
                "content": (
                    f"Summarize the following conversation in 2-3 sentences, "
                    f"preserving all factual details (names, numbers, decisions made):\n\n{history_text}"
                )
            }]
        )
        
        summary_text = summary_response.content[0].text
        
        # Replace compressed turns with a single synthetic assistant turn
        summary_turn = ConversationTurn(
            role="user",
            content=[{
                "type": "text",
                "text": f"[Earlier conversation summary: {summary_text}]"
            }],
            timestamp=old_turns[0].timestamp
        )
        
        # Keep summary + recent turns
        session.turns = [summary_turn] + session.turns[compress_count:]
        
        logger.info(f"Compressed {compress_count} turns to 1 summary for session {session.session_id}")
    
    def _extract_text(self, content: list[dict]) -> str:
        parts = []
        for block in content:
            if block.get("type") == "text":
                parts.append(block["text"][:500])  # Truncate long blocks
            elif block.get("type") == "tool_use":
                parts.append(f"[called tool: {block['name']}]")
        return " ".join(parts)

The compression step uses claude-haiku (the fastest, cheapest model) to produce a summary, then replaces the old turns with that summary as a context note. This keeps the conversation window manageable without losing the thread of the conversation. The recent COMPRESSION_TARGET turns are always kept verbatim so the model has full detail about recent context.

One practical consideration: images don't compress well via text summary. If your agent handles image-heavy conversations, you may want to track image hashes and store them separately rather than trying to summarize them into text.

Testing Multimodal Agents: Strategies That Actually Work

Testing deterministic code is straightforward. Testing LLM-powered agents is not — the same input can produce different outputs across runs. Here are patterns that work in practice.

Behavior-Level Assertions, Not Output Matching

Don't assert that the agent returns an exact string. Assert that it exhibits the correct behavior:

import pytest
from unittest.mock import MagicMock, patch
 
class TestInventoryAgent:
    
    @pytest.fixture
    def agent(self):
        # Use real client with test API key for integration tests,
        # or mock client for unit tests
        client = MagicMock()
        tools = {"check_inventory": MagicMock()}
        return InventoryAgent(client, tools)
    
    def test_single_sku_lookup(self, agent):
        """Agent should call check_inventory exactly once for a single-SKU query."""
        agent.tools["check_inventory"].execute.return_value = {
            "inventory": {
                "SKU-123": {
                    "status": "success",
                    "quantity_on_hand": 42,
                    "reorder_point": 10,
                    "supplier_lead_time_days": 7
                }
            }
        }
        
        # Mock the API client to return a tool_use response then a final response
        agent.client.messages.create.side_effect = [
            _make_tool_use_response("check_inventory", {"skus": ["SKU-123"]}),
            _make_text_response("SKU-123 has 42 units in stock.")
        ]
        
        result = agent.run("How many units of SKU-123 do we have?")
        
        # Behavior assertions
        assert agent.tools["check_inventory"].execute.call_count == 1
        called_skus = agent.tools["check_inventory"].execute.call_args[1]["skus"]
        assert "SKU-123" in called_skus
        assert result["success"] is True
    
    def test_not_found_sku_informs_user(self, agent):
        """Agent should not hallucinate inventory data when SKU is not found."""
        agent.tools["check_inventory"].execute.return_value = {
            "inventory": {
                "SKU-999": {
                    "status": "error",
                    "error_code": "not_found",
                    "error_message": "SKU 'SKU-999' not found in inventory system"
                }
            }
        }
        
        agent.client.messages.create.side_effect = [
            _make_tool_use_response("check_inventory", {"skus": ["SKU-999"]}),
            _make_text_response("SKU-999 was not found in the inventory system.")
        ]
        
        result = agent.run("How many units of SKU-999 do we have?")
        
        # The agent should not claim to know the inventory level
        final_text = result.get("response", "").lower()
        assert "not found" in final_text or "doesn't exist" in final_text
        assert result["success"] is True  # Agent handled gracefully
    
    def test_respects_iteration_limit(self, agent):
        """Agent should stop after MAX_ITERATIONS even if task is unresolved."""
        # Return tool_use response forever (simulating a stuck agent)
        agent.client.messages.create.return_value = _make_tool_use_response(
            "check_inventory", {"skus": ["SKU-123"]}
        )
        agent.tools["check_inventory"].execute.return_value = {
            "inventory": {"SKU-123": {"status": "success", "quantity_on_hand": 42}}
        }
        
        result = agent.run("Check SKU-123 repeatedly until I tell you to stop")
        
        # Should stop at MAX_ITERATIONS
        assert agent.client.messages.create.call_count <= agent.MAX_ITERATIONS + 1
        assert result["success"] is False
        assert "iteration" in result.get("error", "").lower()

Integration Tests with Real Image Fixtures

For multimodal behavior, maintain a set of test images with known expected behaviors:

TEST_IMAGES_DIR = Path("tests/fixtures/images")
 
class TestImageAnalysis:
    
    def test_product_damage_detection(self, real_agent):
        """Agent should detect obvious physical damage in product images."""
        damaged_image = PIL.Image.open(TEST_IMAGES_DIR / "cracked_product.jpg")
        
        result = real_agent.run(
            "Is there any damage to this product?",
            images=[damaged_image]
        )
        
        response_text = result["response"].lower()
        # The model should mention damage/crack/defect
        damage_keywords = ["damage", "crack", "defect", "broken", "chip"]
        assert any(kw in response_text for kw in damage_keywords), \
            f"Agent failed to detect damage. Response: {result['response']}"
    
    def test_clean_product_no_false_positive(self, real_agent):
        """Agent should not report damage on a clean product image."""
        clean_image = PIL.Image.open(TEST_IMAGES_DIR / "clean_product.jpg")
        
        result = real_agent.run(
            "Is there any damage to this product?",
            images=[clean_image]
        )
        
        response_text = result["response"].lower()
        negative_indicators = ["no damage", "no defect", "appears undamaged", "looks good", "in good condition"]
        assert any(ind in response_text for ind in negative_indicators), \
            f"Possible false positive. Response: {result['response']}"

Integration tests with real images and real API calls are slow and cost money, so they don't belong in your fast unit test suite. Run them in a separate CI stage (nightly or pre-release), and keep the fixture images small to control token costs.

Putting It All Together: A Production-Ready Agent

A complete production multimodal agent integrates all of these layers:

Request → SessionManager.load() → Context assembly
       → Agent.run() → Tool execution loop
       → SessionManager.compress_if_needed() → SessionManager.save()
       → Response + cost/token metadata

The observability layer wraps the entire flow and emits structured logs that feed into your monitoring system. A Grafana dashboard with panels for request_latency_p99, tool_calls_per_request, estimated_cost_per_request, and error_rate_by_tool_error_code will tell you within minutes if something has regressed.

One last piece of operational advice: set up a "canary" request that runs every few minutes and exercises the full agent pipeline with a known input and expected output shape. Automated regression checks like this catch API changes (the Antigravity API occasionally adjusts how tool_use blocks are formatted) before your users do.

Permission Design: What to Delegate and What to Keep

Everything so far assumes the agent calls tools autonomously. But in production, a question always surfaces: should a state-changing operation like add_to_cart really run on the model's judgment alone, with no gate in front of it?

Antigravity's unified permission model, introduced in v2.2.1 (2026-06-25), gave that unease a concrete shape. You declare the operational scope an agent is allowed, and anything outside that scope requires human confirmation. The same idea carries directly into agents you build yourself against the API.

I classify tools by the weight of their side effects, in three tiers. Read-only operations (search_product_database, get_product_image) are auto-approved. Reversible mutations (add_to_cart) run automatically but must be audited. Irreversible or money-moving operations (checkout confirmation, inventory allocation) require explicit human approval. That line gets enforced as a single gate in the execution layer.

from enum import Enum
from typing import Callable
 
class Risk(Enum):
    READ = "read"                  # read-only, auto-approved
    REVERSIBLE = "reversible"      # reversible change, auto but audited
    IRREVERSIBLE = "irreversible"  # irreversible / money, human approval required
 
# tool name -> risk tier (bringing the unified permission idea to a custom agent)
TOOL_RISK: dict[str, Risk] = {
    "search_product_database": Risk.READ,
    "get_product_image": Risk.READ,
    "add_to_cart": Risk.REVERSIBLE,
    "checkout": Risk.IRREVERSIBLE,
}
 
class PermissionDenied(Exception):
    pass
 
def guarded_execute(
    tool_name: str,
    tool_args: dict,
    approver: Callable[[str, dict], bool] | None = None,
) -> Any:
    """Run a tool only after passing the permission gate."""
    risk = TOOL_RISK.get(tool_name, Risk.IRREVERSIBLE)  # unknown tools => strictest
 
    if risk is Risk.IRREVERSIBLE:
        if approver is None or not approver(tool_name, tool_args):
            logger.warning(json.dumps({
                "event": "tool_blocked", "tool": tool_name, "risk": risk.value
            }))
            raise PermissionDenied(f"{tool_name} requires approval")
 
    result = execute_tool(tool_name, tool_args)
 
    if risk is not Risk.READ:
        logger.info(json.dumps({
            "event": "tool_mutation", "tool": tool_name,
            "risk": risk.value, "args": tool_args
        }, ensure_ascii=False))
    return result

guarded_execute drops in as a replacement for execute_tool in the agent loop. The important detail is that unknown tools fall back to the strictest tier. If you add a new tool and forget to register it, the system fails safe rather than open. Deny-by-default is the insurance that keeps a slip from becoming an incident.

Why be this careful? Because with multimodal, the model's reasoning includes a black box — its interpretation of an image. With text you can re-read the input and verify, but a misread image is hard to reconstruct after the fact. So the heavier the side effect, the more of the reins you keep on the human side. That asymmetry is the biggest lesson multimodal taught me.

Three Axes for Deciding What to Delegate

Before encoding permission tiers, it helps to hold a few axes for judging what is even safe to automate. I look at three.

Axis	Safe to automate	Keep with a human
Impact of failure	Limited to a preview or draft	Reaches money, inventory, or user trust
Recovery documented	Rollback is automated	Recovery is manual and tribal
Observability	Reproducible from logs	Rationale buried in image interpretation

This table is not an absolute standard. The line shifts with the stage of your business and the nature of the assets an agent touches. But the habit of asking about impact, recovery, and observability every time is what prevents runaway behavior before it happens.

The Path to Production

Once the design settles, I never open everything at once. I move toward production in the same order every time.

First, I shrink every gate to the IRREVERSIBLE tier and run a shadow deployment limited to myself or the internal team. Collecting the "here's what the agent wanted to do" rejection logs reveals where the tiers are too loose. Next, I install observability metrics — tool call counts, rejection rate, latency, estimated cost — and alerts, so abnormal call patterns surface immediately. Only then comes a limited rollout: open to a slice of users or hours, and watch behavior on real traffic.

Running these three stages also validates how far the numbers from your cost-estimation code drift from reality. Paper cost math is surprisingly sensitive to the real distribution of conversation length and image counts.

Building reliable multimodal agents is a discipline that combines model understanding, software engineering, and operational practices. The patterns in this article represent the layer between "it works in a notebook" and "it runs in production without waking anyone up at 3am." Start with the ones most relevant to your current pain points, and add the others as your system matures.

Thank You for Reading

Antigravity Lab is ad-free, supported entirely by members like you. We publish practical guides daily with implementation code, benchmarks, and production-ready patterns. If you've found it useful, we'd love to have you on board.