Add additional_span_attributes param to PipelineTask for extra otel… (#1992)

CHANGELOG.md

@@ -7,6 +7,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- Adds a parameter called `additional_span_attributes` to PipelineTask that
+  lets you add any additional attributes you'd like to the conversation span.
+
 ### Fixed
 
 - Fixed an issue with `CartesiaSTTService` initialization.

src/pipecat/pipeline/task.py

@@ -184,7 +184,9 @@ class PipelineTask(BaseTask):
             the idle timeout is reached.
         enable_turn_tracking: Whether to enable turn tracking.
         enable_turn_tracing: Whether to enable turn tracing.
-
+        conversation_id: Optional custom ID for the conversation.
+        additional_span_attributes: Optional dictionary of attributes to propagate as
+            OpenTelemetry conversation span attributes.
     """
 
     def __init__(
@@ -205,6 +207,7 @@ class PipelineTask(BaseTask):
         enable_turn_tracking: bool = True,
         enable_tracing: bool = False,
         conversation_id: Optional[str] = None,
+        additional_span_attributes: Optional[dict] = None,
     ):
         super().__init__()
         self._pipeline = pipeline
@@ -217,6 +220,7 @@ class PipelineTask(BaseTask):
         self._enable_turn_tracking = enable_turn_tracking
         self._enable_tracing = enable_tracing and is_tracing_available()
         self._conversation_id = conversation_id
+        self._additional_span_attributes = additional_span_attributes or {}
         if self._params.observers:
             import warnings
 
@@ -235,7 +239,9 @@ class PipelineTask(BaseTask):
             observers.append(self._turn_tracking_observer)
         if self._enable_tracing and self._turn_tracking_observer:
             self._turn_trace_observer = TurnTraceObserver(
-                self._turn_tracking_observer, conversation_id=self._conversation_id
+                self._turn_tracking_observer,
+                conversation_id=self._conversation_id,
+                additional_span_attributes=self._additional_span_attributes,
             )
             observers.append(self._turn_trace_observer)
         self._finished = False

src/pipecat/utils/tracing/turn_trace_observer.py

@@ -35,7 +35,11 @@ class TurnTraceObserver(BaseObserver):
     """
 
     def __init__(
-        self, turn_tracker: TurnTrackingObserver, conversation_id: Optional[str] = None, **kwargs
+        self,
+        turn_tracker: TurnTrackingObserver,
+        conversation_id: Optional[str] = None,
+        additional_span_attributes: Optional[dict] = None,
+        **kwargs,
     ):
         super().__init__(**kwargs)
         self._turn_tracker = turn_tracker
@@ -47,6 +51,7 @@ class TurnTraceObserver(BaseObserver):
         # Conversation tracking properties
         self._conversation_span: Optional["Span"] = None
         self._conversation_id = conversation_id
+        self._additional_span_attributes = additional_span_attributes or {}
 
         if turn_tracker:
 
@@ -89,6 +94,9 @@ class TurnTraceObserver(BaseObserver):
         # Set span attributes
         self._conversation_span.set_attribute("conversation.id", conversation_id)
         self._conversation_span.set_attribute("conversation.type", "voice")
+        # Set custom otel attributes if provided
+        for k, v in (self._additional_span_attributes or {}).items():
+            self._conversation_span.set_attribute(k, v)
 
         # Update the conversation context provider
         context_provider.set_current_conversation_context(