diff --git a/changelog/4467.fixed.2.md b/changelog/4467.fixed.2.md new file mode 100644 index 000000000..80de8a289 --- /dev/null +++ b/changelog/4467.fixed.2.md @@ -0,0 +1 @@ +- Fixed incorrect `metrics.ttfb` on STT OpenTelemetry spans, and parented them to the current turn span. diff --git a/changelog/4467.fixed.3.md b/changelog/4467.fixed.3.md new file mode 100644 index 000000000..628623c16 --- /dev/null +++ b/changelog/4467.fixed.3.md @@ -0,0 +1 @@ +- Fixed missing `output` attribute on LLM OpenTelemetry spans when the LLM call is interrupted mid-stream. diff --git a/changelog/4467.fixed.md b/changelog/4467.fixed.md new file mode 100644 index 000000000..a365da620 --- /dev/null +++ b/changelog/4467.fixed.md @@ -0,0 +1 @@ +- Fixed incorrect `metrics.ttfb` on TTS OpenTelemetry spans for streaming services. diff --git a/src/pipecat/utils/tracing/service_decorators.py b/src/pipecat/utils/tracing/service_decorators.py index f4de1dd85..c7e7583d0 100644 --- a/src/pipecat/utils/tracing/service_decorators.py +++ b/src/pipecat/utils/tracing/service_decorators.py @@ -11,7 +11,6 @@ rich information about service execution including configuration, parameters, and performance metrics. """ -import contextlib import functools import inspect import json @@ -24,7 +23,16 @@ if TYPE_CHECKING: from opentelemetry import context as context_api from opentelemetry import trace +from pipecat.frames.frames import ( + MetricsFrame, + TranscriptionFrame, + TTSStoppedFrame, + UserStoppedSpeakingFrame, + VADUserStartedSpeakingFrame, +) +from pipecat.metrics.metrics import TTFBMetricsData from pipecat.processors.aggregators.llm_context import NOT_GIVEN +from pipecat.processors.frame_processor import FrameDirection from pipecat.utils.tracing.service_attributes import ( add_gemini_live_span_attributes, add_llm_span_attributes, @@ -175,6 +183,13 @@ def traced_tts(func: Callable | None = None, *, name: str | None = None) -> Call - Character count and text content - Performance metrics like TTFB + The span is scoped to the full synthesis operation, from + ``create_audio_context`` until ``TTSStoppedFrame`` (or + ``remove_audio_context`` as a safety net), so TTFB and any other + runtime-computed metrics land on the correct span even when audio + chunks are delivered after ``run_tts`` returns (e.g. WebSocket + streaming TTS services). + Works with both async functions and generators. Args: @@ -190,102 +205,224 @@ def traced_tts(func: Callable | None = None, *, name: str | None = None) -> Call def decorator(f): is_async_generator = inspect.isasyncgenfunction(f) - @contextlib.asynccontextmanager - async def tracing_context(self, text): - """Async context manager for TTS tracing. + def end_tts_span(service, context_id, *, interrupted=False): + """End the TTS span for ``context_id`` if still open. Idempotent.""" + entry = service._tts_spans.pop(context_id, None) + if not entry: + return + try: + span = entry["span"] + if interrupted: + span.set_attribute("tts.interrupted", True) + span.end() + except Exception as e: + logging.warning(f"Error closing TTS span: {e}") - Args: - self: The TTS service instance. - text: The text being synthesized. + def install_audio_context_patches(service): + """Install per-instance wrappers on the audio-context methods. - Yields: - The active span for the TTS operation. + The wrappers own the lifetime of the TTS span: + + - ``create_audio_context``: opens the span and records + baseline attributes. + - ``append_to_audio_context``: ends the span on + ``TTSStoppedFrame``. + - ``push_frame``: records ``metrics.ttfb`` from the + canonical ``TTFBMetricsData`` payload of any + ``MetricsFrame`` pushed by ``stop_ttfb_metrics``. Reading + the value from the metrics event (instead of polling + ``_metrics.ttfb`` when the first audio is queued) avoids + the ``ttfb`` property's in-progress fallback, which would + otherwise report an under-estimate whenever a context's + audio waits behind earlier queued audio before + ``_handle_audio_context`` actually stops the TTFB + measurement. + - ``remove_audio_context``: ends any still-open span as a + safety net for error and cancellation paths. + - ``on_audio_context_completed``: ends the span on natural + completion. Needed because services that rely on the + base class to auto-push ``TTSStoppedFrame`` (via + ``push_frame`` in ``_handle_audio_context``) bypass the + ``append_to_audio_context`` hook entirely. + - ``reset_active_audio_context``: ends the currently + playing context's span if still open. Always called from + ``_handle_interruption``, so this is the interruption + hook. + + The patches check ``_tracing_enabled`` at invocation time, + so they are safe to install regardless of whether tracing + is enabled. """ - # Check if tracing is enabled for this service instance - if not getattr(self, "_tracing_enabled", False): - yield None + if getattr(service, "__tts_tracing_patches_installed__", False): + return + service.__tts_tracing_patches_installed__ = True + service._tts_spans = {} + + orig_create = service.create_audio_context + orig_append = service.append_to_audio_context + orig_remove = service.remove_audio_context + orig_completed = service.on_audio_context_completed + orig_reset_active = service.reset_active_audio_context + orig_push_frame = service.push_frame + + async def traced_create_audio_context(context_id): + if getattr(service, "_tracing_enabled", False): + try: + parent = _get_turn_context(service) or _get_parent_service_context(service) + tracer = trace.get_tracer("pipecat") + span = tracer.start_span("tts", context=parent) + service._tts_spans[context_id] = {"span": span, "ttfb_recorded": False} + + settings = getattr(service, "_settings", None) + add_tts_span_attributes( + span=span, + service_name=service.__class__.__name__, + model=_get_model_name(service), + voice_id=getattr(settings, "voice", "unknown"), + settings=settings, + operation_name="tts", + ) + except Exception as e: + logging.warning(f"Error opening TTS span: {e}") + return await orig_create(context_id) + + async def traced_append_to_audio_context(context_id, frame): + entry = service._tts_spans.get(context_id) + if entry and frame is not None: + try: + if isinstance(frame, TTSStoppedFrame): + entry["span"].end() + service._tts_spans.pop(context_id, None) + except Exception as e: + logging.warning(f"Error updating TTS span: {e}") + return await orig_append(context_id, frame) + + async def traced_push_frame(frame, direction=FrameDirection.DOWNSTREAM): + await orig_push_frame(frame, direction) + if not getattr(service, "_tracing_enabled", False): + return + if not isinstance(frame, MetricsFrame): + return + try: + playing_id = getattr(service, "_playing_context_id", None) + if playing_id is None: + return + entry = service._tts_spans.get(playing_id) + if not entry or entry["ttfb_recorded"]: + return + for data in frame.data: + if isinstance(data, TTFBMetricsData): + entry["span"].set_attribute("metrics.ttfb", data.value) + entry["ttfb_recorded"] = True + break + except Exception as e: + logging.warning(f"Error recording TTS ttfb from MetricsFrame: {e}") + + async def traced_remove_audio_context(context_id): + entry = service._tts_spans.pop(context_id, None) + if entry: + try: + entry["span"].end() + except Exception as e: + logging.warning(f"Error closing TTS span: {e}") + return await orig_remove(context_id) + + async def traced_on_audio_context_completed(context_id): + end_tts_span(service, context_id) + return await orig_completed(context_id) + + def traced_reset_active_audio_context(): + playing_id = getattr(service, "_playing_context_id", None) + if playing_id is not None: + end_tts_span(service, playing_id, interrupted=True) + return orig_reset_active() + + service.create_audio_context = traced_create_audio_context + service.append_to_audio_context = traced_append_to_audio_context + service.push_frame = traced_push_frame + service.remove_audio_context = traced_remove_audio_context + service.on_audio_context_completed = traced_on_audio_context_completed + service.reset_active_audio_context = traced_reset_active_audio_context + + def patch_setup(owner): + """Wrap ``owner.setup`` so audio-context patches install per-instance. + + Idempotent: if a parent class has already been wrapped, + skip. The patches check ``_tracing_enabled`` at invocation + time, so wrapping is always safe. + """ + original_setup = owner.setup + if getattr(original_setup, "__tts_tracing_setup_wrapped__", False): return - service_class_name = self.__class__.__name__ - span_name = "tts" + @functools.wraps(original_setup) + async def patched_setup(self, setup): + await original_setup(self, setup) + install_audio_context_patches(self) - # Get parent context - parent_context = _get_turn_context(self) or _get_parent_service_context(self) + setattr(patched_setup, "__tts_tracing_setup_wrapped__", True) + owner.setup = patched_setup - # Create span - tracer = trace.get_tracer("pipecat") - with tracer.start_as_current_span(span_name, context=parent_context) as span: - try: - settings = getattr(self, "_settings", None) - add_tts_span_attributes( - span=span, - service_name=service_class_name, - model=_get_model_name(self), - voice_id=getattr(settings, "voice", "unknown"), - text=text, - settings=settings, - character_count=len(text), - operation_name="tts", - cartesia_version=getattr(self, "_cartesia_version", None), - context_id=getattr(self, "_context_id", None), - ) + def attach_run_tts_attributes(service, text, args, kwargs): + """Attach text-specific attributes to the in-flight TTS span.""" + if not getattr(service, "_tracing_enabled", False): + return + try: + context_id = args[0] if args else kwargs.get("context_id") + entry = getattr(service, "_tts_spans", {}).get(context_id) + if entry and text: + span = entry["span"] + span.set_attribute("text", text) + span.set_attribute("metrics.character_count", len(text)) + except Exception as e: + logging.warning(f"Error attaching TTS text to span: {e}") - yield span + def make_run_tts_wrapper(): + """Build the wrapper around ``run_tts`` that adds per-call attributes. - except Exception as e: - logging.warning(f"Error in TTS tracing: {e}") - raise - finally: - # Update TTFB metric at the end - ttfb: float | None = getattr(getattr(self, "_metrics", None), "ttfb", None) - if ttfb is not None: - span.set_attribute("metrics.ttfb", ttfb) + Span lifetime is owned by the audio-context patches. This + wrapper only attaches the text and character count to the + span that was opened by ``create_audio_context`` just + before ``run_tts`` was invoked. + """ + if is_async_generator: - if is_async_generator: - - @functools.wraps(f) - async def gen_wrapper(self, text, *args, **kwargs): - if not getattr(self, "_tracing_enabled", False): - async for item in f(self, text, *args, **kwargs): - yield item - return - - fn_called = False - try: - async with tracing_context(self, text): - fn_called = True - async for item in f(self, text, *args, **kwargs): - yield item - except Exception as e: - if fn_called: - raise - logging.error(f"Error in TTS tracing (continuing without tracing): {e}") + @functools.wraps(f) + async def gen_wrapper(self, text, *args, **kwargs): + attach_run_tts_attributes(self, text, args, kwargs) async for item in f(self, text, *args, **kwargs): yield item - return gen_wrapper - else: + return gen_wrapper @functools.wraps(f) - async def wrapper(self, text, *args, **kwargs): - if not getattr(self, "_tracing_enabled", False): - return await f(self, text, *args, **kwargs) + async def coro_wrapper(self, text, *args, **kwargs): + attach_run_tts_attributes(self, text, args, kwargs) + return await f(self, text, *args, **kwargs) - fn_called = False - try: - async with tracing_context(self, text): - fn_called = True - return await f(self, text, *args, **kwargs) - except Exception as e: - if fn_called: - raise - logging.error(f"Error in TTS tracing (continuing without tracing): {e}") - return await f(self, text, *args, **kwargs) + return coro_wrapper - return wrapper + class _TracedTTSDescriptor: + """Class-level descriptor that wires up TTS tracing at class definition time. + + ``__set_name__`` fires when the class body finishes evaluating, + giving us a chance to wrap the owner's ``setup()`` so that the + audio-context patches install on every instance before any + ``create_audio_context`` call (including the very first one). + """ + + def __set_name__(self, owner, attr_name): + patch_setup(owner) + setattr(owner, attr_name, make_run_tts_wrapper()) + + return _TracedTTSDescriptor() if func is not None: - return decorator(func) + # ``decorator(func)`` returns a descriptor placeholder that + # Python replaces with the real wrapped function once + # ``__set_name__`` runs at class definition time. Pyright sees + # only the descriptor instance, hence the ignore. + return decorator(func) # type: ignore[return-value] return decorator @@ -299,72 +436,285 @@ def traced_stt(func: Callable | None = None, *, name: str | None = None) -> Call - Language information - Performance metrics like TTFB + The span is scoped to one STT segment, from + ``VADUserStartedSpeakingFrame`` (or the first ``TranscriptionFrame`` + when VAD did not fire, e.g. whispered speech) until a finalized + ``TranscriptionFrame``. Multiple finalized transcripts in a single + user turn produce multiple sequential spans, each anchored at the + point speech for that segment began. ``metrics.ttfb`` is read after + the base ``push_frame`` runs ``stop_ttfb_metrics`` for the + finalized frame, so the value is correct for the closing span. + Args: func: The STT method to trace. name: Custom span name. Defaults to function name. Returns: - Wrapped method with STT-specific tracing. + The original method unchanged. The decorator's class-definition- + time work is to install a ``push_frame`` wrapper on the owning + class that owns the span lifetime. """ if not is_tracing_available(): return _noop_decorator if func is None else _noop_decorator(func) def decorator(f): - @functools.wraps(f) - async def wrapper(self, transcript, is_final, language=None): - if not getattr(self, "_tracing_enabled", False): - return await f(self, transcript, is_final, language) + def patch_push_frame(owner): + """Wrap ``owner.push_frame`` to drive the STT span lifecycle. - fn_called = False - try: - service_class_name = self.__class__.__name__ - span_name = "stt" + Idempotent: if a parent class has already been wrapped, skip. + The wrapper checks ``_tracing_enabled`` at invocation time, + so it is safe to install regardless of whether tracing is + enabled. + """ + original_push_frame = owner.push_frame + if getattr(original_push_frame, "__stt_tracing_push_frame_wrapped__", False): + return - # Get the turn context first, then fall back to service context - parent_context = _get_turn_context(self) or _get_parent_service_context(self) + def update_transcript(state, new_text): + """Append or extend the current segment in ``state['segments']``. - # Create a new span as child of the turn span or service span + If ``new_text`` starts with the last recorded segment, + treat it as a continuation (interim accumulation) and + replace the last segment. Otherwise treat it as a new + segment and append. Some STT services (Deepgram with + utterance_end_ms enabled, for example) emit several + ``TranscriptionFrame``s per turn where each carries a + different segment rather than a cumulative update — + without this logic the span's transcript would only + show the last segment and the beginning would be lost. + """ + if not new_text: + return + segments = state["segments"] + if not segments: + segments.append(new_text) + elif new_text.startswith(segments[-1]): + segments[-1] = new_text + else: + segments.append(new_text) + + def open_span(service, state): + """Open the STT span, anchored at ``segment_start_time`` if set.""" + parent = _get_turn_context(service) or _get_parent_service_context(service) tracer = trace.get_tracer("pipecat") - with tracer.start_as_current_span( - span_name, context=parent_context - ) as current_span: + start_time_ns = ( + int(state["segment_start_time"] * 1e9) + if state["segment_start_time"] is not None + else None + ) + span = tracer.start_span("stt", context=parent, start_time=start_time_ns) + try: + settings = getattr(service, "_settings", None) + add_stt_span_attributes( + span=span, + service_name=service.__class__.__name__, + model=_get_model_name(service), + settings=settings, + vad_enabled=getattr(service, "vad_enabled", False), + ) + except Exception as e: + logging.warning(f"Error setting STT span baseline attributes: {e}") + state["span"] = span + + def handle_pre_push(service, frame, state): + """Record speech-start anchor; lazy-open span on first transcript. + + Lazy-opening on ``TranscriptionFrame`` (rather than on + ``VADUserStartedSpeakingFrame`` or + ``UserStartedSpeakingFrame``) avoids racing with + ``TurnTraceObserver._handle_turn_started``, which runs + in a background task fired by ``_call_event_handler`` + (``base_object.py:232``) and may not have set the new + turn's context yet — that produces STT spans parented + to the previous turn. By the time STT actually emits + a transcript, the turn observer has run. + + Opening happens in pre-push (rather than post-push) so + that the recursive ``push_frame`` that + ``STTService.push_frame`` triggers for the + ``MetricsFrame`` (via ``stop_ttfb_metrics`` at + ``stt_service.py:465``) sees the span already open and + can attribute ``metrics.ttfb`` to it. + """ + if isinstance(frame, VADUserStartedSpeakingFrame): + # Anchor the next span at the moment speech began. + # Skip if we already have an anchor (intra-turn VAD + # re-trigger) or a span open. + if state["span"] is None and state["segment_start_time"] is None: + state["segment_start_time"] = frame.timestamp - frame.start_secs + elif isinstance(frame, TranscriptionFrame) and state["span"] is None: + open_span(service, state) + + async def handle_post_push(service, frame, state): + """Attach per-frame attrs; close on finalized; record TTFB from MetricsFrame. + + ``metrics.ttfb`` is read off the ``TTFBMetricsData`` + payload of any ``MetricsFrame`` pushed by + ``stop_ttfb_metrics`` — the canonical value the rest + of the system uses — rather than from + ``_metrics.ttfb``, which has an in-progress fallback + branch (``frame_processor_metrics.py:48-62``) that + would return an under-estimate if read at the wrong + time. + + One STT span per finalized transcript: the span opens + lazily on the first ``TranscriptionFrame`` (pre-push, + anchored at speech start via ``segment_start_time``) + and closes on ``finalized=True``. Multiple finalized + transcripts in a single turn produce multiple spans. + + For services that never set ``frame.finalized=True`` + (e.g. Deepgram, which only marks it via + ``confirm_finalize()``), the span closes on + ``UserStoppedSpeakingFrame``. To capture + ``metrics.ttfb`` for those spans we force-stop any + pending TTFB measurement before closing — that pushes + a ``MetricsFrame``, our post-push attributes the + value, and ``patched_stop_ttfb_metrics`` closes the + span. The ``stt.incomplete=true`` flag is only set if + neither a finalized transcript nor a TTFB measurement + ever finalized for the span. + """ + if isinstance(frame, UserStoppedSpeakingFrame): + prev_span = state["span"] + if prev_span is None: + return + metrics = getattr(service, "_metrics", None) + if metrics is not None and getattr(metrics, "_start_ttfb_time", 0) > 0: + last_transcript_time = getattr(service, "_last_transcript_time", 0) or None + try: + await service.stop_ttfb_metrics(end_time=last_transcript_time) + except Exception as e: + logging.warning(f"Error force-stopping STT TTFB on user turn end: {e}") + # patched_stop_ttfb_metrics may have closed the span + # via the timeout path; re-check. + if state["span"] is None: + state["segments"] = [] + return + state["span"].set_attribute("stt.incomplete", True) + state["span"].end() + state["span"] = None + state["segment_start_time"] = None + state["segments"] = [] + elif isinstance(frame, MetricsFrame): + span = state["span"] + if span is None: + return + for data in frame.data: + if isinstance(data, TTFBMetricsData): + span.set_attribute("metrics.ttfb", data.value) + break + elif isinstance(frame, TranscriptionFrame): + span = state["span"] + if span is None: + return + if frame.text: + update_transcript(state, frame.text) + span.set_attribute("transcript", " ".join(state["segments"]).strip()) + span.set_attribute("is_final", bool(frame.finalized)) + if frame.language: + span.set_attribute("language", str(frame.language)) + if frame.user_id: + span.set_attribute("user_id", frame.user_id) + if frame.finalized: + span.end() + state["span"] = None + state["segment_start_time"] = None + state["segments"] = [] + + @functools.wraps(original_push_frame) + async def patched_push_frame(self, frame, direction=FrameDirection.DOWNSTREAM): + state = getattr(self, "_stt_span_state", None) + if state is None: + state = {"span": None, "segment_start_time": None, "segments": []} + self._stt_span_state = state + + if getattr(self, "_tracing_enabled", False): try: - # Get TTFB metric if available - ttfb: float | None = getattr(getattr(self, "_metrics", None), "ttfb", None) - - # Use settings from the service if available - settings = getattr(self, "_settings", None) - - add_stt_span_attributes( - span=current_span, - service_name=service_class_name, - model=_get_model_name(self), - transcript=transcript, - is_final=is_final, - language=str(language) if language else None, - user_id=getattr(self, "_user_id", None), - vad_enabled=getattr(self, "vad_enabled", False), - settings=settings, - ttfb=ttfb, - ) - - # Call the original function - fn_called = True - return await f(self, transcript, is_final, language) + handle_pre_push(self, frame, state) except Exception as e: - # Log any exception but don't disrupt the main flow - logging.warning(f"Error in STT transcription tracing: {e}") - raise - except Exception as e: - if fn_called: - raise - logging.error(f"Error in STT tracing (continuing without tracing): {e}") - return await f(self, transcript, is_final, language) + logging.warning(f"Error in STT pre-push tracing: {e}") - return wrapper + await original_push_frame(self, frame, direction) + + if getattr(self, "_tracing_enabled", False): + try: + await handle_post_push(self, frame, state) + except Exception as e: + logging.warning(f"Error in STT post-push tracing: {e}") + + setattr(patched_push_frame, "__stt_tracing_push_frame_wrapped__", True) + owner.push_frame = patched_push_frame + + def patch_stop_ttfb_metrics(owner): + """Wrap ``owner.stop_ttfb_metrics`` to close the span on the timeout path. + + When ``stop_ttfb_metrics`` is invoked with ``end_time`` set, + that signals the TTFB-timeout handler firing + (`stt_service.py:566`), or our own force-stop from the + ``UserStoppedSpeakingFrame`` handler. In either case we + anchor the span's end at ``end_time`` + (= ``_last_transcript_time``) rather than at whenever the + coroutine resumed. + + ``metrics.ttfb`` attribution is not done here — the + ``MetricsFrame`` that ``stop_ttfb_metrics`` pushes flows + through ``push_frame`` and gets recorded by + ``handle_post_push``, which reads the canonical + ``TTFBMetricsData.value`` rather than the in-progress + ``_metrics.ttfb`` property. + """ + original_stop = owner.stop_ttfb_metrics + if getattr(original_stop, "__stt_tracing_stop_ttfb_wrapped__", False): + return + + @functools.wraps(original_stop) + async def patched_stop(self, *, end_time=None): + await original_stop(self, end_time=end_time) + if end_time is None: + return + if not getattr(self, "_tracing_enabled", False): + return + state = getattr(self, "_stt_span_state", None) + if not state or state["span"] is None: + return + try: + span = state["span"] + span.end(end_time=int(end_time * 1e9)) + state["span"] = None + state["segment_start_time"] = None + state["segments"] = [] + except Exception as e: + logging.warning(f"Error in STT stop_ttfb_metrics tracing: {e}") + + setattr(patched_stop, "__stt_tracing_stop_ttfb_wrapped__", True) + owner.stop_ttfb_metrics = patched_stop + + class _TracedSTTDescriptor: + """Class-level descriptor that wires up STT tracing at class definition time. + + ``__set_name__`` fires when the class body finishes evaluating, + giving us a chance to wrap the owner's ``push_frame`` so that + VAD, transcription, and finalization events drive the span + lifecycle, and to wrap ``stop_ttfb_metrics`` so the + TTFB-timeout path can attach metrics and close the span when + no finalized transcript ever arrives. The decorated method + itself runs unchanged. + """ + + def __set_name__(self, owner, attr_name): + patch_push_frame(owner) + patch_stop_ttfb_metrics(owner) + setattr(owner, attr_name, f) + + return _TracedSTTDescriptor() if func is not None: - return decorator(func) + # ``decorator(func)`` returns a descriptor placeholder that + # Python replaces with the real wrapped function once + # ``__set_name__`` runs at class definition time. Pyright sees + # only the descriptor instance, hence the ignore. + return decorator(func) # type: ignore[return-value] return decorator @@ -549,10 +899,6 @@ def traced_llm(func: Callable | None = None, *, name: str | None = None) -> Call fn_called = True result = await f(self, context, *args, **kwargs) - # Add aggregated output after function completes, if available - if output_text: - current_span.set_attribute("output", output_text) - return result finally: @@ -565,6 +911,15 @@ def traced_llm(func: Callable | None = None, *, name: str | None = None) -> Call ): self.start_llm_usage_metrics = original_start_llm_usage_metrics + # Attach whatever output text we accumulated so + # far. Doing this in finally captures partial + # output when ``f`` is cancelled or raises mid- + # stream (e.g. interruption during LLM + # generation), rather than only on clean + # completion. + if output_text: + current_span.set_attribute("output", output_text) + # Update TTFB metric ttfb: float | None = getattr(getattr(self, "_metrics", None), "ttfb", None) if ttfb is not None: