openinference-instrumentation-google-adk

import os import asyncio from google.adk.agents import Agent from google.adk.runners import InMemoryRunner from google.genai import types from opentelemetry import trace from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import SimpleSpanProcessor, ConsoleSpanExporter from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter from openinference.instrumentation.google_adk import GoogleADKInstrumentor # Set your Google API Key (replace with your actual key or use env var) os.environ["GOOGLE_API_KEY"] = os.environ.get("GOOGLE_API_KEY", "YOUR_GEMINI_API_KEY_HERE") # 1. Configure OpenTelemetry Tracer Provider # For simplicity, using ConsoleSpanExporter here. In a real app, use OTLPSpanExporter # to send traces to an observability backend like Arize Phoenix, SigNoz, or Langfuse. resource = Resource.create({"service.name": "my-adk-agent"}) tracer_provider = TracerProvider(resource=resource) # To export to console (for quick verification): tracer_provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter())) # To export to an OTLP endpoint (e.g., a collector, Phoenix, SigNoz, Langfuse): # endpoint = os.environ.get("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", "http://localhost:4317") # tracer_provider.add_span_processor(SimpleSpanProcessor(OTLPSpanExporter(endpoint))) trace.set_tracer_provider(tracer_provider) # 2. Instrument Google ADK GoogleADKInstrumentor().instrument(tracer_provider=tracer_provider) # 3. Define a sample Google ADK Agent def get_current_time(city: str) -> dict: """Returns the current time for a specified city.""" # This is a dummy tool, replace with actual logic return {"status": "success", "time": f"The current time in {city} is 12:00 PM."} agent = Agent( name="TimeAgent", model="gemini-pro", # or "gemini-2.0-flash-exp" as per docs description="Agent to provide current time in a city.", instruction="You must use the get_current_time tool to find the answer.", tools=[get_current_time] ) async def main(): runner = InMemoryRunner(agent=agent) print("Agent started. Ask a question about time.") user_prompt = "What time is it in London?" print(f"User: {user_prompt}") response = await runner.run(types.Message(text=user_prompt)) print(f"Agent: {response.text}") user_prompt_2 = "What about New York?" print(f"User: {user_prompt_2}") response_2 = await runner.run(types.Message(text=user_prompt_2)) print(f"Agent: {response_2.text}") if __name__ == "__main__": asyncio.run(main())