From d3df75aaa067088cfebd27535bdc04d1de577858 Mon Sep 17 00:00:00 2001 From: Sunah Suh Date: Tue, 10 Jun 2025 16:32:24 -0500 Subject: [PATCH] =?UTF-8?q?Add=20`additional=5Fspan=5Fattributes`=20param?= =?UTF-8?q?=20to=20PipelineTask=20for=20extra=20otel=E2=80=A6=20(#1992)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- CHANGELOG.md | 5 +++++ src/pipecat/pipeline/task.py | 10 ++++++++-- src/pipecat/utils/tracing/turn_trace_observer.py | 10 +++++++++- 3 files changed, 22 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 54bb45013..be7440cf2 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 6bf5dd688..736f98244 100644 --- a/src/pipecat/pipeline/task.py +++ b/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 diff --git a/src/pipecat/utils/tracing/turn_trace_observer.py b/src/pipecat/utils/tracing/turn_trace_observer.py index 26fc5b38a..f67ca3b28 100644 --- a/src/pipecat/utils/tracing/turn_trace_observer.py +++ b/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(