Merge pull request #3885 from pipecat-ai/mb/latency-breakdown

Add latency breakdown to UserBotLatencyObserver

src/pipecat/observers/user_bot_latency_observer.py

@@ -1,22 +1,146 @@
+#
+# 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 typing import Dict, List, Optional
+
+from pydantic import BaseModel, Field
 
 from pipecat.frames.frames import (
     BotStartedSpeakingFrame,
+    ClientConnectedFrame,
+    FunctionCallInProgressFrame,
+    FunctionCallResultFrame,
+    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
 
 
+class TTFBBreakdownMetrics(BaseModel):
+    """TTFB measurement with timestamp for timeline placement.
+
+    Parameters:
+        processor: Name of the processor that reported the TTFB.
+        model: Optional model name associated with the metric.
+        start_time: Unix timestamp when the TTFB measurement started.
+        duration_secs: TTFB duration in seconds.
+    """
+
+    processor: str
+    model: Optional[str] = None
+    start_time: float
+    duration_secs: float
+
+
+class TextAggregationBreakdownMetrics(BaseModel):
+    """Text aggregation measurement with timestamp for timeline placement.
+
+    Parameters:
+        processor: Name of the processor that reported the metric.
+        start_time: Unix timestamp when text aggregation started.
+        duration_secs: Aggregation duration in seconds.
+    """
+
+    processor: str
+    start_time: float
+    duration_secs: float
+
+
+class FunctionCallMetrics(BaseModel):
+    """Latency for a single function call execution.
+
+    Parameters:
+        function_name: Name of the function that was called.
+        start_time: Unix timestamp when execution started.
+        duration_secs: Time in seconds from execution start to result.
+    """
+
+    function_name: str
+    start_time: float
+    duration_secs: float
+
+
+class LatencyBreakdown(BaseModel):
+    """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_start_time: Unix timestamp when the user turn started
+            (actual user silence, adjusted for VAD stop_secs). ``None`` if
+            no ``VADUserStoppedSpeakingFrame`` was observed.
+        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).
+        function_calls: Latency for each function call executed during
+            this cycle. Empty if no function calls occurred.
+    """
+
+    ttfb: List[TTFBBreakdownMetrics] = Field(default_factory=list)
+    text_aggregation: Optional[TextAggregationBreakdownMetrics] = None
+    user_turn_start_time: Optional[float] = None
+    user_turn_secs: Optional[float] = None
+    function_calls: List[FunctionCallMetrics] = Field(default_factory=list)
+
+    def chronological_events(self) -> List[str]:
+        """Return human-readable event labels sorted by start time.
+
+        Collects all sub-metrics into a flat list, sorts by ``start_time``,
+        and returns formatted strings suitable for logging.
+
+        Returns:
+            List of formatted strings, one per event, in chronological order.
+        """
+        events: List[tuple] = []
+
+        if self.user_turn_start_time is not None and self.user_turn_secs is not None:
+            events.append((self.user_turn_start_time, f"User turn: {self.user_turn_secs:.3f}s"))
+
+        for t in self.ttfb:
+            events.append((t.start_time, f"{t.processor}: TTFB {t.duration_secs:.3f}s"))
+
+        for fc in self.function_calls:
+            events.append((fc.start_time, f"{fc.function_name}: {fc.duration_secs:.3f}s"))
+
+        if self.text_aggregation:
+            ta = self.text_aggregation
+            events.append(
+                (ta.start_time, f"{ta.processor}: text aggregation {ta.duration_secs:.3f}s")
+            )
+
+        events.sort(key=lambda e: e[0])
+        return [label for _, label in events]
+
+
 class UserBotLatencyObserver(BaseObserver):
     """Observer that tracks user-to-bot response latency.
 
@@ -25,34 +149,66 @@ 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.
+        on_first_bot_speech_latency(observer, latency_seconds): Emitted once,
+            the first time ``BotStartedSpeakingFrame`` arrives after
+            ``ClientConnectedFrame``. Measures the time from client connection
+            to the first bot speech.
     """
 
-    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_start_time: Optional[float] = None
+        self._user_turn: Optional[float] = None
+
+        # First bot speech tracking
+        self._client_connected_time: Optional[float] = None
+        self._first_bot_speech_measured: bool = False
+
+        # 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[TTFBBreakdownMetrics] = []
+        self._text_aggregation: Optional[TextAggregationBreakdownMetrics] = None
+        self._function_call_starts: Dict[str, tuple[str, float]] = {}
+        self._function_call_metrics: List[FunctionCallMetrics] = []
 
         self._register_event_handler("on_latency_measured")
+        self._register_event_handler("on_latency_breakdown")
+        self._register_event_handler("on_first_bot_speech_latency")
 
     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 +217,135 @@ 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 client connection (first occurrence only)
+        if isinstance(data.frame, ClientConnectedFrame):
+            if self._client_connected_time is None:
+                self._client_connected_time = time.time()
+            return
+
+        # 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_start_time = None
+            self._user_turn = None
+            self._reset_accumulators()
+            # If user speaks before the bot's first speech, abandon the
+            # first-bot-speech measurement — it's only meaningful for greetings.
+            self._first_bot_speech_measured = True
         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
+            self._user_turn_start_time = self._user_stopped_time
+        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, FunctionCallInProgressFrame):
+            self._function_call_starts[data.frame.tool_call_id] = (
+                data.frame.function_name,
+                time.time(),
+            )
+        elif isinstance(data.frame, FunctionCallResultFrame):
+            start = self._function_call_starts.pop(data.frame.tool_call_id, None)
+            if start is not None:
+                function_name, start_time = start
+                self._function_call_metrics.append(
+                    FunctionCallMetrics(
+                        function_name=function_name,
+                        start_time=start_time,
+                        duration_secs=time.time() - start_time,
+                    )
+                )
+        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."""
+        emit_breakdown = False
+
+        # One-time first bot speech measurement (client connect → first speech)
+        if self._client_connected_time is not None and not self._first_bot_speech_measured:
+            self._first_bot_speech_measured = True
+            latency = time.time() - self._client_connected_time
+            await self._call_event_handler("on_first_bot_speech_latency", latency)
+            emit_breakdown = True
+
+        if self._user_stopped_time is not None:
             latency = time.time() - self._user_stopped_time
             self._user_stopped_time = None
             await self._call_event_handler("on_latency_measured", latency)
+            emit_breakdown = True
+
+        if emit_breakdown:
+            breakdown = LatencyBreakdown(
+                ttfb=list(self._ttfb),
+                text_aggregation=self._text_aggregation,
+                user_turn_start_time=self._user_turn_start_time,
+                user_turn_secs=self._user_turn,
+                function_calls=list(self._function_call_metrics),
+            )
+            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.
+
+        Accumulates metrics when a measurement is in progress: either a
+        user→bot cycle (after ``VADUserStoppedSpeakingFrame``) or the
+        first-bot-speech window (after ``ClientConnectedFrame``).
+        """
+        waiting_for_first_speech = (
+            self._client_connected_time is not None and not self._first_bot_speech_measured
+        )
+        if self._user_stopped_time is None and not waiting_for_first_speech:
+            return
+
+        now = time.time()
+        for metrics_data in frame.data:
+            if isinstance(metrics_data, TTFBMetricsData) and metrics_data.value > 0:
+                self._ttfb.append(
+                    TTFBBreakdownMetrics(
+                        processor=metrics_data.processor,
+                        model=metrics_data.model,
+                        start_time=now - metrics_data.value,
+                        duration_secs=metrics_data.value,
+                    )
+                )
+            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 = TextAggregationBreakdownMetrics(
+                        processor=metrics_data.processor,
+                        start_time=now - metrics_data.value,
+                        duration_secs=metrics_data.value,
+                    )
+
+    def _reset_accumulators(self):
+        """Clear per-cycle metric accumulators."""
+        self._ttfb = []
+        self._text_aggregation = None
+        self._user_turn_start_time = None
+        self._user_turn = None
+        self._function_call_starts = {}
+        self._function_call_metrics = []