From 793ed8f9e3f59a07ee0499d032d958ca864fd109 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 2 Apr 2026 21:54:36 -0400 Subject: [PATCH] Remove deprecated UserBotLatencyLogObserver and UserIdleProcessor UserBotLatencyLogObserver (deprecated 0.0.102) is replaced by UserBotLatencyObserver. UserIdleProcessor (deprecated 0.0.100) is replaced by LLMUserAggregator with user_idle_timeout. --- .../loggers/user_bot_latency_log_observer.py | 109 --------- src/pipecat/processors/user_idle_processor.py | 209 ---------------- tests/test_user_idle_processor.py | 224 ------------------ 3 files changed, 542 deletions(-) delete mode 100644 src/pipecat/observers/loggers/user_bot_latency_log_observer.py delete mode 100644 src/pipecat/processors/user_idle_processor.py delete mode 100644 tests/test_user_idle_processor.py diff --git a/src/pipecat/observers/loggers/user_bot_latency_log_observer.py b/src/pipecat/observers/loggers/user_bot_latency_log_observer.py deleted file mode 100644 index 2323a36ee..000000000 --- a/src/pipecat/observers/loggers/user_bot_latency_log_observer.py +++ /dev/null @@ -1,109 +0,0 @@ -# -# Copyright (c) 2024-2026, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - -"""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 - -from pipecat.frames.frames import ( - BotStartedSpeakingFrame, - CancelFrame, - EndFrame, - VADUserStartedSpeakingFrame, - VADUserStoppedSpeakingFrame, -) -from pipecat.observers.base_observer import BaseObserver, FramePushed -from pipecat.processors.frame_processor import FrameDirection - - -class UserBotLatencyLogObserver(BaseObserver): - """Observer that measures time between user stopping speech and bot starting speech. - - 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): - """Initialize the latency observer. - - 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 - self._latencies = [] - - async def on_push_frame(self, data: FramePushed): - """Process frames to track speech timing and calculate latency. - - 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._user_bot_latency_processed_frames: - return - - self._user_bot_latency_processed_frames.add(data.frame.id) - - if isinstance(data.frame, VADUserStartedSpeakingFrame): - self._user_stopped_time = 0 - elif isinstance(data.frame, VADUserStoppedSpeakingFrame): - self._user_stopped_time = data.frame.timestamp - data.frame.stop_secs - elif isinstance(data.frame, (EndFrame, CancelFrame)): - self._log_summary() - elif isinstance(data.frame, BotStartedSpeakingFrame) and self._user_stopped_time: - latency = time.time() - self._user_stopped_time - self._user_stopped_time = 0 - self._latencies.append(latency) - self._log_latency(latency) - - def _log_summary(self): - if not self._latencies: - return - avg_latency = mean(self._latencies) - min_latency = min(self._latencies) - max_latency = max(self._latencies) - logger.info( - f"⏱️ LATENCY FROM USER STOPPED SPEAKING TO BOT STARTED SPEAKING - Avg: {avg_latency:.3f}s, Min: {min_latency:.3f}s, Max: {max_latency:.3f}s" - ) - - def _log_latency(self, latency: float): - """Log the latency. - - Args: - latency: The latency to log. - """ - logger.debug( - f"⏱️ LATENCY FROM USER STOPPED SPEAKING TO BOT STARTED SPEAKING: {latency:.3f}s" - ) diff --git a/src/pipecat/processors/user_idle_processor.py b/src/pipecat/processors/user_idle_processor.py deleted file mode 100644 index f7ea48599..000000000 --- a/src/pipecat/processors/user_idle_processor.py +++ /dev/null @@ -1,209 +0,0 @@ -# -# Copyright (c) 2024-2026, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - -"""User idle detection and timeout handling for Pipecat.""" - -import asyncio -import inspect -import warnings -from typing import Awaitable, Callable, Union - -from pipecat.frames.frames import ( - BotSpeakingFrame, - CancelFrame, - EndFrame, - Frame, - FunctionCallInProgressFrame, - FunctionCallResultFrame, - UserStartedSpeakingFrame, - UserStoppedSpeakingFrame, -) -from pipecat.processors.frame_processor import FrameDirection, FrameProcessor - - -class UserIdleProcessor(FrameProcessor): - """Monitors user inactivity and triggers callbacks after timeout periods. - - .. deprecated:: 0.0.100 - UserIdleProcessor is deprecated in 0.0.100 and will be removed in a future version. - Use LLMUserAggregator with user_idle_timeout parameter instead. - - This processor tracks user activity and triggers configurable callbacks when - users become idle. It starts monitoring only after the first conversation - activity and supports both basic and retry-based callback patterns. - - Example:: - - # Retry callback: - async def handle_idle(processor: "UserIdleProcessor", retry_count: int) -> bool: - if retry_count < 3: - await send_reminder("Are you still there?") - return True - return False - - # Basic callback: - async def handle_idle(processor: "UserIdleProcessor") -> None: - await send_reminder("Are you still there?") - - processor = UserIdleProcessor( - callback=handle_idle, - timeout=5.0 - ) - """ - - def __init__( - self, - *, - callback: Union[ - Callable[["UserIdleProcessor"], Awaitable[None]], # Basic - Callable[["UserIdleProcessor", int], Awaitable[bool]], # Retry - ], - timeout: float, - **kwargs, - ): - """Initialize the user idle processor. - - Args: - callback: Function to call when user is idle. Can be either a basic - callback taking only the processor, or a retry callback taking - the processor and retry count. Retry callbacks should return - True to continue monitoring or False to stop. - timeout: Seconds to wait before considering user idle. - **kwargs: Additional arguments passed to FrameProcessor. - """ - super().__init__(**kwargs) - - warnings.warn( - "UserIdleProcessor is deprecated in 0.0.100 and will be removed in a " - "future version. Use LLMUserAggregator with user_idle_timeout parameter " - "instead.", - DeprecationWarning, - ) - - self._callback = self._wrap_callback(callback) - self._timeout = timeout - self._retry_count = 0 - self._interrupted = False - self._conversation_started = False - self._idle_task = None - self._idle_event = asyncio.Event() - - def _wrap_callback( - self, - callback: Union[ - Callable[["UserIdleProcessor"], Awaitable[None]], - Callable[["UserIdleProcessor", int], Awaitable[bool]], - ], - ) -> Callable[["UserIdleProcessor", int], Awaitable[bool]]: - """Wraps callback to support both basic and retry signatures. - - Args: - callback: The callback function to wrap. - - Returns: - A wrapped callback that returns bool to indicate whether to continue monitoring. - """ - sig = inspect.signature(callback) - param_count = len(sig.parameters) - - async def wrapper(processor: "UserIdleProcessor", retry_count: int) -> bool: - if param_count == 1: - # Basic callback - await callback(processor) # type: ignore - return True - else: - # Retry callback - return await callback(processor, retry_count) # type: ignore - - return wrapper - - def _create_idle_task(self) -> None: - """Creates the idle task if it hasn't been created yet.""" - if not self._idle_task: - self._idle_task = self.create_task(self._idle_task_handler()) - - @property - def retry_count(self) -> int: - """Get the current retry count. - - Returns: - The number of times the idle callback has been triggered. - """ - return self._retry_count - - async def _stop(self) -> None: - """Stops and cleans up the idle monitoring task.""" - if self._idle_task: - await self.cancel_task(self._idle_task) - self._idle_task = None - - async def process_frame(self, frame: Frame, direction: FrameDirection) -> None: - """Processes incoming frames and manages idle monitoring state. - - Args: - frame: The frame to process. - direction: Direction of the frame flow. - """ - await super().process_frame(frame, direction) - - # Check for end frames before processing - if isinstance(frame, (EndFrame, CancelFrame)): - # Stop the idle task, if it exists - await self._stop() - # Push the frame down the pipeline - await self.push_frame(frame, direction) - return - - await self.push_frame(frame, direction) - - # Start monitoring on first conversation activity - if not self._conversation_started and isinstance( - frame, (UserStartedSpeakingFrame, BotSpeakingFrame) - ): - self._conversation_started = True - self._create_idle_task() - - # Only process these events if conversation has started - if self._conversation_started: - # We shouldn't call the idle callback if the user or the bot are speaking - if isinstance(frame, UserStartedSpeakingFrame): - self._retry_count = 0 # Reset retry count when user speaks - self._interrupted = True - self._idle_event.set() - elif isinstance(frame, UserStoppedSpeakingFrame): - self._interrupted = False - self._idle_event.set() - elif isinstance(frame, BotSpeakingFrame): - self._idle_event.set() - elif isinstance(frame, FunctionCallInProgressFrame): - # Function calls can take longer than the timeout, so we want to prevent idle callbacks - self._interrupted = True - self._idle_event.set() - elif isinstance(frame, FunctionCallResultFrame): - self._interrupted = False - self._idle_event.set() - - async def cleanup(self) -> None: - """Cleans up resources when processor is shutting down.""" - await super().cleanup() - if self._idle_task: # Only stop if task exists - await self._stop() - - async def _idle_task_handler(self) -> None: - """Monitors for idle timeout and triggers callbacks. - - Runs in a loop until cancelled or callback indicates completion. - """ - running = True - while running: - try: - await asyncio.wait_for(self._idle_event.wait(), timeout=self._timeout) - except asyncio.TimeoutError: - if not self._interrupted: - self._retry_count += 1 - running = await self._callback(self, self._retry_count) - finally: - self._idle_event.clear() diff --git a/tests/test_user_idle_processor.py b/tests/test_user_idle_processor.py deleted file mode 100644 index 3a0169ec7..000000000 --- a/tests/test_user_idle_processor.py +++ /dev/null @@ -1,224 +0,0 @@ -# -# Copyright (c) 2024-2026, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - -import asyncio -import unittest - -from pipecat.frames.frames import ( - BotSpeakingFrame, - BotStartedSpeakingFrame, - BotStoppedSpeakingFrame, - UserStartedSpeakingFrame, - UserStoppedSpeakingFrame, -) -from pipecat.processors.user_idle_processor import UserIdleProcessor -from pipecat.tests.utils import SleepFrame, run_test - - -class TestUserIdleProcessor(unittest.IsolatedAsyncioTestCase): - async def test_basic_idle_detection(self): - """Test that idle callback is triggered after timeout when user stops speaking.""" - callback_called = asyncio.Event() - - async def idle_callback(processor: UserIdleProcessor) -> None: - callback_called.set() - - # Create processor with a short timeout for testing - processor = UserIdleProcessor(callback=idle_callback, timeout=0.1) # 100ms timeout - - frames_to_send = [ - # Start conversation - UserStartedSpeakingFrame(), - UserStoppedSpeakingFrame(), - # Wait 200ms - double the idle timeout to ensure it triggers - SleepFrame(sleep=0.2), - ] - - expected_down_frames = [ - UserStartedSpeakingFrame, - UserStoppedSpeakingFrame, - ] - - await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - ) - - assert callback_called.is_set(), "Idle callback was not called" - - async def test_active_listening_resets_idle(self): - """Test that bot speaking frames reset the idle timer because user is actively listening.""" - callback_called = asyncio.Event() - - async def idle_callback(processor: UserIdleProcessor) -> None: - callback_called.set() - - processor = UserIdleProcessor(callback=idle_callback, timeout=0.2) - - frames_to_send = [ - # Start conversation - UserStartedSpeakingFrame(), - UserStoppedSpeakingFrame(), - # Wait almost long enough for idle timeout - SleepFrame(sleep=0.1), - # Bot speaking frame should reset idle timer - BotSpeakingFrame(), - # Wait almost long enough for idle timeout again - SleepFrame(sleep=0.1), - # Another bot speaking frame resets timer again - BotSpeakingFrame(), - # Give some time for the idle timeout task to start (Python 3.10 - # doesn't really like when you create a task and then cancel it - # right away). - SleepFrame(sleep=0.1), - ] - - expected_down_frames = [ - UserStartedSpeakingFrame, - UserStoppedSpeakingFrame, - BotSpeakingFrame, - BotSpeakingFrame, - ] - - await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - ) - - assert not callback_called.is_set(), ( - "Idle callback was called even though bot speaking frames reset the timer" - ) - - async def test_idle_retry_callback(self): - """Test that retry count increases until user activity resets it.""" - retry_counts = [] - - async def retry_callback(processor: UserIdleProcessor, retry_count: int) -> bool: - retry_counts.append(retry_count) - return True # Keep monitoring for idle events - - processor = UserIdleProcessor(callback=retry_callback, timeout=0.4) - - frames_to_send = [ - # Start conversation - UserStartedSpeakingFrame(), - UserStoppedSpeakingFrame(), - # Wait for first idle timeout (count=1) - SleepFrame(sleep=0.5), - # Wait for second idle timeout (count=2) - SleepFrame(sleep=0.5), - # User activity resets the count - UserStartedSpeakingFrame(), - UserStoppedSpeakingFrame(), - # Wait for new idle timeout (count should be 1 again) - SleepFrame(sleep=0.5), - ] - - expected_down_frames = [ - UserStartedSpeakingFrame, - UserStoppedSpeakingFrame, - UserStartedSpeakingFrame, - UserStoppedSpeakingFrame, - ] - - await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - ) - - assert retry_counts == [1, 2, 1], f"Expected retry counts [1, 2, 1], got {retry_counts}" - - async def test_idle_monitoring_stops_on_false_return(self): - """Test that idle monitoring stops when callback returns False.""" - retry_counts = [] - - async def retry_callback(processor: UserIdleProcessor, retry_count: int) -> bool: - retry_counts.append(retry_count) - return retry_count < 2 # Stop after second retry - - processor = UserIdleProcessor(callback=retry_callback, timeout=0.4) - - frames_to_send = [ - UserStartedSpeakingFrame(), - UserStoppedSpeakingFrame(), - SleepFrame(sleep=0.5), # First retry (count=1, returns True) - SleepFrame(sleep=0.5), # Second retry (count=2, returns False) - SleepFrame(sleep=0.5), # Should not trigger callback - ] - - expected_down_frames = [ - UserStartedSpeakingFrame, - UserStoppedSpeakingFrame, - ] - - await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - ) - - assert retry_counts == [1, 2], f"Expected retry counts [1, 2], got {retry_counts}" - - async def test_no_idle_before_conversation(self): - """Test that idle monitoring doesn't start before first conversation activity.""" - callback_called = asyncio.Event() - - async def idle_callback(processor: UserIdleProcessor) -> None: - callback_called.set() - - processor = UserIdleProcessor(callback=idle_callback, timeout=0.1) - - frames_to_send = [ - SleepFrame(sleep=0.2), # Should not trigger callback - # No conversation activity yet - ] - - expected_down_frames = [] - - await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - ) - - assert not callback_called.is_set(), "Idle callback was called before conversation started" - - async def test_idle_starts_with_bot_speech(self): - """Test that monitoring starts with bot speaking frames, not just user speech.""" - callback_called = asyncio.Event() - - async def idle_callback(processor: UserIdleProcessor) -> None: - callback_called.set() - - processor = UserIdleProcessor(callback=idle_callback, timeout=0.1) - - frames_to_send = [ - BotStartedSpeakingFrame(), - BotSpeakingFrame(), - BotStoppedSpeakingFrame(), - SleepFrame(sleep=0.2), - ] - - expected_down_frames = [ - BotStartedSpeakingFrame, - BotSpeakingFrame, - BotStoppedSpeakingFrame, - ] - - await run_test( - processor, - frames_to_send=frames_to_send, - expected_down_frames=expected_down_frames, - ) - - assert callback_called.is_set(), "Idle callback not called after bot speech" - - -if __name__ == "__main__": - unittest.main()