diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 5634d79ee..8d237defc 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -42,6 +42,7 @@ from pipecat.utils.utils import obj_count, obj_id if TYPE_CHECKING: from pipecat.processors.aggregators.llm_context import LLMContext, NotGiven from pipecat.processors.frame_processor import FrameProcessor + from pipecat.utils.tracing.tracing_context import TracingContext class DeprecatedKeypadEntry: @@ -1036,6 +1037,7 @@ class StartFrame(SystemFrame): Use `LLMUserAggregator`'s new `user_turn_strategies` parameter instead. report_only_initial_ttfb: Whether to report only initial time-to-first-byte. + tracing_context: Pipeline-scoped tracing context for span hierarchy. """ audio_in_sample_rate: int = 16000 @@ -1046,6 +1048,7 @@ class StartFrame(SystemFrame): enable_usage_metrics: bool = False interruption_strategies: List[BaseInterruptionStrategy] = field(default_factory=list) report_only_initial_ttfb: bool = False + tracing_context: Optional["TracingContext"] = None @dataclass diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 8bda19b18..2cfe26606 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -53,6 +53,7 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, F from pipecat.processors.frameworks.rtvi import RTVIObserver, RTVIObserverParams, RTVIProcessor from pipecat.utils.asyncio.task_manager import BaseTaskManager, TaskManager, TaskManagerParams from pipecat.utils.tracing.setup import is_tracing_available +from pipecat.utils.tracing.tracing_context import TracingContext from pipecat.utils.tracing.turn_trace_observer import TurnTraceObserver HEARTBEAT_SECS = 1.0 @@ -290,10 +291,13 @@ class PipelineTask(BasePipelineTask): self._turn_tracking_observer: Optional[TurnTrackingObserver] = None self._user_bot_latency_observer: Optional[UserBotLatencyObserver] = None self._turn_trace_observer: Optional[TurnTraceObserver] = None + self._tracing_context: Optional[TracingContext] = None if self._enable_turn_tracking: self._turn_tracking_observer = TurnTrackingObserver() observers.append(self._turn_tracking_observer) if self._enable_tracing and self._turn_tracking_observer: + # Create pipeline-scoped tracing context + self._tracing_context = TracingContext() # Create latency observer for tracing self._user_bot_latency_observer = UserBotLatencyObserver() observers.append(self._user_bot_latency_observer) @@ -303,6 +307,7 @@ class PipelineTask(BasePipelineTask): latency_tracker=self._user_bot_latency_observer, conversation_id=self._conversation_id, additional_span_attributes=self._additional_span_attributes, + tracing_context=self._tracing_context, ) observers.append(self._turn_trace_observer) @@ -813,6 +818,7 @@ class PipelineTask(BasePipelineTask): enable_usage_metrics=self._params.enable_usage_metrics, report_only_initial_ttfb=self._params.report_only_initial_ttfb, interruption_strategies=self._params.interruption_strategies, + tracing_context=self._tracing_context, ) start_frame.metadata = self._create_start_metadata() await self._pipeline.queue_frame(start_frame) diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index af7e691b0..acbd6baae 100644 --- a/src/pipecat/services/llm_service.py +++ b/src/pipecat/services/llm_service.py @@ -286,6 +286,7 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService): if not self._run_in_parallel: await self._create_sequential_runner_task() self._tracing_enabled = frame.enable_tracing + self._tracing_context = frame.tracing_context async def stop(self, frame: EndFrame): """Stop the LLM service. diff --git a/src/pipecat/services/stt_service.py b/src/pipecat/services/stt_service.py index b556bb23a..81f369572 100644 --- a/src/pipecat/services/stt_service.py +++ b/src/pipecat/services/stt_service.py @@ -203,6 +203,7 @@ class STTService(AIService): await super().start(frame) self._sample_rate = self._init_sample_rate or frame.audio_in_sample_rate self._tracing_enabled = frame.enable_tracing + self._tracing_context = frame.tracing_context async def cleanup(self): """Clean up STT service resources.""" diff --git a/src/pipecat/services/tts_service.py b/src/pipecat/services/tts_service.py index 239d2398b..00e3e81f8 100644 --- a/src/pipecat/services/tts_service.py +++ b/src/pipecat/services/tts_service.py @@ -350,6 +350,7 @@ class TTSService(AIService): if self._push_stop_frames and not self._stop_frame_task: self._stop_frame_task = self.create_task(self._stop_frame_handler()) self._tracing_enabled = frame.enable_tracing + self._tracing_context = frame.tracing_context async def stop(self, frame: EndFrame): """Stop the TTS service. diff --git a/src/pipecat/utils/tracing/conversation_context_provider.py b/src/pipecat/utils/tracing/conversation_context_provider.py deleted file mode 100644 index 4bb88fe14..000000000 --- a/src/pipecat/utils/tracing/conversation_context_provider.py +++ /dev/null @@ -1,114 +0,0 @@ -# -# Copyright (c) 2024-2026, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - -"""Conversation context provider for OpenTelemetry tracing in Pipecat. - -This module provides a singleton context provider that manages the current -conversation's tracing context, allowing services to create child spans -that are properly associated with the conversation. -""" - -import uuid -from typing import TYPE_CHECKING, Optional - -# Import types for type checking only -if TYPE_CHECKING: - from opentelemetry.context import Context - from opentelemetry.trace import SpanContext - -from pipecat.utils.tracing.setup import is_tracing_available - -if is_tracing_available(): - from opentelemetry.context import Context - from opentelemetry.trace import NonRecordingSpan, SpanContext, set_span_in_context - - -class ConversationContextProvider: - """Provides access to the current conversation's tracing context. - - This is a singleton that can be used to get the current conversation's - span context to create child spans (like turns). - """ - - _instance = None - _current_conversation_context: Optional["Context"] = None - _conversation_id: Optional[str] = None - - @classmethod - def get_instance(cls): - """Get the singleton instance. - - Returns: - The singleton ConversationContextProvider instance. - """ - if cls._instance is None: - cls._instance = ConversationContextProvider() - return cls._instance - - def set_current_conversation_context( - self, span_context: Optional["SpanContext"], conversation_id: Optional[str] = None - ): - """Set the current conversation context. - - Args: - span_context: The span context for the current conversation or None to clear it. - conversation_id: Optional ID for the conversation. - """ - if not is_tracing_available(): - return - - self._conversation_id = conversation_id - - if span_context: - # Create a non-recording span from the span context - non_recording_span = NonRecordingSpan(span_context) - self._current_conversation_context = set_span_in_context(non_recording_span) - else: - self._current_conversation_context = None - - def get_current_conversation_context(self) -> Optional["Context"]: - """Get the OpenTelemetry context for the current conversation. - - Returns: - The current conversation context or None if not available. - """ - return self._current_conversation_context - - def get_conversation_id(self) -> Optional[str]: - """Get the ID for the current conversation. - - Returns: - The current conversation ID or None if not available. - """ - return self._conversation_id - - def generate_conversation_id(self) -> str: - """Generate a new conversation ID. - - Returns: - A new randomly generated UUID string. - """ - return str(uuid.uuid4()) - - -def get_current_conversation_context() -> Optional["Context"]: - """Get the OpenTelemetry context for the current conversation. - - Returns: - The current conversation context or None if not available. - """ - provider = ConversationContextProvider.get_instance() - return provider.get_current_conversation_context() - - -def get_conversation_id() -> Optional[str]: - """Get the ID for the current conversation. - - Returns: - The current conversation ID or None if not available. - """ - provider = ConversationContextProvider.get_instance() - return provider.get_conversation_id() diff --git a/src/pipecat/utils/tracing/service_decorators.py b/src/pipecat/utils/tracing/service_decorators.py index 42babd5cd..a4f0d8c37 100644 --- a/src/pipecat/utils/tracing/service_decorators.py +++ b/src/pipecat/utils/tracing/service_decorators.py @@ -25,7 +25,6 @@ if TYPE_CHECKING: from pipecat.processors.aggregators.llm_context import NOT_GIVEN, LLMContext from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext -from pipecat.utils.tracing.conversation_context_provider import get_current_conversation_context from pipecat.utils.tracing.service_attributes import ( add_gemini_live_span_attributes, add_llm_span_attributes, @@ -34,7 +33,6 @@ from pipecat.utils.tracing.service_attributes import ( add_tts_span_attributes, ) from pipecat.utils.tracing.setup import is_tracing_available -from pipecat.utils.tracing.turn_context_provider import get_current_turn_context if is_tracing_available(): from opentelemetry import context as context_api @@ -76,7 +74,8 @@ def _get_parent_service_context(self): return trace.set_span_in_context(self._span) # Fall back to conversation context if available - conversation_context = get_current_conversation_context() + tracing_ctx = getattr(self, "_tracing_context", None) + conversation_context = tracing_ctx.get_conversation_context() if tracing_ctx else None if conversation_context: return conversation_context @@ -183,7 +182,8 @@ def traced_tts(func: Optional[Callable] = None, *, name: Optional[str] = None) - span_name = "tts" # Get parent context - turn_context = get_current_turn_context() + tracing_ctx = getattr(self, "_tracing_context", None) + turn_context = tracing_ctx.get_turn_context() if tracing_ctx else None parent_context = turn_context or _get_parent_service_context(self) # Create span @@ -290,7 +290,8 @@ def traced_stt(func: Optional[Callable] = None, *, name: Optional[str] = None) - span_name = "stt" # Get the turn context first, then fall back to service context - turn_context = get_current_turn_context() + tracing_ctx = getattr(self, "_tracing_context", None) + turn_context = tracing_ctx.get_turn_context() if tracing_ctx else None parent_context = turn_context or _get_parent_service_context(self) # Create a new span as child of the turn span or service span @@ -372,7 +373,8 @@ def traced_llm(func: Optional[Callable] = None, *, name: Optional[str] = None) - span_name = "llm" # Get the parent context - turn context if available, otherwise service context - turn_context = get_current_turn_context() + tracing_ctx = getattr(self, "_tracing_context", None) + turn_context = tracing_ctx.get_turn_context() if tracing_ctx else None parent_context = turn_context or _get_parent_service_context(self) # Create a new span as child of the turn span or service span @@ -582,7 +584,8 @@ def traced_gemini_live(operation: str) -> Callable: span_name = f"{operation}" # Get the parent context - turn context if available, otherwise service context - turn_context = get_current_turn_context() + tracing_ctx = getattr(self, "_tracing_context", None) + turn_context = tracing_ctx.get_turn_context() if tracing_ctx else None parent_context = turn_context or _get_parent_service_context(self) # Create a new span as child of the turn span or service span @@ -889,7 +892,8 @@ def traced_openai_realtime(operation: str) -> Callable: span_name = f"{operation}" # Get the parent context - turn context if available, otherwise service context - turn_context = get_current_turn_context() + tracing_ctx = getattr(self, "_tracing_context", None) + turn_context = tracing_ctx.get_turn_context() if tracing_ctx else None parent_context = turn_context or _get_parent_service_context(self) # Create a new span as child of the turn span or service span diff --git a/src/pipecat/utils/tracing/tracing_context.py b/src/pipecat/utils/tracing/tracing_context.py new file mode 100644 index 000000000..c84299701 --- /dev/null +++ b/src/pipecat/utils/tracing/tracing_context.py @@ -0,0 +1,109 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Pipeline-scoped tracing context for OpenTelemetry tracing in Pipecat. + +This module provides a per-pipeline tracing context that holds the current +conversation and turn span contexts. Each PipelineTask creates its own +TracingContext, ensuring concurrent pipelines do not interfere with each other. +""" + +import uuid +from typing import TYPE_CHECKING, Optional + +if TYPE_CHECKING: + from opentelemetry.context import Context + from opentelemetry.trace import SpanContext + +from pipecat.utils.tracing.setup import is_tracing_available + +if is_tracing_available(): + from opentelemetry.context import Context + from opentelemetry.trace import NonRecordingSpan, SpanContext, set_span_in_context + + +class TracingContext: + """Pipeline-scoped tracing context. + + Holds the current conversation and turn span contexts for a single pipeline. + Created by PipelineTask, passed to TurnTraceObserver (writer) and services + (readers) via StartFrame. + """ + + def __init__(self): + """Initialize the tracing context with empty state.""" + self._conversation_context: Optional["Context"] = None + self._turn_context: Optional["Context"] = None + self._conversation_id: Optional[str] = None + + def set_conversation_context( + self, span_context: Optional["SpanContext"], conversation_id: Optional[str] = None + ): + """Set the current conversation context. + + Args: + span_context: The span context for the current conversation or None to clear it. + conversation_id: Optional ID for the conversation. + """ + if not is_tracing_available(): + return + + self._conversation_id = conversation_id + + if span_context: + non_recording_span = NonRecordingSpan(span_context) + self._conversation_context = set_span_in_context(non_recording_span) + else: + self._conversation_context = None + + def get_conversation_context(self) -> Optional["Context"]: + """Get the OpenTelemetry context for the current conversation. + + Returns: + The current conversation context or None if not available. + """ + return self._conversation_context + + def set_turn_context(self, span_context: Optional["SpanContext"]): + """Set the current turn context. + + Args: + span_context: The span context for the current turn or None to clear it. + """ + if not is_tracing_available(): + return + + if span_context: + non_recording_span = NonRecordingSpan(span_context) + self._turn_context = set_span_in_context(non_recording_span) + else: + self._turn_context = None + + def get_turn_context(self) -> Optional["Context"]: + """Get the OpenTelemetry context for the current turn. + + Returns: + The current turn context or None if not available. + """ + return self._turn_context + + @property + def conversation_id(self) -> Optional[str]: + """Get the ID for the current conversation. + + Returns: + The current conversation ID or None if not available. + """ + return self._conversation_id + + @staticmethod + def generate_conversation_id() -> str: + """Generate a new conversation ID. + + Returns: + A new randomly generated UUID string. + """ + return str(uuid.uuid4()) diff --git a/src/pipecat/utils/tracing/turn_context_provider.py b/src/pipecat/utils/tracing/turn_context_provider.py deleted file mode 100644 index edb165561..000000000 --- a/src/pipecat/utils/tracing/turn_context_provider.py +++ /dev/null @@ -1,81 +0,0 @@ -# -# Copyright (c) 2024-2026, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - -"""Turn context provider for OpenTelemetry tracing in Pipecat. - -This module provides a singleton context provider that manages the current -turn's tracing context, allowing services to create child spans that are -properly associated with the conversation turn. -""" - -from typing import TYPE_CHECKING, Optional - -# Import types for type checking only -if TYPE_CHECKING: - from opentelemetry.context import Context - from opentelemetry.trace import SpanContext - -from pipecat.utils.tracing.setup import is_tracing_available - -if is_tracing_available(): - from opentelemetry.context import Context - from opentelemetry.trace import NonRecordingSpan, SpanContext, set_span_in_context - - -class TurnContextProvider: - """Provides access to the current turn's tracing context. - - This is a singleton that services can use to get the current turn's - span context to create child spans. - """ - - _instance = None - _current_turn_context: Optional["Context"] = None - - @classmethod - def get_instance(cls): - """Get the singleton instance. - - Returns: - The singleton TurnContextProvider instance. - """ - if cls._instance is None: - cls._instance = TurnContextProvider() - return cls._instance - - def set_current_turn_context(self, span_context: Optional["SpanContext"]): - """Set the current turn context. - - Args: - span_context: The span context for the current turn or None to clear it. - """ - if not is_tracing_available(): - return - - if span_context: - # Create a non-recording span from the span context - non_recording_span = NonRecordingSpan(span_context) - self._current_turn_context = set_span_in_context(non_recording_span) - else: - self._current_turn_context = None - - def get_current_turn_context(self) -> Optional["Context"]: - """Get the OpenTelemetry context for the current turn. - - Returns: - The current turn context or None if not available. - """ - return self._current_turn_context - - -def get_current_turn_context() -> Optional["Context"]: - """Get the OpenTelemetry context for the current turn. - - Returns: - The current turn context or None if not available. - """ - provider = TurnContextProvider.get_instance() - return provider.get_current_turn_context() diff --git a/src/pipecat/utils/tracing/turn_trace_observer.py b/src/pipecat/utils/tracing/turn_trace_observer.py index 78a9c9ea9..83c2bcdc2 100644 --- a/src/pipecat/utils/tracing/turn_trace_observer.py +++ b/src/pipecat/utils/tracing/turn_trace_observer.py @@ -19,9 +19,8 @@ from pipecat.frames.frames import StartFrame from pipecat.observers.base_observer import BaseObserver, FramePushed from pipecat.observers.turn_tracking_observer import TurnTrackingObserver from pipecat.observers.user_bot_latency_observer import UserBotLatencyObserver -from pipecat.utils.tracing.conversation_context_provider import ConversationContextProvider from pipecat.utils.tracing.setup import is_tracing_available -from pipecat.utils.tracing.turn_context_provider import TurnContextProvider +from pipecat.utils.tracing.tracing_context import TracingContext # Import types for type checking only if TYPE_CHECKING: @@ -49,6 +48,7 @@ class TurnTraceObserver(BaseObserver): latency_tracker: UserBotLatencyObserver, conversation_id: Optional[str] = None, additional_span_attributes: Optional[dict] = None, + tracing_context: Optional[TracingContext] = None, **kwargs, ): """Initialize the turn trace observer. @@ -58,11 +58,13 @@ class TurnTraceObserver(BaseObserver): latency_tracker: The latency tracking observer for user-bot latency. conversation_id: Optional conversation ID for grouping turns. additional_span_attributes: Additional attributes to add to spans. + tracing_context: Pipeline-scoped tracing context for span hierarchy. **kwargs: Additional arguments passed to parent class. """ super().__init__(**kwargs) self._turn_tracker = turn_tracker self._latency_tracker = latency_tracker + self._tracing_context = tracing_context or TracingContext() self._current_span: Optional["Span"] = None self._current_turn_number: int = 0 self._trace_context_map: Dict[int, "SpanContext"] = {} @@ -123,9 +125,8 @@ class TurnTraceObserver(BaseObserver): return # Generate a conversation ID if not provided - context_provider = ConversationContextProvider.get_instance() if conversation_id is None: - conversation_id = context_provider.generate_conversation_id() + conversation_id = TracingContext.generate_conversation_id() logger.debug(f"Generated new conversation ID: {conversation_id}") self._conversation_id = conversation_id @@ -140,8 +141,8 @@ class TurnTraceObserver(BaseObserver): 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( + # Update the tracing context + self._tracing_context.set_conversation_context( self._conversation_span.get_span_context(), conversation_id ) @@ -161,9 +162,8 @@ class TurnTraceObserver(BaseObserver): self._current_span.end() self._current_span = None - # Clear the turn context provider - context_provider = TurnContextProvider.get_instance() - context_provider.set_current_turn_context(None) + # Clear the turn context + self._tracing_context.set_turn_context(None) # Now end the conversation span if it exists if self._conversation_span: @@ -171,9 +171,8 @@ class TurnTraceObserver(BaseObserver): self._conversation_span.end() self._conversation_span = None - # Clear the context provider - context_provider = ConversationContextProvider.get_instance() - context_provider.set_current_conversation_context(None) + # Clear the conversation context + self._tracing_context.set_conversation_context(None) logger.debug(f"Ended tracing for Conversation {self._conversation_id}") self._conversation_id = None @@ -189,8 +188,7 @@ class TurnTraceObserver(BaseObserver): # Get the parent context - conversation if available, otherwise use root context parent_context = None if self._conversation_span: - context_provider = ConversationContextProvider.get_instance() - parent_context = context_provider.get_current_conversation_context() + parent_context = self._tracing_context.get_conversation_context() # Create a new span for this turn self._current_span = self._tracer.start_span("turn", context=parent_context) @@ -207,9 +205,8 @@ class TurnTraceObserver(BaseObserver): # Store the span context so services can become children of this span self._trace_context_map[turn_number] = self._current_span.get_span_context() - # Update the context provider so services can access this span - context_provider = TurnContextProvider.get_instance() - context_provider.set_current_turn_context(self._current_span.get_span_context()) + # Update the tracing context so services can access this span + self._tracing_context.set_turn_context(self._current_span.get_span_context()) logger.debug(f"Started tracing for Turn {turn_number}") @@ -228,9 +225,8 @@ class TurnTraceObserver(BaseObserver): self._current_span.end() self._current_span = None - # Clear the context provider - context_provider = TurnContextProvider.get_instance() - context_provider.set_current_turn_context(None) + # Clear the turn context + self._tracing_context.set_turn_context(None) logger.debug(f"Ended tracing for Turn {turn_number}")