wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Replace singleton context providers with pipeline-scoped TracingContext

Browse Source 
ConversationContextProvider and TurnContextProvider were singletons that
stored tracing context as class-level state. When two PipelineTask instances
ran concurrently, they would overwrite each other's context, causing service
spans to attach to the wrong pipeline's turn span.

Replace both singletons with a single TracingContext object owned by each
PipelineTask, threaded to services via StartFrame.
This commit is contained in:
Mark Backman
2026-02-11 21:58:10 -05:00
parent d99a256715
commit 358f237507
10 changed files with 149 additions and 223 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
src/pipecat/frames/frames.py
View File

@@ -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

6
src/pipecat/pipeline/task.py
View File

@@ -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)

1
src/pipecat/services/llm_service.py
View File

@@ -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.

1
src/pipecat/services/stt_service.py
View File

@@ -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."""

1
src/pipecat/services/tts_service.py
View File

@@ -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.

114
src/pipecat/utils/tracing/conversation_context_provider.py
View File

@@ -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()

20
src/pipecat/utils/tracing/service_decorators.py
View File

@@ -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

109
src/pipecat/utils/tracing/tracing_context.py Normal file
View File

@@ -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())

81
src/pipecat/utils/tracing/turn_context_provider.py
View File

@@ -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()

36
src/pipecat/utils/tracing/turn_trace_observer.py
View File

@@ -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}")