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..4a695ec7b 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 + # Capture text from LLMTextFrame during streaming + 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,24 @@ 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) - except Exception as e: - logging.warning(f"Error adding initial LLM attributes: {e}") - # Call the original function - return await f(self, context, *args, **kwargs) + except Exception as e: + 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: - # 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