Use this file to discover all available pages before exploring further.
The galileo-adk package provides native observability for Google ADK agents. It automatically traces agent runs, LLM calls, and tool executions with minimal setup.
The simplest way to add Galileo observability to your Google ADK application is using the GalileoADKPlugin. This plugin attaches to the ADK Runner and automatically captures all agent, LLM, and tool events.
import asynciofrom galileo_adk import GalileoADKPluginfrom google.adk.runners import Runnerfrom google.adk.agents import LlmAgentfrom google.genai import typesasync def main(): # Create the Galileo ADK plugin plugin = GalileoADKPlugin(project="my-project", log_stream="production") # Create your agent agent = LlmAgent( name="assistant", model="gemini-2.0-flash", instruction="You are helpful." ) # Pass the plugin to the Runner runner = Runner(agent=agent, plugins=[plugin]) # Run the agent message = types.Content( parts=[types.Part(text="Hello! What can you help me with?")] ) async for event in runner.run_async( user_id="user-123", session_id="session-456", new_message=message ): if event.is_final_response(): print(event.content.parts[0].text)if __name__ == "__main__": # Set environment variables: GALILEO_API_KEY, GOOGLE_API_KEY asyncio.run(main())
Once the plugin is configured, all agent runs are automatically logged to Galileo, including:
All traces with the same ADK session_id are automatically grouped into a Galileo session. This enables conversation-level tracking across multiple interactions:
import asynciofrom galileo_adk import GalileoADKPluginfrom google.adk.runners import Runnerfrom google.adk.agents import LlmAgentfrom google.genai import typesasync def main(): plugin = GalileoADKPlugin(project="my-project", log_stream="production") agent = LlmAgent( name="assistant", model="gemini-2.0-flash", instruction="You are helpful." ) runner = Runner(agent=agent, plugins=[plugin]) # All traces in this conversation are grouped together session_id = "conversation-abc" # First message message1 = types.Content( parts=[types.Part(text="Hello! What's the capital of France?")] ) async for event in runner.run_async( user_id="user-123", session_id=session_id, new_message=message1 ): if event.is_final_response(): print(f"Response 1: {event.content.parts[0].text}") # Follow-up in same session message2 = types.Content( parts=[types.Part(text="What about Germany?")] ) async for event in runner.run_async( user_id="user-123", session_id=session_id, new_message=message2 ): if event.is_final_response(): print(f"Response 2: {event.content.parts[0].text}")if __name__ == "__main__": # Set environment variables: GALILEO_API_KEY, GOOGLE_API_KEY asyncio.run(main())
Attach custom metadata to traces using ADK’s native RunConfig.custom_metadata. Metadata is propagated to all spans (agent, LLM, tool) within the invocation:
import asynciofrom galileo_adk import GalileoADKPluginfrom google.adk.runners import Runnerfrom google.adk.agents import LlmAgentfrom google.adk.agents.run_config import RunConfigfrom google.genai import typesasync def main(): plugin = GalileoADKPlugin(project="my-project", log_stream="production") agent = LlmAgent( name="assistant", model="gemini-2.0-flash", instruction="You are helpful." ) runner = Runner(agent=agent, plugins=[plugin]) # Attach custom metadata to the trace run_config = RunConfig( custom_metadata={ "user_tier": "premium", "conversation_id": "conv-abc", "turn": 1, "experiment_group": "A", } ) message = types.Content( parts=[types.Part(text="Hello! Tell me a fun fact.")] ) async for event in runner.run_async( user_id="user-123", session_id="session-456", new_message=message, run_config=run_config, # Pass the run config with metadata ): if event.is_final_response(): print(event.content.parts[0].text)if __name__ == "__main__": # Set environment variables: GALILEO_API_KEY, GOOGLE_API_KEY asyncio.run(main())
For granular control over which callbacks to use, you can attach them directly to your agent instead of using the plugin. This is useful when you need to customize behavior for specific agents:
import asynciofrom galileo_adk import GalileoADKCallbackfrom google.adk.runners import Runnerfrom google.adk.agents import LlmAgentfrom google.genai import typesasync def main(): # Create the Galileo ADK callback callback = GalileoADKCallback(project="my-project", log_stream="production") # Attach callbacks directly to your agent agent = LlmAgent( name="assistant", model="gemini-2.0-flash", instruction="You are helpful.", before_agent_callback=callback.before_agent_callback, after_agent_callback=callback.after_agent_callback, before_model_callback=callback.before_model_callback, after_model_callback=callback.after_model_callback, before_tool_callback=callback.before_tool_callback, after_tool_callback=callback.after_tool_callback, ) # Create the runner without the plugin runner = Runner(agent=agent) message = types.Content( parts=[types.Part(text="Hello! How are you?")] ) async for event in runner.run_async( user_id="user-123", session_id="session-456", new_message=message ): if event.is_final_response(): print(event.content.parts[0].text)if __name__ == "__main__": # Set environment variables: GALILEO_API_KEY, GOOGLE_API_KEY asyncio.run(main())
By default, all FunctionTool calls are logged as tool spans. To log a retriever function as a retriever span (enabling RAG quality metrics in Galileo), decorate it with @galileo_retriever:
from galileo_adk import galileo_retrieverfrom google.adk.tools import FunctionTool@galileo_retrieverdef search_docs(query: str) -> str: """Search the knowledge base.""" # Your retrieval logic here results = my_vector_db.search(query) return "\n".join(r["content"] for r in results)# Wrap in a FunctionTool - Galileo will log this as a retriever spantool = FunctionTool(search_docs)
This enables RAG-specific metrics like chunk attribution and context relevance to be calculated for your retrieval operations.
Recommendation: Use GalileoADKPlugin for most applications. Use GalileoADKCallback only when you need fine-grained control over individual agent callbacks.