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

Merge pull request #4232 from pipecat-ai/mb/more-deprecation-removals

Browse Source
This commit is contained in:
Mark Backman
2026-04-03 06:52:56 -04:00
committed by GitHub
parent d8ea33e1a4 33f145d70a
commit b94071d37f
15 changed files with 10 additions and 688 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
changelog/4232.fixed.md Normal file
View File

@@ -0,0 +1 @@
- Fixed undefined `_warn_deprecated_param` calls in `OpenAIRealtimeLLMService` and `GrokRealtimeLLMService` for the deprecated `session_properties` init parameter.

1
changelog/4232.removed.2.md Normal file
View File

@@ -0,0 +1 @@
- ⚠️ Removed deprecated `UserIdleProcessor` (deprecated in 0.0.100). Use `LLMUserAggregator` with the `user_idle_timeout` parameter instead.

1
changelog/4232.removed.3.md Normal file
View File

@@ -0,0 +1 @@
- ⚠️ Removed deprecated `TranscriptionUserTurnStopStrategy` alias (deprecated in 0.0.102). Use `SpeechTimeoutUserTurnStopStrategy` instead.

1
changelog/4232.removed.4.md Normal file
View File

@@ -0,0 +1 @@
- ⚠️ Removed deprecated `send_transcription_frames` parameter from `OpenAIRealtimeLLMService` (deprecated in 0.0.92). Transcription frames are always sent.

1
changelog/4232.removed.5.md Normal file
View File

@@ -0,0 +1 @@
- ⚠️ Removed deprecated `vad_events` setting and `should_interrupt` parameter from `DeepgramSTTService` (deprecated in 0.0.99). Use Silero VAD for voice activity detection instead.

1
changelog/4232.removed.md Normal file
View File

@@ -0,0 +1 @@
- ⚠️ Removed deprecated `UserBotLatencyLogObserver` (deprecated in 0.0.102). Use `UserBotLatencyObserver` with its `on_latency_measured` event handler instead.

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

79
src/pipecat/services/deepgram/stt.py
View File

@@ -19,8 +19,6 @@ from pipecat.frames.frames import (
InterimTranscriptionFrame,
StartFrame,
TranscriptionFrame,
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
VADUserStartedSpeakingFrame,
VADUserStoppedSpeakingFrame,
)
@@ -45,8 +43,6 @@ try:
ListenV1Finalize,
ListenV1KeepAlive,
ListenV1Results,
ListenV1SpeechStarted,
ListenV1UtteranceEnd,
)
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
@@ -94,7 +90,6 @@ class LiveOptions:
smart_format: Optional[bool] = None,
tag: Optional[Any] = None,
utterance_end_ms: Optional[int] = None,
vad_events: Optional[bool] = None,
version: Optional[str] = None,
**kwargs,
):
@@ -127,7 +122,6 @@ class LiveOptions:
smart_format: Apply smart formatting to transcripts.
tag: Custom billing tag (str or list of str).
utterance_end_ms: Silence duration in ms before an utterance-end event.
vad_events: Enable Deepgram VAD speech-started / utterance-end events.
version: Model version (e.g. ``"latest"``).
**kwargs: Any additional Deepgram query parameters.
"""
@@ -157,7 +151,6 @@ class LiveOptions:
self.smart_format = smart_format
self.tag = tag
self.utterance_end_ms = utterance_end_ms
self.vad_events = vad_events
self.version = version
self._extra = kwargs
@@ -201,7 +194,6 @@ class DeepgramSTTSettings(STTSettings):
search: Search terms to highlight (str or list of str).
smart_format: Apply smart formatting to transcripts.
utterance_end_ms: Silence duration in ms before an utterance-end event.
vad_events: Enable Deepgram VAD speech-started / utterance-end events.
"""
detect_entities: bool | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
@@ -219,7 +211,6 @@ class DeepgramSTTSettings(STTSettings):
search: Any | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
smart_format: bool | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
utterance_end_ms: int | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
vad_events: bool | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
def _sync_extra_to_fields(self) -> None:
"""Sync values from extra dict to declared fields.
@@ -294,17 +285,6 @@ class DeepgramSTTService(STTService):
Provides real-time speech recognition using Deepgram's WebSocket API.
Supports configurable models, languages, and various audio processing options.
Event handlers available (in addition to STTService events):
- on_speech_started(service): Deepgram detected start of speech
- on_utterance_end(service): Deepgram detected end of utterance
Example::
@stt.event_handler("on_speech_started")
async def on_speech_started(service):
...
"""
Settings = DeepgramSTTSettings
@@ -325,7 +305,6 @@ class DeepgramSTTService(STTService):
mip_opt_out: Optional[bool] = None,
live_options: Optional[LiveOptions] = None,
addons: Optional[dict] = None,
should_interrupt: bool = True,
settings: Optional[Settings] = None,
ttfs_p99_latency: Optional[float] = DEEPGRAM_TTFS_P99,
**kwargs,
@@ -352,21 +331,12 @@ class DeepgramSTTService(STTService):
fields and direct init parameters for connection-level config.
addons: Additional Deepgram features to enable.
should_interrupt: Whether to interrupt the bot when Deepgram VAD
detects the user is speaking.
.. deprecated:: 0.0.99
This parameter will be removed along with `vad_events` support.
settings: Runtime-updatable settings. When provided alongside
``live_options``, ``settings`` values take precedence (applied
after the ``live_options`` merge).
ttfs_p99_latency: P99 latency from speech end to final transcript in seconds.
Override for your deployment. See https://github.com/pipecat-ai/stt-benchmark
**kwargs: Additional arguments passed to the parent STTService.
Note:
The `vad_events` option in LiveOptions is deprecated as of version 0.0.99 and will be removed in a future version. Please use the Silero VAD instead.
"""
# 1. Initialize default_settings with hardcoded defaults
default_settings = self.Settings(
@@ -387,7 +357,6 @@ class DeepgramSTTService(STTService):
search=None,
smart_format=False,
utterance_end_ms=None,
vad_events=False,
)
# 2. (No step 2, as there are no deprecated direct args)
@@ -444,7 +413,6 @@ class DeepgramSTTService(STTService):
)
self._addons = addons
self._should_interrupt = should_interrupt
self._encoding = encoding
self._channels = channels
self._multichannel = multichannel
@@ -453,18 +421,6 @@ class DeepgramSTTService(STTService):
self._tag = tag
self._mip_opt_out = mip_opt_out
if self._settings.vad_events:
import warnings
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"The 'vad_events' parameter is deprecated and will be removed in a future version. "
"Please use the Silero VAD instead.",
DeprecationWarning,
stacklevel=2,
)
# Build client - support optional custom base URL via DeepgramClientEnvironment
if base_url:
try:
@@ -488,19 +444,6 @@ class DeepgramSTTService(STTService):
self._connection = None
self._connection_task = None
if self.vad_enabled:
self._register_event_handler("on_speech_started")
self._register_event_handler("on_utterance_end")
@property
def vad_enabled(self):
"""Check if Deepgram VAD events are enabled.
Returns:
True if VAD events are enabled in the current settings.
"""
return self._settings.vad_events
def can_generate_metrics(self) -> bool:
"""Check if this service can generate processing metrics.
@@ -705,17 +648,6 @@ class DeepgramSTTService(STTService):
# Reconnection is handled automatically by the retry loop in
# _connection_handler once start_listening() exits after the error.
async def _on_speech_started(self, message):
await self._start_metrics()
await self._call_event_handler("on_speech_started", message)
await self.broadcast_frame(UserStartedSpeakingFrame)
if self._should_interrupt:
await self.broadcast_interruption()
async def _on_utterance_end(self, message):
await self._call_event_handler("on_utterance_end", message)
await self.broadcast_frame(UserStoppedSpeakingFrame)
@traced_stt
async def _handle_transcription(
self, transcript: str, is_final: bool, language: Optional[Language] = None
@@ -724,13 +656,7 @@ class DeepgramSTTService(STTService):
pass
async def _on_message(self, message):
if isinstance(message, ListenV1SpeechStarted):
if self.vad_enabled:
await self._on_speech_started(message)
elif isinstance(message, ListenV1UtteranceEnd):
if self.vad_enabled:
await self._on_utterance_end(message)
elif isinstance(message, ListenV1Results):
if isinstance(message, ListenV1Results):
if not message.channel or len(message.channel.alternatives) == 0:
return
is_final = message.is_final
@@ -778,8 +704,7 @@ class DeepgramSTTService(STTService):
"""
await super().process_frame(frame, direction)
if isinstance(frame, VADUserStartedSpeakingFrame) and not self.vad_enabled:
# Start metrics if Deepgram VAD is disabled & pipeline VAD has detected speech
if isinstance(frame, VADUserStartedSpeakingFrame):
await self._start_metrics()
elif isinstance(frame, VADUserStoppedSpeakingFrame):
# https://developers.deepgram.com/docs/finalize

25
src/pipecat/services/openai/realtime/llm.py
View File

@@ -218,7 +218,6 @@ class OpenAIRealtimeLLMService(LLMService):
start_audio_paused: bool = False,
start_video_paused: bool = False,
video_frame_detail: str = "auto",
send_transcription_frames: Optional[bool] = None,
**kwargs,
):
"""Initialize the OpenAI Realtime LLM service.
@@ -247,26 +246,8 @@ class OpenAIRealtimeLLMService(LLMService):
This sets the image_detail parameter in the OpenAI Realtime API.
"auto" lets the model decide, "low" is faster and uses fewer tokens,
"high" provides more detail. Defaults to "auto".
send_transcription_frames: Whether to emit transcription frames.
.. deprecated:: 0.0.92
This parameter is deprecated and will be removed in a future version.
Transcription frames are always sent.
**kwargs: Additional arguments passed to parent LLMService.
"""
if send_transcription_frames is not None:
import warnings
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"`send_transcription_frames` is deprecated and will be removed in a future version. "
"Transcription frames are always sent.",
DeprecationWarning,
stacklevel=2,
)
# 1. Initialize default_settings with hardcoded defaults
default_settings = self.Settings(
model="gpt-realtime-1.5",
@@ -289,11 +270,7 @@ class OpenAIRealtimeLLMService(LLMService):
default_settings.model = model
if session_properties is not None:
_warn_deprecated_param(
"session_properties",
self.Settings,
"session_properties",
)
self._warn_init_param_moved_to_settings("session_properties", "session_properties")
default_settings.session_properties = session_properties
# Sync model/instructions from the deprecated SP arg to top-level,
# but only if the deprecated `model` arg didn't already set it.

6
src/pipecat/services/xai/realtime/llm.py
View File

@@ -252,11 +252,7 @@ class GrokRealtimeLLMService(LLMService):
# 2. Apply direct init arg overrides (deprecated)
if session_properties is not None:
_warn_deprecated_param(
"session_properties",
self.Settings,
"session_properties",
)
self._warn_init_param_moved_to_settings("session_properties", "session_properties")
default_settings.session_properties = session_properties
# Sync instructions from the deprecated SP arg to top-level
if session_properties.instructions is not None:

7
src/pipecat/transports/base_transport.py
View File

@@ -18,8 +18,6 @@ from pydantic import BaseModel, ConfigDict, Field
from pipecat.audio.filters.base_audio_filter import BaseAudioFilter
from pipecat.audio.mixers.base_audio_mixer import BaseAudioMixer
from pipecat.audio.turn.base_turn_analyzer import BaseTurnAnalyzer
from pipecat.audio.vad.vad_analyzer import VADAnalyzer
from pipecat.processors.frame_processor import FrameProcessor
from pipecat.utils.base_object import BaseObject
@@ -54,11 +52,6 @@ class TransportParams(BaseModel):
video_out_color_format: Video output color format string.
video_out_codec: Preferred video codec for output (e.g., 'VP8', 'H264', 'H265').
video_out_destinations: List of video output destination identifiers.
turn_analyzer: Turn-taking analyzer instance for conversation management.
.. deprecated:: 0.0.99
The `turn_analyzer` parameter is deprecated, use `LLMUSerAggregator`'s
new `user_turn_strategies` parameter instead.
"""
model_config = ConfigDict(arbitrary_types_allowed=True)

31
src/pipecat/turns/user_stop/transcription_user_turn_stop_strategy.py
View File

@@ -1,31 +0,0 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Transcription-based user turn stop strategy (deprecated).
.. deprecated:: 0.0.102
This module is deprecated. Please use
``pipecat.turns.user_stop.speech_timeout_user_turn_stop_strategy.SpeechTimeoutUserTurnStopStrategy``
instead.
"""
import warnings
from pipecat.turns.user_stop.speech_timeout_user_turn_stop_strategy import (
SpeechTimeoutUserTurnStopStrategy,
)
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"TranscriptionUserTurnStopStrategy is deprecated. "
"Please use SpeechTimeoutUserTurnStopStrategy from "
"pipecat.turns.user_stop.speech_timeout_user_turn_stop_strategy instead.",
DeprecationWarning,
stacklevel=2,
)
TranscriptionUserTurnStopStrategy = SpeechTimeoutUserTurnStopStrategy

2
tests/test_settings.py
View File

@@ -334,7 +334,6 @@ class TestDeepgramSTTSettingsApplyUpdate:
smart_format=False,
punctuate=True,
profanity_filter=True,
vad_events=False,
)
defaults.update(kwargs)
return DeepgramSTTSettings(**defaults)
@@ -430,7 +429,6 @@ class TestDeepgramSTTSettingsFromMapping:
interim_results=True,
punctuate=True,
profanity_filter=True,
vad_events=False,
)
raw = {"punctuate": False, "diarize": True}

224
tests/test_user_idle_processor.py
View File

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