Google ADK + Opik Integration Cookbook

1
%pip install opik google-adk litellm --upgrade

1
import opik
2
opik.configure()

1
import os
2
import getpass
3
if "OPENAI_API_KEY" not in os.environ:
4
    os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter your OpenAI API key: ")

1
import asyncio
2
import datetime
3
from zoneinfo import ZoneInfo
4
5
from google.adk.agents import LlmAgent
6
from google.adk.models.lite_llm import LiteLlm
7
from google.adk.sessions import InMemorySessionService
8
from google.adk.runners import Runner
9
from google.genai import types
10
from opik.integrations.adk import OpikTracer, track_adk_agent_recursive

1
def get_weather(city: str) -> dict:
2
    """Get weather information for a city."""
3
    if city.lower() == "new york":
4
        return {
5
            "status": "success",
6
            "report": "The weather in New York is sunny with a temperature of 25 °C (77 °F).",
7
        }
8
    elif city.lower() == "london":
9
        return {
10
            "status": "success",
11
            "report": "The weather in London is cloudy with a temperature of 18 °C (64 °F).",
12
        }
13
    return {"status": "error", "error_message": f"Weather info for '{city}' is unavailable."}
14
15
def get_current_time(city: str) -> dict:
16
    """Get current time for a city."""
17
    if city.lower() == "new york":
18
        tz = ZoneInfo("America/New_York")
19
        now = datetime.datetime.now(tz)
20
        return {
21
            "status": "success",
22
            "report": now.strftime(f"The current time in {city} is %Y-%m-%d %H:%M:%S %Z%z."),
23
        }
24
    elif city.lower() == "london":
25
        tz = ZoneInfo("Europe/London")
26
        now = datetime.datetime.now(tz)
27
        return {
28
            "status": "success",
29
            "report": now.strftime(f"The current time in {city} is %Y-%m-%d %H:%M:%S %Z%z."),
30
        }
31
    return {"status": "error", "error_message": f"No timezone info for '{city}'."}

1
basic_tracer = OpikTracer(
2
    name="basic-weather-agent",
3
    tags=["basic", "weather", "time", "single-agent"],
4
    metadata={
5
        "environment": "development",
6
        "model": "gpt-4o",
7
        "framework": "google-adk",
8
        "example": "basic"
9
    },
10
    project_name="adk-basic-demo"
11
)

1
# Initialize LiteLLM with OpenAI gpt-4o
2
llm = LiteLlm(model="openai/gpt-4o")
3
4
# Create the basic agent with Opik callbacks
5
basic_agent = LlmAgent(
6
    name="weather_time_agent",
7
    model=llm,
8
    description="Agent for answering time & weather questions",
9
    instruction="Answer questions about the time or weather in a city. Be helpful and provide clear information.",
10
    tools=[get_weather, get_current_time],
11
    before_agent_callback=basic_tracer.before_agent_callback,
12
    after_agent_callback=basic_tracer.after_agent_callback,
13
    before_model_callback=basic_tracer.before_model_callback,
14
    after_model_callback=basic_tracer.after_model_callback,
15
    before_tool_callback=basic_tracer.before_tool_callback,
16
    after_tool_callback=basic_tracer.after_tool_callback,
17
)

1
basic_session_service = InMemorySessionService()
2
basic_runner = Runner(
3
    agent=basic_agent,
4
    app_name="basic_weather_app",
5
    session_service=basic_session_service,
6
)

1
async def setup_basic_session():
2
    """Create a new session for the basic example."""
3
    sess = await basic_session_service.create_session(
4
        app_name="basic_weather_app",
5
        user_id="user_basic",
6
        session_id="session_basic_001"
7
    )
8
    return sess.id
9
10
async def call_basic_agent(user_msg: str, session_id: str):
11
    """Send a message to the basic agent and get the response."""
12
    print(f"User: {user_msg}")
13
    content = types.Content(role="user", parts=[types.Part(text=user_msg)])
14
    async for event in basic_runner.run_async(user_id="user_basic", session_id=session_id, new_message=content):
15
        if event.is_final_response():
16
            print(f"Assistant: {event.content.parts[0].text}")
17
            print()

1
# Create a session for basic example
2
basic_session_id = await setup_basic_session()
3
print(f"Created basic session: {basic_session_id}")
4
print()
5
6
# Test weather query
7
await call_basic_agent("What's the weather like in New York?", basic_session_id)
8
9
# Test time query  
10
await call_basic_agent("What time is it in London?", basic_session_id)
11
12
# Test combined query
13
await call_basic_agent("Can you tell me both the weather and time in New York?", basic_session_id)

1
def get_detailed_weather(city: str) -> dict:
2
    """Get detailed weather information including forecast."""
3
    weather_data = {
4
        "new york": {
5
            "current": "Sunny, 25°C (77°F)",
6
            "humidity": "65%",
7
            "wind": "10 km/h NW",
8
            "forecast": "Partly cloudy tomorrow, high of 27°C"
9
        },
10
        "london": {
11
            "current": "Cloudy, 18°C (64°F)", 
12
            "humidity": "78%",
13
            "wind": "15 km/h SW",
14
            "forecast": "Light rain expected tomorrow, high of 16°C"
15
        },
16
        "tokyo": {
17
            "current": "Partly cloudy, 22°C (72°F)",
18
            "humidity": "70%", 
19
            "wind": "8 km/h E",
20
            "forecast": "Sunny tomorrow, high of 25°C"
21
        }
22
    }
23
    
24
    city_lower = city.lower()
25
    if city_lower in weather_data:
26
        data = weather_data[city_lower]
27
        return {
28
            "status": "success",
29
            "report": f"Weather in {city}: {data['current']}. Humidity: {data['humidity']}, Wind: {data['wind']}. {data['forecast']}"
30
        }
31
    return {"status": "error", "error_message": f"Detailed weather for '{city}' is unavailable."}
32
33
def get_world_time(city: str) -> dict:
34
    """Get time information for major world cities."""
35
    timezones = {
36
        "new york": "America/New_York",
37
        "london": "Europe/London", 
38
        "tokyo": "Asia/Tokyo",
39
        "sydney": "Australia/Sydney",
40
        "paris": "Europe/Paris"
41
    }
42
    
43
    city_lower = city.lower()
44
    if city_lower in timezones:
45
        tz = ZoneInfo(timezones[city_lower])
46
        now = datetime.datetime.now(tz)
47
        return {
48
            "status": "success",
49
            "report": now.strftime(f"Current time in {city}: %A, %B %d, %Y at %I:%M %p %Z")
50
        }
51
    return {"status": "error", "error_message": f"Time zone info for '{city}' is unavailable."}
52
53
def get_travel_info(from_city: str, to_city: str) -> dict:
54
    """Get basic travel information between cities."""
55
    travel_data = {
56
        ("new york", "london"): {"flight_time": "7 hours", "time_diff": "+5 hours"},
57
        ("london", "new york"): {"flight_time": "8 hours", "time_diff": "-5 hours"},
58
        ("new york", "tokyo"): {"flight_time": "14 hours", "time_diff": "+14 hours"},
59
        ("tokyo", "new york"): {"flight_time": "13 hours", "time_diff": "-14 hours"},
60
        ("london", "tokyo"): {"flight_time": "12 hours", "time_diff": "+9 hours"},
61
        ("tokyo", "london"): {"flight_time": "11 hours", "time_diff": "-9 hours"},
62
    }
63
    
64
    route = (from_city.lower(), to_city.lower())
65
    if route in travel_data:
66
        data = travel_data[route]
67
        return {
68
            "status": "success",
69
            "report": f"Travel from {from_city} to {to_city}: Approximately {data['flight_time']} flight time. Time difference: {data['time_diff']}"
70
        }
71
    return {"status": "error", "error_message": f"Travel info for '{from_city}' to '{to_city}' is unavailable."}

1
# Weather specialist agent (no Opik callbacks needed)
2
weather_agent = LlmAgent(
3
    name="weather_specialist",
4
    model=llm,
5
    description="Specialized agent for detailed weather information",
6
    instruction="Provide comprehensive weather information including current conditions and forecasts. Be detailed and informative.",
7
    tools=[get_detailed_weather]
8
)
9
10
# Time specialist agent (no Opik callbacks needed)  
11
time_agent = LlmAgent(
12
    name="time_specialist",
13
    model=llm,
14
    description="Specialized agent for world time information",
15
    instruction="Provide accurate time information for cities around the world. Include day of week and full date.",
16
    tools=[get_world_time]
17
)
18
19
# Travel specialist agent (no Opik callbacks needed)
20
travel_agent = LlmAgent(
21
    name="travel_specialist", 
22
    model=llm,
23
    description="Specialized agent for travel information",
24
    instruction="Provide helpful travel information including flight times and time zone differences.",
25
    tools=[get_travel_info]
26
)

1
# Configure Opik tracer for multi-agent example
2
multi_agent_tracer = OpikTracer(
3
    name="multi-agent-coordinator",
4
    tags=["multi-agent", "coordinator", "weather", "time", "travel"],
5
    metadata={
6
        "environment": "development",
7
        "model": "gpt-4o",
8
        "framework": "google-adk", 
9
        "example": "multi-agent",
10
        "agent_count": 4
11
    },
12
    project_name="adk-multi-agent-demo"
13
)
14
15
# Coordinator agent with sub-agents
16
coordinator_agent = LlmAgent(
17
    name="travel_coordinator",
18
    model=llm,
19
    description="Coordinator agent that delegates to specialized agents for weather, time, and travel information",
20
    instruction="""You are a travel coordinator that helps users with weather, time, and travel information.
21
    
22
    You have access to three specialized agents:
23
    - weather_specialist: For detailed weather information
24
    - time_specialist: For world time information  
25
    - travel_specialist: For travel planning information
26
    
27
    Delegate appropriate queries to the right specialist agents and compile comprehensive responses for the user.""",
28
    tools=[],  # No direct tools, delegates to sub-agents
29
    sub_agents=[weather_agent, time_agent, travel_agent],
30
)
31
32
# Use the experimental recursive tracking feature to instrument all agents at once
33
from opik.integrations.adk import track_adk_agent_recursive
34
track_adk_agent_recursive(coordinator_agent, multi_agent_tracer)

1
multi_session_service = InMemorySessionService()
2
multi_runner = Runner(
3
    agent=coordinator_agent,
4
    app_name="multi_agent_travel_app",
5
    session_service=multi_session_service,
6
)

1
async def setup_multi_session():
2
    """Create a new session for the multi-agent example."""
3
    sess = await multi_session_service.create_session(
4
        app_name="multi_agent_travel_app",
5
        user_id="user_multi",
6
        session_id="session_multi_001"
7
    )
8
    return sess.id
9
10
async def call_multi_agent(user_msg: str, session_id: str):
11
    """Send a message to the coordinator agent and get the response."""
12
    print(f"User: {user_msg}")
13
    content = types.Content(role="user", parts=[types.Part(text=user_msg)])
14
    async for event in multi_runner.run_async(user_id="user_multi", session_id=session_id, new_message=content):
15
        if event.is_final_response():
16
            print(f"Coordinator: {event.content.parts[0].text}")
17
            print()

1
# Demo: Each question in a separate session to show individual traces
2
print("=== Creating separate sessions for individual traces ===")
3
4
# Question 1: Weather (separate session)
5
session_1 = await setup_multi_session()
6
await call_multi_agent("I need detailed weather information for Tokyo", session_1)
7
8
# Question 2: Time (separate session) 
9
session_2 = await setup_multi_session()
10
await call_multi_agent("What time is it in Paris right now?", session_2)
11
12
# Question 3: Travel (separate session)
13
session_3 = await setup_multi_session()
14
await call_multi_agent("I'm planning to travel from London to New York. Can you help with travel time and time zones?", session_3)
15
16
# Question 4: Complex multi-agent (separate session)
17
session_4 = await setup_multi_session()
18
await call_multi_agent("I'm traveling from New York to Tokyo tomorrow. Can you give me the weather in both cities, current times, and travel information?", session_4)

1
@opik.track(name="weather_data_processing", tags=["data-processing", "weather"])
2
def process_weather_data(raw_data: dict) -> dict:
3
    """Process raw weather data with additional computations."""
4
    # Simulate some data processing steps that we want to trace separately
5
    processed = {
6
        "temperature_celsius": raw_data.get("temp_c", 0),
7
        "temperature_fahrenheit": raw_data.get("temp_c", 0) * 9/5 + 32,
8
        "conditions": raw_data.get("condition", "unknown"),
9
        "comfort_index": "comfortable" if 18 <= raw_data.get("temp_c", 0) <= 25 else "less comfortable"
10
    }
11
    return processed
12
13
@opik.track(name="location_validation", tags=["validation", "location"])  
14
def validate_location(city: str) -> dict:
15
    """Validate and normalize city names."""
16
    # Simulate location validation logic that we want to trace
17
    normalized_cities = {
18
        "nyc": "New York",
19
        "ny": "New York", 
20
        "new york city": "New York",
21
        "london uk": "London",
22
        "london england": "London",
23
        "tokyo japan": "Tokyo"
24
    }
25
    
26
    city_lower = city.lower().strip()
27
    validated_city = normalized_cities.get(city_lower, city.title())
28
    
29
    return {
30
        "original": city,
31
        "validated": validated_city,
32
        "is_valid": city_lower in ["new york", "london", "tokyo"] or city_lower in normalized_cities
33
    }
34
35
@opik.track(name="advanced_weather_lookup", tags=["weather", "api-simulation"])
36
def get_advanced_weather(city: str) -> dict:
37
    """Get weather with internal processing steps tracked by Opik decorators."""
38
    
39
    # Step 1: Validate location (traced by @opik.track)
40
    location_result = validate_location(city)
41
    
42
    if not location_result["is_valid"]:
43
        return {
44
            "status": "error", 
45
            "error_message": f"Invalid location: {city}"
46
        }
47
    
48
    validated_city = location_result["validated"]
49
    
50
    # Step 2: Get raw weather data (simulated)
51
    raw_weather_data = {
52
        "New York": {"temp_c": 25, "condition": "sunny", "humidity": 65},
53
        "London": {"temp_c": 18, "condition": "cloudy", "humidity": 78}, 
54
        "Tokyo": {"temp_c": 22, "condition": "partly cloudy", "humidity": 70}
55
    }
56
    
57
    if validated_city not in raw_weather_data:
58
        return {
59
            "status": "error",
60
            "error_message": f"Weather data unavailable for {validated_city}"
61
        }
62
    
63
    raw_data = raw_weather_data[validated_city]
64
    
65
    # Step 3: Process the data (traced by @opik.track)
66
    processed_data = process_weather_data(raw_data)
67
    
68
    return {
69
        "status": "success",
70
        "city": validated_city,
71
        "report": f"Weather in {validated_city}: {processed_data['conditions']}, {processed_data['temperature_celsius']}°C ({processed_data['temperature_fahrenheit']:.1f}°F). Comfort level: {processed_data['comfort_index']}.",
72
        "raw_humidity": raw_data["humidity"]
73
    }
74
75
@opik.track(name="time_calculation", tags=["time", "calculation"])
76
def calculate_time_info(timezone_name: str) -> dict:
77
    """Calculate detailed time information with internal steps."""
78
    try:
79
        tz = ZoneInfo(timezone_name)
80
        now = datetime.datetime.now(tz)
81
        
82
        # Calculate additional time info
83
        time_info = {
84
            "current_time": now,
85
            "hour_24": now.hour,
86
            "is_business_hours": 9 <= now.hour <= 17,
87
            "day_of_week": now.strftime("%A"),
88
            "week_number": now.isocalendar()[1],
89
            "is_weekend": now.weekday() >= 5
90
        }
91
        
92
        return time_info
93
    except Exception as e:
94
        return {"error": str(e)}
95
96
@opik.track(name="advanced_time_lookup", tags=["time", "timezone"])  
97
def get_advanced_time(city: str) -> dict:
98
    """Get time information with advanced calculations."""
99
    
100
    # Step 1: Validate location
101
    location_result = validate_location(city)
102
    
103
    if not location_result["is_valid"]:
104
        return {
105
            "status": "error",
106
            "error_message": f"Invalid location for time lookup: {city}"
107
        }
108
    
109
    validated_city = location_result["validated"]
110
    
111
    # Step 2: Map to timezone
112
    timezone_map = {
113
        "New York": "America/New_York",
114
        "London": "Europe/London",
115
        "Tokyo": "Asia/Tokyo"
116
    }
117
    
118
    if validated_city not in timezone_map:
119
        return {
120
            "status": "error", 
121
            "error_message": f"Timezone mapping unavailable for {validated_city}"
122
        }
123
    
124
    # Step 3: Calculate time info (traced by @opik.track)
125
    time_info = calculate_time_info(timezone_map[validated_city])
126
    
127
    if "error" in time_info:
128
        return {"status": "error", "error_message": time_info["error"]}
129
    
130
    current_time = time_info["current_time"]
131
    business_status = "during business hours" if time_info["is_business_hours"] else "outside business hours"
132
    weekend_status = "on a weekend" if time_info["is_weekend"] else "on a weekday"
133
    
134
    return {
135
        "status": "success",
136
        "city": validated_city,
137
        "report": f"Current time in {validated_city}: {current_time.strftime('%A, %B %d, %Y at %I:%M %p %Z')} ({business_status}, {weekend_status})",
138
        "metadata": {
139
            "hour_24": time_info["hour_24"],
140
            "week_number": time_info["week_number"],
141
            "is_business_hours": time_info["is_business_hours"]
142
        }
143
    }

1
# Configure Opik tracer for hybrid example
2
hybrid_tracer = OpikTracer(
3
    name="hybrid-tracing-agent",
4
    tags=["hybrid", "decorators", "callbacks", "advanced"],
5
    metadata={
6
        "environment": "development",
7
        "model": "gpt-4o",
8
        "framework": "google-adk",
9
        "example": "hybrid-tracing",
10
        "tracing_methods": ["decorators", "callbacks"]
11
    },
12
    project_name="adk-hybrid-demo"
13
)
14
15
# Create hybrid agent that combines both tracing approaches
16
hybrid_agent = LlmAgent(
17
    name="advanced_weather_time_agent",
18
    model=llm,
19
    description="Advanced agent with hybrid Opik tracing using both decorators and callbacks",
20
    instruction="""You are an advanced weather and time agent that provides detailed information with comprehensive internal processing.
21
    
22
    Your tools perform multi-step operations that are individually traced, giving detailed visibility into the processing pipeline.
23
    Use the advanced weather and time tools to provide thorough, well-processed information to users.""",
24
    tools=[get_advanced_weather, get_advanced_time],
25
    # ADK callbacks for agent-level tracing
26
    before_agent_callback=hybrid_tracer.before_agent_callback,
27
    after_agent_callback=hybrid_tracer.after_agent_callback,
28
    before_model_callback=hybrid_tracer.before_model_callback,
29
    after_model_callback=hybrid_tracer.after_model_callback,
30
    before_tool_callback=hybrid_tracer.before_tool_callback,
31
    after_tool_callback=hybrid_tracer.after_tool_callback,
32
)

1
hybrid_session_service = InMemorySessionService()
2
hybrid_runner = Runner(
3
    agent=hybrid_agent,
4
    app_name="hybrid_tracing_app",
5
    session_service=hybrid_session_service,
6
)

1
async def setup_hybrid_session():
2
    """Create a new session for the hybrid tracing example."""
3
    sess = await hybrid_session_service.create_session(
4
        app_name="hybrid_tracing_app",
5
        user_id="user_hybrid",
6
        session_id="session_hybrid_001"
7
    )
8
    return sess.id
9
10
async def call_hybrid_agent(user_msg: str, session_id: str):
11
    """Send a message to the hybrid agent and get the response."""
12
    print(f"User: {user_msg}")
13
    content = types.Content(role="user", parts=[types.Part(text=user_msg)])
14
    async for event in hybrid_runner.run_async(user_id="user_hybrid", session_id=session_id, new_message=content):
15
        if event.is_final_response():
16
            print(f"Advanced Assistant: {event.content.parts[0].text}")
17
            print()

1
# Create a session for hybrid example
2
hybrid_session_id = await setup_hybrid_session()
3
print(f"Created hybrid tracing session: {hybrid_session_id}")
4
print()
5
6
# Test weather with internal processing steps
7
print("=== Testing Advanced Weather Lookup ===")
8
await call_hybrid_agent("What's the weather like in NYC?", hybrid_session_id)
9
10
# Test time with internal calculations
11
print("=== Testing Advanced Time Lookup ===") 
12
await call_hybrid_agent("What time is it in London right now? Include business hours info.", hybrid_session_id)
13
14
# Test invalid location to see validation tracing
15
print("=== Testing Location Validation ===")
16
await call_hybrid_agent("What's the weather in InvalidCity?", hybrid_session_id)
17
18
# Test complex query that triggers multiple internal operations
19
print("=== Testing Complex Multi-Step Query ===")
20
await call_hybrid_agent("I need detailed weather and time information for Tokyo, including whether it's business hours", hybrid_session_id)