Fix orphan spans in tracing during flow initialization and transitions
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
@@ -32,6 +32,7 @@ from pipecat.utils.tracing.service_attributes import (
|
||||
add_stt_span_attributes,
|
||||
add_tts_span_attributes,
|
||||
)
|
||||
from pipecat.utils.tracing.conversation_context_provider import get_current_conversation_context
|
||||
from pipecat.utils.tracing.setup import is_tracing_available
|
||||
from pipecat.utils.tracing.turn_context_provider import get_current_turn_context
|
||||
|
||||
@@ -58,7 +59,8 @@ def _noop_decorator(func):
|
||||
def _get_parent_service_context(self):
|
||||
"""Get the parent service span context (internal use only).
|
||||
|
||||
This looks for the service span that was created when the service was initialized.
|
||||
This looks for the service span that was created when the service was initialized,
|
||||
or falls back to the conversation context if available.
|
||||
|
||||
Args:
|
||||
self: The service instance.
|
||||
@@ -73,7 +75,12 @@ def _get_parent_service_context(self):
|
||||
if hasattr(self, "_span") and self._span:
|
||||
return trace.set_span_in_context(self._span)
|
||||
|
||||
# If we can't find a stored span, default to current context
|
||||
# Fall back to conversation context if available
|
||||
conversation_context = get_current_conversation_context()
|
||||
if conversation_context:
|
||||
return conversation_context
|
||||
|
||||
# Last resort: use current context (may create orphan spans)
|
||||
return context_api.get_current()
|
||||
|
||||
|
||||
|
||||
@@ -15,6 +15,7 @@ from typing import TYPE_CHECKING, Dict, Optional
|
||||
|
||||
from loguru import logger
|
||||
|
||||
from pipecat.frames.frames import StartFrame
|
||||
from pipecat.observers.base_observer import BaseObserver, FramePushed
|
||||
from pipecat.observers.turn_tracking_observer import TurnTrackingObserver
|
||||
from pipecat.utils.tracing.conversation_context_provider import ConversationContextProvider
|
||||
@@ -81,13 +82,15 @@ class TurnTraceObserver(BaseObserver):
|
||||
async def on_push_frame(self, data: FramePushed):
|
||||
"""Process a frame without modifying it.
|
||||
|
||||
This observer doesn't need to process individual frames as it
|
||||
relies on turn start/end events from the turn tracker.
|
||||
Handles StartFrame to begin conversation tracing early, ensuring
|
||||
that any spans created before Turn 1 (e.g., from flow initialization)
|
||||
are properly attached to the conversation trace.
|
||||
|
||||
Args:
|
||||
data: The frame push event data.
|
||||
"""
|
||||
pass
|
||||
if isinstance(data.frame, StartFrame) and not self._conversation_span:
|
||||
self.start_conversation_tracing(self._conversation_id)
|
||||
|
||||
def start_conversation_tracing(self, conversation_id: Optional[str] = None):
|
||||
"""Start a new conversation span.
|
||||
@@ -159,8 +162,6 @@ class TurnTraceObserver(BaseObserver):
|
||||
if not is_tracing_available() or not self._tracer:
|
||||
return
|
||||
|
||||
# If this is the first turn and no conversation span exists yet,
|
||||
# start the conversation tracing (will generate ID if needed)
|
||||
if turn_number == 1 and not self._conversation_span:
|
||||
self.start_conversation_tracing(self._conversation_id)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user