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/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index 25377e183..6dd93720b 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, traced_stt, traced_tts from . import events @@ -378,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 @@ -471,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( @@ -752,6 +755,8 @@ class GeminiMultimodalLiveLLMService(LLMService): logger.debug(f"Creating initial response: {messages}") + await self.start_ttfb_metrics() + evt = events.ClientContentMessage.model_validate( { "clientContent": { @@ -793,6 +798,8 @@ class GeminiMultimodalLiveLLMService(LLMService): return logger.debug(f"Creating response: {messages}") + await self.start_ttfb_metrics() + evt = events.ClientContentMessage.model_validate( { "clientContent": { @@ -803,6 +810,7 @@ class GeminiMultimodalLiveLLMService(LLMService): ) await self.send_client_event(evt) + @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. @@ -827,6 +835,7 @@ class GeminiMultimodalLiveLLMService(LLMService): await self._websocket.send(response_message) # await self._websocket.send(json.dumps({"clientContent": {"turnComplete": True}})) + @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 @@ -840,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: @@ -873,6 +884,7 @@ class GeminiMultimodalLiveLLMService(LLMService): ) await self.push_frame(frame) + @traced_gemini_live(operation="llm_tool_call") async def _handle_evt_tool_call(self, evt): function_calls = evt.toolCall.functionCalls if not function_calls: @@ -887,12 +899,28 @@ class GeminiMultimodalLiveLLMService(LLMService): arguments=call.args, ) + @traced_gemini_live(operation="llm_response") async def _handle_evt_turn_complete(self, evt): self._bot_is_speaking = False text = self._bot_text_buffer - self._bot_text_buffer = "" - # Only push the TTSStoppedFrame the bot is outputting audio + # Determine output and modality for tracing + if text: + # TEXT modality + output_text = text + output_modality = "TEXT" + else: + # AUDIO modality + 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 if the bot is outputting audio # when text is found, modalities is set to TEXT and no audio # is produced. if not text: @@ -900,6 +928,13 @@ class GeminiMultimodalLiveLLMService(LLMService): await self.push_frame(LLMFullResponseEndFrame()) + @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. @@ -935,6 +970,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, @@ -957,6 +995,9 @@ 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)) 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 619dfbf11..8f90ad3be 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,207 @@ 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) + + +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 4341d308c..c016827d6 100644 --- a/src/pipecat/utils/tracing/service_decorators.py +++ b/src/pipecat/utils/tracing/service_decorators.py @@ -24,7 +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_openai_realtime_span_attributes, add_stt_span_attributes, add_tts_span_attributes, ) @@ -477,3 +479,525 @@ 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: + - 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) + + 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 == "llm_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 == "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: + 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 == "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): + # 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 == "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"): + operation_attrs["tokens.completion"] = ( + usage.responseTokenCount or 0 + ) + 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_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 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 + # 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) + + # Capture TTFB metric if available + 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) + + 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 + + +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: + - 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) + + 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 and function calls + assistant_transcript = "" + function_calls = [] + + 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 + " " + + 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, + 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 = 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) + + 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