From 25ca2964777311d81b955816c7d5ac67da86f793 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 13 Feb 2026 11:21:24 -0500 Subject: [PATCH] Move tracing fields to AIService and extract _get_turn_context helper Consolidate _tracing_enabled and _tracing_context from LLMService, STTService, and TTSService into the shared AIService base class. Extract _get_turn_context() helper in service_decorators.py to encapsulate the repeated pattern across all traced decorators. --- src/pipecat/services/ai_service.py | 5 ++- src/pipecat/services/llm_service.py | 3 -- src/pipecat/services/stt_service.py | 3 -- src/pipecat/services/tts_service.py | 4 --- .../utils/tracing/service_decorators.py | 33 ++++++++++--------- 5 files changed, 22 insertions(+), 26 deletions(-) diff --git a/src/pipecat/services/ai_service.py b/src/pipecat/services/ai_service.py index c03ab9d0e..52b42663f 100644 --- a/src/pipecat/services/ai_service.py +++ b/src/pipecat/services/ai_service.py @@ -44,6 +44,8 @@ class AIService(FrameProcessor): self._model_name: str = "" self._settings: Dict[str, Any] = {} self._session_properties: Dict[str, Any] = {} + self._tracing_enabled: bool = False + self._tracing_context = None @property def model_name(self) -> str: @@ -72,7 +74,8 @@ class AIService(FrameProcessor): Args: frame: The start frame containing initialization parameters. """ - pass + self._tracing_enabled = frame.enable_tracing + self._tracing_context = frame.tracing_context async def stop(self, frame: EndFrame): """Stop the AI service. diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index acbd6baae..c8af00b80 100644 --- a/src/pipecat/services/llm_service.py +++ b/src/pipecat/services/llm_service.py @@ -198,7 +198,6 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService): self._functions: Dict[Optional[str], FunctionCallRegistryItem] = {} self._function_call_tasks: Dict[Optional[asyncio.Task], FunctionCallRunnerItem] = {} self._sequential_runner_task: Optional[asyncio.Task] = None - self._tracing_enabled: bool = False self._skip_tts: Optional[bool] = None self._summary_task: Optional[asyncio.Task] = None @@ -285,8 +284,6 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService): await super().start(frame) 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 81f369572..1d8d1590f 100644 --- a/src/pipecat/services/stt_service.py +++ b/src/pipecat/services/stt_service.py @@ -102,7 +102,6 @@ class STTService(AIService): self._init_sample_rate = sample_rate self._sample_rate = 0 self._settings: Dict[str, Any] = {} - self._tracing_enabled: bool = False self._muted: bool = False self._user_id: str = "" self._ttfs_p99_latency = ttfs_p99_latency @@ -202,8 +201,6 @@ 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 00e3e81f8..02c799d0f 100644 --- a/src/pipecat/services/tts_service.py +++ b/src/pipecat/services/tts_service.py @@ -208,8 +208,6 @@ class TTSService(AIService): # TODO: Deprecate _text_filters when added to LLMTextProcessor self._text_filters: Sequence[BaseTextFilter] = text_filters or [] self._transport_destination: Optional[str] = transport_destination - self._tracing_enabled: bool = False - if text_filter: import warnings @@ -349,8 +347,6 @@ class TTSService(AIService): self._sample_rate = self._init_sample_rate or frame.audio_out_sample_rate 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/service_decorators.py b/src/pipecat/utils/tracing/service_decorators.py index a4f0d8c37..fae7f7e77 100644 --- a/src/pipecat/utils/tracing/service_decorators.py +++ b/src/pipecat/utils/tracing/service_decorators.py @@ -54,6 +54,19 @@ def _noop_decorator(func): return func +def _get_turn_context(self): + """Get the current turn's tracing context if available. + + Args: + self: The service instance. + + Returns: + The turn context, or None if unavailable. + """ + tracing_ctx = getattr(self, "_tracing_context", None) + return tracing_ctx.get_turn_context() if tracing_ctx else None + + def _get_parent_service_context(self): """Get the parent service span context (internal use only). @@ -182,9 +195,7 @@ def traced_tts(func: Optional[Callable] = None, *, name: Optional[str] = None) - span_name = "tts" # Get parent 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) + parent_context = _get_turn_context(self) or _get_parent_service_context(self) # Create span tracer = trace.get_tracer("pipecat") @@ -290,9 +301,7 @@ 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 - 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) + parent_context = _get_turn_context(self) or _get_parent_service_context(self) # Create a new span as child of the turn span or service span tracer = trace.get_tracer("pipecat") @@ -373,9 +382,7 @@ 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 - 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) + parent_context = _get_turn_context(self) or _get_parent_service_context(self) # Create a new span as child of the turn span or service span tracer = trace.get_tracer("pipecat") @@ -584,9 +591,7 @@ def traced_gemini_live(operation: str) -> Callable: span_name = f"{operation}" # Get the parent context - turn context if available, otherwise service 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) + parent_context = _get_turn_context(self) or _get_parent_service_context(self) # Create a new span as child of the turn span or service span tracer = trace.get_tracer("pipecat") @@ -892,9 +897,7 @@ def traced_openai_realtime(operation: str) -> Callable: span_name = f"{operation}" # Get the parent context - turn context if available, otherwise service 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) + parent_context = _get_turn_context(self) or _get_parent_service_context(self) # Create a new span as child of the turn span or service span tracer = trace.get_tracer("pipecat")