diff --git a/changelog/3355.added.md b/changelog/3355.added.md new file mode 100644 index 000000000..ec8209f57 --- /dev/null +++ b/changelog/3355.added.md @@ -0,0 +1 @@ +- Added `UserBotLatencyObserver` for tracking user-to-bot response latency. When tracing is enabled, latency measurements are automatically recorded as `turn.user_bot_latency_seconds` attributes on OpenTelemetry turn spans. diff --git a/changelog/3355.deprecated.md b/changelog/3355.deprecated.md new file mode 100644 index 000000000..7406268a7 --- /dev/null +++ b/changelog/3355.deprecated.md @@ -0,0 +1 @@ +- Deprecated `UserBotLatencyLogObserver`. Use `UserBotLatencyObserver` directly with its `on_latency_measured` event handler instead. diff --git a/examples/foundational/29-turn-tracking-observer.py b/examples/foundational/29-turn-tracking-observer.py index b6d8dcd88..272254eb6 100644 --- a/examples/foundational/29-turn-tracking-observer.py +++ b/examples/foundational/29-turn-tracking-observer.py @@ -14,7 +14,7 @@ from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnal from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.audio.vad.vad_analyzer import VADParams from pipecat.frames.frames import LLMRunFrame -from pipecat.observers.loggers.user_bot_latency_log_observer import UserBotLatencyLogObserver +from pipecat.observers.user_bot_latency_observer import UserBotLatencyObserver from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -96,6 +96,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): ] ) + # Create latency tracking observer + latency_observer = UserBotLatencyObserver() + task = PipelineTask( pipeline, params=PipelineParams( @@ -103,9 +106,14 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): enable_usage_metrics=True, ), idle_timeout_secs=runner_args.pipeline_idle_timeout_secs, - observers=[UserBotLatencyLogObserver()], + observers=[latency_observer], ) + # Log latency measurements using the event handler + @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") + turn_observer = task.turn_tracking_observer if turn_observer: diff --git a/src/pipecat/observers/loggers/user_bot_latency_log_observer.py b/src/pipecat/observers/loggers/user_bot_latency_log_observer.py index b8dff734e..044d4dea6 100644 --- a/src/pipecat/observers/loggers/user_bot_latency_log_observer.py +++ b/src/pipecat/observers/loggers/user_bot_latency_log_observer.py @@ -4,9 +4,15 @@ # SPDX-License-Identifier: BSD 2-Clause License # -"""Observer for measuring user-to-bot response latency.""" +"""Observer for measuring user-to-bot response latency. + +.. deprecated:: 0.0.102 + This module is deprecated. Use :class:`UserBotLatencyObserver` directly + with its ``on_latency_measured`` event handler instead. +""" import time +import warnings from statistics import mean from loguru import logger @@ -27,6 +33,10 @@ class UserBotLatencyLogObserver(BaseObserver): This helps measure how quickly the AI services respond by tracking conversation turn timing and logging latency metrics. + + .. deprecated:: 0.0.102 + This class is deprecated. Use :class:`UserBotLatencyObserver` directly + with its ``on_latency_measured`` event handler for custom logging. """ def __init__(self): @@ -34,7 +44,17 @@ class UserBotLatencyLogObserver(BaseObserver): Sets up tracking for processed frames and user speech timing to calculate response latencies. + + .. deprecated:: 0.0.102 + This class is deprecated. Use :class:`UserBotLatencyObserver` + directly with its ``on_latency_measured`` event handler. """ + warnings.warn( + "UserBotLatencyLogObserver is deprecated and will be removed in a future version. " + "Use UserBotLatencyObserver directly with its on_latency_measured event handler instead.", + DeprecationWarning, + stacklevel=2, + ) super().__init__() self._user_bot_latency_processed_frames = set() self._user_stopped_time = 0 diff --git a/src/pipecat/observers/user_bot_latency_observer.py b/src/pipecat/observers/user_bot_latency_observer.py new file mode 100644 index 000000000..475da0c5d --- /dev/null +++ b/src/pipecat/observers/user_bot_latency_observer.py @@ -0,0 +1,81 @@ +"""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. +""" + +import time +from typing import Optional, Set + +from pipecat.frames.frames import ( + BotStartedSpeakingFrame, + VADUserStartedSpeakingFrame, + VADUserStoppedSpeakingFrame, +) +from pipecat.observers.base_observer import BaseObserver, FramePushed +from pipecat.processors.frame_processor import FrameDirection + + +class UserBotLatencyObserver(BaseObserver): + """Observer that tracks user-to-bot response latency. + + Measures the time between when a user stops speaking (VADUserStoppedSpeakingFrame) + and when the bot starts speaking (BotStartedSpeakingFrame). Emits events when + latency is measured, allowing consumers to log, trace, or otherwise process + the latency data. + + 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. + """ + + def __init__(self, **kwargs): + """Initialize the user-bot latency observer. + + Sets up tracking for processed frames and user speech timing + to calculate response latencies. + + Args: + **kwargs: Additional arguments passed to parent class. + """ + super().__init__(**kwargs) + self._user_stopped_time: Optional[float] = None + self._processed_frames: Set[str] = set() + + self._register_event_handler("on_latency_measured") + + 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. + + Args: + data: Frame push event containing the frame and direction information. + """ + # Only process downstream frames + if data.direction != FrameDirection.DOWNSTREAM: + return + + # Skip already processed frames + if data.frame.id in self._processed_frames: + return + + self._processed_frames.add(data.frame.id) + + # Track VAD and bot speaking events for latency + if isinstance(data.frame, VADUserStartedSpeakingFrame): + # Reset when user starts speaking + self._user_stopped_time = None + elif isinstance(data.frame, VADUserStoppedSpeakingFrame): + # Record timestamp when user stops speaking + self._user_stopped_time = time.time() + 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) diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 4ccc879b8..e643164eb 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -43,6 +43,7 @@ from pipecat.frames.frames import ( from pipecat.metrics.metrics import ProcessingMetricsData, TTFBMetricsData from pipecat.observers.base_observer import BaseObserver, FramePushed from pipecat.observers.turn_tracking_observer import TurnTrackingObserver +from pipecat.observers.user_bot_latency_observer import UserBotLatencyObserver from pipecat.pipeline.base_pipeline import BasePipeline from pipecat.pipeline.base_task import BasePipelineTask, PipelineTaskParams from pipecat.pipeline.pipeline import Pipeline, PipelineSink, PipelineSource @@ -287,13 +288,19 @@ class PipelineTask(BasePipelineTask): observers = self._params.observers observers = observers or [] self._turn_tracking_observer: Optional[TurnTrackingObserver] = None + self._user_bot_latency_observer: Optional[UserBotLatencyObserver] = None self._turn_trace_observer: Optional[TurnTraceObserver] = None if self._enable_turn_tracking: self._turn_tracking_observer = TurnTrackingObserver() observers.append(self._turn_tracking_observer) if self._enable_tracing and self._turn_tracking_observer: + # Create latency observer for tracing + self._user_bot_latency_observer = UserBotLatencyObserver() + observers.append(self._user_bot_latency_observer) + # Create turn trace observer with latency tracking self._turn_trace_observer = TurnTraceObserver( self._turn_tracking_observer, + latency_tracker=self._user_bot_latency_observer, conversation_id=self._conversation_id, additional_span_attributes=self._additional_span_attributes, ) diff --git a/src/pipecat/utils/tracing/turn_trace_observer.py b/src/pipecat/utils/tracing/turn_trace_observer.py index 59b0bd2ae..78a9c9ea9 100644 --- a/src/pipecat/utils/tracing/turn_trace_observer.py +++ b/src/pipecat/utils/tracing/turn_trace_observer.py @@ -18,6 +18,7 @@ from loguru import logger from pipecat.frames.frames import StartFrame from pipecat.observers.base_observer import BaseObserver, FramePushed from pipecat.observers.turn_tracking_observer import TurnTrackingObserver +from pipecat.observers.user_bot_latency_observer import UserBotLatencyObserver from pipecat.utils.tracing.conversation_context_provider import ConversationContextProvider from pipecat.utils.tracing.setup import is_tracing_available from pipecat.utils.tracing.turn_context_provider import TurnContextProvider @@ -45,6 +46,7 @@ class TurnTraceObserver(BaseObserver): def __init__( self, turn_tracker: TurnTrackingObserver, + latency_tracker: UserBotLatencyObserver, conversation_id: Optional[str] = None, additional_span_attributes: Optional[dict] = None, **kwargs, @@ -53,12 +55,14 @@ class TurnTraceObserver(BaseObserver): Args: turn_tracker: The turn tracking observer to monitor. + latency_tracker: The latency tracking observer for user-bot latency. conversation_id: Optional conversation ID for grouping turns. additional_span_attributes: Additional attributes to add to spans. **kwargs: Additional arguments passed to parent class. """ super().__init__(**kwargs) self._turn_tracker = turn_tracker + self._latency_tracker = latency_tracker self._current_span: Optional["Span"] = None self._current_turn_number: int = 0 self._trace_context_map: Dict[int, "SpanContext"] = {} @@ -69,15 +73,32 @@ class TurnTraceObserver(BaseObserver): self._conversation_id = conversation_id self._additional_span_attributes = additional_span_attributes or {} - if turn_tracker: + @turn_tracker.event_handler("on_turn_started") + async def on_turn_started(tracker, turn_number): + await self._handle_turn_started(turn_number) - @turn_tracker.event_handler("on_turn_started") - async def on_turn_started(tracker, turn_number): - await self._handle_turn_started(turn_number) + @turn_tracker.event_handler("on_turn_ended") + async def on_turn_ended(tracker, turn_number, duration, was_interrupted): + await self._handle_turn_ended(turn_number, duration, was_interrupted) - @turn_tracker.event_handler("on_turn_ended") - async def on_turn_ended(tracker, turn_number, duration, was_interrupted): - await self._handle_turn_ended(turn_number, duration, was_interrupted) + @latency_tracker.event_handler("on_latency_measured") + async def on_latency_measured(tracker, latency_seconds): + await self._handle_latency_measured(latency_seconds) + + async def _handle_latency_measured(self, latency_seconds: float): + """Handle latency measurement events. + + Called when the latency tracker measures user-to-bot latency. + Adds the latency as an attribute to the current turn span. + + Args: + latency_seconds: The measured latency in seconds. + """ + if self._current_span and is_tracing_available(): + self._current_span.set_attribute("turn.user_bot_latency_seconds", latency_seconds) + logger.debug( + f"Turn {self._current_turn_number} user-bot latency: {latency_seconds:.3f}s" + ) async def on_push_frame(self, data: FramePushed): """Process a frame without modifying it. diff --git a/tests/test_user_bot_latency_observer.py b/tests/test_user_bot_latency_observer.py new file mode 100644 index 000000000..1b7325d14 --- /dev/null +++ b/tests/test_user_bot_latency_observer.py @@ -0,0 +1,137 @@ +import unittest + +from pipecat.frames.frames import ( + BotStartedSpeakingFrame, + VADUserStartedSpeakingFrame, + VADUserStoppedSpeakingFrame, +) +from pipecat.observers.user_bot_latency_observer import UserBotLatencyObserver +from pipecat.processors.filters.identity_filter import IdentityFilter +from pipecat.tests.utils import run_test + + +class TestUserBotLatencyObserver(unittest.IsolatedAsyncioTestCase): + """Tests for UserBotLatencyObserver.""" + + async def test_normal_latency_measurement(self): + """Test basic latency measurement from user stop to bot start.""" + # Create observer + observer = UserBotLatencyObserver() + + # Create identity filter (passes all frames through) + processor = IdentityFilter() + + # Capture latency events + latencies = [] + + @observer.event_handler("on_latency_measured") + async def on_latency(obs, latency_seconds): + latencies.append(latency_seconds) + + # Define frame sequence + frames_to_send = [ + VADUserStoppedSpeakingFrame(), + BotStartedSpeakingFrame(), + ] + + expected_down_frames = [ + VADUserStoppedSpeakingFrame, + BotStartedSpeakingFrame, + ] + + # Run test + await run_test( + processor, + frames_to_send=frames_to_send, + expected_down_frames=expected_down_frames, + observers=[observer], + ) + + # Verify latency was measured + self.assertEqual(len(latencies), 1) + self.assertGreater(latencies[0], 0) + self.assertLess(latencies[0], 1.0) # Should be very quick + + async def test_multiple_latency_measurements(self): + """Test that multiple user-bot exchanges produce separate latency events.""" + # Create observer + observer = UserBotLatencyObserver() + + # Create identity filter + processor = IdentityFilter() + + # Capture latency events + latencies = [] + + @observer.event_handler("on_latency_measured") + async def on_latency(obs, latency_seconds): + latencies.append(latency_seconds) + + # Define frame sequence with two complete cycles + frames_to_send = [ + # First cycle + VADUserStoppedSpeakingFrame(), + BotStartedSpeakingFrame(), + # Second cycle + VADUserStoppedSpeakingFrame(), + BotStartedSpeakingFrame(), + ] + + expected_down_frames = [ + VADUserStoppedSpeakingFrame, + BotStartedSpeakingFrame, + VADUserStoppedSpeakingFrame, + BotStartedSpeakingFrame, + ] + + # Run test + await run_test( + processor, + frames_to_send=frames_to_send, + expected_down_frames=expected_down_frames, + observers=[observer], + ) + + # Verify two separate latencies were measured + self.assertEqual(len(latencies), 2) + 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 + observer = UserBotLatencyObserver() + + # Create identity filter + processor = IdentityFilter() + + # Capture latency events + latencies = [] + + @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 + frames_to_send = [ + BotStartedSpeakingFrame(), + ] + + expected_down_frames = [ + BotStartedSpeakingFrame, + ] + + # Run test + await run_test( + processor, + frames_to_send=frames_to_send, + expected_down_frames=expected_down_frames, + observers=[observer], + ) + + # Verify no latency was measured + self.assertEqual(len(latencies), 0) + + +if __name__ == "__main__": + unittest.main()