observers: added UserBotLatencyLogObserver

CHANGELOG.md

@@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Added `UserBotLatencyLogObserver`. This is an observer that logs the latency
+  between when the user stops speaking and when the bot starts speaking. This
+  gives you an initial idea on how quickly the AI services respond.
+
 - Added `SarvamTTSService`, which implements Sarvam AI's TTS API:
   https://docs.sarvam.ai/api-reference-docs/text-to-speech/convert.
 

examples/foundational/29-turn-tracking-observer.py

@@ -11,6 +11,7 @@ from dotenv import load_dotenv
 from loguru import logger
 
 from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.observers.loggers.user_bot_latency_log_observer import UserBotLatencyLogObserver
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
@@ -76,6 +77,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac
             enable_usage_metrics=True,
             report_only_initial_ttfb=True,
         ),
+        observers=[UserBotLatencyLogObserver()],
     )
 
     turn_observer = task.turn_tracking_observer

src/pipecat/observers/loggers/user_bot_latency_log_observer.py

@@ -0,0 +1,50 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import time
+
+from loguru import logger
+
+from pipecat.frames.frames import (
+    BotStartedSpeakingFrame,
+    UserStartedSpeakingFrame,
+    UserStoppedSpeakingFrame,
+)
+from pipecat.observers.base_observer import BaseObserver, FramePushed
+from pipecat.processors.frame_processor import FrameDirection
+
+
+class UserBotLatencyLogObserver(BaseObserver):
+    """Observer that logs the latency between when the user stops speaking and
+    when the bot starts speaking.
+
+    This helps measure how quickly the AI services respond.
+
+    """
+
+    def __init__(self):
+        super().__init__()
+        self._processed_frames = set()
+        self._user_stopped_time = 0
+
+    async def on_push_frame(self, data: FramePushed):
+        # 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)
+
+        if isinstance(data.frame, UserStartedSpeakingFrame):
+            self._user_stopped_time = 0
+        elif isinstance(data.frame, UserStoppedSpeakingFrame):
+            self._user_stopped_time = time.time()
+        elif isinstance(data.frame, BotStartedSpeakingFrame) and self._user_stopped_time:
+            latency = time.time() - self._user_stopped_time
+            logger.debug(f"⏱️ LATENCY FROM USER STOPPED SPEAKING TO BOT STARTED SPEAKING: {latency}")