From 7b1a937d4c0176a0290be832738620cb553470b3 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 29 May 2025 18:40:37 -0400 Subject: [PATCH] Add tracing for Gemini Live --- examples/open-telemetry/langfuse/bot.py | 68 ++-- .../services/gemini_multimodal_live/gemini.py | 7 + .../utils/tracing/service_attributes.py | 107 +++++- .../utils/tracing/service_decorators.py | 308 ++++++++++++++++++ 4 files changed, 458 insertions(+), 32 deletions(-) diff --git a/examples/open-telemetry/langfuse/bot.py b/examples/open-telemetry/langfuse/bot.py index 22ff399a0..872a4d979 100644 --- a/examples/open-telemetry/langfuse/bot.py +++ b/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 = [ - { - "role": "system", - "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.", - }, - ] + llm = GeminiMultimodalLiveLLMService( + api_key=os.getenv("GOOGLE_API_KEY"), + system_instruction=system_instruction, + 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(), ] diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index 25377e183..696d4acbf 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/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 diff --git a/src/pipecat/utils/tracing/service_attributes.py b/src/pipecat/utils/tracing/service_attributes.py index 619dfbf11..bb7dc8761 100644 --- a/src/pipecat/utils/tracing/service_attributes.py +++ b/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) diff --git a/src/pipecat/utils/tracing/service_decorators.py b/src/pipecat/utils/tracing/service_decorators.py index 4341d308c..4d6cac694 100644 --- a/src/pipecat/utils/tracing/service_decorators.py +++ b/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