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
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
# Configure Opik tracer
41
opik_tracer = OpikTracer(
42
    name="basic-weather-agent",
43
    tags=["basic", "weather", "time", "single-agent"],
44
    metadata={
45
        "environment": "development",
46
        "model": "gpt-4o",
47
        "framework": "google-adk",
48
        "example": "basic"
49
    },
50
    project_name="adk-basic-demo"
51
)
52
53
# Initialize LiteLLM with OpenAI gpt-4o
54
llm = LiteLlm(model="openai/gpt-4o")
55
56
# Create the basic agent with Opik callbacks
57
basic_agent = LlmAgent(
58
    name="weather_time_agent",
59
    model=llm,
60
    description="Agent for answering time & weather questions",
61
    instruction="Answer questions about the time or weather in a city. Be helpful and provide clear information.",
62
    tools=[get_weather, get_current_time],
63
    before_agent_callback=opik_tracer.before_agent_callback,
64
    after_agent_callback=opik_tracer.after_agent_callback,
65
    before_model_callback=opik_tracer.before_model_callback,
66
    after_model_callback=opik_tracer.after_model_callback,
67
    before_tool_callback=opik_tracer.before_tool_callback,
68
    after_tool_callback=opik_tracer.after_tool_callback,
69
)

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 the experimental recursive tracking feature to instrument all agents at once
132
from opik.integrations.adk import track_adk_agent_recursive
133
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
    # ADK callbacks for agent-level tracing
102
    before_agent_callback=hybrid_tracer.before_agent_callback,
103
    after_agent_callback=hybrid_tracer.after_agent_callback,
104
    before_model_callback=hybrid_tracer.before_model_callback,
105
    after_model_callback=hybrid_tracer.after_model_callback,
106
    before_tool_callback=hybrid_tracer.before_tool_callback,
107
    after_tool_callback=hybrid_tracer.after_tool_callback,
108
)

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()