Add tracing for Gemini Live

examples/open-telemetry/langfuse/bot.py

@@ -12,7 +12,7 @@ from loguru import logger
 from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
 
 from pipecat.adapters.schemas.function_schema import FunctionSchema
-from pipecat.adapters.schemas.tools_schema import ToolsSchema
+from pipecat.adapters.schemas.tools_schema import AdapterType, ToolsSchema
 from pipecat.audio.vad.silero import SileroVADAnalyzer
 from pipecat.frames.frames import TTSSpeakFrame
 from pipecat.pipeline.pipeline import Pipeline
@@ -21,10 +21,12 @@ from pipecat.pipeline.task import PipelineParams, PipelineTask
 from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.services.cartesia.tts import CartesiaTTSService
 from pipecat.services.deepgram.stt import DeepgramSTTService
+from pipecat.services.gemini_multimodal_live.gemini import GeminiMultimodalLiveLLMService
 from pipecat.services.llm_service import FunctionCallParams
 from pipecat.services.openai.llm import OpenAILLMService
 from pipecat.transports.base_transport import BaseTransport, TransportParams
 from pipecat.transports.services.daily import DailyParams
+from pipecat.utils.time import time_now_iso8601
 from pipecat.utils.tracing.setup import setup_tracing
 
 load_dotenv(override=True)
@@ -45,11 +47,6 @@ if IS_TRACING_ENABLED:
     logger.info("OpenTelemetry tracing initialized")
 
 
-async def fetch_weather_from_api(params: FunctionCallParams):
-    await params.llm.push_frame(TTSSpeakFrame("Let me check on that."))
-    await params.result_callback({"conditions": "nice", "temperature": "75"})
-
-
 # We store functions so objects (e.g. SileroVADAnalyzer) don't get
 # instantiated. The function will be called when the desired transport gets
 # selected.
@@ -72,24 +69,29 @@ transport_params = {
 }
 
 
+async def fetch_weather_from_api(params: FunctionCallParams):
+    temperature = 75 if params.arguments["format"] == "fahrenheit" else 24
+    await params.result_callback(
+        {
+            "conditions": "nice",
+            "temperature": temperature,
+            "format": params.arguments["format"],
+            "timestamp": time_now_iso8601(),
+        }
+    )
+
+
+system_instruction = """
+You are a helpful assistant who can answer questions and use tools.
+
+You have a tool called "get_current_weather" that can be used to get the current weather. If the user asks
+for the weather, call this function.
+"""
+
+
 async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_sigint: bool):
     logger.info(f"Starting bot")
 
-    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
-
-    tts = CartesiaTTSService(
-        api_key=os.getenv("CARTESIA_API_KEY"),
-        voice_id="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
-    )
-
-    llm = OpenAILLMService(
-        api_key=os.getenv("OPENAI_API_KEY"), params=OpenAILLMService.InputParams(temperature=0.5)
-    )
-
-    # You can also register a function_name of None to get all functions
-    # sent to the same callback with an additional function_name parameter.
-    llm.register_function("get_current_weather", fetch_weather_from_api)
-
     weather_function = FunctionSchema(
         name="get_current_weather",
         description="Get the current weather",
@@ -106,25 +108,29 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si
         },
         required=["location", "format"],
     )
-    tools = ToolsSchema(standard_tools=[weather_function])
+    search_tool = {"google_search": {}}
+    tools = ToolsSchema(
+        standard_tools=[weather_function], custom_tools={AdapterType.GEMINI: [search_tool]}
+    )
 
-    messages = [
+    llm = GeminiMultimodalLiveLLMService(
-        {
+        api_key=os.getenv("GOOGLE_API_KEY"),
-            "role": "system",
+        system_instruction=system_instruction,
-            "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
+        tools=tools,
-        },
+    )
-    ]
 
-    context = OpenAILLMContext(messages, tools)
+    llm.register_function("get_current_weather", fetch_weather_from_api)
+
+    context = OpenAILLMContext(
+        [{"role": "user", "content": "Say hello."}],
+    )
     context_aggregator = llm.create_context_aggregator(context)
 
     pipeline = Pipeline(
         [
             transport.input(),
-            stt,
             context_aggregator.user(),
             llm,
-            tts,
             transport.output(),
             context_aggregator.assistant(),
         ]

src/pipecat/services/gemini_multimodal_live/gemini.py

@@ -60,6 +60,7 @@ from pipecat.services.openai.llm import (
 from pipecat.transcriptions.language import Language
 from pipecat.utils.string import match_endofsentence
 from pipecat.utils.time import time_now_iso8601
+from pipecat.utils.tracing.service_decorators import traced_gemini_live
 
 from . import events
 
@@ -803,6 +804,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
         )
         await self.send_client_event(evt)
 
+    @traced_gemini_live(operation="tool_result")
     async def _tool_result(self, tool_result_message):
         # For now we're shoving the name into the tool_call_id field, so this
         # will work until we revisit that.
@@ -827,6 +829,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
         await self._websocket.send(response_message)
         # await self._websocket.send(json.dumps({"clientContent": {"turnComplete": True}}))
 
+    @traced_gemini_live(operation="setup")
     async def _handle_evt_setup_complete(self, evt):
         # If this is our first context frame, run the LLM
         self._api_session_ready = True
@@ -873,6 +876,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
         )
         await self.push_frame(frame)
 
+    @traced_gemini_live(operation="tool_call")
     async def _handle_evt_tool_call(self, evt):
         function_calls = evt.toolCall.functionCalls
         if not function_calls:
@@ -900,6 +904,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
 
         await self.push_frame(LLMFullResponseEndFrame())
 
+    @traced_gemini_live(operation="input_transcription")
     async def _handle_evt_input_transcription(self, evt):
         """Handle the input transcription event.
 
@@ -945,6 +950,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
                 FrameDirection.UPSTREAM,
             )
 
+    @traced_gemini_live(operation="output_transcription")
     async def _handle_evt_output_transcription(self, evt):
         if not evt.serverContent.outputTranscription:
             return
@@ -960,6 +966,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
         await self.push_frame(LLMTextFrame(text=text))
         await self.push_frame(TTSTextFrame(text=text))
 
+    @traced_gemini_live(operation="usage_metadata")
     async def _handle_evt_usage_metadata(self, evt):
         if not evt.usageMetadata:
             return

src/pipecat/utils/tracing/service_attributes.py

@@ -6,7 +6,7 @@
 
 """Functions for adding attributes to OpenTelemetry spans."""
 
-from typing import TYPE_CHECKING, Any, Dict, Optional
+from typing import TYPE_CHECKING, Any, Dict, List, Optional
 
 # Import for type checking only
 if TYPE_CHECKING:
@@ -256,3 +256,108 @@ def add_llm_span_attributes(
     for key, value in kwargs.items():
         if isinstance(value, (str, int, float, bool)):
             span.set_attribute(key, value)
+
+
+def add_gemini_live_span_attributes(
+    span: "Span",
+    service_name: str,
+    model: str,
+    operation_name: str,
+    voice_id: Optional[str] = None,
+    language: Optional[str] = None,
+    modalities: Optional[str] = None,
+    settings: Optional[Dict[str, Any]] = None,
+    tools: Optional[List[Dict]] = None,
+    tools_serialized: Optional[str] = None,
+    transcript: Optional[str] = None,
+    is_input: Optional[bool] = None,
+    text_output: Optional[str] = None,
+    audio_data_size: Optional[int] = None,
+    **kwargs,
+) -> None:
+    """Add Gemini Live specific attributes to a span.
+
+    Args:
+        span: The span to add attributes to
+        service_name: Name of the service
+        model: Model name/identifier
+        operation_name: Name of the operation (setup, model_turn, tool_call, etc.)
+        voice_id: Voice identifier used for output
+        language: Language code for the session
+        modalities: Supported modalities (e.g., "AUDIO", "TEXT")
+        settings: Service configuration settings
+        tools: Available tools/functions list
+        tools_serialized: JSON-serialized tools for detailed inspection
+        transcript: Transcription text
+        is_input: Whether transcript is input (True) or output (False)
+        text_output: Text output from model
+        audio_data_size: Size of audio data in bytes
+        **kwargs: Additional attributes to add
+    """
+    # Add standard attributes
+    span.set_attribute("gen_ai.system", "gcp.gemini")
+    span.set_attribute("gen_ai.request.model", model)
+    span.set_attribute("gen_ai.operation.name", operation_name)
+    span.set_attribute("service.operation", operation_name)
+
+    # Add optional attributes
+    if voice_id:
+        span.set_attribute("voice_id", voice_id)
+
+    if language:
+        span.set_attribute("language", language)
+
+    if modalities:
+        span.set_attribute("modalities", modalities)
+
+    if transcript:
+        span.set_attribute("transcript", transcript)
+        if is_input is not None:
+            span.set_attribute("transcript.is_input", is_input)
+
+    if text_output:
+        span.set_attribute("text_output", text_output)
+
+    if audio_data_size is not None:
+        span.set_attribute("audio.data_size_bytes", audio_data_size)
+
+    if tools:
+        span.set_attribute("tools.count", len(tools))
+        span.set_attribute("tools.available", True)
+
+        # Add individual tool names for easier filtering
+        tool_names = []
+        for tool in tools:
+            if isinstance(tool, dict) and "name" in tool:
+                tool_names.append(tool["name"])
+            elif hasattr(tool, "name"):
+                tool_name = getattr(tool, "name", None)
+                if tool_name is not None:
+                    tool_names.append(tool_name)
+
+        if tool_names:
+            span.set_attribute("tools.names", ",".join(tool_names))
+
+    if tools_serialized:
+        span.set_attribute("tools.definitions", tools_serialized)
+
+    # Add settings if provided
+    if settings:
+        for key, value in settings.items():
+            if isinstance(value, (str, int, float, bool)):
+                span.set_attribute(f"settings.{key}", value)
+            elif key == "vad" and value:
+                # Handle VAD settings specially
+                if hasattr(value, "disabled") and value.disabled is not None:
+                    span.set_attribute("settings.vad.disabled", value.disabled)
+                if hasattr(value, "start_sensitivity") and value.start_sensitivity:
+                    span.set_attribute(
+                        "settings.vad.start_sensitivity", value.start_sensitivity.value
+                    )
+                if hasattr(value, "end_sensitivity") and value.end_sensitivity:
+                    span.set_attribute("settings.vad.end_sensitivity", value.end_sensitivity.value)
+
+    # Add any additional keyword arguments as attributes
+    for key, value in kwargs.items():
+        if isinstance(value, (str, int, float, bool)):
+            span.set_attribute(key, value)

src/pipecat/utils/tracing/service_decorators.py

@@ -24,6 +24,7 @@ if TYPE_CHECKING:
     from opentelemetry import trace
 
 from pipecat.utils.tracing.service_attributes import (
+    add_gemini_live_span_attributes,
     add_llm_span_attributes,
     add_stt_span_attributes,
     add_tts_span_attributes,
@@ -477,3 +478,310 @@ def traced_llm(func: Optional[Callable] = None, *, name: Optional[str] = None) -
     if func is not None:
         return decorator(func)
     return decorator
+
+
+def traced_gemini_live(operation: str) -> Callable:
+    """Traces Gemini Live service methods with operation-specific attributes.
+
+    This decorator automatically captures relevant information based on the operation type:
+    - setup_complete: Configuration, tools definitions, and system instructions
+    - model_turn: Text and audio output
+    - tool_call: Function call information
+    - tool_result: Function execution results
+    - input_transcription: User transcription
+    - output_transcription: Assistant transcription
+    - usage_metadata: Token usage metrics
+
+    Args:
+        operation: The operation name (matches the event type being handled)
+
+    Returns:
+        Wrapped method with Gemini Live specific tracing.
+    """
+    if not is_tracing_available():
+        return _noop_decorator
+
+    def decorator(func):
+        @functools.wraps(func)
+        async def wrapper(self, *args, **kwargs):
+            try:
+                if not is_tracing_available():
+                    return await func(self, *args, **kwargs)
+
+                service_class_name = self.__class__.__name__
+                span_name = f"{operation}"
+
+                # Get the parent context - turn context if available, otherwise service context
+                turn_context = get_current_turn_context()
+                parent_context = turn_context or _get_parent_service_context(self)
+
+                # Create a new span as child of the turn span or service span
+                tracer = trace.get_tracer("pipecat")
+                with tracer.start_as_current_span(
+                    span_name, context=parent_context
+                ) as current_span:
+                    try:
+                        # Base service attributes
+                        model_name = getattr(
+                            self, "model_name", getattr(self, "_model_name", "unknown")
+                        )
+                        voice_id = getattr(self, "_voice_id", None)
+                        language_code = getattr(self, "_language_code", None)
+                        settings = getattr(self, "_settings", {})
+
+                        # Get modalities if available
+                        modalities = None
+                        if hasattr(self, "_settings") and "modalities" in self._settings:
+                            modality_obj = self._settings["modalities"]
+                            if hasattr(modality_obj, "value"):
+                                modalities = modality_obj.value
+                            else:
+                                modalities = str(modality_obj)
+
+                        # Operation-specific attribute collection
+                        operation_attrs = {}
+
+                        if operation == "setup":
+                            # Capture detailed tool information
+                            tools = getattr(self, "_tools", None)
+                            if tools:
+                                # Handle different tool formats
+                                tools_list = []
+                                tools_serialized = None
+
+                                try:
+                                    if hasattr(tools, "standard_tools"):
+                                        # ToolsSchema object
+                                        tools_list = tools.standard_tools
+                                        # Serialize the tools for detailed inspection
+                                        tools_serialized = json.dumps(
+                                            [
+                                                {
+                                                    "name": tool.name
+                                                    if hasattr(tool, "name")
+                                                    else tool.get("name", "unknown"),
+                                                    "description": tool.description
+                                                    if hasattr(tool, "description")
+                                                    else tool.get("description", ""),
+                                                    "properties": tool.properties
+                                                    if hasattr(tool, "properties")
+                                                    else tool.get("properties", {}),
+                                                    "required": tool.required
+                                                    if hasattr(tool, "required")
+                                                    else tool.get("required", []),
+                                                }
+                                                for tool in tools_list
+                                            ]
+                                        )
+                                    elif isinstance(tools, list):
+                                        # List of tool dictionaries or objects
+                                        tools_list = tools
+                                        tools_serialized = json.dumps(
+                                            [
+                                                {
+                                                    "name": tool.get("name", "unknown")
+                                                    if isinstance(tool, dict)
+                                                    else getattr(tool, "name", "unknown"),
+                                                    "description": tool.get("description", "")
+                                                    if isinstance(tool, dict)
+                                                    else getattr(tool, "description", ""),
+                                                    "properties": tool.get("properties", {})
+                                                    if isinstance(tool, dict)
+                                                    else getattr(tool, "properties", {}),
+                                                    "required": tool.get("required", [])
+                                                    if isinstance(tool, dict)
+                                                    else getattr(tool, "required", []),
+                                                }
+                                                for tool in tools_list
+                                            ]
+                                        )
+
+                                    if tools_list:
+                                        operation_attrs["tools"] = tools_list
+                                        operation_attrs["tools_serialized"] = tools_serialized
+
+                                except Exception as e:
+                                    logging.warning(f"Error serializing tools for tracing: {e}")
+                                    # Fallback to basic tool count
+                                    if tools_list:
+                                        operation_attrs["tools"] = tools_list
+
+                            # Capture system instruction information
+                            system_instruction = getattr(self, "_system_instruction", None)
+                            if system_instruction:
+                                operation_attrs["system_instruction"] = system_instruction[
+                                    :500
+                                ]  # Truncate if very long
+
+                            # Capture context system instructions if available
+                            if hasattr(self, "_context") and self._context:
+                                try:
+                                    context_system = self._context.extract_system_instructions()
+                                    if context_system:
+                                        operation_attrs["context_system_instruction"] = (
+                                            context_system[:500]
+                                        )  # Truncate if very long
+                                except Exception as e:
+                                    logging.warning(
+                                        f"Error extracting context system instructions: {e}"
+                                    )
+
+                        elif operation == "input_transcription" and args:
+                            # Extract input transcription
+                            evt = args[0] if args else None
+                            if (
+                                evt
+                                and hasattr(evt, "serverContent")
+                                and evt.serverContent.inputTranscription
+                            ):
+                                text = evt.serverContent.inputTranscription.text
+                                if text:
+                                    operation_attrs["transcript"] = text
+                                    operation_attrs["is_input"] = True
+
+                        elif operation == "output_transcription" and args:
+                            # Extract output transcription
+                            evt = args[0] if args else None
+                            if (
+                                evt
+                                and hasattr(evt, "serverContent")
+                                and evt.serverContent.outputTranscription
+                            ):
+                                text = evt.serverContent.outputTranscription.text
+                                if text:
+                                    operation_attrs["transcript"] = text
+                                    operation_attrs["is_input"] = False
+
+                        elif operation == "tool_call" and args:
+                            # Extract tool call information
+                            evt = args[0] if args else None
+                            if evt and hasattr(evt, "toolCall") and evt.toolCall.functionCalls:
+                                function_calls = evt.toolCall.functionCalls
+                                if function_calls:
+                                    # Add information about the first function call
+                                    call = function_calls[0]
+                                    operation_attrs["tool.function_name"] = call.name
+                                    operation_attrs["tool.call_id"] = call.id
+                                    operation_attrs["tool.calls_count"] = len(function_calls)
+
+                                    # Add all function names being called
+                                    all_function_names = [c.name for c in function_calls]
+                                    operation_attrs["tool.all_function_names"] = ",".join(
+                                        all_function_names
+                                    )
+
+                                    # Add arguments for the first call (truncated if too long)
+                                    try:
+                                        args_str = json.dumps(call.args) if call.args else "{}"
+                                        if len(args_str) > 1000:
+                                            args_str = args_str[:1000] + "..."
+                                        operation_attrs["tool.arguments"] = args_str
+                                    except Exception:
+                                        operation_attrs["tool.arguments"] = str(call.args)[:1000]
+
+                        elif operation == "tool_result" and args:
+                            # Extract tool result information
+                            tool_result_message = args[0] if args else None
+                            if tool_result_message and isinstance(tool_result_message, dict):
+                                # Extract the tool call information
+                                tool_call_id = tool_result_message.get("tool_call_id")
+                                tool_call_name = tool_result_message.get("tool_call_name")
+                                result_content = tool_result_message.get("content")
+
+                                if tool_call_id:
+                                    operation_attrs["tool.call_id"] = tool_call_id
+                                if tool_call_name:
+                                    operation_attrs["tool.function_name"] = tool_call_name
+
+                                # Parse and capture the result
+                                if result_content:
+                                    try:
+                                        result = json.loads(result_content)
+                                        # Serialize the result, truncating if too long
+                                        result_str = json.dumps(result)
+                                        if len(result_str) > 2000:  # Larger limit for results
+                                            result_str = result_str[:2000] + "..."
+                                        operation_attrs["tool.result"] = result_str
+
+                                        # Add result status/success indicator if present
+                                        if isinstance(result, dict):
+                                            if "error" in result:
+                                                operation_attrs["tool.result_status"] = "error"
+                                            elif "success" in result:
+                                                operation_attrs["tool.result_status"] = "success"
+                                            else:
+                                                operation_attrs["tool.result_status"] = "completed"
+
+                                    except json.JSONDecodeError as e:
+                                        operation_attrs["tool.result"] = (
+                                            f"Invalid JSON: {str(result_content)[:500]}"
+                                        )
+                                        operation_attrs["tool.result_status"] = "parse_error"
+                                    except Exception as e:
+                                        operation_attrs["tool.result"] = (
+                                            f"Error processing result: {str(e)}"
+                                        )
+                                        operation_attrs["tool.result_status"] = "processing_error"
+
+                        elif operation == "usage_metadata" and args:
+                            # Token usage will be handled by the original start_llm_usage_metrics method
+                            evt = args[0] if args else None
+                            if evt and hasattr(evt, "usageMetadata") and evt.usageMetadata:
+                                usage = evt.usageMetadata
+                                if hasattr(usage, "promptTokenCount"):
+                                    operation_attrs["tokens.prompt"] = usage.promptTokenCount or 0
+                                if hasattr(usage, "responseTokenCount"):
+                                    operation_attrs["tokens.completion"] = (
+                                        usage.responseTokenCount or 0
+                                    )
+                                if hasattr(usage, "totalTokenCount"):
+                                    operation_attrs["tokens.total"] = usage.totalTokenCount or 0
+
+                        # Add all attributes to the span
+                        add_gemini_live_span_attributes(
+                            span=current_span,
+                            service_name=service_class_name,
+                            model=model_name,
+                            operation_name=operation,
+                            voice_id=voice_id,
+                            language=language_code,
+                            modalities=modalities,
+                            settings=settings,
+                            **operation_attrs,
+                        )
+
+                        # For usage_metadata operation, also handle token usage metrics
+                        if operation == "usage_metadata" and hasattr(
+                            self, "start_llm_usage_metrics"
+                        ):
+                            evt = args[0] if args else None
+                            if evt and hasattr(evt, "usageMetadata") and evt.usageMetadata:
+                                usage = evt.usageMetadata
+                                # Create LLMTokenUsage object
+                                from pipecat.metrics.metrics import LLMTokenUsage
+
+                                tokens = LLMTokenUsage(
+                                    prompt_tokens=usage.promptTokenCount or 0,
+                                    completion_tokens=usage.responseTokenCount or 0,
+                                    total_tokens=usage.totalTokenCount or 0,
+                                )
+                                _add_token_usage_to_span(current_span, tokens)
+
+                        # Run the original function
+                        result = await func(self, *args, **kwargs)
+
+                        return result
+
+                    except Exception as e:
+                        current_span.record_exception(e)
+                        current_span.set_status(trace.Status(trace.StatusCode.ERROR, str(e)))
+                        raise
+
+            except Exception as e:
+                logging.error(f"Error in Gemini Live tracing (continuing without tracing): {e}")
+                # If tracing fails, fall back to the original function
+                return await func(self, *args, **kwargs)
+
+        return wrapper
+
+    return decorator