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

Move tracing fields to AIService and extract _get_turn_context helper

Browse Source 
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.
This commit is contained in:
Mark Backman
2026-02-13 11:21:24 -05:00
parent 71a752c971
commit 25ca296477
5 changed files with 22 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
src/pipecat/services/ai_service.py
View File

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

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

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

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

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

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

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

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

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