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/changelog/3885.added.2.md b/changelog/3885.added.2.md new file mode 100644 index 000000000..5a6adce12 --- /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. An `on_latency_breakdown` is also emitted for this first speech event. diff --git a/changelog/3885.added.md b/changelog/3885.added.md new file mode 100644 index 000000000..96f8cc2cd --- /dev/null +++ b/changelog/3885.added.md @@ -0,0 +1 @@ +- 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 4af28f1ed..cf85972e1 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()), @@ -101,6 +148,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") @@ -131,6 +182,11 @@ 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): + for event in breakdown.chronological_events(): + logger.info(f" {event}") + @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..0672b689c 100644 --- a/src/pipecat/observers/user_bot_latency_observer.py +++ b/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 = [] diff --git a/tests/test_user_bot_latency_observer.py b/tests/test_user_bot_latency_observer.py index 1b7325d14..96c24724b 100644 --- a/tests/test_user_bot_latency_observer.py +++ b/tests/test_user_bot_latency_observer.py @@ -2,12 +2,28 @@ import unittest from pipecat.frames.frames import ( BotStartedSpeakingFrame, + ClientConnectedFrame, + FunctionCallInProgressFrame, + FunctionCallResultFrame, + InterruptionFrame, + MetricsFrame, + UserStoppedSpeakingFrame, VADUserStartedSpeakingFrame, VADUserStoppedSpeakingFrame, ) -from pipecat.observers.user_bot_latency_observer import UserBotLatencyObserver +from pipecat.metrics.metrics import ( + TextAggregationMetricsData, + TTFBMetricsData, +) +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 run_test +from pipecat.tests.utils import SleepFrame, run_test class TestUserBotLatencyObserver(unittest.IsolatedAsyncioTestCase): @@ -97,22 +113,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.duration_secs, 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].duration_secs, 0.224) + self.assertEqual(bd.ttfb[1].processor, "CartesiaTTSService#0") + 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.""" + 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.duration_secs, 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 +341,6 @@ class TestUserBotLatencyObserver(unittest.IsolatedAsyncioTestCase): BotStartedSpeakingFrame, ] - # Run test await run_test( processor, frames_to_send=frames_to_send, @@ -129,8 +348,283 @@ class TestUserBotLatencyObserver(unittest.IsolatedAsyncioTestCase): observers=[observer], ) - # Verify no latency was measured 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) + + 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) + + +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__":