From 18155b6a633868b35c4a5210d1c0283cd143d6b0 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Sun, 1 Mar 2026 07:40:02 -0500 Subject: [PATCH] Add latency breakdown to UserBotLatencyObserver MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add per-service latency breakdown metrics alongside existing user-to-bot latency measurement. When enable_metrics=True, the observer now emits an on_latency_breakdown event with TTFB, text aggregation, and user turn duration metrics collected between VADUserStoppedSpeakingFrame and BotStartedSpeakingFrame. - Add LatencyBreakdown dataclass with ttfb, text_aggregation, user_turn_secs fields - Accumulate MetricsFrame data during user→bot cycles - Reset accumulators on InterruptionFrame to discard stale metrics - Measure user_turn_secs from actual user silence (VAD timestamp - stop_secs) to turn release (UserStoppedSpeakingFrame) - Filter zero-value TTFB entries from startup metric resets - Add frame deduplication using bounded deque + set pattern - Update example 29 with latency breakdown display --- changelog/3885.added.md | 1 + .../foundational/29-turn-tracking-observer.py | 20 ++ .../observers/user_bot_latency_observer.py | 144 +++++++++-- tests/test_user_bot_latency_observer.py | 230 +++++++++++++++++- 4 files changed, 371 insertions(+), 24 deletions(-) create mode 100644 changelog/3885.added.md diff --git a/changelog/3885.added.md b/changelog/3885.added.md new file mode 100644 index 000000000..0713bbd45 --- /dev/null +++ b/changelog/3885.added.md @@ -0,0 +1 @@ +- Added `LatencyBreakdown` dataclass and `on_latency_breakdown` event to `UserBotLatencyObserver` for per-service latency metrics (TTFB, text aggregation, user turn duration) collected during each user-to-bot response cycle. diff --git a/examples/foundational/29-turn-tracking-observer.py b/examples/foundational/29-turn-tracking-observer.py index 4af28f1ed..736c68c55 100644 --- a/examples/foundational/29-turn-tracking-observer.py +++ b/examples/foundational/29-turn-tracking-observer.py @@ -131,6 +131,26 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): else: logger.info(f"🏁 Turn {turn_number} completed in {duration:.2f}s") + @latency_observer.event_handler("on_latency_breakdown") + async def on_latency_breakdown(observer, breakdown): + # Display a sequential waterfall that roughly adds up to the total. + # User turn is the first stage: user silence → turn release. + # The STT TTFB is shown as context within the user turn since + # it's a component of that time (along with VAD silence and any + # turn analyzer delay). + stt_ttfb = next((t for t in breakdown.ttfb if "STT" in t.processor), None) + if breakdown.user_turn_secs is not None: + stt_note = f" (STT: {stt_ttfb.value:.3f}s)" if stt_ttfb else "" + logger.info(f" User turn: {breakdown.user_turn_secs:.3f}s{stt_note}") + + for ttfb in breakdown.ttfb: + if ttfb is not stt_ttfb: + logger.info(f" {ttfb.processor}: TTFB {ttfb.value:.3f}s") + + if breakdown.text_aggregation: + ta = breakdown.text_aggregation + logger.info(f" {ta.processor}: text aggregation {ta.value:.3f}s") + @transport.event_handler("on_client_connected") async def on_client_connected(transport, client): logger.info(f"Client connected") diff --git a/src/pipecat/observers/user_bot_latency_observer.py b/src/pipecat/observers/user_bot_latency_observer.py index 37d5bc1a0..a7ad579f3 100644 --- a/src/pipecat/observers/user_bot_latency_observer.py +++ b/src/pipecat/observers/user_bot_latency_observer.py @@ -1,22 +1,63 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + """Observer for tracking user-to-bot response latency. This module provides an observer that monitors the time between when a user stops speaking and when the bot starts speaking, emitting events when latency -is measured. +is measured. Optionally collects per-service latency breakdown metrics +(TTFB, text aggregation) when ``enable_metrics=True``. """ import time -from typing import Optional, Set +from collections import deque +from dataclasses import dataclass, field +from typing import List, Optional from pipecat.frames.frames import ( BotStartedSpeakingFrame, + InterruptionFrame, + MetricsFrame, + UserStoppedSpeakingFrame, VADUserStartedSpeakingFrame, VADUserStoppedSpeakingFrame, ) +from pipecat.metrics.metrics import ( + TextAggregationMetricsData, + TTFBMetricsData, +) from pipecat.observers.base_observer import BaseObserver, FramePushed from pipecat.processors.frame_processor import FrameDirection +@dataclass +class LatencyBreakdown: + """Per-service latency breakdown for a single user-to-bot cycle. + + Collected between ``VADUserStoppedSpeakingFrame`` and + ``BotStartedSpeakingFrame`` when ``enable_metrics=True`` in + :class:`~pipecat.pipeline.task.PipelineParams`. + + Parameters: + ttfb: Time-to-first-byte metrics from each service in the pipeline. + text_aggregation: First text aggregation measurement, representing + the latency cost of sentence aggregation in the TTS pipeline. + user_turn_secs: Duration in seconds of the user's turn, measured + from when the user actually stopped speaking to when the turn + was released (``UserStoppedSpeakingFrame``). This includes + VAD silence detection, STT finalization, and any turn analyzer + wait. ``None`` if no ``UserStoppedSpeakingFrame`` was observed + (e.g. no turn analyzer configured). + """ + + ttfb: List[TTFBMetricsData] = field(default_factory=list) + text_aggregation: Optional[TextAggregationMetricsData] = None + user_turn_secs: Optional[float] = None + + class UserBotLatencyObserver(BaseObserver): """Observer that tracks user-to-bot response latency. @@ -25,34 +66,54 @@ class UserBotLatencyObserver(BaseObserver): latency is measured, allowing consumers to log, trace, or otherwise process the latency data. + When ``enable_metrics=True`` in pipeline params, also collects per-service + latency breakdown (TTFB, text aggregation) and emits an + ``on_latency_breakdown`` event alongside the existing latency measurement. + This observer follows the composition pattern used by TurnTrackingObserver, acting as a reusable component for latency measurement. Events: - on_latency_measured(observer, latency_seconds): Emitted when user-to-bot - latency is calculated. Includes the latency value in seconds as a float. + on_latency_measured(observer, latency_seconds): Emitted when + time-to-first-bot-speech is calculated. Measures the time from + when the user stopped speaking to when the bot starts speaking. + on_latency_breakdown(observer, breakdown): Emitted at each + ``BotStartedSpeakingFrame`` with a :class:`LatencyBreakdown` + containing per-service metrics collected during the user→bot cycle. """ - def __init__(self, **kwargs): + def __init__(self, *, max_frames=100, **kwargs): """Initialize the user-bot latency observer. Sets up tracking for processed frames and user speech timing to calculate response latencies. Args: + max_frames: Maximum number of frame IDs to keep in history for + duplicate detection. Defaults to 100. **kwargs: Additional arguments passed to parent class. """ super().__init__(**kwargs) self._user_stopped_time: Optional[float] = None - self._processed_frames: Set[str] = set() + self._user_turn: Optional[float] = None + + # Frame deduplication (bounded deque + set pattern) + self._processed_frames: set = set() + self._frame_history: deque = deque(maxlen=max_frames) + + # Per-cycle metric accumulators + self._ttfb: List[TTFBMetricsData] = [] + self._text_aggregation: Optional[TextAggregationMetricsData] = None self._register_event_handler("on_latency_measured") + self._register_event_handler("on_latency_breakdown") async def on_push_frame(self, data: FramePushed): """Process frames to track speech timing and calculate latency. Tracks VAD events and bot speaking events to measure the time between - user stopping speech and bot starting speech. + user stopping speech and bot starting speech. Also accumulates metrics + from MetricsFrame for the latency breakdown. Args: data: Frame push event containing the frame and direction information. @@ -61,23 +122,78 @@ class UserBotLatencyObserver(BaseObserver): if data.direction != FrameDirection.DOWNSTREAM: return - # Skip already processed frames + # Skip already processed frames (bounded deque + set) if data.frame.id in self._processed_frames: return self._processed_frames.add(data.frame.id) + self._frame_history.append(data.frame.id) - # Track VAD and bot speaking events for latency + if len(self._processed_frames) > len(self._frame_history): + self._processed_frames = set(self._frame_history) + + # Track speech and pipeline events for latency if isinstance(data.frame, VADUserStartedSpeakingFrame): # Reset when user starts speaking self._user_stopped_time = None + self._user_turn = None + self._reset_accumulators() elif isinstance(data.frame, VADUserStoppedSpeakingFrame): # Record the actual time the user stopped speaking, which is # the VAD determination time minus the stop_secs silence duration # that had to elapse before the VAD confirmed speech ended. self._user_stopped_time = data.frame.timestamp - data.frame.stop_secs - elif isinstance(data.frame, BotStartedSpeakingFrame) and self._user_stopped_time: - # Calculate and emit latency - latency = time.time() - self._user_stopped_time - self._user_stopped_time = None - await self._call_event_handler("on_latency_measured", latency) + elif isinstance(data.frame, UserStoppedSpeakingFrame): + # Measure the user turn duration: from actual user silence to + # turn release. Includes VAD silence detection, STT finalization, + # and any turn analyzer wait. + if self._user_stopped_time is not None: + self._user_turn = time.time() - self._user_stopped_time + elif isinstance(data.frame, InterruptionFrame): + # Discard stale metrics from cancelled LLM/TTS cycles + self._reset_accumulators() + elif isinstance(data.frame, MetricsFrame): + self._handle_metrics_frame(data.frame) + elif isinstance(data.frame, BotStartedSpeakingFrame): + await self._handle_bot_started_speaking() + + async def _handle_bot_started_speaking(self): + """Handle BotStartedSpeakingFrame to emit latency and breakdown.""" + if self._user_stopped_time is None: + return + + latency = time.time() - self._user_stopped_time + self._user_stopped_time = None + await self._call_event_handler("on_latency_measured", latency) + + breakdown = LatencyBreakdown( + ttfb=list(self._ttfb), + text_aggregation=self._text_aggregation, + user_turn_secs=self._user_turn, + ) + await self._call_event_handler("on_latency_breakdown", breakdown) + self._reset_accumulators() + + def _handle_metrics_frame(self, frame: MetricsFrame): + """Extract latency metrics from a MetricsFrame. + + Only accumulates metrics when a user→bot measurement is in progress + (after ``VADUserStoppedSpeakingFrame``). + """ + if self._user_stopped_time is None: + return + + for metrics_data in frame.data: + if isinstance(metrics_data, TTFBMetricsData) and metrics_data.value > 0: + self._ttfb.append(metrics_data) + elif isinstance(metrics_data, TextAggregationMetricsData): + # Only keep the first measurement — it's the one that + # impacts the initial speaking latency. + if self._text_aggregation is None: + self._text_aggregation = metrics_data + + def _reset_accumulators(self): + """Clear per-cycle metric accumulators.""" + self._ttfb = [] + self._text_aggregation = None + self._user_turn = None diff --git a/tests/test_user_bot_latency_observer.py b/tests/test_user_bot_latency_observer.py index 1b7325d14..8f8b2893d 100644 --- a/tests/test_user_bot_latency_observer.py +++ b/tests/test_user_bot_latency_observer.py @@ -2,12 +2,19 @@ import unittest from pipecat.frames.frames import ( BotStartedSpeakingFrame, + InterruptionFrame, + MetricsFrame, + UserStoppedSpeakingFrame, VADUserStartedSpeakingFrame, VADUserStoppedSpeakingFrame, ) +from pipecat.metrics.metrics import ( + TextAggregationMetricsData, + TTFBMetricsData, +) from pipecat.observers.user_bot_latency_observer import UserBotLatencyObserver from pipecat.processors.filters.identity_filter import IdentityFilter -from pipecat.tests.utils import run_test +from pipecat.tests.utils import SleepFrame, run_test class TestUserBotLatencyObserver(unittest.IsolatedAsyncioTestCase): @@ -97,22 +104,226 @@ class TestUserBotLatencyObserver(unittest.IsolatedAsyncioTestCase): self.assertGreater(latencies[0], 0) self.assertGreater(latencies[1], 0) - async def test_no_measurement_without_user_stop(self): - """Test that latency is not measured if bot starts without user stopping first.""" - # Create observer + async def test_breakdown_with_metrics(self): + """Test that metrics collected between VADUserStopped and BotStarted appear in breakdown.""" + observer = UserBotLatencyObserver() + processor = IdentityFilter() + + breakdowns = [] + + @observer.event_handler("on_latency_breakdown") + async def on_breakdown(obs, breakdown): + breakdowns.append(breakdown) + + stt_ttfb = TTFBMetricsData(processor="DeepgramSTTService#0", value=0.080) + llm_ttfb = TTFBMetricsData(processor="OpenAILLMService#0", model="gpt-4o", value=0.250) + tts_ttfb = TTFBMetricsData(processor="CartesiaTTSService#0", value=0.070) + text_agg = TextAggregationMetricsData(processor="CartesiaTTSService#0", value=0.030) + + frames_to_send = [ + VADUserStoppedSpeakingFrame(), + MetricsFrame(data=[stt_ttfb]), + MetricsFrame(data=[llm_ttfb, text_agg]), + MetricsFrame(data=[tts_ttfb]), + BotStartedSpeakingFrame(), + ] + + expected_down_frames = [ + VADUserStoppedSpeakingFrame, + MetricsFrame, + MetricsFrame, + MetricsFrame, + BotStartedSpeakingFrame, + ] + + await run_test( + processor, + frames_to_send=frames_to_send, + expected_down_frames=expected_down_frames, + observers=[observer], + ) + + self.assertEqual(len(breakdowns), 1) + bd = breakdowns[0] + self.assertEqual(len(bd.ttfb), 3) + self.assertEqual(bd.ttfb[0].processor, "DeepgramSTTService#0") + self.assertEqual(bd.ttfb[1].processor, "OpenAILLMService#0") + self.assertEqual(bd.ttfb[2].processor, "CartesiaTTSService#0") + self.assertIsNotNone(bd.text_aggregation) + self.assertEqual(bd.text_aggregation.value, 0.030) + + async def test_interruption_resets_accumulators(self): + """Test that InterruptionFrame clears stale metrics from earlier cycles.""" + observer = UserBotLatencyObserver() + processor = IdentityFilter() + + breakdowns = [] + + @observer.event_handler("on_latency_breakdown") + async def on_breakdown(obs, breakdown): + breakdowns.append(breakdown) + + # First cycle metrics (will be interrupted) + stale_llm = TTFBMetricsData(processor="OpenAILLMService#0", value=0.245) + # Second cycle metrics (the ones that matter) + final_llm = TTFBMetricsData(processor="OpenAILLMService#0", value=0.224) + final_tts = TTFBMetricsData(processor="CartesiaTTSService#0", value=0.142) + + frames_to_send = [ + VADUserStoppedSpeakingFrame(), + MetricsFrame(data=[stale_llm]), + InterruptionFrame(), + MetricsFrame(data=[final_llm]), + MetricsFrame(data=[final_tts]), + BotStartedSpeakingFrame(), + ] + + expected_down_frames = [ + VADUserStoppedSpeakingFrame, + MetricsFrame, + InterruptionFrame, + MetricsFrame, + MetricsFrame, + BotStartedSpeakingFrame, + ] + + await run_test( + processor, + frames_to_send=frames_to_send, + expected_down_frames=expected_down_frames, + observers=[observer], + ) + + self.assertEqual(len(breakdowns), 1) + bd = breakdowns[0] + # Only the post-interruption metrics should be present + self.assertEqual(len(bd.ttfb), 2) + self.assertEqual(bd.ttfb[0].processor, "OpenAILLMService#0") + self.assertEqual(bd.ttfb[0].value, 0.224) + self.assertEqual(bd.ttfb[1].processor, "CartesiaTTSService#0") + self.assertEqual(bd.ttfb[1].value, 0.142) + + async def test_only_first_text_aggregation_kept(self): + """Test that only the first text aggregation metric is kept per cycle.""" + observer = UserBotLatencyObserver() + processor = IdentityFilter() + + breakdowns = [] + + @observer.event_handler("on_latency_breakdown") + async def on_breakdown(obs, breakdown): + breakdowns.append(breakdown) + + text_agg_1 = TextAggregationMetricsData(processor="CartesiaTTSService#0", value=0.030) + text_agg_2 = TextAggregationMetricsData(processor="CartesiaTTSService#0", value=0.080) + + frames_to_send = [ + VADUserStoppedSpeakingFrame(), + MetricsFrame(data=[text_agg_1]), + MetricsFrame(data=[text_agg_2]), + BotStartedSpeakingFrame(), + ] + + expected_down_frames = [ + VADUserStoppedSpeakingFrame, + MetricsFrame, + MetricsFrame, + BotStartedSpeakingFrame, + ] + + await run_test( + processor, + frames_to_send=frames_to_send, + expected_down_frames=expected_down_frames, + observers=[observer], + ) + + self.assertEqual(len(breakdowns), 1) + self.assertIsNotNone(breakdowns[0].text_aggregation) + self.assertEqual(breakdowns[0].text_aggregation.value, 0.030) + + async def test_user_turn_measured(self): + """Test that pre-LLM wait from user silence to UserStopped is captured.""" + observer = UserBotLatencyObserver() + processor = IdentityFilter() + + breakdowns = [] + + @observer.event_handler("on_latency_breakdown") + async def on_breakdown(obs, breakdown): + breakdowns.append(breakdown) + + frames_to_send = [ + VADUserStoppedSpeakingFrame(), + SleepFrame(sleep=0.1), # Simulate turn analyzer wait + UserStoppedSpeakingFrame(), + BotStartedSpeakingFrame(), + ] + + expected_down_frames = [ + VADUserStoppedSpeakingFrame, + UserStoppedSpeakingFrame, + BotStartedSpeakingFrame, + ] + + await run_test( + processor, + frames_to_send=frames_to_send, + expected_down_frames=expected_down_frames, + observers=[observer], + ) + + self.assertEqual(len(breakdowns), 1) + self.assertIsNotNone(breakdowns[0].user_turn_secs) + self.assertGreaterEqual(breakdowns[0].user_turn_secs, 0.1) + + async def test_user_turn_none_without_user_stopped(self): + """Test that user_turn is None when no UserStoppedSpeakingFrame arrives.""" + observer = UserBotLatencyObserver() + processor = IdentityFilter() + + breakdowns = [] + + @observer.event_handler("on_latency_breakdown") + async def on_breakdown(obs, breakdown): + breakdowns.append(breakdown) + + frames_to_send = [ + VADUserStoppedSpeakingFrame(), + BotStartedSpeakingFrame(), + ] + + expected_down_frames = [ + VADUserStoppedSpeakingFrame, + BotStartedSpeakingFrame, + ] + + await run_test( + processor, + frames_to_send=frames_to_send, + expected_down_frames=expected_down_frames, + observers=[observer], + ) + + self.assertEqual(len(breakdowns), 1) + self.assertIsNone(breakdowns[0].user_turn_secs) + + async def test_no_measurement_without_user_stop(self): + """Test that BotStartedSpeaking without prior user stop emits nothing.""" observer = UserBotLatencyObserver() - - # Create identity filter processor = IdentityFilter() - # Capture latency events latencies = [] + breakdowns = [] @observer.event_handler("on_latency_measured") async def on_latency(obs, latency_seconds): latencies.append(latency_seconds) - # Define frame sequence - bot starts without user stop + @observer.event_handler("on_latency_breakdown") + async def on_breakdown(obs, breakdown): + breakdowns.append(breakdown) + frames_to_send = [ BotStartedSpeakingFrame(), ] @@ -121,7 +332,6 @@ class TestUserBotLatencyObserver(unittest.IsolatedAsyncioTestCase): BotStartedSpeakingFrame, ] - # Run test await run_test( processor, frames_to_send=frames_to_send, @@ -129,8 +339,8 @@ class TestUserBotLatencyObserver(unittest.IsolatedAsyncioTestCase): observers=[observer], ) - # Verify no latency was measured self.assertEqual(len(latencies), 0) + self.assertEqual(len(breakdowns), 0) if __name__ == "__main__":