From 98bd530574abe8e9b456a24fcb1b38ddbdd84ce7 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Sat, 28 Feb 2026 14:05:25 -0500 Subject: [PATCH 1/8] Add changelog for #3881 --- changelog/3881.added.md | 2 +- .../utils/tracing/service_attributes.py | 14 ++++++-------- .../utils/tracing/service_decorators.py | 19 ++++++++++++------- 3 files changed, 19 insertions(+), 16 deletions(-) diff --git a/changelog/3881.added.md b/changelog/3881.added.md index cbf6d0293..c71475675 100644 --- a/changelog/3881.added.md +++ b/changelog/3881.added.md @@ -1 +1 @@ -- Added `StartupTimingObserver` for measuring how long each processor's `start()` method takes during pipeline startup. Also measures transport readiness — the time from `StartFrame` to first client connection — via the `on_transport_readiness_measured` event. Useful for diagnosing cold start slowness and identifying initialization bottlenecks. +- Added `StartupTimingObserver` for measuring how long each processor's `start()` method takes during pipeline startup. Also measures transport readiness — the time from `StartFrame` to first client connection — via the `on_transport_timing_report` event. diff --git a/src/pipecat/utils/tracing/service_attributes.py b/src/pipecat/utils/tracing/service_attributes.py index 97ac49d87..c8471a03b 100644 --- a/src/pipecat/utils/tracing/service_attributes.py +++ b/src/pipecat/utils/tracing/service_attributes.py @@ -17,8 +17,6 @@ from typing import TYPE_CHECKING, Any, Dict, List, Optional if TYPE_CHECKING: from opentelemetry.trace import Span - from pipecat.services.settings import ServiceSettings - from pipecat.utils.tracing.setup import is_tracing_available if is_tracing_available(): @@ -70,7 +68,7 @@ def add_tts_span_attributes( model: str, voice_id: str, text: Optional[str] = None, - settings: Optional["ServiceSettings"] = None, + settings: Optional[Dict[str, Any]] = None, character_count: Optional[int] = None, operation_name: str = "tts", ttfb: Optional[float] = None, @@ -109,7 +107,7 @@ def add_tts_span_attributes( # Add settings if provided if settings: - for key, value in settings.given_fields().items(): + for key, value in settings.items(): if isinstance(value, (str, int, float, bool)): span.set_attribute(f"settings.{key}", value) @@ -128,7 +126,7 @@ def add_stt_span_attributes( is_final: Optional[bool] = None, language: Optional[str] = None, user_id: Optional[str] = None, - settings: Optional["ServiceSettings"] = None, + settings: Optional[Dict[str, Any]] = None, vad_enabled: bool = False, ttfb: Optional[float] = None, **kwargs, @@ -173,7 +171,7 @@ def add_stt_span_attributes( # Add settings if provided if settings: - for key, value in settings.given_fields().items(): + for key, value in settings.items(): if isinstance(value, (str, int, float, bool)): span.set_attribute(f"settings.{key}", value) @@ -284,7 +282,7 @@ def add_gemini_live_span_attributes( voice_id: Optional[str] = None, language: Optional[str] = None, modalities: Optional[str] = None, - settings: Optional["ServiceSettings"] = None, + settings: Optional[Dict[str, Any]] = None, tools: Optional[List[Dict]] = None, tools_serialized: Optional[str] = None, transcript: Optional[str] = None, @@ -361,7 +359,7 @@ def add_gemini_live_span_attributes( # Add settings if provided if settings: - for key, value in settings.given_fields().items(): + for key, value in settings.items(): if isinstance(value, (str, int, float, bool)): span.set_attribute(f"settings.{key}", value) elif key == "vad" and value: diff --git a/src/pipecat/utils/tracing/service_decorators.py b/src/pipecat/utils/tracing/service_decorators.py index 304ecb5e8..601cad53d 100644 --- a/src/pipecat/utils/tracing/service_decorators.py +++ b/src/pipecat/utils/tracing/service_decorators.py @@ -219,7 +219,7 @@ def traced_tts(func: Optional[Callable] = None, *, name: Optional[str] = None) - tracer = trace.get_tracer("pipecat") with tracer.start_as_current_span(span_name, context=parent_context) as span: try: - settings = getattr(self, "_settings", None) + settings = getattr(self, "_settings", {}) add_tts_span_attributes( span=span, service_name=service_class_name, @@ -338,7 +338,7 @@ def traced_stt(func: Optional[Callable] = None, *, name: Optional[str] = None) - ) # Use settings from the service if available - settings = getattr(self, "_settings", None) + settings = getattr(self, "_settings", {}) add_stt_span_attributes( span=current_span, @@ -510,10 +510,15 @@ def traced_llm(func: Optional[Callable] = None, *, name: Optional[str] = None) - # Get settings from the service params = {} if hasattr(self, "_settings"): - for key, value in self._settings.given_fields().items(): + for key, value in self._settings.items(): + if key == "extra": + continue + # Add value directly if it's a basic type if isinstance(value, (int, float, bool, str)): params[key] = value - elif value is None: + elif value is None or ( + hasattr(value, "__name__") and value.__name__ == "NOT_GIVEN" + ): params[key] = "NOT_GIVEN" # Add all available attributes to the span @@ -622,12 +627,12 @@ def traced_gemini_live(operation: str) -> Callable: model_name = _get_model_name(self) voice_id = getattr(self, "_voice_id", None) language_code = getattr(self, "_language_code", None) - settings = getattr(self, "_settings", None) + settings = getattr(self, "_settings", {}) # Get modalities if available modalities = None - if settings and hasattr(settings, "modalities"): - modality_obj = settings.modalities + if hasattr(self, "_settings") and "modalities" in self._settings: + modality_obj = self._settings["modalities"] if hasattr(modality_obj, "value"): modalities = modality_obj.value else: From ac69b3441e1161120138d43af11ca6cb12aac8c7 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 27 Feb 2026 22:35:29 -0500 Subject: [PATCH 2/8] Fix tracing to use ServiceSettings API instead of dict access The ServiceSettings refactor (PR #3714) changed self._settings from dicts to dataclass subclasses, but tracing code still used .items(), in containment, and subscript access, causing AttributeError on every traced call. Use given_fields() for iteration and attribute access for named fields. --- .../utils/tracing/service_attributes.py | 14 ++++++++------ .../utils/tracing/service_decorators.py | 19 +++++++------------ 2 files changed, 15 insertions(+), 18 deletions(-) diff --git a/src/pipecat/utils/tracing/service_attributes.py b/src/pipecat/utils/tracing/service_attributes.py index c8471a03b..97ac49d87 100644 --- a/src/pipecat/utils/tracing/service_attributes.py +++ b/src/pipecat/utils/tracing/service_attributes.py @@ -17,6 +17,8 @@ from typing import TYPE_CHECKING, Any, Dict, List, Optional if TYPE_CHECKING: from opentelemetry.trace import Span + from pipecat.services.settings import ServiceSettings + from pipecat.utils.tracing.setup import is_tracing_available if is_tracing_available(): @@ -68,7 +70,7 @@ def add_tts_span_attributes( model: str, voice_id: str, text: Optional[str] = None, - settings: Optional[Dict[str, Any]] = None, + settings: Optional["ServiceSettings"] = None, character_count: Optional[int] = None, operation_name: str = "tts", ttfb: Optional[float] = None, @@ -107,7 +109,7 @@ def add_tts_span_attributes( # Add settings if provided if settings: - for key, value in settings.items(): + for key, value in settings.given_fields().items(): if isinstance(value, (str, int, float, bool)): span.set_attribute(f"settings.{key}", value) @@ -126,7 +128,7 @@ def add_stt_span_attributes( is_final: Optional[bool] = None, language: Optional[str] = None, user_id: Optional[str] = None, - settings: Optional[Dict[str, Any]] = None, + settings: Optional["ServiceSettings"] = None, vad_enabled: bool = False, ttfb: Optional[float] = None, **kwargs, @@ -171,7 +173,7 @@ def add_stt_span_attributes( # Add settings if provided if settings: - for key, value in settings.items(): + for key, value in settings.given_fields().items(): if isinstance(value, (str, int, float, bool)): span.set_attribute(f"settings.{key}", value) @@ -282,7 +284,7 @@ def add_gemini_live_span_attributes( voice_id: Optional[str] = None, language: Optional[str] = None, modalities: Optional[str] = None, - settings: Optional[Dict[str, Any]] = None, + settings: Optional["ServiceSettings"] = None, tools: Optional[List[Dict]] = None, tools_serialized: Optional[str] = None, transcript: Optional[str] = None, @@ -359,7 +361,7 @@ def add_gemini_live_span_attributes( # Add settings if provided if settings: - for key, value in settings.items(): + for key, value in settings.given_fields().items(): if isinstance(value, (str, int, float, bool)): span.set_attribute(f"settings.{key}", value) elif key == "vad" and value: diff --git a/src/pipecat/utils/tracing/service_decorators.py b/src/pipecat/utils/tracing/service_decorators.py index 601cad53d..304ecb5e8 100644 --- a/src/pipecat/utils/tracing/service_decorators.py +++ b/src/pipecat/utils/tracing/service_decorators.py @@ -219,7 +219,7 @@ def traced_tts(func: Optional[Callable] = None, *, name: Optional[str] = None) - tracer = trace.get_tracer("pipecat") with tracer.start_as_current_span(span_name, context=parent_context) as span: try: - settings = getattr(self, "_settings", {}) + settings = getattr(self, "_settings", None) add_tts_span_attributes( span=span, service_name=service_class_name, @@ -338,7 +338,7 @@ def traced_stt(func: Optional[Callable] = None, *, name: Optional[str] = None) - ) # Use settings from the service if available - settings = getattr(self, "_settings", {}) + settings = getattr(self, "_settings", None) add_stt_span_attributes( span=current_span, @@ -510,15 +510,10 @@ def traced_llm(func: Optional[Callable] = None, *, name: Optional[str] = None) - # Get settings from the service params = {} if hasattr(self, "_settings"): - for key, value in self._settings.items(): - if key == "extra": - continue - # Add value directly if it's a basic type + for key, value in self._settings.given_fields().items(): if isinstance(value, (int, float, bool, str)): params[key] = value - elif value is None or ( - hasattr(value, "__name__") and value.__name__ == "NOT_GIVEN" - ): + elif value is None: params[key] = "NOT_GIVEN" # Add all available attributes to the span @@ -627,12 +622,12 @@ def traced_gemini_live(operation: str) -> Callable: model_name = _get_model_name(self) voice_id = getattr(self, "_voice_id", None) language_code = getattr(self, "_language_code", None) - settings = getattr(self, "_settings", {}) + settings = getattr(self, "_settings", None) # Get modalities if available modalities = None - if hasattr(self, "_settings") and "modalities" in self._settings: - modality_obj = self._settings["modalities"] + if settings and hasattr(settings, "modalities"): + modality_obj = settings.modalities if hasattr(modality_obj, "value"): modalities = modality_obj.value else: From 18155b6a633868b35c4a5210d1c0283cd143d6b0 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Sun, 1 Mar 2026 07:40:02 -0500 Subject: [PATCH 3/8] 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__": From ddba1b84a937cceb6ebe602ebd67b42f3a2a374b Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Sun, 1 Mar 2026 09:11:10 -0500 Subject: [PATCH 4/8] Add first-bot-speech latency to UserBotLatencyObserver Measure time from ClientConnectedFrame to first BotStartedSpeakingFrame, emitting a one-time on_first_bot_speech_latency event with breakdown. --- changelog/3885.added.2.md | 1 + changelog/3885.added.md | 2 +- .../foundational/29-turn-tracking-observer.py | 4 + .../observers/user_bot_latency_observer.py | 62 ++++++--- tests/test_user_bot_latency_observer.py | 121 ++++++++++++++++++ 5 files changed, 174 insertions(+), 16 deletions(-) create mode 100644 changelog/3885.added.2.md diff --git a/changelog/3885.added.2.md b/changelog/3885.added.2.md new file mode 100644 index 000000000..a9562b883 --- /dev/null +++ b/changelog/3885.added.2.md @@ -0,0 +1 @@ +- Added `on_first_bot_speech_latency` event to `UserBotLatencyObserver` measuring the time from client connection to first bot speech, including a latency breakdown with per-service metrics. diff --git a/changelog/3885.added.md b/changelog/3885.added.md index 0713bbd45..87cbd7824 100644 --- a/changelog/3885.added.md +++ b/changelog/3885.added.md @@ -1 +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. +- Added `on_latency_breakdown` event to `UserBotLatencyObserver` providing per-service TTFB, text aggregation, and user turn duration metrics for 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 736c68c55..2c8ec1f84 100644 --- a/examples/foundational/29-turn-tracking-observer.py +++ b/examples/foundational/29-turn-tracking-observer.py @@ -101,6 +101,10 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): observers=[latency_observer, startup_observer], ) + @latency_observer.event_handler("on_first_bot_speech_latency") + async def on_first_bot_speech_latency(observer, latency_seconds): + logger.info(f"First bot speech: {latency_seconds:.3f}s after client connected") + @latency_observer.event_handler("on_latency_measured") async def on_latency_measured(observer, latency_seconds): logger.info(f"⏱️ User-to-bot latency: {latency_seconds:.3f}s") diff --git a/src/pipecat/observers/user_bot_latency_observer.py b/src/pipecat/observers/user_bot_latency_observer.py index a7ad579f3..cdc7c43d0 100644 --- a/src/pipecat/observers/user_bot_latency_observer.py +++ b/src/pipecat/observers/user_bot_latency_observer.py @@ -19,6 +19,7 @@ from typing import List, Optional from pipecat.frames.frames import ( BotStartedSpeakingFrame, + ClientConnectedFrame, InterruptionFrame, MetricsFrame, UserStoppedSpeakingFrame, @@ -80,6 +81,10 @@ class UserBotLatencyObserver(BaseObserver): 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, *, max_frames=100, **kwargs): @@ -97,6 +102,10 @@ class UserBotLatencyObserver(BaseObserver): self._user_stopped_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) @@ -107,6 +116,7 @@ class UserBotLatencyObserver(BaseObserver): 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. @@ -132,12 +142,21 @@ class UserBotLatencyObserver(BaseObserver): 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 = 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 @@ -159,28 +178,41 @@ class UserBotLatencyObserver(BaseObserver): async def _handle_bot_started_speaking(self): """Handle BotStartedSpeakingFrame to emit latency and breakdown.""" - if self._user_stopped_time is None: - return + emit_breakdown = False - latency = time.time() - self._user_stopped_time - self._user_stopped_time = None - await self._call_event_handler("on_latency_measured", latency) + # 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 - 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() + 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_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``). + Accumulates metrics when a measurement is in progress: either a + user→bot cycle (after ``VADUserStoppedSpeakingFrame``) or the + first-bot-speech window (after ``ClientConnectedFrame``). """ - if self._user_stopped_time is None: + 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 for metrics_data in frame.data: diff --git a/tests/test_user_bot_latency_observer.py b/tests/test_user_bot_latency_observer.py index 8f8b2893d..ab00bfd61 100644 --- a/tests/test_user_bot_latency_observer.py +++ b/tests/test_user_bot_latency_observer.py @@ -2,6 +2,7 @@ import unittest from pipecat.frames.frames import ( BotStartedSpeakingFrame, + ClientConnectedFrame, InterruptionFrame, MetricsFrame, UserStoppedSpeakingFrame, @@ -342,6 +343,126 @@ class TestUserBotLatencyObserver(unittest.IsolatedAsyncioTestCase): self.assertEqual(len(latencies), 0) self.assertEqual(len(breakdowns), 0) + async def test_first_bot_speech_latency(self): + """Test first bot speech latency and breakdown from ClientConnected to BotStartedSpeaking.""" + observer = UserBotLatencyObserver() + processor = IdentityFilter() + + first_speech_latencies = [] + breakdowns = [] + + @observer.event_handler("on_first_bot_speech_latency") + async def on_first_bot_speech(obs, latency_seconds): + first_speech_latencies.append(latency_seconds) + + @observer.event_handler("on_latency_breakdown") + async def on_breakdown(obs, breakdown): + breakdowns.append(breakdown) + + llm_ttfb = TTFBMetricsData(processor="OpenAILLMService#0", value=0.250) + tts_ttfb = TTFBMetricsData(processor="CartesiaTTSService#0", value=0.070) + + frames_to_send = [ + ClientConnectedFrame(), + MetricsFrame(data=[llm_ttfb]), + MetricsFrame(data=[tts_ttfb]), + BotStartedSpeakingFrame(), + ] + + expected_down_frames = [ + ClientConnectedFrame, + MetricsFrame, + MetricsFrame, + BotStartedSpeakingFrame, + ] + + await run_test( + processor, + frames_to_send=frames_to_send, + expected_down_frames=expected_down_frames, + observers=[observer], + ) + + self.assertEqual(len(first_speech_latencies), 1) + self.assertGreater(first_speech_latencies[0], 0) + self.assertLess(first_speech_latencies[0], 1.0) + + # Breakdown should also be emitted with the accumulated metrics + self.assertEqual(len(breakdowns), 1) + self.assertEqual(len(breakdowns[0].ttfb), 2) + self.assertEqual(breakdowns[0].ttfb[0].processor, "OpenAILLMService#0") + self.assertEqual(breakdowns[0].ttfb[1].processor, "CartesiaTTSService#0") + + async def test_first_bot_speech_only_once(self): + """Test that first bot speech latency is only emitted once.""" + observer = UserBotLatencyObserver() + processor = IdentityFilter() + + first_speech_latencies = [] + + @observer.event_handler("on_first_bot_speech_latency") + async def on_first_bot_speech(obs, latency_seconds): + first_speech_latencies.append(latency_seconds) + + frames_to_send = [ + ClientConnectedFrame(), + BotStartedSpeakingFrame(), + # Second bot speech should not trigger the event again + VADUserStoppedSpeakingFrame(), + BotStartedSpeakingFrame(), + ] + + expected_down_frames = [ + ClientConnectedFrame, + BotStartedSpeakingFrame, + VADUserStoppedSpeakingFrame, + BotStartedSpeakingFrame, + ] + + await run_test( + processor, + frames_to_send=frames_to_send, + expected_down_frames=expected_down_frames, + observers=[observer], + ) + + self.assertEqual(len(first_speech_latencies), 1) + + async def test_first_bot_speech_skipped_when_user_speaks_first(self): + """Test that first bot speech event is not emitted when user speaks before the bot.""" + observer = UserBotLatencyObserver() + processor = IdentityFilter() + + first_speech_latencies = [] + + @observer.event_handler("on_first_bot_speech_latency") + async def on_first_bot_speech(obs, latency_seconds): + first_speech_latencies.append(latency_seconds) + + frames_to_send = [ + ClientConnectedFrame(), + # User speaks before bot has a chance to greet + VADUserStartedSpeakingFrame(), + VADUserStoppedSpeakingFrame(), + BotStartedSpeakingFrame(), + ] + + expected_down_frames = [ + ClientConnectedFrame, + VADUserStartedSpeakingFrame, + VADUserStoppedSpeakingFrame, + BotStartedSpeakingFrame, + ] + + await run_test( + processor, + frames_to_send=frames_to_send, + expected_down_frames=expected_down_frames, + observers=[observer], + ) + + self.assertEqual(len(first_speech_latencies), 0) + if __name__ == "__main__": unittest.main() From a738a4d82b44d6b6e0c670376db769127abd41e3 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Sun, 1 Mar 2026 11:14:17 -0500 Subject: [PATCH 5/8] Add function call latency tracking to LatencyBreakdown --- changelog/3885.added.md | 2 +- .../foundational/29-turn-tracking-observer.py | 62 +++++++++++++- .../observers/user_bot_latency_observer.py | 40 ++++++++- tests/test_user_bot_latency_observer.py | 81 +++++++++++++++++++ 4 files changed, 179 insertions(+), 6 deletions(-) diff --git a/changelog/3885.added.md b/changelog/3885.added.md index 87cbd7824..96f8cc2cd 100644 --- a/changelog/3885.added.md +++ b/changelog/3885.added.md @@ -1 +1 @@ -- Added `on_latency_breakdown` event to `UserBotLatencyObserver` providing per-service TTFB, text aggregation, and user turn duration metrics for each user-to-bot response cycle. +- Added `on_latency_breakdown` event to `UserBotLatencyObserver` providing per-service TTFB, text aggregation, user turn duration, and function call latency metrics for 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 2c8ec1f84..8bec9e2bc 100644 --- a/examples/foundational/29-turn-tracking-observer.py +++ b/examples/foundational/29-turn-tracking-observer.py @@ -5,11 +5,14 @@ # +import asyncio import os from dotenv import load_dotenv from loguru import logger +from pipecat.adapters.schemas.function_schema import FunctionSchema +from pipecat.adapters.schemas.tools_schema import ToolsSchema from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.frames.frames import LLMRunFrame from pipecat.observers.startup_timing_observer import StartupTimingObserver @@ -26,6 +29,7 @@ from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport from pipecat.services.cartesia.tts import CartesiaTTSService from pipecat.services.deepgram.stt import DeepgramSTTService +from pipecat.services.llm_service import FunctionCallParams from pipecat.services.openai.llm import OpenAILLMService from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.daily.transport import DailyParams @@ -33,6 +37,17 @@ from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams load_dotenv(override=True) + +async def fetch_weather_from_api(params: FunctionCallParams): + await asyncio.sleep(0.25) + await params.result_callback({"conditions": "nice", "temperature": "75"}) + + +async def fetch_restaurant_recommendation(params: FunctionCallParams): + await asyncio.sleep(0.1) + await params.result_callback({"name": "The Golden Dragon"}) + + # We use lambdas to defer transport parameter creation until the transport # type is selected at runtime. transport_params = { @@ -63,6 +78,38 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) + llm.register_function("get_current_weather", fetch_weather_from_api) + llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation) + + weather_function = FunctionSchema( + name="get_current_weather", + description="Get the current weather", + properties={ + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + }, + "format": { + "type": "string", + "enum": ["celsius", "fahrenheit"], + "description": "The temperature unit to use. Infer this from the user's location.", + }, + }, + required=["location", "format"], + ) + restaurant_function = FunctionSchema( + name="get_restaurant_recommendation", + description="Get a restaurant recommendation", + properties={ + "location": { + "type": "string", + "description": "The city and state, e.g. San Francisco, CA", + }, + }, + required=["location"], + ) + tools = ToolsSchema(standard_tools=[weather_function, restaurant_function]) + messages = [ { "role": "system", @@ -70,7 +117,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): }, ] - context = LLMContext(messages) + context = LLMContext(messages, tools) user_aggregator, assistant_aggregator = LLMContextAggregatorPair( context, user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()), @@ -147,9 +194,16 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): 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") + # Show non-STT TTFBs, inserting function calls after the first + # LLM TTFB (which triggered the calls) for a chronological waterfall. + non_stt = [t for t in breakdown.ttfb if t is not stt_ttfb] + fc_shown = False + for ttfb in non_stt: + logger.info(f" {ttfb.processor}: TTFB {ttfb.value:.3f}s") + if not fc_shown and breakdown.function_calls: + for fc in breakdown.function_calls: + logger.info(f" {fc.function_name}: {fc.duration_secs:.3f}s") + fc_shown = True if breakdown.text_aggregation: ta = breakdown.text_aggregation diff --git a/src/pipecat/observers/user_bot_latency_observer.py b/src/pipecat/observers/user_bot_latency_observer.py index cdc7c43d0..46dbbd0f1 100644 --- a/src/pipecat/observers/user_bot_latency_observer.py +++ b/src/pipecat/observers/user_bot_latency_observer.py @@ -15,11 +15,13 @@ is measured. Optionally collects per-service latency breakdown metrics import time from collections import deque from dataclasses import dataclass, field -from typing import List, Optional +from typing import Dict, List, Optional from pipecat.frames.frames import ( BotStartedSpeakingFrame, ClientConnectedFrame, + FunctionCallInProgressFrame, + FunctionCallResultFrame, InterruptionFrame, MetricsFrame, UserStoppedSpeakingFrame, @@ -34,6 +36,19 @@ from pipecat.observers.base_observer import BaseObserver, FramePushed from pipecat.processors.frame_processor import FrameDirection +@dataclass +class FunctionCallMetrics: + """Latency for a single function call execution. + + Parameters: + function_name: Name of the function that was called. + duration_secs: Time in seconds from execution start to result. + """ + + function_name: str + duration_secs: float + + @dataclass class LatencyBreakdown: """Per-service latency breakdown for a single user-to-bot cycle. @@ -52,11 +67,14 @@ class LatencyBreakdown: 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[TTFBMetricsData] = field(default_factory=list) text_aggregation: Optional[TextAggregationMetricsData] = None user_turn_secs: Optional[float] = None + function_calls: List[FunctionCallMetrics] = field(default_factory=list) class UserBotLatencyObserver(BaseObserver): @@ -113,6 +131,8 @@ class UserBotLatencyObserver(BaseObserver): # Per-cycle metric accumulators self._ttfb: List[TTFBMetricsData] = [] self._text_aggregation: Optional[TextAggregationMetricsData] = 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") @@ -171,6 +191,21 @@ class UserBotLatencyObserver(BaseObserver): 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, + duration_secs=time.time() - start_time, + ) + ) elif isinstance(data.frame, MetricsFrame): self._handle_metrics_frame(data.frame) elif isinstance(data.frame, BotStartedSpeakingFrame): @@ -198,6 +233,7 @@ class UserBotLatencyObserver(BaseObserver): ttfb=list(self._ttfb), text_aggregation=self._text_aggregation, 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() @@ -229,3 +265,5 @@ class UserBotLatencyObserver(BaseObserver): self._ttfb = [] self._text_aggregation = None self._user_turn = None + self._function_call_starts = {} + self._function_call_metrics = [] diff --git a/tests/test_user_bot_latency_observer.py b/tests/test_user_bot_latency_observer.py index ab00bfd61..8e63647cf 100644 --- a/tests/test_user_bot_latency_observer.py +++ b/tests/test_user_bot_latency_observer.py @@ -3,6 +3,8 @@ import unittest from pipecat.frames.frames import ( BotStartedSpeakingFrame, ClientConnectedFrame, + FunctionCallInProgressFrame, + FunctionCallResultFrame, InterruptionFrame, MetricsFrame, UserStoppedSpeakingFrame, @@ -463,6 +465,85 @@ class TestUserBotLatencyObserver(unittest.IsolatedAsyncioTestCase): self.assertEqual(len(first_speech_latencies), 0) + async def test_function_call_latency_in_breakdown(self): + """Test that function call duration appears in the latency breakdown.""" + observer = UserBotLatencyObserver() + processor = IdentityFilter() + + breakdowns = [] + + @observer.event_handler("on_latency_breakdown") + async def on_breakdown(obs, breakdown): + breakdowns.append(breakdown) + + tool_call_id = "call_abc123" + + frames_to_send = [ + VADUserStoppedSpeakingFrame(), + FunctionCallInProgressFrame( + function_name="get_weather", + tool_call_id=tool_call_id, + arguments={"location": "Atlanta"}, + ), + SleepFrame(sleep=0.1), + FunctionCallResultFrame( + function_name="get_weather", + tool_call_id=tool_call_id, + arguments={"location": "Atlanta"}, + result={"temperature": "75"}, + ), + BotStartedSpeakingFrame(), + ] + + await run_test( + processor, + frames_to_send=frames_to_send, + observers=[observer], + ) + + self.assertEqual(len(breakdowns), 1) + self.assertEqual(len(breakdowns[0].function_calls), 1) + fc = breakdowns[0].function_calls[0] + self.assertEqual(fc.function_name, "get_weather") + self.assertGreaterEqual(fc.duration_secs, 0.1) + + async def test_function_call_reset_on_interruption(self): + """Test that function call metrics are cleared on interruption.""" + observer = UserBotLatencyObserver() + processor = IdentityFilter() + + breakdowns = [] + + @observer.event_handler("on_latency_breakdown") + async def on_breakdown(obs, breakdown): + breakdowns.append(breakdown) + + frames_to_send = [ + VADUserStoppedSpeakingFrame(), + FunctionCallInProgressFrame( + function_name="get_weather", + tool_call_id="call_1", + arguments={}, + ), + FunctionCallResultFrame( + function_name="get_weather", + tool_call_id="call_1", + arguments={}, + result={}, + ), + InterruptionFrame(), + BotStartedSpeakingFrame(), + ] + + await run_test( + processor, + frames_to_send=frames_to_send, + observers=[observer], + ) + + self.assertEqual(len(breakdowns), 1) + self.assertEqual(len(breakdowns[0].function_calls), 0) + if __name__ == "__main__": unittest.main() From ff5b9850096d2d2226b19f5d905f6417958e1af4 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Sun, 1 Mar 2026 11:51:27 -0500 Subject: [PATCH 6/8] Convert observer data models to Pydantic BaseModel with timestamps Enables .model_dump() serialization for Pipecat Cloud collection. All metrics now include start_time (Unix timestamp) for timeline plotting alongside duration_secs. --- .../foundational/29-turn-tracking-observer.py | 6 +- .../observers/user_bot_latency_observer.py | 77 ++++++++++++++++--- tests/test_user_bot_latency_observer.py | 8 +- 3 files changed, 72 insertions(+), 19 deletions(-) diff --git a/examples/foundational/29-turn-tracking-observer.py b/examples/foundational/29-turn-tracking-observer.py index 8bec9e2bc..476dc4612 100644 --- a/examples/foundational/29-turn-tracking-observer.py +++ b/examples/foundational/29-turn-tracking-observer.py @@ -191,7 +191,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): # 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 "" + stt_note = f" (STT: {stt_ttfb.duration_secs:.3f}s)" if stt_ttfb else "" logger.info(f" User turn: {breakdown.user_turn_secs:.3f}s{stt_note}") # Show non-STT TTFBs, inserting function calls after the first @@ -199,7 +199,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): non_stt = [t for t in breakdown.ttfb if t is not stt_ttfb] fc_shown = False for ttfb in non_stt: - logger.info(f" {ttfb.processor}: TTFB {ttfb.value:.3f}s") + logger.info(f" {ttfb.processor}: TTFB {ttfb.duration_secs:.3f}s") if not fc_shown and breakdown.function_calls: for fc in breakdown.function_calls: logger.info(f" {fc.function_name}: {fc.duration_secs:.3f}s") @@ -207,7 +207,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): if breakdown.text_aggregation: ta = breakdown.text_aggregation - logger.info(f" {ta.processor}: text aggregation {ta.value:.3f}s") + logger.info(f" {ta.processor}: text aggregation {ta.duration_secs:.3f}s") @transport.event_handler("on_client_connected") async def on_client_connected(transport, client): diff --git a/src/pipecat/observers/user_bot_latency_observer.py b/src/pipecat/observers/user_bot_latency_observer.py index 46dbbd0f1..aa0887e30 100644 --- a/src/pipecat/observers/user_bot_latency_observer.py +++ b/src/pipecat/observers/user_bot_latency_observer.py @@ -14,9 +14,10 @@ is measured. Optionally collects per-service latency breakdown metrics import time from collections import deque -from dataclasses import dataclass, field from typing import Dict, List, Optional +from pydantic import BaseModel, Field + from pipecat.frames.frames import ( BotStartedSpeakingFrame, ClientConnectedFrame, @@ -36,21 +37,51 @@ from pipecat.observers.base_observer import BaseObserver, FramePushed from pipecat.processors.frame_processor import FrameDirection -@dataclass -class FunctionCallMetrics: +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 -@dataclass -class LatencyBreakdown: +class LatencyBreakdown(BaseModel): """Per-service latency breakdown for a single user-to-bot cycle. Collected between ``VADUserStoppedSpeakingFrame`` and @@ -61,6 +92,9 @@ class LatencyBreakdown: 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 @@ -71,10 +105,11 @@ class LatencyBreakdown: this cycle. Empty if no function calls occurred. """ - ttfb: List[TTFBMetricsData] = field(default_factory=list) - text_aggregation: Optional[TextAggregationMetricsData] = None + 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) + function_calls: List[FunctionCallMetrics] = Field(default_factory=list) class UserBotLatencyObserver(BaseObserver): @@ -118,6 +153,7 @@ class UserBotLatencyObserver(BaseObserver): """ super().__init__(**kwargs) self._user_stopped_time: Optional[float] = None + self._user_turn_start_time: Optional[float] = None self._user_turn: Optional[float] = None # First bot speech tracking @@ -129,8 +165,8 @@ class UserBotLatencyObserver(BaseObserver): self._frame_history: deque = deque(maxlen=max_frames) # Per-cycle metric accumulators - self._ttfb: List[TTFBMetricsData] = [] - self._text_aggregation: Optional[TextAggregationMetricsData] = None + 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] = [] @@ -172,6 +208,7 @@ class UserBotLatencyObserver(BaseObserver): 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 @@ -182,6 +219,7 @@ class UserBotLatencyObserver(BaseObserver): # 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 + 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, @@ -203,6 +241,7 @@ class UserBotLatencyObserver(BaseObserver): self._function_call_metrics.append( FunctionCallMetrics( function_name=function_name, + start_time=start_time, duration_secs=time.time() - start_time, ) ) @@ -232,6 +271,7 @@ class UserBotLatencyObserver(BaseObserver): 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), ) @@ -251,19 +291,32 @@ class UserBotLatencyObserver(BaseObserver): 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(metrics_data) + 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 = metrics_data + 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 = [] diff --git a/tests/test_user_bot_latency_observer.py b/tests/test_user_bot_latency_observer.py index 8e63647cf..42d5d3367 100644 --- a/tests/test_user_bot_latency_observer.py +++ b/tests/test_user_bot_latency_observer.py @@ -153,7 +153,7 @@ class TestUserBotLatencyObserver(unittest.IsolatedAsyncioTestCase): 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) + self.assertEqual(bd.text_aggregation.duration_secs, 0.030) async def test_interruption_resets_accumulators(self): """Test that InterruptionFrame clears stale metrics from earlier cycles.""" @@ -202,9 +202,9 @@ class TestUserBotLatencyObserver(unittest.IsolatedAsyncioTestCase): # 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[0].duration_secs, 0.224) self.assertEqual(bd.ttfb[1].processor, "CartesiaTTSService#0") - self.assertEqual(bd.ttfb[1].value, 0.142) + self.assertEqual(bd.ttfb[1].duration_secs, 0.142) async def test_only_first_text_aggregation_kept(self): """Test that only the first text aggregation metric is kept per cycle.""" @@ -243,7 +243,7 @@ class TestUserBotLatencyObserver(unittest.IsolatedAsyncioTestCase): self.assertEqual(len(breakdowns), 1) self.assertIsNotNone(breakdowns[0].text_aggregation) - self.assertEqual(breakdowns[0].text_aggregation.value, 0.030) + self.assertEqual(breakdowns[0].text_aggregation.duration_secs, 0.030) async def test_user_turn_measured(self): """Test that pre-LLM wait from user silence to UserStopped is captured.""" From 8f66272de7079818aba54b0ea74804052b030ae4 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Mon, 2 Mar 2026 16:16:38 -0500 Subject: [PATCH 7/8] Update changelog --- changelog/3885.added.2.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/changelog/3885.added.2.md b/changelog/3885.added.2.md index a9562b883..5a6adce12 100644 --- a/changelog/3885.added.2.md +++ b/changelog/3885.added.2.md @@ -1 +1 @@ -- Added `on_first_bot_speech_latency` event to `UserBotLatencyObserver` measuring the time from client connection to first bot speech, including a latency breakdown with per-service metrics. +- Added `on_first_bot_speech_latency` event to `UserBotLatencyObserver` measuring the time from client connection to first bot speech. An `on_latency_breakdown` is also emitted for this first speech event. From 7dbb130666acd5ca9177f4a65b82194d60077b52 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Mon, 2 Mar 2026 19:23:42 -0500 Subject: [PATCH 8/8] Add chronological_events utility function to display UserBotLatencyObserver report --- .../foundational/29-turn-tracking-observer.py | 26 +----- .../observers/user_bot_latency_observer.py | 29 +++++++ tests/test_user_bot_latency_observer.py | 84 ++++++++++++++++++- 3 files changed, 114 insertions(+), 25 deletions(-) diff --git a/examples/foundational/29-turn-tracking-observer.py b/examples/foundational/29-turn-tracking-observer.py index 476dc4612..cf85972e1 100644 --- a/examples/foundational/29-turn-tracking-observer.py +++ b/examples/foundational/29-turn-tracking-observer.py @@ -184,30 +184,8 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): @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.duration_secs:.3f}s)" if stt_ttfb else "" - logger.info(f" User turn: {breakdown.user_turn_secs:.3f}s{stt_note}") - - # Show non-STT TTFBs, inserting function calls after the first - # LLM TTFB (which triggered the calls) for a chronological waterfall. - non_stt = [t for t in breakdown.ttfb if t is not stt_ttfb] - fc_shown = False - for ttfb in non_stt: - logger.info(f" {ttfb.processor}: TTFB {ttfb.duration_secs:.3f}s") - if not fc_shown and breakdown.function_calls: - for fc in breakdown.function_calls: - logger.info(f" {fc.function_name}: {fc.duration_secs:.3f}s") - fc_shown = True - - if breakdown.text_aggregation: - ta = breakdown.text_aggregation - logger.info(f" {ta.processor}: text aggregation {ta.duration_secs:.3f}s") + for event in breakdown.chronological_events(): + logger.info(f" {event}") @transport.event_handler("on_client_connected") async def on_client_connected(transport, client): diff --git a/src/pipecat/observers/user_bot_latency_observer.py b/src/pipecat/observers/user_bot_latency_observer.py index aa0887e30..0672b689c 100644 --- a/src/pipecat/observers/user_bot_latency_observer.py +++ b/src/pipecat/observers/user_bot_latency_observer.py @@ -111,6 +111,35 @@ class LatencyBreakdown(BaseModel): 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. diff --git a/tests/test_user_bot_latency_observer.py b/tests/test_user_bot_latency_observer.py index 42d5d3367..96c24724b 100644 --- a/tests/test_user_bot_latency_observer.py +++ b/tests/test_user_bot_latency_observer.py @@ -15,7 +15,13 @@ from pipecat.metrics.metrics import ( TextAggregationMetricsData, TTFBMetricsData, ) -from pipecat.observers.user_bot_latency_observer import UserBotLatencyObserver +from pipecat.observers.user_bot_latency_observer import ( + FunctionCallMetrics, + LatencyBreakdown, + TextAggregationBreakdownMetrics, + TTFBBreakdownMetrics, + UserBotLatencyObserver, +) from pipecat.processors.filters.identity_filter import IdentityFilter from pipecat.tests.utils import SleepFrame, run_test @@ -545,5 +551,81 @@ class TestUserBotLatencyObserver(unittest.IsolatedAsyncioTestCase): self.assertEqual(len(breakdowns[0].function_calls), 0) +class TestLatencyBreakdownChronologicalEvents(unittest.TestCase): + """Tests for LatencyBreakdown.chronological_events().""" + + def test_events_sorted_by_start_time(self): + """Test that events are returned in chronological order.""" + breakdown = LatencyBreakdown( + user_turn_start_time=100.0, + user_turn_secs=0.150, + ttfb=[ + TTFBBreakdownMetrics( + processor="OpenAILLMService#0", + model="gpt-4o", + start_time=100.200, + duration_secs=0.250, + ), + TTFBBreakdownMetrics( + processor="DeepgramSTTService#0", + start_time=100.050, + duration_secs=0.080, + ), + TTFBBreakdownMetrics( + processor="CartesiaTTSService#0", + start_time=100.500, + duration_secs=0.070, + ), + ], + function_calls=[ + FunctionCallMetrics( + function_name="get_weather", + start_time=100.450, + duration_secs=0.120, + ), + ], + text_aggregation=TextAggregationBreakdownMetrics( + processor="CartesiaTTSService#0", + start_time=100.480, + duration_secs=0.030, + ), + ) + + events = breakdown.chronological_events() + + self.assertEqual(len(events), 6) + self.assertEqual(events[0], "User turn: 0.150s") + self.assertEqual(events[1], "DeepgramSTTService#0: TTFB 0.080s") + self.assertEqual(events[2], "OpenAILLMService#0: TTFB 0.250s") + self.assertEqual(events[3], "get_weather: 0.120s") + self.assertEqual(events[4], "CartesiaTTSService#0: text aggregation 0.030s") + self.assertEqual(events[5], "CartesiaTTSService#0: TTFB 0.070s") + + def test_empty_breakdown(self): + """Test that an empty breakdown returns no events.""" + breakdown = LatencyBreakdown() + self.assertEqual(breakdown.chronological_events(), []) + + def test_user_turn_requires_both_fields(self): + """Test that user turn is only included when both start_time and secs are set.""" + # Only start_time, no duration + breakdown = LatencyBreakdown(user_turn_start_time=100.0) + self.assertEqual(breakdown.chronological_events(), []) + + # Only duration, no start_time + breakdown = LatencyBreakdown(user_turn_secs=0.150) + self.assertEqual(breakdown.chronological_events(), []) + + def test_ttfb_only(self): + """Test breakdown with only TTFB metrics.""" + breakdown = LatencyBreakdown( + ttfb=[ + TTFBBreakdownMetrics(processor="LLM#0", start_time=100.0, duration_secs=0.200), + ], + ) + events = breakdown.chronological_events() + self.assertEqual(events, ["LLM#0: TTFB 0.200s"]) + + if __name__ == "__main__": unittest.main()