From 7b1a937d4c0176a0290be832738620cb553470b3 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 29 May 2025 18:40:37 -0400 Subject: [PATCH 1/5] 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 From ec39e794d3b64c705ad4d19b55d6439a64d2d94f Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 29 May 2025 18:57:49 -0400 Subject: [PATCH 2/5] _handle_transcription --- .../services/gemini_multimodal_live/gemini.py | 38 +++++++++++++++---- 1 file changed, 31 insertions(+), 7 deletions(-) diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index 696d4acbf..1b25bb80e 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -60,7 +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 pipecat.utils.tracing.service_decorators import traced_multimodal_llm, traced_stt, traced_tts from . import events @@ -379,6 +379,7 @@ class GeminiMultimodalLiveLLMService(LLMService): self._last_transcription_sent = "" self._bot_audio_buffer = bytearray() self._bot_text_buffer = "" + self._llm_output_buffer = "" self._sample_rate = 24000 @@ -804,7 +805,7 @@ class GeminiMultimodalLiveLLMService(LLMService): ) await self.send_client_event(evt) - @traced_gemini_live(operation="tool_result") + @traced_multimodal_llm(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. @@ -829,7 +830,7 @@ class GeminiMultimodalLiveLLMService(LLMService): await self._websocket.send(response_message) # await self._websocket.send(json.dumps({"clientContent": {"turnComplete": True}})) - @traced_gemini_live(operation="setup") + @traced_multimodal_llm(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 @@ -876,7 +877,7 @@ class GeminiMultimodalLiveLLMService(LLMService): ) await self.push_frame(frame) - @traced_gemini_live(operation="tool_call") + @traced_multimodal_llm(operation="tool_call") async def _handle_evt_tool_call(self, evt): function_calls = evt.toolCall.functionCalls if not function_calls: @@ -891,10 +892,22 @@ class GeminiMultimodalLiveLLMService(LLMService): arguments=call.args, ) + @traced_tts + async def _handle_bot_transcription(self, text: str): + """Handle a transcription result with tracing.""" + pass + async def _handle_evt_turn_complete(self, evt): self._bot_is_speaking = False text = self._bot_text_buffer + if text: + # TEXT modality + await self._handle_bot_transcription(text) + else: + # AUDIO modality + await self._handle_bot_transcription(self._llm_output_buffer) self._bot_text_buffer = "" + self._llm_output_buffer = "" # Only push the TTSStoppedFrame the bot is outputting audio # when text is found, modalities is set to TEXT and no audio @@ -904,7 +917,13 @@ class GeminiMultimodalLiveLLMService(LLMService): await self.push_frame(LLMFullResponseEndFrame()) - @traced_gemini_live(operation="input_transcription") + @traced_stt + async def _handle_user_transcription( + self, transcript: str, is_final: bool, language: Optional[Language] = None + ): + """Handle a transcription result with tracing.""" + pass + async def _handle_evt_input_transcription(self, evt): """Handle the input transcription event. @@ -940,6 +959,9 @@ class GeminiMultimodalLiveLLMService(LLMService): # Send a TranscriptionFrame with the complete sentence logger.debug(f"[Transcription:user] [{complete_sentence}]") + await self._handle_user_transcription( + complete_sentence, True, self._settings["language"] + ) await self.push_frame( TranscriptionFrame( text=complete_sentence, @@ -950,7 +972,6 @@ 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 @@ -963,10 +984,13 @@ class GeminiMultimodalLiveLLMService(LLMService): if not text: return + # Collect text for tracing + self._llm_output_buffer += text + await self.push_frame(LLMTextFrame(text=text)) await self.push_frame(TTSTextFrame(text=text)) - @traced_gemini_live(operation="usage_metadata") + @traced_multimodal_llm(operation="usage_metadata") async def _handle_evt_usage_metadata(self, evt): if not evt.usageMetadata: return From dd1f7d0875a6a1d58e811af545a9072b205fcb48 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 29 May 2025 22:35:06 -0400 Subject: [PATCH 3/5] Add tracking to OpenAI Realtime --- .../services/openai_realtime_beta/openai.py | 14 ++ .../utils/tracing/service_attributes.py | 101 +++++++- .../utils/tracing/service_decorators.py | 229 +++++++++++++++--- 3 files changed, 314 insertions(+), 30 deletions(-) diff --git a/src/pipecat/services/openai_realtime_beta/openai.py b/src/pipecat/services/openai_realtime_beta/openai.py index 579be2ebe..9957cb134 100644 --- a/src/pipecat/services/openai_realtime_beta/openai.py +++ b/src/pipecat/services/openai_realtime_beta/openai.py @@ -50,7 +50,9 @@ from pipecat.processors.aggregators.openai_llm_context import ( from pipecat.processors.frame_processor import FrameDirection from pipecat.services.llm_service import LLMService from pipecat.services.openai.llm import OpenAIContextAggregatorPair +from pipecat.transcriptions.language import Language from pipecat.utils.time import time_now_iso8601 +from pipecat.utils.tracing.service_decorators import traced_openai_realtime, traced_stt, traced_tts from . import events from .context import ( @@ -100,6 +102,7 @@ class OpenAIRealtimeBetaLLMService(LLMService): self.api_key = api_key self.base_url = full_url + self.set_model_name(model) self._session_properties: events.SessionProperties = ( session_properties or events.SessionProperties() @@ -402,6 +405,7 @@ class OpenAIRealtimeBetaLLMService(LLMService): # errors are fatal, so exit the receive loop return + @traced_openai_realtime(operation="llm_setup") async def _handle_evt_session_created(self, evt): # session.created is received right after connecting. Send a message # to configure the session properties. @@ -467,6 +471,13 @@ class OpenAIRealtimeBetaLLMService(LLMService): InterimTranscriptionFrame(evt.delta, "", time_now_iso8601(), result=evt) ) + @traced_stt + async def _handle_user_transcription( + self, transcript: str, is_final: bool, language: Optional[Language] = None + ): + """Handle a transcription result with tracing.""" + pass + async def handle_evt_input_audio_transcription_completed(self, evt): await self._call_event_handler("on_conversation_item_updated", evt.item_id, None) @@ -475,6 +486,7 @@ class OpenAIRealtimeBetaLLMService(LLMService): # no way to get a language code? TranscriptionFrame(evt.transcript, "", time_now_iso8601(), result=evt) ) + await self._handle_user_transcription(evt.transcript, True, Language.EN) pair = self._user_and_response_message_tuple if pair: user, assistant = pair @@ -493,6 +505,7 @@ class OpenAIRealtimeBetaLLMService(LLMService): for future in futures: future.set_result(evt.item) + @traced_openai_realtime(operation="llm_response") async def _handle_evt_response_done(self, evt): # todo: figure out whether there's anything we need to do for "cancelled" events # usage metrics @@ -609,6 +622,7 @@ class OpenAIRealtimeBetaLLMService(LLMService): self._context.llm_needs_initial_messages = True await self._connect() + @traced_openai_realtime(operation="llm_request") async def _create_response(self): if not self._api_session_ready: self._run_llm_when_api_session_ready = True diff --git a/src/pipecat/utils/tracing/service_attributes.py b/src/pipecat/utils/tracing/service_attributes.py index bb7dc8761..ea18bc8fa 100644 --- a/src/pipecat/utils/tracing/service_attributes.py +++ b/src/pipecat/utils/tracing/service_attributes.py @@ -258,7 +258,7 @@ def add_llm_span_attributes( span.set_attribute(key, value) -def add_gemini_live_span_attributes( +def add_multimodal_llm_span_attributes( span: "Span", service_name: str, model: str, @@ -361,3 +361,102 @@ def add_gemini_live_span_attributes( for key, value in kwargs.items(): if isinstance(value, (str, int, float, bool)): span.set_attribute(key, value) + + +def add_openai_realtime_span_attributes( + span: "Span", + service_name: str, + model: str, + operation_name: str, + session_properties: Optional[Dict[str, Any]] = None, + transcript: Optional[str] = None, + is_input: Optional[bool] = None, + context_messages: Optional[str] = None, + function_calls: Optional[List[Dict]] = None, + tools: Optional[List[Dict]] = None, + tools_serialized: Optional[str] = None, + audio_data_size: Optional[int] = None, + **kwargs, +) -> None: + """Add OpenAI Realtime 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, transcription, response, etc.) + session_properties: Session configuration properties + transcript: Transcription text + is_input: Whether transcript is input (True) or output (False) + context_messages: JSON-serialized context messages + function_calls: Function calls being made + tools: Available tools/functions list + tools_serialized: JSON-serialized tools for detailed inspection + audio_data_size: Size of audio data in bytes + **kwargs: Additional attributes to add + """ + # Add standard attributes + span.set_attribute("gen_ai.system", "openai") + 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 transcript: + span.set_attribute("transcript", transcript) + if is_input is not None: + span.set_attribute("transcript.is_input", is_input) + + if context_messages: + span.set_attribute("input", context_messages) + + 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_names.append(tool.name) + elif isinstance(tool, dict) and "function" in tool and "name" in tool["function"]: + tool_names.append(tool["function"]["name"]) + + if tool_names: + span.set_attribute("tools.names", ",".join(tool_names)) + + if tools_serialized: + span.set_attribute("tools.definitions", tools_serialized) + + if function_calls: + span.set_attribute("function_calls.count", len(function_calls)) + if function_calls: + call = function_calls[0] + if hasattr(call, "name"): + span.set_attribute("function_calls.first_name", call.name) + elif isinstance(call, dict) and "name" in call: + span.set_attribute("function_calls.first_name", call["name"]) + + # Add session properties if provided + if session_properties: + for key, value in session_properties.items(): + if isinstance(value, (str, int, float, bool)): + span.set_attribute(f"session.{key}", value) + elif key == "turn_detection" and value is not None: + if isinstance(value, bool): + span.set_attribute("session.turn_detection.enabled", value) + elif isinstance(value, dict): + span.set_attribute("session.turn_detection.enabled", True) + for td_key, td_value in value.items(): + if isinstance(td_value, (str, int, float, bool)): + span.set_attribute(f"session.turn_detection.{td_key}", td_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 4d6cac694..a41138714 100644 --- a/src/pipecat/utils/tracing/service_decorators.py +++ b/src/pipecat/utils/tracing/service_decorators.py @@ -24,8 +24,9 @@ if TYPE_CHECKING: from opentelemetry import trace from pipecat.utils.tracing.service_attributes import ( - add_gemini_live_span_attributes, add_llm_span_attributes, + add_multimodal_llm_span_attributes, + add_openai_realtime_span_attributes, add_stt_span_attributes, add_tts_span_attributes, ) @@ -480,7 +481,7 @@ def traced_llm(func: Optional[Callable] = None, *, name: Optional[str] = None) - return decorator -def traced_gemini_live(operation: str) -> Callable: +def traced_multimodal_llm(operation: str) -> Callable: """Traces Gemini Live service methods with operation-specific attributes. This decorator automatically captures relevant information based on the operation type: @@ -626,32 +627,6 @@ def traced_gemini_live(operation: str) -> Callable: 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 @@ -738,7 +713,7 @@ def traced_gemini_live(operation: str) -> Callable: operation_attrs["tokens.total"] = usage.totalTokenCount or 0 # Add all attributes to the span - add_gemini_live_span_attributes( + add_multimodal_llm_span_attributes( span=current_span, service_name=service_class_name, model=model_name, @@ -785,3 +760,199 @@ def traced_gemini_live(operation: str) -> Callable: return wrapper return decorator + + +def traced_openai_realtime(operation: str) -> Callable: + """Traces OpenAI Realtime service methods with operation-specific attributes. + + This decorator automatically captures relevant information based on the operation type: + - setup: Session configuration and tools + - transcription_completed: User transcription + - response_create: Context and input messages + - response_done: Usage metadata and output + - function_call: Function call information + + Args: + operation: The operation name (matches the event type being handled) + + Returns: + Wrapped method with OpenAI Realtime 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") + ) + + # Operation-specific attribute collection + operation_attrs = {} + + if operation == "llm_setup": + # Capture session properties and tools + session_properties = getattr(self, "_session_properties", None) + if session_properties: + try: + # Convert to dict for easier processing + if hasattr(session_properties, "model_dump"): + props_dict = session_properties.model_dump() + elif hasattr(session_properties, "__dict__"): + props_dict = session_properties.__dict__ + else: + props_dict = {} + + operation_attrs["session_properties"] = props_dict + + # Extract tools if available + tools = props_dict.get("tools") + if tools: + operation_attrs["tools"] = tools + try: + operation_attrs["tools_serialized"] = json.dumps(tools) + except Exception as e: + logging.warning(f"Error serializing OpenAI tools: {e}") + + # Extract instructions + instructions = props_dict.get("instructions") + if instructions: + operation_attrs["instructions"] = instructions[:500] + + except Exception as e: + logging.warning(f"Error processing session properties: {e}") + + # Also check context for tools + if hasattr(self, "_context") and self._context: + try: + context_tools = getattr(self._context, "tools", None) + if context_tools and not operation_attrs.get("tools"): + operation_attrs["tools"] = context_tools + operation_attrs["tools_serialized"] = json.dumps( + context_tools + ) + except Exception as e: + logging.warning(f"Error extracting context tools: {e}") + + elif operation == "llm_request": + # Capture context messages being sent + if hasattr(self, "_context") and self._context: + try: + messages = self._context.get_messages_for_logging() + if messages: + operation_attrs["context_messages"] = json.dumps(messages) + except Exception as e: + logging.warning(f"Error getting context messages: {e}") + + elif operation == "llm_response" and args: + # Extract usage and response metadata + evt = args[0] if args else None + if evt and hasattr(evt, "response"): + response = evt.response + + # Token usage - basic attributes for span visibility + if hasattr(response, "usage"): + usage = response.usage + if hasattr(usage, "input_tokens"): + operation_attrs["tokens.prompt"] = usage.input_tokens + if hasattr(usage, "output_tokens"): + operation_attrs["tokens.completion"] = usage.output_tokens + if hasattr(usage, "total_tokens"): + operation_attrs["tokens.total"] = usage.total_tokens + + # Response status and metadata + if hasattr(response, "status"): + operation_attrs["response.status"] = response.status + + if hasattr(response, "id"): + operation_attrs["response.id"] = response.id + + # Output items and extract transcript for output field + if hasattr(response, "output") and response.output: + operation_attrs["response.output_items"] = len(response.output) + + # Extract assistant transcript for output field + assistant_transcript = "" + for item in response.output: + if ( + hasattr(item, "content") + and item.content + and hasattr(item, "role") + and item.role == "assistant" + ): + for content in item.content: + if ( + hasattr(content, "transcript") + and content.transcript + ): + assistant_transcript += content.transcript + " " + + if assistant_transcript.strip(): + operation_attrs["output"] = assistant_transcript.strip() + + # Add all attributes to the span + add_openai_realtime_span_attributes( + span=current_span, + service_name=service_class_name, + model=model_name, + operation_name=operation, + **operation_attrs, + ) + + # For llm_response operation, also handle token usage metrics + if operation == "llm_response" and hasattr(self, "start_llm_usage_metrics"): + evt = args[0] if args else None + if evt and hasattr(evt, "response") and hasattr(evt.response, "usage"): + usage = evt.response.usage + # Create LLMTokenUsage object + from pipecat.metrics.metrics import LLMTokenUsage + + tokens = LLMTokenUsage( + prompt_tokens=getattr(usage, "input_tokens", 0), + completion_tokens=getattr(usage, "output_tokens", 0), + total_tokens=getattr(usage, "total_tokens", 0), + ) + _add_token_usage_to_span(current_span, tokens) + + # Capture TTFB metric if available + ttfb_ms = getattr(getattr(self, "_metrics", None), "ttfb_ms", None) + if ttfb_ms is not None: + current_span.set_attribute("metrics.ttfb_ms", ttfb_ms) + + # 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 OpenAI Realtime tracing (continuing without tracing): {e}") + # If tracing fails, fall back to the original function + return await func(self, *args, **kwargs) + + return wrapper + + return decorator From 86025579853c6a1975152b67c41a5fa7546ba989 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 29 May 2025 23:39:13 -0400 Subject: [PATCH 4/5] Refactor Gemini tracing to more closely match OpenAI Realtime, add TTFB metrics --- examples/open-telemetry/langfuse/bot.py | 68 ++++++------- .../services/gemini_multimodal_live/gemini.py | 36 ++++--- .../utils/tracing/service_attributes.py | 2 +- .../utils/tracing/service_decorators.py | 95 ++++++++++++++----- 4 files changed, 125 insertions(+), 76 deletions(-) diff --git a/examples/open-telemetry/langfuse/bot.py b/examples/open-telemetry/langfuse/bot.py index 872a4d979..22ff399a0 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 AdapterType, ToolsSchema +from pipecat.adapters.schemas.tools_schema import ToolsSchema from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.frames.frames import TTSSpeakFrame from pipecat.pipeline.pipeline import Pipeline @@ -21,12 +21,10 @@ 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) @@ -47,6 +45,11 @@ 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. @@ -69,29 +72,24 @@ 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", @@ -108,29 +106,25 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si }, required=["location", "format"], ) - search_tool = {"google_search": {}} - tools = ToolsSchema( - standard_tools=[weather_function], custom_tools={AdapterType.GEMINI: [search_tool]} - ) + tools = ToolsSchema(standard_tools=[weather_function]) - llm = GeminiMultimodalLiveLLMService( - api_key=os.getenv("GOOGLE_API_KEY"), - system_instruction=system_instruction, - tools=tools, - ) + 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.register_function("get_current_weather", fetch_weather_from_api) - - context = OpenAILLMContext( - [{"role": "user", "content": "Say hello."}], - ) + context = OpenAILLMContext(messages, tools) 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 1b25bb80e..6dd93720b 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -60,7 +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_multimodal_llm, traced_stt, traced_tts +from pipecat.utils.tracing.service_decorators import traced_gemini_live, traced_stt, traced_tts from . import events @@ -473,6 +473,7 @@ class GeminiMultimodalLiveLLMService(LLMService): async def _handle_user_stopped_speaking(self, frame): self._user_is_speaking = False self._user_audio_buffer = bytearray() + await self.start_ttfb_metrics() if self._needs_turn_complete_message: self._needs_turn_complete_message = False evt = events.ClientContentMessage.model_validate( @@ -754,6 +755,8 @@ class GeminiMultimodalLiveLLMService(LLMService): logger.debug(f"Creating initial response: {messages}") + await self.start_ttfb_metrics() + evt = events.ClientContentMessage.model_validate( { "clientContent": { @@ -795,6 +798,8 @@ class GeminiMultimodalLiveLLMService(LLMService): return logger.debug(f"Creating response: {messages}") + await self.start_ttfb_metrics() + evt = events.ClientContentMessage.model_validate( { "clientContent": { @@ -805,7 +810,7 @@ class GeminiMultimodalLiveLLMService(LLMService): ) await self.send_client_event(evt) - @traced_multimodal_llm(operation="tool_result") + @traced_gemini_live(operation="llm_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. @@ -830,7 +835,7 @@ class GeminiMultimodalLiveLLMService(LLMService): await self._websocket.send(response_message) # await self._websocket.send(json.dumps({"clientContent": {"turnComplete": True}})) - @traced_multimodal_llm(operation="setup") + @traced_gemini_live(operation="llm_setup") async def _handle_evt_setup_complete(self, evt): # If this is our first context frame, run the LLM self._api_session_ready = True @@ -844,6 +849,8 @@ class GeminiMultimodalLiveLLMService(LLMService): if not part: return + await self.stop_ttfb_metrics() + # part.text is added when `modalities` is set to TEXT; otherwise, it's None text = part.text if text: @@ -877,7 +884,7 @@ class GeminiMultimodalLiveLLMService(LLMService): ) await self.push_frame(frame) - @traced_multimodal_llm(operation="tool_call") + @traced_gemini_live(operation="llm_tool_call") async def _handle_evt_tool_call(self, evt): function_calls = evt.toolCall.functionCalls if not function_calls: @@ -892,24 +899,28 @@ class GeminiMultimodalLiveLLMService(LLMService): arguments=call.args, ) - @traced_tts - async def _handle_bot_transcription(self, text: str): - """Handle a transcription result with tracing.""" - pass - + @traced_gemini_live(operation="llm_response") async def _handle_evt_turn_complete(self, evt): self._bot_is_speaking = False text = self._bot_text_buffer + + # Determine output and modality for tracing if text: # TEXT modality - await self._handle_bot_transcription(text) + output_text = text + output_modality = "TEXT" else: # AUDIO modality - await self._handle_bot_transcription(self._llm_output_buffer) + output_text = self._llm_output_buffer + output_modality = "AUDIO" + + # Trace the complete LLM response (this will be handled by the decorator) + # The decorator will extract the output text and usage metadata from the event + self._bot_text_buffer = "" self._llm_output_buffer = "" - # Only push the TTSStoppedFrame the bot is outputting audio + # Only push the TTSStoppedFrame if the bot is outputting audio # when text is found, modalities is set to TEXT and no audio # is produced. if not text: @@ -990,7 +1001,6 @@ class GeminiMultimodalLiveLLMService(LLMService): await self.push_frame(LLMTextFrame(text=text)) await self.push_frame(TTSTextFrame(text=text)) - @traced_multimodal_llm(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 ea18bc8fa..8f90ad3be 100644 --- a/src/pipecat/utils/tracing/service_attributes.py +++ b/src/pipecat/utils/tracing/service_attributes.py @@ -258,7 +258,7 @@ def add_llm_span_attributes( span.set_attribute(key, value) -def add_multimodal_llm_span_attributes( +def add_gemini_live_span_attributes( span: "Span", service_name: str, model: str, diff --git a/src/pipecat/utils/tracing/service_decorators.py b/src/pipecat/utils/tracing/service_decorators.py index a41138714..e4d50846a 100644 --- a/src/pipecat/utils/tracing/service_decorators.py +++ b/src/pipecat/utils/tracing/service_decorators.py @@ -24,8 +24,8 @@ if TYPE_CHECKING: from opentelemetry import trace from pipecat.utils.tracing.service_attributes import ( + add_gemini_live_span_attributes, add_llm_span_attributes, - add_multimodal_llm_span_attributes, add_openai_realtime_span_attributes, add_stt_span_attributes, add_tts_span_attributes, @@ -481,17 +481,14 @@ def traced_llm(func: Optional[Callable] = None, *, name: Optional[str] = None) - return decorator -def traced_multimodal_llm(operation: str) -> Callable: +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 + - llm_setup: Configuration, tools definitions, and system instructions + - llm_tool_call: Function call information + - llm_tool_result: Function execution results + - llm_response: Complete LLM response with usage and output Args: operation: The operation name (matches the event type being handled) @@ -542,7 +539,7 @@ def traced_multimodal_llm(operation: str) -> Callable: # Operation-specific attribute collection operation_attrs = {} - if operation == "setup": + if operation == "llm_setup": # Capture detailed tool information tools = getattr(self, "_tools", None) if tools: @@ -627,7 +624,7 @@ def traced_multimodal_llm(operation: str) -> Callable: f"Error extracting context system instructions: {e}" ) - elif operation == "tool_call" and args: + elif operation == "llm_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: @@ -654,7 +651,7 @@ def traced_multimodal_llm(operation: str) -> Callable: except Exception: operation_attrs["tool.arguments"] = str(call.args)[:1000] - elif operation == "tool_result" and args: + elif operation == "llm_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): @@ -698,11 +695,13 @@ def traced_multimodal_llm(operation: str) -> Callable: ) 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 + elif operation == "llm_response" and args: + # Extract usage and response metadata from turn complete event evt = args[0] if args else None if evt and hasattr(evt, "usageMetadata") and evt.usageMetadata: usage = evt.usageMetadata + + # Token usage - basic attributes for span visibility if hasattr(usage, "promptTokenCount"): operation_attrs["tokens.prompt"] = usage.promptTokenCount or 0 if hasattr(usage, "responseTokenCount"): @@ -712,8 +711,29 @@ def traced_multimodal_llm(operation: str) -> Callable: if hasattr(usage, "totalTokenCount"): operation_attrs["tokens.total"] = usage.totalTokenCount or 0 + # Get output text and modality from service state + text = getattr(self, "_bot_text_buffer", "") + audio_text = getattr(self, "_llm_output_buffer", "") + + if text: + # TEXT modality + operation_attrs["output"] = text + operation_attrs["output_modality"] = "TEXT" + elif audio_text: + # AUDIO modality + operation_attrs["output"] = audio_text + operation_attrs["output_modality"] = "AUDIO" + + # Add turn completion status + if ( + evt + and hasattr(evt, "serverContent") + and evt.serverContent.turnComplete + ): + operation_attrs["turn_complete"] = True + # Add all attributes to the span - add_multimodal_llm_span_attributes( + add_gemini_live_span_attributes( span=current_span, service_name=service_class_name, model=model_name, @@ -725,10 +745,8 @@ def traced_multimodal_llm(operation: str) -> Callable: **operation_attrs, ) - # For usage_metadata operation, also handle token usage metrics - if operation == "usage_metadata" and hasattr( - self, "start_llm_usage_metrics" - ): + # For llm_response operation, also handle token usage metrics + if operation == "llm_response" 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 @@ -742,6 +760,11 @@ def traced_multimodal_llm(operation: str) -> Callable: ) _add_token_usage_to_span(current_span, tokens) + # Capture TTFB metric if available + ttfb_ms = getattr(getattr(self, "_metrics", None), "ttfb_ms", None) + if ttfb_ms is not None: + current_span.set_attribute("metrics.ttfb_ms", ttfb_ms) + # Run the original function result = await func(self, *args, **kwargs) @@ -766,11 +789,9 @@ def traced_openai_realtime(operation: str) -> Callable: """Traces OpenAI Realtime service methods with operation-specific attributes. This decorator automatically captures relevant information based on the operation type: - - setup: Session configuration and tools - - transcription_completed: User transcription - - response_create: Context and input messages - - response_done: Usage metadata and output - - function_call: Function call information + - llm_setup: Session configuration and tools + - llm_request: Context and input messages + - llm_response: Usage metadata, output, and function calls Args: operation: The operation name (matches the event type being handled) @@ -890,8 +911,10 @@ def traced_openai_realtime(operation: str) -> Callable: if hasattr(response, "output") and response.output: operation_attrs["response.output_items"] = len(response.output) - # Extract assistant transcript for output field + # Extract assistant transcript and function calls assistant_transcript = "" + function_calls = [] + for item in response.output: if ( hasattr(item, "content") @@ -906,9 +929,31 @@ def traced_openai_realtime(operation: str) -> Callable: ): assistant_transcript += content.transcript + " " + elif hasattr(item, "type") and item.type == "function_call": + function_call_info = { + "name": getattr(item, "name", "unknown"), + "call_id": getattr(item, "call_id", "unknown"), + } + if hasattr(item, "arguments"): + args_str = item.arguments + if len(args_str) > 500: + args_str = args_str[:500] + "..." + function_call_info["arguments"] = args_str + function_calls.append(function_call_info) + if assistant_transcript.strip(): operation_attrs["output"] = assistant_transcript.strip() + if function_calls: + operation_attrs["function_calls"] = function_calls + operation_attrs["function_calls.count"] = len( + function_calls + ) + all_names = [call["name"] for call in function_calls] + operation_attrs["function_calls.all_names"] = ",".join( + all_names + ) + # Add all attributes to the span add_openai_realtime_span_attributes( span=current_span, From 43719ec7373855f385d52f28b502e258220a58b6 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 30 May 2025 11:08:08 -0400 Subject: [PATCH 5/5] Update CHANGELOG --- CHANGELOG.md | 8 ++++++++ src/pipecat/utils/tracing/service_decorators.py | 12 ++++++------ 2 files changed, 14 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c74400860..9c4940edf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added OpenTelemetry tracing for `GeminiMultimodalLiveLLMService` and + `OpenAIRealtimeBetaLLMService`. + - Added `interruption_strategies` to `PipelineParams` using `MinWordsInterruptionStrategy` to specify minimum words required to interrupt the bot when it's speaking. Use @@ -48,6 +51,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 `DailyTransport.stop_transcription()` to be able to start and stop Daily transcription dynamically (maybe with different settings). +### Changed + +- Updated OpenTelemetry tracing attribute `metrics.ttfb_ms` to `metrics.ttfb`. + The attribute reports TTFB in seconds. + ### Deprecated - `DailyTransport.send_dtmf()` is deprecated, push an `OutputDTMFFrame` or an diff --git a/src/pipecat/utils/tracing/service_decorators.py b/src/pipecat/utils/tracing/service_decorators.py index e4d50846a..c016827d6 100644 --- a/src/pipecat/utils/tracing/service_decorators.py +++ b/src/pipecat/utils/tracing/service_decorators.py @@ -761,9 +761,9 @@ def traced_gemini_live(operation: str) -> Callable: _add_token_usage_to_span(current_span, tokens) # Capture TTFB metric if available - ttfb_ms = getattr(getattr(self, "_metrics", None), "ttfb_ms", None) - if ttfb_ms is not None: - current_span.set_attribute("metrics.ttfb_ms", ttfb_ms) + ttfb = getattr(getattr(self, "_metrics", None), "ttfb", None) + if ttfb is not None: + current_span.set_attribute("metrics.ttfb", ttfb) # Run the original function result = await func(self, *args, **kwargs) @@ -979,9 +979,9 @@ def traced_openai_realtime(operation: str) -> Callable: _add_token_usage_to_span(current_span, tokens) # Capture TTFB metric if available - ttfb_ms = getattr(getattr(self, "_metrics", None), "ttfb_ms", None) - if ttfb_ms is not None: - current_span.set_attribute("metrics.ttfb_ms", ttfb_ms) + ttfb = getattr(getattr(self, "_metrics", None), "ttfb", None) + if ttfb is not None: + current_span.set_attribute("metrics.ttfb", ttfb) # Run the original function result = await func(self, *args, **kwargs)