Observability for Google Agent Development Kit (Python) with Opik

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

1
import datetime
2
from zoneinfo import ZoneInfo
3
4
from google.adk.agents import LlmAgent
5
from google.adk.models.lite_llm import LiteLlm
6
from opik.integrations.adk import OpikTracer, track_adk_agent_recursive
7
8
def get_weather(city: str) -> dict:
9
    """Get weather information for a city."""
10
    if city.lower() == "new york":
11
        return {
12
            "status": "success",
13
            "report": "The weather in New York is sunny with a temperature of 25 °C (77 °F).",
14
        }
15
    elif city.lower() == "london":
16
        return {
17
            "status": "success",
18
            "report": "The weather in London is cloudy with a temperature of 18 °C (64 °F).",
19
        }
20
    return {"status": "error", "error_message": f"Weather info for '{city}' is unavailable."}
21
22
def get_current_time(city: str) -> dict:
23
    """Get current time for a city."""
24
    if city.lower() == "new york":
25
        tz = ZoneInfo("America/New_York")
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
    elif city.lower() == "london":
32
        tz = ZoneInfo("Europe/London")
33
        now = datetime.datetime.now(tz)
34
        return {
35
            "status": "success",
36
            "report": now.strftime(f"The current time in {city} is %Y-%m-%d %H:%M:%S %Z%z."),
37
        }
38
    return {"status": "error", "error_message": f"No timezone info for '{city}'."}
39
40
# Initialize LiteLLM with OpenAI gpt-4o
41
llm = LiteLlm(model="openai/gpt-4o")
42
43
# Create the basic agent
44
basic_agent = LlmAgent(
45
    name="weather_time_agent",
46
    model=llm,
47
    description="Agent for answering time & weather questions",
48
    instruction="Answer questions about the time or weather in a city. Be helpful and provide clear information.",
49
    tools=[get_weather, get_current_time],
50
)
51
52
# Configure Opik tracer
53
opik_tracer = OpikTracer(
54
    name="basic-weather-agent",
55
    tags=["basic", "weather", "time", "single-agent"],
56
    metadata={
57
        "environment": "development",
58
        "model": "gpt-4o",
59
        "framework": "google-adk",
60
        "example": "basic"
61
    },
62
    project_name="adk-basic-demo"
63
)
64
65
# Instrument the agent with a single function call - this is the recommended approach
66
track_adk_agent_recursive(basic_agent, opik_tracer)

1
# Configure Opik tracer (same as before)
2
opik_tracer = OpikTracer(
3
    name="basic-weather-agent",
4
    tags=["basic", "weather", "time", "single-agent"],
5
    metadata={
6
        "environment": "development",
7
        "model": "gpt-4o",
8
        "framework": "google-adk",
9
        "example": "basic"
10
    },
11
    project_name="adk-basic-demo"
12
)
13
14
# Create the agent with explicit callback configuration
15
basic_agent = LlmAgent(
16
    name="weather_time_agent",
17
    model=llm,
18
    description="Agent for answering time & weather questions",
19
    instruction="Answer questions about the time or weather in a city. Be helpful and provide clear information.",
20
    tools=[get_weather, get_current_time],
21
    before_agent_callback=opik_tracer.before_agent_callback,
22
    after_agent_callback=opik_tracer.after_agent_callback,
23
    before_model_callback=opik_tracer.before_model_callback,
24
    after_model_callback=opik_tracer.after_model_callback,
25
    before_tool_callback=opik_tracer.before_tool_callback,
26
    after_tool_callback=opik_tracer.after_tool_callback,
27
)

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."}
72
73
# Weather specialist agent (no Opik callbacks needed)
74
weather_agent = LlmAgent(
75
    name="weather_specialist",
76
    model=llm,
77
    description="Specialized agent for detailed weather information",
78
    instruction="Provide comprehensive weather information including current conditions and forecasts. Be detailed and informative.",
79
    tools=[get_detailed_weather]
80
)
81
82
# Time specialist agent (no Opik callbacks needed)
83
time_agent = LlmAgent(
84
    name="time_specialist",
85
    model=llm,
86
    description="Specialized agent for world time information",
87
    instruction="Provide accurate time information for cities around the world. Include day of week and full date.",
88
    tools=[get_world_time]
89
)
90
91
# Travel specialist agent (no Opik callbacks needed)
92
travel_agent = LlmAgent(
93
    name="travel_specialist",
94
    model=llm,
95
    description="Specialized agent for travel information",
96
    instruction="Provide helpful travel information including flight times and time zone differences.",
97
    tools=[get_travel_info]
98
)
99
100
# Configure Opik tracer for multi-agent example
101
multi_agent_tracer = OpikTracer(
102
    name="multi-agent-coordinator",
103
    tags=["multi-agent", "coordinator", "weather", "time", "travel"],
104
    metadata={
105
        "environment": "development",
106
        "model": "gpt-4o",
107
        "framework": "google-adk",
108
        "example": "multi-agent",
109
        "agent_count": 4
110
    },
111
    project_name="adk-multi-agent-demo"
112
)
113
114
# Coordinator agent with sub-agents
115
coordinator_agent = LlmAgent(
116
    name="travel_coordinator",
117
    model=llm,
118
    description="Coordinator agent that delegates to specialized agents for weather, time, and travel information",
119
    instruction="""You are a travel coordinator that helps users with weather, time, and travel information.
120
121
    You have access to three specialized agents:
122
    - weather_specialist: For detailed weather information
123
    - time_specialist: For world time information
124
    - travel_specialist: For travel planning information
125
126
    Delegate appropriate queries to the right specialist agents and compile comprehensive responses for the user.""",
127
    tools=[],  # No direct tools, delegates to sub-agents
128
    sub_agents=[weather_agent, time_agent, travel_agent],
129
)
130
131
# Use track_adk_agent_recursive to instrument all agents at once
132
# This automatically adds callbacks to the coordinator and ALL sub-agents
133
from opik.integrations.adk import track_adk_agent_recursive
134
track_adk_agent_recursive(coordinator_agent, multi_agent_tracer)

1
from opik import track
2
3
@track(name="weather_data_processing", tags=["data-processing", "weather"])
4
def process_weather_data(raw_data: dict) -> dict:
5
    """Process raw weather data with additional computations."""
6
    # Simulate some data processing steps that we want to trace separately
7
    processed = {
8
        "temperature_celsius": raw_data.get("temp_c", 0),
9
        "temperature_fahrenheit": raw_data.get("temp_c", 0) * 9/5 + 32,
10
        "conditions": raw_data.get("condition", "unknown"),
11
        "comfort_index": "comfortable" if 18 <= raw_data.get("temp_c", 0) <= 25 else "less comfortable"
12
    }
13
    return processed
14
15
@track(name="location_validation", tags=["validation", "location"])
16
def validate_location(city: str) -> dict:
17
    """Validate and normalize city names."""
18
    # Simulate location validation logic that we want to trace
19
    normalized_cities = {
20
        "nyc": "New York",
21
        "ny": "New York",
22
        "new york city": "New York",
23
        "london uk": "London",
24
        "london england": "London",
25
        "tokyo japan": "Tokyo"
26
    }
27
28
    city_lower = city.lower().strip()
29
    validated_city = normalized_cities.get(city_lower, city.title())
30
31
    return {
32
        "original": city,
33
        "validated": validated_city,
34
        "is_valid": city_lower in ["new york", "london", "tokyo"] or city_lower in normalized_cities
35
    }
36
37
@track(name="advanced_weather_lookup", tags=["weather", "api-simulation"])
38
def get_advanced_weather(city: str) -> dict:
39
    """Get weather with internal processing steps tracked by Opik decorators."""
40
41
    # Step 1: Validate location (traced by @opik.track)
42
    location_result = validate_location(city)
43
44
    if not location_result["is_valid"]:
45
        return {
46
            "status": "error",
47
            "error_message": f"Invalid location: {city}"
48
        }
49
50
    validated_city = location_result["validated"]
51
52
    # Step 2: Get raw weather data (simulated)
53
    raw_weather_data = {
54
        "New York": {"temp_c": 25, "condition": "sunny", "humidity": 65},
55
        "London": {"temp_c": 18, "condition": "cloudy", "humidity": 78},
56
        "Tokyo": {"temp_c": 22, "condition": "partly cloudy", "humidity": 70}
57
    }
58
59
    if validated_city not in raw_weather_data:
60
        return {
61
            "status": "error",
62
            "error_message": f"Weather data unavailable for {validated_city}"
63
        }
64
65
    raw_data = raw_weather_data[validated_city]
66
67
    # Step 3: Process the data (traced by @opik.track)
68
    processed_data = process_weather_data(raw_data)
69
70
    return {
71
        "status": "success",
72
        "city": validated_city,
73
        "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']}.",
74
        "raw_humidity": raw_data["humidity"]
75
    }
76
77
# Configure Opik tracer for hybrid example
78
hybrid_tracer = OpikTracer(
79
    name="hybrid-tracing-agent",
80
    tags=["hybrid", "decorators", "callbacks", "advanced"],
81
    metadata={
82
        "environment": "development",
83
        "model": "gpt-4o",
84
        "framework": "google-adk",
85
        "example": "hybrid-tracing",
86
        "tracing_methods": ["decorators", "callbacks"]
87
    },
88
    project_name="adk-hybrid-demo"
89
)
90
91
# Create hybrid agent that combines both tracing approaches
92
hybrid_agent = LlmAgent(
93
    name="advanced_weather_time_agent",
94
    model=llm,
95
    description="Advanced agent with hybrid Opik tracing using both decorators and callbacks",
96
    instruction="""You are an advanced weather and time agent that provides detailed information with comprehensive internal processing.
97
98
    Your tools perform multi-step operations that are individually traced, giving detailed visibility into the processing pipeline.
99
    Use the advanced weather and time tools to provide thorough, well-processed information to users.""",
100
    tools=[get_advanced_weather],
101
)
102
103
# Instrument the agent with track_adk_agent_recursive
104
# The @opik.track decorators in your tools will automatically create child spans
105
from opik.integrations.adk import track_adk_agent_recursive
106
track_adk_agent_recursive(hybrid_agent, hybrid_tracer)

1
from opik.integrations.adk import OpikTracer
2
from google.adk import sessions as adk_sessions, runners as adk_runners
3
4
# ADK session management
5
session_service = adk_sessions.InMemorySessionService()
6
session = session_service.create_session_sync(
7
    app_name="my_app",
8
    user_id="user_123",
9
    session_id="conversation_456"
10
)
11
12
opik_tracer = OpikTracer()
13
runner = adk_runners.Runner(
14
    agent=your_agent,
15
    app_name="my_app",
16
    session_service=session_service
17
)
18
19
# All traces will be automatically grouped by session_id as thread_id

1
async for event in runner.run_async(user_id=user_id, session_id=session_id, new_message=content):
2
    if event.is_final_response():
3
        ...
4
        break  # Stop processing events once the final response is found

1
from opik.integrations.adk import OpikTracer
2
3
opik_tracer = OpikTracer()
4
5
# Your ADK agent execution code here...
6
7
# Ensure all traces are sent before script exits
8
opik_tracer.flush()

1
import datetime
2
import uuid
3
from typing import Iterator, Optional
4
from zoneinfo import ZoneInfo
5
6
from google.adk import Agent, Runner
7
from google.adk.agents import LlmAgent
8
from google.adk.models.lite_llm import LiteLlm
9
from google.adk.sessions import InMemorySessionService
10
from google.genai import types as genai_types
11
from google.adk import events as adk_events
12
13
import opik
14
from opik.opik_context import update_current_trace
15
from opik.integrations.adk import OpikTracer, track_adk_agent_recursive
16
17
# Create prompt
18
system_prompt = opik.Prompt(
19
    name="system-prompt",
20
    prompt="You are a helpful assistant that provides accurate and concise answers."
21
)
22
23
# Get prompt from the Prompt library
24
client = opik.Opik()
25
user_prompt = client.get_prompt(name="user-prompt")
26
27
def get_weather(city: str) -> dict:
28
    """Get weather information for a city."""
29
    # Add prompts to the current trace
30
    update_current_trace(
31
        prompts=[system_prompt, user_prompt]
32
    )
33
34
    if city.lower() == "new york":
35
        return {
36
            "status": "success",
37
            "report": "The weather in New York is sunny with a temperature of 25 °C (77 °F).",
38
        }
39
    elif city.lower() == "london":
40
        return {
41
            "status": "success",
42
            "report": "The weather in London is cloudy with a temperature of 18 °C (64 °F).",
43
        }
44
    return {"status": "error", "error_message": f"Weather info for '{city}' is unavailable."}
45
46
def get_current_time(city: str) -> dict:
47
    """Get current time for a city."""
48
    if city.lower() == "new york":
49
        tz = ZoneInfo("America/New_York")
50
        now = datetime.datetime.now(tz)
51
        return {
52
            "status": "success",
53
            "report": now.strftime(f"The current time in {city} is %Y-%m-%d %H:%M:%S %Z%z."),
54
        }
55
    elif city.lower() == "london":
56
        tz = ZoneInfo("Europe/London")
57
        now = datetime.datetime.now(tz)
58
        return {
59
            "status": "success",
60
            "report": now.strftime(f"The current time in {city} is %Y-%m-%d %H:%M:%S %Z%z."),
61
        }
62
    return {"status": "error", "error_message": f"No timezone info for '{city}'."}
63
64
# Initialize LiteLLM with OpenAI gpt-4o
65
llm = LiteLlm(model="openai/gpt-4o")
66
67
# Create the basic agent
68
basic_agent = LlmAgent(
69
    name="weather_time_agent",
70
    model=llm,
71
    description="Agent for answering time & weather questions",
72
    instruction="Answer questions about the time or weather in a city. Be helpful and provide clear information.",
73
    tools=[get_weather, get_current_time],
74
)
75
76
# Configure Opik tracer
77
opik_tracer = OpikTracer(
78
    name="basic-weather-agent",
79
    tags=["basic", "weather", "time", "single-agent"],
80
    metadata={
81
        "environment": "development",
82
        "model": "gpt-4o",
83
        "framework": "google-adk",
84
        "example": "basic"
85
    },
86
    project_name="adk-basic-demo"
87
)
88
89
# Instrument the agent with a single function call - this is the recommended approach
90
track_adk_agent_recursive(basic_agent, opik_tracer)