From 6f524fb81671eaa27b6f68be0549b517f5c00c13 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Sat, 24 May 2025 09:15:39 -0400 Subject: [PATCH 1/2] Add LLM response to OTel tracing --- .../utils/tracing/service_attributes.py | 5 +++ .../utils/tracing/service_decorators.py | 42 +++++++++++++++++-- 2 files changed, 43 insertions(+), 4 deletions(-) diff --git a/src/pipecat/utils/tracing/service_attributes.py b/src/pipecat/utils/tracing/service_attributes.py index b30cf5272..af192ea61 100644 --- a/src/pipecat/utils/tracing/service_attributes.py +++ b/src/pipecat/utils/tracing/service_attributes.py @@ -171,6 +171,7 @@ def add_llm_span_attributes( model: str, stream: bool = True, messages: Optional[str] = None, + output: Optional[str] = None, tools: Optional[str] = None, tool_count: Optional[int] = None, tool_choice: Optional[str] = None, @@ -188,6 +189,7 @@ def add_llm_span_attributes( model: Model name/identifier stream: Whether streaming is enabled messages: JSON-serialized messages + output: Aggregated output text from the LLM tools: JSON-serialized tools configuration tool_count: Number of tools available tool_choice: Tool selection configuration @@ -208,6 +210,9 @@ def add_llm_span_attributes( if messages: span.set_attribute("input", messages) + if output: + span.set_attribute("output", output) + if tools: span.set_attribute("tools", tools) diff --git a/src/pipecat/utils/tracing/service_decorators.py b/src/pipecat/utils/tracing/service_decorators.py index 95c08e5ec..123db1ec0 100644 --- a/src/pipecat/utils/tracing/service_decorators.py +++ b/src/pipecat/utils/tracing/service_decorators.py @@ -282,6 +282,7 @@ def traced_llm(func: Optional[Callable] = None, *, name: Optional[str] = None) - - Tool configurations - Token usage metrics - Performance metrics like TTFB + - Aggregated output text Args: func: The LLM method to trace. @@ -313,6 +314,26 @@ def traced_llm(func: Optional[Callable] = None, *, name: Optional[str] = None) - span_name, context=parent_context ) as current_span: try: + # Store original method and output aggregator + original_push_frame = self.push_frame + output_text = "" # Simple string accumulation + + async def traced_push_frame(frame, direction=None): + nonlocal output_text + # Check for LLMTextFrame + if ( + hasattr(frame, "__class__") + and frame.__class__.__name__ == "LLMTextFrame" + and hasattr(frame, "text") + ): + output_text += frame.text + + # Call original + if direction is not None: + return await original_push_frame(frame, direction) + else: + return await original_push_frame(frame) + # For token usage monitoring original_start_llm_usage_metrics = None if hasattr(self, "start_llm_usage_metrics"): @@ -331,6 +352,9 @@ def traced_llm(func: Optional[Callable] = None, *, name: Optional[str] = None) - self.start_llm_usage_metrics = wrapped_start_llm_usage_metrics try: + # Replace push_frame to capture output + self.push_frame = traced_push_frame + # Detect if we're using Google's service is_google_service = "google" in service_class_name.lower() @@ -411,13 +435,23 @@ def traced_llm(func: Optional[Callable] = None, *, name: Optional[str] = None) - # Add all gathered attributes to the span add_llm_span_attributes(span=current_span, **attribute_kwargs) + + # Call the original function + result = await f(self, context, *args, **kwargs) + + # Add output if we captured any + if output_text: + current_span.set_attribute("output", output_text) + + return result + except Exception as e: logging.warning(f"Error adding initial LLM attributes: {e}") - - # Call the original function - return await f(self, context, *args, **kwargs) + raise finally: - # Restore the original methods if we overrode them + # Always restore the original methods + self.push_frame = original_push_frame + if ( "original_start_llm_usage_metrics" in locals() and original_start_llm_usage_metrics From aa7d15beb35c6de59e6bff0d4adfc5dbc6f23cad Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Mon, 26 May 2025 21:54:35 -0400 Subject: [PATCH 2/2] fix: move LLM call outside tracing try block to prevent double execution --- .../utils/tracing/service_decorators.py | 25 ++++++++++--------- 1 file changed, 13 insertions(+), 12 deletions(-) diff --git a/src/pipecat/utils/tracing/service_decorators.py b/src/pipecat/utils/tracing/service_decorators.py index 123db1ec0..4a695ec7b 100644 --- a/src/pipecat/utils/tracing/service_decorators.py +++ b/src/pipecat/utils/tracing/service_decorators.py @@ -320,7 +320,7 @@ def traced_llm(func: Optional[Callable] = None, *, name: Optional[str] = None) - async def traced_push_frame(frame, direction=None): nonlocal output_text - # Check for LLMTextFrame + # Capture text from LLMTextFrame during streaming if ( hasattr(frame, "__class__") and frame.__class__.__name__ == "LLMTextFrame" @@ -436,18 +436,19 @@ def traced_llm(func: Optional[Callable] = None, *, name: Optional[str] = None) - # Add all gathered attributes to the span add_llm_span_attributes(span=current_span, **attribute_kwargs) - # Call the original function - result = await f(self, context, *args, **kwargs) - - # Add output if we captured any - if output_text: - current_span.set_attribute("output", output_text) - - return result - except Exception as e: - logging.warning(f"Error adding initial LLM attributes: {e}") - raise + logging.warning(f"Error setting up LLM tracing: {e}") + # Don't raise - let the function execute anyway + + # Run function with modified push_frame to capture the output + result = await f(self, context, *args, **kwargs) + + # Add aggregated output after function completes, if available + if output_text: + current_span.set_attribute("output", output_text) + + return result + finally: # Always restore the original methods self.push_frame = original_push_frame