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