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

Add STT changelog entry for #4467

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2026-05-12 12:01:00 -07:00
parent 111e59a7b1
commit e70ee603b2
2 changed files with 120 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
changelog/4467.fixed.2.md Normal file
View File

@@ -0,0 +1 @@
- Fixed incorrect `metrics.ttfb` on STT OpenTelemetry spans, and parented them to the current turn span.

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

@@ -466,6 +466,29 @@ def traced_stt(func: Callable | None = None, *, name: str | None = None) -> Call
if getattr(original_push_frame, "__stt_tracing_push_frame_wrapped__", False): if getattr(original_push_frame, "__stt_tracing_push_frame_wrapped__", False):
return return
def update_transcript(state, new_text):
"""Append or extend the current segment in ``state['segments']``.
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): def open_span(service, state):
"""Open the STT span, anchored at ``segment_start_time`` if set.""" """Open the STT span, anchored at ``segment_start_time`` if set."""
parent = _get_turn_context(service) or _get_parent_service_context(service) parent = _get_turn_context(service) or _get_parent_service_context(service)
@@ -488,18 +511,24 @@ def traced_stt(func: Callable | None = None, *, name: str | None = None) -> Call
state["span"] = span state["span"] = span
def handle_pre_push(service, frame, state): def handle_pre_push(service, frame, state):
"""Record speech-start anchor. """Record speech-start anchor; lazy-open span on first transcript.
The span itself is lazy-opened in ``handle_post_push`` Lazy-opening on ``TranscriptionFrame`` (rather than on
when the first ``TranscriptionFrame`` arrives. Opening ``VADUserStartedSpeakingFrame`` or
on ``VADUserStartedSpeakingFrame`` or ``UserStartedSpeakingFrame``) avoids racing with
``UserStartedSpeakingFrame`` would race with
``TurnTraceObserver._handle_turn_started``, which runs ``TurnTraceObserver._handle_turn_started``, which runs
in a background task fired by ``_call_event_handler`` in a background task fired by ``_call_event_handler``
(``base_object.py:232``) and may not have set the new (``base_object.py:232``) and may not have set the new
turn's context yet — that produces STT spans parented turn's context yet — that produces STT spans parented
to the previous turn. By the time STT actually emits a to the previous turn. By the time STT actually emits
transcript, the turn observer has run. 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): if isinstance(frame, VADUserStartedSpeakingFrame):
# Anchor the next span at the moment speech began. # Anchor the next span at the moment speech began.
@@ -507,52 +536,91 @@ def traced_stt(func: Callable | None = None, *, name: str | None = None) -> Call
# re-trigger) or a span open. # re-trigger) or a span open.
if state["span"] is None and state["segment_start_time"] is None: if state["span"] is None and state["segment_start_time"] is None:
state["segment_start_time"] = frame.timestamp - frame.start_secs state["segment_start_time"] = frame.timestamp - frame.start_secs
elif isinstance(frame, TranscriptionFrame) and state["span"] is None:
open_span(service, state)
def handle_post_push(service, frame, state): async def handle_post_push(service, frame, state):
"""Lazy-open span on first transcript; attach attrs; close on finalized. """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 One STT span per finalized transcript: the span opens
on the first ``TranscriptionFrame`` (anchored at lazily on the first ``TranscriptionFrame`` (pre-push,
speech start via ``segment_start_time``) and closes on anchored at speech start via ``segment_start_time``)
``finalized=True``. Multiple finalized transcripts in and closes on ``finalized=True``. Multiple finalized
a single turn produce multiple spans; subsequent transcripts in a single turn produce multiple spans.
spans lazy-open with no explicit anchor (start_time
defaults to "now"). Spans that never see a finalized For services that never set ``frame.finalized=True``
transcript are closed by ``UserStoppedSpeakingFrame`` (e.g. Deepgram, which only marks it via
(marked ``stt.incomplete``) or by the TTFB timeout ``confirm_finalize()``), the span closes on
(marked ``stt.timed_out``). ``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): if isinstance(frame, UserStoppedSpeakingFrame):
prev_span = state["span"] prev_span = state["span"]
if prev_span is not None: if prev_span is None:
prev_span.set_attribute("stt.incomplete", True) return
prev_span.end() 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["span"] = None
state["segment_start_time"] = None state["segment_start_time"] = None
elif isinstance(frame, TranscriptionFrame): state["segments"] = []
if state["span"] is None: elif isinstance(frame, MetricsFrame):
open_span(service, state)
span = state["span"] 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: if frame.text:
span.set_attribute("transcript", frame.text) update_transcript(state, frame.text)
span.set_attribute("transcript", " ".join(state["segments"]).strip())
span.set_attribute("is_final", bool(frame.finalized)) span.set_attribute("is_final", bool(frame.finalized))
if frame.language: if frame.language:
span.set_attribute("language", str(frame.language)) span.set_attribute("language", str(frame.language))
if frame.user_id: if frame.user_id:
span.set_attribute("user_id", frame.user_id) span.set_attribute("user_id", frame.user_id)
if frame.finalized: if frame.finalized:
ttfb = getattr(getattr(service, "_metrics", None), "ttfb", None)
if ttfb is not None:
span.set_attribute("metrics.ttfb", ttfb)
span.end() span.end()
state["span"] = None state["span"] = None
state["segment_start_time"] = None state["segment_start_time"] = None
state["segments"] = []
@functools.wraps(original_push_frame) @functools.wraps(original_push_frame)
async def patched_push_frame(self, frame, direction=FrameDirection.DOWNSTREAM): async def patched_push_frame(self, frame, direction=FrameDirection.DOWNSTREAM):
state = getattr(self, "_stt_span_state", None) state = getattr(self, "_stt_span_state", None)
if state is None: if state is None:
state = {"span": None, "segment_start_time": None} state = {"span": None, "segment_start_time": None, "segments": []}
self._stt_span_state = state self._stt_span_state = state
if getattr(self, "_tracing_enabled", False): if getattr(self, "_tracing_enabled", False):
@@ -565,7 +633,7 @@ def traced_stt(func: Callable | None = None, *, name: str | None = None) -> Call
if getattr(self, "_tracing_enabled", False): if getattr(self, "_tracing_enabled", False):
try: try:
handle_post_push(self, frame, state) await handle_post_push(self, frame, state)
except Exception as e: except Exception as e:
logging.warning(f"Error in STT post-push tracing: {e}") logging.warning(f"Error in STT post-push tracing: {e}")
@@ -573,18 +641,22 @@ def traced_stt(func: Callable | None = None, *, name: str | None = None) -> Call
owner.push_frame = patched_push_frame owner.push_frame = patched_push_frame
def patch_stop_ttfb_metrics(owner): def patch_stop_ttfb_metrics(owner):
"""Wrap ``owner.stop_ttfb_metrics`` to handle the TTFB-timeout path. """Wrap ``owner.stop_ttfb_metrics`` to close the span on the timeout path.
When ``stop_ttfb_metrics`` is invoked with ``end_time`` set, When ``stop_ttfb_metrics`` is invoked with ``end_time`` set,
that signals the STT service's TTFB-timeout handler firing that signals the TTFB-timeout handler firing
without a finalized ``TranscriptionFrame`` ever arriving (`stt_service.py:566`), or our own force-stop from the
(`stt_service.py:566`). In that case we attach ``UserStoppedSpeakingFrame`` handler. In either case we
``metrics.ttfb`` and close the span with anchor the span's end at ``end_time``
``stt.timed_out=true``. The default ``end_time=None`` path (= ``_last_transcript_time``) rather than at whenever the
(called from ``push_frame`` on finalized frames) only coroutine resumed.
attaches ``metrics.ttfb``; closing happens in the
``push_frame`` post-hook so the final transcript attributes ``metrics.ttfb`` attribution is not done here — the
land first. ``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 original_stop = owner.stop_ttfb_metrics
if getattr(original_stop, "__stt_tracing_stop_ttfb_wrapped__", False): if getattr(original_stop, "__stt_tracing_stop_ttfb_wrapped__", False):
@@ -593,25 +665,19 @@ def traced_stt(func: Callable | None = None, *, name: str | None = None) -> Call
@functools.wraps(original_stop) @functools.wraps(original_stop)
async def patched_stop(self, *, end_time=None): async def patched_stop(self, *, end_time=None):
await original_stop(self, end_time=end_time) await original_stop(self, end_time=end_time)
if end_time is None:
return
if not getattr(self, "_tracing_enabled", False): if not getattr(self, "_tracing_enabled", False):
return return
state = getattr(self, "_stt_span_state", None) state = getattr(self, "_stt_span_state", None)
if not state or state["span"] is None: if not state or state["span"] is None:
return return
span = state["span"]
try: try:
ttfb = getattr(getattr(self, "_metrics", None), "ttfb", None) span = state["span"]
if ttfb is not None: span.end(end_time=int(end_time * 1e9))
span.set_attribute("metrics.ttfb", ttfb) state["span"] = None
if end_time is not None: state["segment_start_time"] = None
span.set_attribute("stt.timed_out", True) state["segments"] = []
# Use end_time (= _last_transcript_time) so the
# span ends where the timeout handler actually
# finalized TTFB, rather than after the timeout
# sleep.
span.end(end_time=int(end_time * 1e9))
state["span"] = None
state["segment_start_time"] = None
except Exception as e: except Exception as e:
logging.warning(f"Error in STT stop_ttfb_metrics tracing: {e}") logging.warning(f"Error in STT stop_ttfb_metrics tracing: {e}")