From 88e8fcdaca2cd70d9c49a812acc148c38fba8f46 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= <aleix@daily.co>
Date: Wed, 21 May 2025 15:56:38 -0700
Subject: [PATCH] observers: added UserBotLatencyLogObserver

---
 CHANGELOG.md                                  |  4 ++
 .../foundational/29-turn-tracking-observer.py |  2 +
 .../loggers/user_bot_latency_log_observer.py  | 50 +++++++++++++++++++
 3 files changed, 56 insertions(+)
 create mode 100644 src/pipecat/observers/loggers/user_bot_latency_log_observer.py

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 564650905..28e85beb2 100644
--- a/CHANGELOG.md
+++ b/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.
 
diff --git a/examples/foundational/29-turn-tracking-observer.py b/examples/foundational/29-turn-tracking-observer.py
index 0e9c7acec..08c39fe55 100644
--- a/examples/foundational/29-turn-tracking-observer.py
+++ b/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
diff --git a/src/pipecat/observers/loggers/user_bot_latency_log_observer.py b/src/pipecat/observers/loggers/user_bot_latency_log_observer.py
new file mode 100644
index 000000000..f601a9e9e
--- /dev/null
+++ b/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}")