Google ADK Integration

Overview

The Google ADK integration allows you to:

Automatic Instrumentation: Automatically capture traces from ADK agents without code changes
Multi-Agent Monitoring: Track complex agent interactions and workflows
Performance Insights: Monitor agent execution times and resource usage
Metrics Collection: Collect key metrics such as latency, token usage, and cost for each agent or workflow
Error Tracking: Capture and analyze agent failures and exceptions
Custom Logging: Add structured logging to your agent workflows
Node-Level Evaluation: Apply evaluators to components within an agent trace using Maxim’s node-level evaluation capabilities, enabling granular assessment and scoring of each step or node in an agent workflow

Installation

Prerequisites

Python 3.10+
Google ADK installed and configured
Maxim account and API key

Install Dependencies

pip install maxim-py google-adk python-dotenv

Quick Start

1. Environment Setup

Create a .env file in your project directory:

# Google Cloud Configuration
GOOGLE_CLOUD_PROJECT=your-project-id
GOOGLE_CLOUD_LOCATION=us-central1
GOOGLE_API_KEY=your-google-api-key
GOOGLE_GENAI_USE_VERTEXAI=False

# Maxim Configuration
MAXIM_API_KEY=your-maxim-api-key
MAXIM_LOG_REPO_ID=your-log-repository-id

2. Basic Integration

Add Maxim instrumentation to your ADK agent:

# __init__.py
import os
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

# Use Gemini API instead of Vertex AI (simpler setup)
os.environ.setdefault("GOOGLE_GENAI_USE_VERTEXAI", "False")

# Import your agent
from . import agent

# Initialize Maxim instrumentation
try:
    from maxim import Maxim
    from maxim.logger.google_adk import instrument_google_adk

    print("🔌 Initializing Maxim instrumentation for Google ADK...")
    maxim = Maxim()
    maxim_logger = maxim.logger()

    # Apply instrumentation patches to Google ADK
    instrument_google_adk(maxim_logger, debug=True)

    print("✅ Maxim instrumentation complete!")
    
    # Export the instrumented agent
    root_agent = agent.root_agent
    
except ImportError as e:
    print(f"⚠️  Could not initialize Maxim instrumentation: {e}")
    print("💡 Running without Maxim logging")
    # Fall back to using the agent without Maxim
    root_agent = agent.root_agent

3. Run Your Agent

Create a run_with_maxim.py file to run your agent:

#!/usr/bin/env python3
"""
Conversational agent using Google ADK with Maxim tracing.
"""

import asyncio
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent))

from your_agent import root_agent
from google.adk.runners import InMemoryRunner
from google.genai.types import Part, UserContent


async def interactive_session():
    """Run interactive conversation with the agent."""
    print("\n" + "=" * 80)
    print("Agent - Conversational Mode")
    print("=" * 80)
    
    # Create runner - Maxim instrumentation is auto-applied
    runner = InMemoryRunner(agent=root_agent)
    
    session = await runner.session_service.create_session(
        app_name=runner.app_name,
        user_id="user"
    )
    
    print("\nType your message (or 'exit' to quit)")
    print("=" * 80 + "\n")
    
    try:
        while True:
            try:
                user_input = input("You: ").strip()
            except EOFError:
                break
            
            if not user_input:
                continue
            
            if user_input.lower() in ['exit', 'quit']:
                break
            
            # Send message to agent
            content = UserContent(parts=[Part(text=user_input)])
            print("\nAgent: ", end="", flush=True)
            
            try:
                async for event in runner.run_async(
                    user_id=session.user_id,
                    session_id=session.id,
                    new_message=content,
                ):
                    if event.content and event.content.parts:
                        for part in event.content.parts:
                            if part.text:
                                print(part.text, end="", flush=True)
            except Exception as e:
                print(f"\n\n❌ Error: {e}")
                continue
            
            print("\n")
    
    except KeyboardInterrupt:
        pass
    
    finally:
        # End Maxim session to flush remaining traces
        from maxim.logger.google_adk.client import end_maxim_session
        end_maxim_session()
        print("\n" + "=" * 80)
        print("View traces at: https://app.getmaxim.ai")
        print("=" * 80 + "\n")


if __name__ == "__main__":
    asyncio.run(interactive_session())

Then run it:

python3 run_with_maxim.py

Important: When using the Python script approach, always call end_maxim_session() in a finally block to ensure all traces are properly flushed to Maxim before the program exits.

Example: Financial Advisor Agent

Here’s a complete example of a financial advisor agent (Agent Example by Google) with Maxim instrumentation:

# financial_advisor/__init__.py
import os
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

# Use Gemini API instead of Vertex AI
os.environ.setdefault("GOOGLE_GENAI_USE_VERTEXAI", "False")

# Import the agent
from . import agent

# Initialize Maxim instrumentation
try:
    from maxim import Maxim
    from maxim.logger.google_adk import instrument_google_adk

    print("🔌 Initializing Maxim instrumentation for Google ADK...")
    maxim = Maxim()
    maxim_logger = maxim.logger()

    # Apply instrumentation patches
    instrument_google_adk(maxim_logger, debug=True)

    print("✅ Maxim instrumentation complete!")
    
    # Export the instrumented agent
    root_agent = agent.root_agent
    
except ImportError as e:
    print(f"⚠️  Could not initialize Maxim instrumentation: {e}")
    print("💡 Running without Maxim logging")
    root_agent = agent.root_agent

Running the Example

# Navigate to the financial advisor directory
cd financial-advisor

# Install dependencies
uv sync

# Create run_with_maxim.py with the code above
# Make sure to import from 'financial_advisor' instead of 'your_agent'

# Run the agent with Maxim instrumentation
python3 run_with_maxim.py

# Optional: Run tests
uv run pytest tests

# Optional: Run evaluations
uv run pytest eval

Monitoring and Observability

Dashboard Views

Once your agent is running with Maxim instrumentation, you can monitor:

Agent Performance: Response times, success rates, and error rates
Workflow Traces: Complete execution paths through your multi-agent system
Custom Metrics: Business-specific metrics and KPIs
Error Analysis: Detailed error tracking and debugging information

Key Metrics Tracked

Agent execution time
Token usage and costs
Error rates and types
Agent interaction patterns

Troubleshooting

Common Issues

Instrumentation not working:

# Check Maxim installation
pip show maxim-py

# Verify API key
echo $MAXIM_API_KEY

# Check debug logs
export MAXIM_DEBUG=true

Agent not starting:

# Verify environment variables
cat .env

# Check Python path and imports
python3 -c "from your_agent import root_agent; print('Agent imported successfully')"

Missing traces:

Ensure instrument_google_adk() is called before agent initialization
Check that your agent is properly imported after instrumentation
Verify Maxim API key and repository ID are correct

Debug Mode

Enable debug mode for detailed logging:

# Enable debug mode
instrument_google_adk(maxim_logger, debug=True)

# Or set environment variable
os.environ["MAXIM_DEBUG"] = "true"

Advanced Integration with Callbacks

For more control over tracing and custom metrics, you can use callback functions to hook into different stages of agent execution.

Available Callbacks

The instrument_google_adk() function supports the following callbacks:

before_generation_callback: Called before each LLM generation
after_generation_callback: Called after each LLM generation completes
before_trace_callback: Called at the start of a trace
after_trace_callback: Called when a trace completes
before_span_callback: Called when creating a span for an agent
after_span_callback: Called when a span completes

Example with Custom Callbacks

# __init__.py
import os
import time
from dotenv import load_dotenv

load_dotenv()
os.environ.setdefault("GOOGLE_GENAI_USE_VERTEXAI", "False")

from . import agent

try:
    from maxim import Maxim
    from maxim.logger.google_adk import instrument_google_adk

    class MaximCallbacks:
        def __init__(self):
            self.generation_start_times = {}
        
        async def before_generation(self, callback_context, llm_request, model_info, messages):
            """Track generation start time and log model info"""
            gen_id = id(llm_request)
            self.generation_start_times[gen_id] = time.time()
            print(f"🔵 Calling {model_info['model']} with {len(messages)} messages")
        
        async def after_generation(self, callback_context, llm_response, generation, 
                                  generation_result, usage_info, content, tool_calls):
            """Add custom metrics and tags to generation"""
            gen_id = id(callback_context.llm_request) if hasattr(callback_context, 'llm_request') else None
            
            # Calculate latency
            if gen_id and gen_id in self.generation_start_times:
                latency = time.time() - self.generation_start_times[gen_id]
                generation.add_metric("latency_seconds", latency)
                
                # Add tokens per second metric
                total_tokens = usage_info.get('total_tokens', 0)
                if latency > 0:
                    generation.add_metric("tokens_per_second", total_tokens / latency)
                
                del self.generation_start_times[gen_id]
            
            # Add custom tags
            generation.add_tag("model_provider", "google")
            generation.add_tag("has_tool_calls", "yes" if tool_calls else "no")
        
        async def after_trace(self, invocation_context, trace, agent_output, trace_usage):
            """Add custom metadata to trace"""
            # Calculate estimated cost
            total_tokens = trace_usage.get('total_tokens', 0)
            estimated_cost = (total_tokens / 1000) * 0.01
            
            trace.add_tag("estimated_cost_usd", f"${estimated_cost:.4f}")
            trace.add_tag("token_efficiency", 
                         "high" if total_tokens < 2000 else "medium" if total_tokens < 5000 else "low")
            trace.add_metric("estimated_cost", estimated_cost)
        
        async def after_span(self, invocation_context, agent_span, agent_output):
            """Add custom metadata to span"""
            agent_name = invocation_context.agent.name
            output_length = len(agent_output) if agent_output else 0
            
            agent_span.add_tag("agent_name", agent_name)
            agent_span.add_tag("output_length", str(output_length))
            agent_span.add_metadata({
                "output_stats": {
                    "length": output_length,
                    "category": "long" if output_length > 500 else "short"
                }
            })
    
    # Create callbacks instance
    callbacks = MaximCallbacks()
    
    # Initialize Maxim with callbacks
    maxim = Maxim()
    instrument_google_adk(
        maxim.logger(), 
        debug=True,
        before_generation_callback=callbacks.before_generation,
        after_generation_callback=callbacks.after_generation,
        after_trace_callback=callbacks.after_trace,
        after_span_callback=callbacks.after_span,
    )
    
    print("✅ Maxim instrumentation with custom callbacks enabled!")
    root_agent = agent.root_agent
    
except ImportError as e:
    print(f"⚠️  Could not initialize Maxim: {e}")
    root_agent = agent.root_agent

Callback Parameters

Each callback receives specific parameters that you can use:

Generation Callbacks

callback_context: Context object with request/response details
llm_request: The original LLM request
llm_response: The LLM response (after_generation only)
generation: Maxim Generation object for adding metrics/tags (after_generation only)
model_info: Dictionary with model name and configuration
messages: List of messages sent to the LLM
usage_info: Token usage statistics (after_generation only)
content: Generated content (after_generation only)
tool_calls: List of tool calls made (after_generation only)

Trace Callbacks

invocation_context: Agent invocation context
trace: Maxim Trace object for adding metrics/tags (after_trace only)
user_input: The user’s input message (before_trace only)
agent_output: The agent’s output (after_trace only)
trace_usage: Overall token usage for the trace (after_trace only)

Span Callbacks

invocation_context: Agent invocation context with agent details
agent_span: Maxim Span object for adding tags/metadata (after_span only)
parent_context: Parent span context (before_span only)
agent_output: The agent’s output (after_span only)

Adding Custom Metrics and Tags

You can enrich your traces with custom data:

# In after_generation callback
generation.add_metric("latency_seconds", 1.23)
generation.add_metric("tokens_per_second", 45.6)
generation.add_tag("model_provider", "google")

# In after_trace callback
trace.add_metric("estimated_cost", 0.0123)
trace.add_tag("token_efficiency", "high")

# In after_span callback (spans support tags and metadata, not metrics)
agent_span.add_tag("agent_name", "financial_advisor")
agent_span.add_metadata({
    "custom_data": {
        "key": "value"
    }
})

Use Cases for Callbacks

Cost Tracking: Calculate and log estimated costs per trace
Performance Monitoring: Track latency and throughput metrics
Custom Analytics: Add business-specific tags for filtering and analysis
Debugging: Log detailed information about agent execution
A/B Testing: Tag traces with experiment variants
User Segmentation: Add user metadata for cohort analysis

Overview

Python

Typescript

Google ADK Integration

Overview

Installation

Prerequisites

Install Dependencies

Quick Start

1. Environment Setup

2. Basic Integration

3. Run Your Agent

Example: Financial Advisor Agent

Running the Example

Monitoring and Observability

Dashboard Views

Key Metrics Tracked

Troubleshooting

Common Issues

Debug Mode

Advanced Integration with Callbacks

Available Callbacks

Example with Custom Callbacks

Callback Parameters

Generation Callbacks

Trace Callbacks

Span Callbacks

Adding Custom Metrics and Tags

Use Cases for Callbacks

Resources