wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Remove deprecated UserBotLatencyLogObserver and UserIdleProcessor

Browse Source 
UserBotLatencyLogObserver (deprecated 0.0.102) is replaced by
UserBotLatencyObserver. UserIdleProcessor (deprecated 0.0.100) is
replaced by LLMUserAggregator with user_idle_timeout.
This commit is contained in:
Mark Backman
2026-04-02 21:54:36 -04:00
parent d8ea33e1a4
commit 793ed8f9e3
3 changed files with 0 additions and 542 deletions
Download Patch File Download Diff File Expand all files Collapse all files

109
src/pipecat/observers/loggers/user_bot_latency_log_observer.py
View File

@@ -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"
)

209
src/pipecat/processors/user_idle_processor.py
View File

@@ -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()