Add latency breakdown to UserBotLatencyObserver

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

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

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

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
+        on_latency_measured(observer, latency_seconds): Emitted when
-            latency is calculated. Includes the latency value in seconds as a float.
+            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:
+        elif isinstance(data.frame, UserStoppedSpeakingFrame):
-            # Calculate and emit latency
+            # Measure the user turn duration: from actual user silence to
-            latency = time.time() - self._user_stopped_time
+            # turn release. Includes VAD silence detection, STT finalization,
-            self._user_stopped_time = None
+            # and any turn analyzer wait.
-            await self._call_event_handler("on_latency_measured", latency)
+            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

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):
+    async def test_breakdown_with_metrics(self):
-        """Test that latency is not measured if bot starts without user stopping first."""
+        """Test that metrics collected between VADUserStopped and BotStarted appear in breakdown."""
-        # Create observer
+        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__":