From ac0643f82f41b3bdc74035dd063a8340afe38a0d Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Wed, 18 Mar 2026 09:58:11 -0400 Subject: [PATCH] feat: add WakePhraseUserFrameFilter for aggregator-level wake phrase gating Add a new user frame filter concept that runs inside the aggregator, before VAD processing. This blocks transcriptions, VAD events, and interruptions until a wake phrase is detected, solving ordering conflicts that prevented wake phrase detection from working as a user turn start strategy. - Add BaseUserFrameFilter base class and WakePhraseUserFrameFilter with LISTENING/INACTIVE state machine, timeout, and single_activation - Add user_frame_filters param to LLMUserAggregatorParams with _maybe_filter_frame() running after mute check, before VAD - Deprecate WakeCheckFilter in favor of the new filter - Update 10-wake-phrase.py example --- examples/foundational/10-wake-phrase.py | 22 +- .../aggregators/llm_response_universal.py | 50 +++ .../processors/filters/wake_check_filter.py | 16 + src/pipecat/turns/user_filter/__init__.py | 13 + .../user_filter/base_user_frame_filter.py | 66 +++ .../wake_phrase_user_frame_filter.py | 250 +++++++++++ tests/test_wake_phrase_user_frame_filter.py | 394 ++++++++++++++++++ 7 files changed, 805 insertions(+), 6 deletions(-) create mode 100644 src/pipecat/turns/user_filter/__init__.py create mode 100644 src/pipecat/turns/user_filter/base_user_frame_filter.py create mode 100644 src/pipecat/turns/user_filter/wake_phrase_user_frame_filter.py create mode 100644 tests/test_wake_phrase_user_frame_filter.py diff --git a/examples/foundational/10-wake-phrase.py b/examples/foundational/10-wake-phrase.py index 0a4244c33..b6e7568a7 100644 --- a/examples/foundational/10-wake-phrase.py +++ b/examples/foundational/10-wake-phrase.py @@ -19,7 +19,6 @@ from pipecat.processors.aggregators.llm_response_universal import ( LLMContextAggregatorPair, LLMUserAggregatorParams, ) -from pipecat.processors.filters.wake_check_filter import WakeCheckFilter from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport from pipecat.services.cartesia.tts import CartesiaTTSService @@ -28,6 +27,7 @@ from pipecat.services.openai.llm import OpenAILLMService from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.daily.transport import DailyParams from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams +from pipecat.turns.user_filter import WakePhraseUserFrameFilter load_dotenv(override=True) @@ -52,7 +52,12 @@ transport_params = { async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): logger.info(f"Starting bot") - stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + stt = DeepgramSTTService( + api_key=os.getenv("DEEPGRAM_API_KEY"), + settings=DeepgramSTTService.Settings( + keyterm=["pipecat"], + ), + ) tts = CartesiaTTSService( api_key=os.getenv("CARTESIA_API_KEY"), @@ -68,19 +73,24 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): ), ) - hey_robot_filter = WakeCheckFilter(["hey robot", "hey, robot"]) + wake_filter = WakePhraseUserFrameFilter( + phrases=["pipecat"], + timeout=8.0, + ) context = LLMContext() user_aggregator, assistant_aggregator = LLMContextAggregatorPair( context, - user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()), + user_params=LLMUserAggregatorParams( + vad_analyzer=SileroVADAnalyzer(), + user_frame_filters=[wake_filter], + ), ) pipeline = Pipeline( [ transport.input(), # Transport user input stt, # STT - hey_robot_filter, # Filter out speech not directed at the robot user_aggregator, # User responses llm, # LLM tts, # TTS @@ -105,7 +115,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): context.add_message( { "role": "user", - "content": "Please introduce yourself. Tell the user they should say 'Hey Robot' before talking to you.", + "content": "Please introduce yourself. Tell the user they should say 'Pipecat' before talking to you.", } ) await task.queue_frames([LLMRunFrame()]) diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index b8a7061d8..3bf022ede 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/src/pipecat/processors/aggregators/llm_response_universal.py @@ -76,6 +76,7 @@ from pipecat.processors.aggregators.llm_context_summarizer import ( ) from pipecat.processors.frame_processor import FrameCallback, FrameDirection, FrameProcessor from pipecat.services.settings import LLMSettings +from pipecat.turns.user_filter import BaseUserFrameFilter from pipecat.turns.user_idle_controller import UserIdleController from pipecat.turns.user_mute import BaseUserMuteStrategy from pipecat.turns.user_start import BaseUserTurnStartStrategy, UserTurnStartedParams @@ -98,6 +99,10 @@ class LLMUserAggregatorParams: Parameters: user_turn_strategies: User turn start and stop strategies. user_mute_strategies: List of user mute strategies. + user_frame_filters: List of user frame filters. Filters run after mute + check and before VAD processing. Each filter's ``process_frame`` + returns True (pass) or False (block). If any filter blocks, the + frame is dropped. user_turn_stop_timeout: Time in seconds to wait before considering the user's turn finished. user_idle_timeout: Timeout in seconds for detecting user idle state. @@ -116,6 +121,7 @@ class LLMUserAggregatorParams: user_turn_strategies: Optional[UserTurnStrategies] = None user_mute_strategies: List[BaseUserMuteStrategy] = field(default_factory=list) + user_frame_filters: List[BaseUserFrameFilter] = field(default_factory=list) user_turn_stop_timeout: float = 5.0 user_idle_timeout: float = 0 vad_analyzer: Optional[VADAnalyzer] = None @@ -426,6 +432,8 @@ class LLMUserAggregator(LLMContextAggregator): self._register_event_handler("on_user_turn_idle") self._register_event_handler("on_user_mute_started") self._register_event_handler("on_user_mute_stopped") + self._register_event_handler("on_user_frame_filter_started") + self._register_event_handler("on_user_frame_filter_stopped") user_turn_strategies = self._params.user_turn_strategies or UserTurnStrategies() @@ -487,6 +495,9 @@ class LLMUserAggregator(LLMContextAggregator): if await self._maybe_mute_frame(frame): return + if await self._maybe_filter_frame(frame): + return + if self._vad_controller: await self._vad_controller.process_frame(frame) @@ -554,6 +565,9 @@ class LLMUserAggregator(LLMContextAggregator): for s in self._params.user_mute_strategies: await s.setup(self.task_manager) + for f in self._params.user_frame_filters: + await f.setup(self.task_manager) + # Enable incomplete turn filtering on the LLM if configured if self._params.filter_incomplete_user_turns: # Get config or use defaults @@ -584,6 +598,9 @@ class LLMUserAggregator(LLMContextAggregator): for s in self._params.user_mute_strategies: await s.cleanup() + for f in self._params.user_frame_filters: + await f.cleanup() + async def _maybe_mute_frame(self, frame: Frame): # Lifecycle frames should never be muted and should not trigger mute # state changes. Evaluating mute strategies on StartFrame would @@ -627,6 +644,35 @@ class LLMUserAggregator(LLMContextAggregator): return should_mute_frame + async def _maybe_filter_frame(self, frame: Frame) -> bool: + """Check if the frame should be filtered by any user frame filter. + + Lifecycle frames are never filtered. All filters see every frame so + they can maintain internal state, but if any filter returns False + (block) the frame is dropped. + + Args: + frame: The frame to check. + + Returns: + True if the frame should be filtered (dropped), False otherwise. + """ + if not self._params.user_frame_filters: + return False + + if isinstance(frame, (StartFrame, EndFrame, CancelFrame)): + return False + + should_filter = False + for f in self._params.user_frame_filters: + if not await f.process_frame(frame): + should_filter = True + + if should_filter: + logger.trace(f"{frame.name} suppressed - blocked by user frame filter") + + return should_filter + async def _handle_llm_run(self, frame: LLMRunFrame): await self.push_context_frame() @@ -748,6 +794,10 @@ class LLMUserAggregator(LLMContextAggregator): await self._maybe_emit_user_turn_stopped(strategy) + # Reset filters after turn stop (for single_activation mode). + for f in self._params.user_frame_filters: + await f.reset() + async def _on_user_turn_stop_timeout(self, controller): await self._call_event_handler("on_user_turn_stop_timeout") diff --git a/src/pipecat/processors/filters/wake_check_filter.py b/src/pipecat/processors/filters/wake_check_filter.py index ec8f31f53..82bde6822 100644 --- a/src/pipecat/processors/filters/wake_check_filter.py +++ b/src/pipecat/processors/filters/wake_check_filter.py @@ -6,6 +6,11 @@ """Wake phrase detection filter for Pipecat transcription processing. +.. deprecated:: + Use :class:`~pipecat.turns.user_filter.WakePhraseUserFrameFilter` instead. + ``WakePhraseUserFrameFilter`` runs inside the aggregator and blocks both + transcriptions and VAD events, providing more reliable wake phrase gating. + This module provides a frame processor that filters transcription frames, only allowing them through after wake phrases have been detected. Includes keepalive functionality to maintain conversation flow after wake detection. @@ -13,6 +18,7 @@ keepalive functionality to maintain conversation flow after wake detection. import re import time +import warnings from enum import Enum from typing import List @@ -25,6 +31,9 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor class WakeCheckFilter(FrameProcessor): """Frame processor that filters transcription frames based on wake phrase detection. + .. deprecated:: + Use :class:`~pipecat.turns.user_filter.WakePhraseUserFrameFilter` instead. + This filter monitors transcription frames for configured wake phrases and only passes frames through after a wake phrase has been detected. Maintains a keepalive timeout to allow continued conversation after wake detection. @@ -71,6 +80,13 @@ class WakeCheckFilter(FrameProcessor): wake detection. Defaults to 3 seconds. """ super().__init__() + warnings.warn( + "WakeCheckFilter is deprecated. Use WakePhraseUserFrameFilter from " + "pipecat.turns.user_filter instead, which runs inside the aggregator " + "and blocks both transcriptions and VAD events.", + DeprecationWarning, + stacklevel=2, + ) self._participant_states = {} self._keepalive_timeout = keepalive_timeout self._wake_patterns = [] diff --git a/src/pipecat/turns/user_filter/__init__.py b/src/pipecat/turns/user_filter/__init__.py new file mode 100644 index 000000000..a6b17d101 --- /dev/null +++ b/src/pipecat/turns/user_filter/__init__.py @@ -0,0 +1,13 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +from .base_user_frame_filter import BaseUserFrameFilter +from .wake_phrase_user_frame_filter import WakePhraseUserFrameFilter + +__all__ = [ + "BaseUserFrameFilter", + "WakePhraseUserFrameFilter", +] diff --git a/src/pipecat/turns/user_filter/base_user_frame_filter.py b/src/pipecat/turns/user_filter/base_user_frame_filter.py new file mode 100644 index 000000000..aee243c80 --- /dev/null +++ b/src/pipecat/turns/user_filter/base_user_frame_filter.py @@ -0,0 +1,66 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Base filter for deciding whether user frames should be filtered.""" + +from typing import Optional + +from pipecat.frames.frames import Frame +from pipecat.utils.asyncio.task_manager import BaseTaskManager +from pipecat.utils.base_object import BaseObject + + +class BaseUserFrameFilter(BaseObject): + """Base class for filters that decide whether user frames should be blocked. + + A user frame filter examines incoming frames and decides whether each frame + should pass through to the aggregator or be blocked. Unlike mute strategies + which suppress all user frames based on system state, filters can selectively + block frames based on content (e.g., waiting for a wake phrase). + + The filter is evaluated per frame and returns a boolean indicating whether + the frame should pass through (True) or be blocked (False). + + """ + + def __init__(self, **kwargs): + """Initialize the base user frame filter.""" + super().__init__(**kwargs) + self._task_manager: Optional[BaseTaskManager] = None + + @property + def task_manager(self) -> BaseTaskManager: + """Returns the configured task manager.""" + if not self._task_manager: + raise RuntimeError(f"{self} user frame filter was not properly setup") + return self._task_manager + + async def setup(self, task_manager: BaseTaskManager): + """Initialize the filter with the given task manager. + + Args: + task_manager: The task manager to be associated with this instance. + """ + self._task_manager = task_manager + + async def cleanup(self): + """Cleanup the filter.""" + pass + + async def reset(self): + """Reset the filter to its initial state.""" + pass + + async def process_frame(self, frame: Frame) -> bool: + """Process an incoming frame. + + Args: + frame: The frame to be processed. + + Returns: + True if the frame should pass through, False if it should be blocked. + """ + return True diff --git a/src/pipecat/turns/user_filter/wake_phrase_user_frame_filter.py b/src/pipecat/turns/user_filter/wake_phrase_user_frame_filter.py new file mode 100644 index 000000000..df35ffe36 --- /dev/null +++ b/src/pipecat/turns/user_filter/wake_phrase_user_frame_filter.py @@ -0,0 +1,250 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Wake phrase detection filter for gating user interaction. + +This module provides a user frame filter that blocks transcriptions, VAD events, +and interruptions until a configured wake phrase is spoken. After detection, frames +pass through freely until an inactivity timeout returns to the listening state. +""" + +import asyncio +import re +from enum import Enum +from typing import List, Optional + +from loguru import logger + +from pipecat.frames.frames import ( + BotSpeakingFrame, + Frame, + InterimTranscriptionFrame, + InterruptionFrame, + TranscriptionFrame, + UserSpeakingFrame, + VADUserStartedSpeakingFrame, + VADUserStoppedSpeakingFrame, +) +from pipecat.turns.user_filter.base_user_frame_filter import BaseUserFrameFilter +from pipecat.utils.asyncio.task_manager import BaseTaskManager + +# Regex to strip punctuation, keeping letters, digits, underscores, and whitespace. +_PUNCTUATION_RE = re.compile(r"[^\w\s]") + + +class WakePhraseUserFrameFilter(BaseUserFrameFilter): + """User frame filter that gates interaction behind a wake phrase. + + While in the LISTENING state, transcriptions, VAD events, and interruptions + are blocked. When a configured wake phrase is detected in a transcription, + the filter transitions to INACTIVE (pass-through) and starts a timeout. + If no activity occurs before the timeout expires, the filter returns to + LISTENING. + + Event handlers available: + + - on_wake_phrase_detected: Called when a wake phrase is matched. + - on_wake_phrase_timeout: Called when the inactivity timeout expires. + + Example:: + + filter = WakePhraseUserFrameFilter(phrases=["hey robot"]) + + @filter.event_handler("on_wake_phrase_detected") + async def on_wake_phrase_detected(filter, phrase): + ... + + @filter.event_handler("on_wake_phrase_timeout") + async def on_wake_phrase_timeout(filter): + ... + + """ + + class State(Enum): + """Filter states. + + Parameters: + LISTENING: Blocking user frames, waiting for wake phrase. + INACTIVE: Passing all frames through after wake phrase detected. + """ + + LISTENING = "listening" + INACTIVE = "inactive" + + def __init__( + self, + *, + phrases: List[str], + timeout: float = 10.0, + single_activation: bool = False, + **kwargs, + ): + """Initialize the wake phrase filter. + + Args: + phrases: List of wake phrases to detect. + timeout: Inactivity timeout in seconds before returning to LISTENING. + single_activation: If True, require wake phrase every turn. If False, + timeout self-manages via activity detection. + **kwargs: Additional arguments passed to parent. + """ + super().__init__(**kwargs) + + self._timeout = timeout + self._single_activation = single_activation + + self._state = self.State.LISTENING + + self._timeout_event = asyncio.Event() + self._timeout_task: Optional[asyncio.Task] = None + + # Build regex patterns from phrases. + self._patterns = [] + for phrase in phrases: + pattern = re.compile( + r"\b" + r"\s*".join(re.escape(word) for word in phrase.split()) + r"\b", + re.IGNORECASE, + ) + self._patterns.append(pattern) + + self._register_event_handler("on_wake_phrase_detected") + self._register_event_handler("on_wake_phrase_timeout") + + @property + def state(self) -> "WakePhraseUserFrameFilter.State": + """Returns the current filter state.""" + return self._state + + async def setup(self, task_manager: BaseTaskManager): + """Initialize the filter and start the timeout task. + + Args: + task_manager: The task manager to be associated with this instance. + """ + await super().setup(task_manager) + + if not self._timeout_task: + self._timeout_task = self.task_manager.create_task( + self._timeout_task_handler(), + f"{self}::_timeout_task_handler", + ) + + async def cleanup(self): + """Cleanup the filter and cancel the timeout task.""" + if self._timeout_task: + await self.task_manager.cancel_task(self._timeout_task) + self._timeout_task = None + + async def reset(self): + """Reset the filter state after a turn stop. + + With single_activation=True, transitions back to LISTENING. Otherwise + the timeout self-manages via activity detection. + """ + if self._single_activation: + self._transition_to_listening() + + async def process_frame(self, frame: Frame) -> bool: + """Process a frame and decide whether it should pass through. + + Args: + frame: The frame to process. + + Returns: + True if the frame should pass through, False if blocked. + """ + if self._state == self.State.LISTENING: + return await self._process_listening(frame) + else: + return await self._process_inactive(frame) + + async def _process_listening(self, frame: Frame) -> bool: + """Process a frame while in LISTENING state. + + Blocks user interaction frames. Checks transcriptions for wake phrase. + """ + if isinstance(frame, TranscriptionFrame): + return await self._check_wake_phrase(frame) + + if isinstance( + frame, + ( + InterimTranscriptionFrame, + VADUserStartedSpeakingFrame, + VADUserStoppedSpeakingFrame, + InterruptionFrame, + ), + ): + return False + + # Everything else passes through. + return True + + async def _process_inactive(self, frame: Frame) -> bool: + """Process a frame while in INACTIVE state. + + Refreshes timeout on activity frames. All frames pass through. + """ + if isinstance( + frame, + ( + UserSpeakingFrame, + BotSpeakingFrame, + ), + ): + self._refresh_timeout() + + return True + + async def _check_wake_phrase(self, frame: TranscriptionFrame) -> bool: + """Check a transcription frame for wake phrase matches. + + Args: + frame: The transcription frame to check. + + Returns: + True if wake phrase matched (pass frame), False otherwise (block). + """ + text = _PUNCTUATION_RE.sub("", frame.text) + + for pattern in self._patterns: + match = pattern.search(text) + if match: + matched_text = match.group() + logger.debug(f"{self}: Wake phrase detected: '{matched_text}'") + self._state = self.State.INACTIVE + self._refresh_timeout() + await self._call_event_handler("on_wake_phrase_detected", matched_text) + return True + + return False + + def _refresh_timeout(self): + """Signal the timeout task to restart its countdown.""" + self._timeout_event.set() + + def _transition_to_listening(self): + """Transition to LISTENING state.""" + self._state = self.State.LISTENING + + async def _timeout_task_handler(self): + """Background task that manages the inactivity timeout. + + Waits for activity signals and transitions to LISTENING when the + timeout expires without activity. + """ + while True: + try: + await asyncio.wait_for( + self._timeout_event.wait(), + timeout=self._timeout, + ) + self._timeout_event.clear() + except asyncio.TimeoutError: + if self._state == self.State.INACTIVE: + logger.debug(f"{self}: Wake phrase timeout, returning to LISTENING") + self._transition_to_listening() + await self._call_event_handler("on_wake_phrase_timeout") diff --git a/tests/test_wake_phrase_user_frame_filter.py b/tests/test_wake_phrase_user_frame_filter.py new file mode 100644 index 000000000..88242b0fc --- /dev/null +++ b/tests/test_wake_phrase_user_frame_filter.py @@ -0,0 +1,394 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import unittest + +from pipecat.frames.frames import ( + InterimTranscriptionFrame, + InterruptionFrame, + LLMContextFrame, + TextFrame, + TranscriptionFrame, + UserSpeakingFrame, + UserStartedSpeakingFrame, + UserStoppedSpeakingFrame, + VADUserStartedSpeakingFrame, + VADUserStoppedSpeakingFrame, +) +from pipecat.pipeline.pipeline import Pipeline +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.aggregators.llm_response_universal import ( + LLMUserAggregator, + LLMUserAggregatorParams, +) +from pipecat.tests.utils import SleepFrame, run_test +from pipecat.turns.user_filter import WakePhraseUserFrameFilter +from pipecat.turns.user_stop import SpeechTimeoutUserTurnStopStrategy +from pipecat.turns.user_turn_strategies import UserTurnStrategies + +TRANSCRIPTION_TIMEOUT = 0.1 +WAKE_TIMEOUT = 0.3 + + +def _make_transcription(text: str) -> TranscriptionFrame: + return TranscriptionFrame(text=text, user_id="user1", timestamp="now") + + +class TestWakePhraseMatching(unittest.IsolatedAsyncioTestCase): + """Test wake phrase detection logic in isolation via the aggregator.""" + + async def test_basic_match(self): + """Wake phrase in a single transcription frame passes through.""" + wake_filter = WakePhraseUserFrameFilter(phrases=["hey robot"], timeout=WAKE_TIMEOUT) + self.assertEqual(wake_filter.state, WakePhraseUserFrameFilter.State.LISTENING) + + result = await wake_filter.process_frame(_make_transcription("hey robot")) + self.assertTrue(result) + self.assertEqual(wake_filter.state, WakePhraseUserFrameFilter.State.INACTIVE) + + async def test_case_insensitive(self): + """Wake phrase matching is case-insensitive.""" + wake_filter = WakePhraseUserFrameFilter(phrases=["hey robot"], timeout=WAKE_TIMEOUT) + + result = await wake_filter.process_frame(_make_transcription("HEY ROBOT")) + self.assertTrue(result) + self.assertEqual(wake_filter.state, WakePhraseUserFrameFilter.State.INACTIVE) + + async def test_wake_phrase_within_sentence(self): + """Wake phrase embedded in a longer transcription is detected.""" + wake_filter = WakePhraseUserFrameFilter(phrases=["hey robot"], timeout=WAKE_TIMEOUT) + + result = await wake_filter.process_frame( + _make_transcription("so I said hey robot what time is it") + ) + self.assertTrue(result) + self.assertEqual(wake_filter.state, WakePhraseUserFrameFilter.State.INACTIVE) + + async def test_multiple_phrases(self): + """Any of multiple configured phrases can trigger activation.""" + wake_filter = WakePhraseUserFrameFilter( + phrases=["hey robot", "ok computer"], timeout=WAKE_TIMEOUT + ) + + result = await wake_filter.process_frame(_make_transcription("ok computer")) + self.assertTrue(result) + self.assertEqual(wake_filter.state, WakePhraseUserFrameFilter.State.INACTIVE) + + async def test_punctuation_stripped(self): + """Common punctuation is stripped before matching.""" + wake_filter = WakePhraseUserFrameFilter(phrases=["hey robot"], timeout=WAKE_TIMEOUT) + + result = await wake_filter.process_frame(_make_transcription("hey, robot!")) + self.assertTrue(result) + self.assertEqual(wake_filter.state, WakePhraseUserFrameFilter.State.INACTIVE) + + async def test_no_match_blocks(self): + """Transcription without wake phrase is blocked.""" + wake_filter = WakePhraseUserFrameFilter(phrases=["hey robot"], timeout=WAKE_TIMEOUT) + + result = await wake_filter.process_frame(_make_transcription("hello world")) + self.assertFalse(result) + self.assertEqual(wake_filter.state, WakePhraseUserFrameFilter.State.LISTENING) + + +class TestWakePhraseBlocking(unittest.IsolatedAsyncioTestCase): + """Test that frames are correctly blocked/passed in each state.""" + + async def test_listening_blocks_vad_frames(self): + """VAD frames are blocked while LISTENING.""" + wake_filter = WakePhraseUserFrameFilter(phrases=["hey robot"], timeout=WAKE_TIMEOUT) + + self.assertFalse(await wake_filter.process_frame(VADUserStartedSpeakingFrame())) + self.assertFalse(await wake_filter.process_frame(VADUserStoppedSpeakingFrame())) + + async def test_listening_blocks_interim_transcription(self): + """Interim transcriptions are blocked while LISTENING.""" + wake_filter = WakePhraseUserFrameFilter(phrases=["hey robot"], timeout=WAKE_TIMEOUT) + + frame = InterimTranscriptionFrame(text="hey", user_id="user1", timestamp="now") + self.assertFalse(await wake_filter.process_frame(frame)) + + async def test_listening_blocks_interruption(self): + """InterruptionFrame is blocked while LISTENING.""" + wake_filter = WakePhraseUserFrameFilter(phrases=["hey robot"], timeout=WAKE_TIMEOUT) + + self.assertFalse(await wake_filter.process_frame(InterruptionFrame())) + + async def test_listening_passes_user_speaking(self): + """UserSpeakingFrame passes through while LISTENING (activity frame, not gated).""" + wake_filter = WakePhraseUserFrameFilter(phrases=["hey robot"], timeout=WAKE_TIMEOUT) + + self.assertTrue(await wake_filter.process_frame(UserSpeakingFrame())) + + async def test_listening_passes_other_frames(self): + """Non-user-interaction frames pass through while LISTENING.""" + wake_filter = WakePhraseUserFrameFilter(phrases=["hey robot"], timeout=WAKE_TIMEOUT) + + self.assertTrue(await wake_filter.process_frame(TextFrame(text="hello"))) + + async def test_inactive_passes_all_frames(self): + """All frames pass through while INACTIVE.""" + wake_filter = WakePhraseUserFrameFilter(phrases=["hey robot"], timeout=WAKE_TIMEOUT) + + # Activate + await wake_filter.process_frame(_make_transcription("hey robot")) + self.assertEqual(wake_filter.state, WakePhraseUserFrameFilter.State.INACTIVE) + + # All frame types should pass + self.assertTrue(await wake_filter.process_frame(_make_transcription("hello"))) + self.assertTrue(await wake_filter.process_frame(VADUserStartedSpeakingFrame())) + self.assertTrue(await wake_filter.process_frame(VADUserStoppedSpeakingFrame())) + self.assertTrue( + await wake_filter.process_frame( + InterimTranscriptionFrame(text="hi", user_id="user1", timestamp="now") + ) + ) + self.assertTrue(await wake_filter.process_frame(InterruptionFrame())) + self.assertTrue(await wake_filter.process_frame(UserSpeakingFrame())) + + +class TestWakePhraseTimeout(unittest.IsolatedAsyncioTestCase): + """Test timeout behavior.""" + + async def test_timeout_returns_to_listening(self): + """Filter returns to LISTENING after inactivity timeout.""" + wake_filter = WakePhraseUserFrameFilter(phrases=["hey robot"], timeout=WAKE_TIMEOUT) + context = LLMContext() + user_aggregator = LLMUserAggregator( + context, + params=LLMUserAggregatorParams(user_frame_filters=[wake_filter]), + ) + + pipeline = Pipeline([user_aggregator]) + + frames_to_send = [ + _make_transcription("hey robot"), + # Wait for timeout to expire + SleepFrame(sleep=WAKE_TIMEOUT + 0.2), + ] + await run_test(pipeline, frames_to_send=frames_to_send) + + self.assertEqual(wake_filter.state, WakePhraseUserFrameFilter.State.LISTENING) + + async def test_activity_refreshes_timeout(self): + """Speaking activity while INACTIVE refreshes the timeout.""" + wake_filter = WakePhraseUserFrameFilter(phrases=["hey robot"], timeout=WAKE_TIMEOUT) + context = LLMContext() + user_aggregator = LLMUserAggregator( + context, + params=LLMUserAggregatorParams(user_frame_filters=[wake_filter]), + ) + + pipeline = Pipeline([user_aggregator]) + + frames_to_send = [ + _make_transcription("hey robot"), + # Wait just under timeout + SleepFrame(sleep=WAKE_TIMEOUT * 0.7), + # UserSpeakingFrame activity should refresh the timeout + UserSpeakingFrame(), + # Wait just under timeout again + SleepFrame(sleep=WAKE_TIMEOUT * 0.7), + ] + await run_test(pipeline, frames_to_send=frames_to_send) + + # Should still be INACTIVE because activity refreshed timeout + self.assertEqual(wake_filter.state, WakePhraseUserFrameFilter.State.INACTIVE) + + async def test_timeout_event_fires(self): + """on_wake_phrase_timeout event fires when timeout expires.""" + wake_filter = WakePhraseUserFrameFilter(phrases=["hey robot"], timeout=WAKE_TIMEOUT) + timeout_fired = False + + @wake_filter.event_handler("on_wake_phrase_timeout") + async def on_timeout(filter): + nonlocal timeout_fired + timeout_fired = True + + context = LLMContext() + user_aggregator = LLMUserAggregator( + context, + params=LLMUserAggregatorParams(user_frame_filters=[wake_filter]), + ) + pipeline = Pipeline([user_aggregator]) + + frames_to_send = [ + _make_transcription("hey robot"), + SleepFrame(sleep=WAKE_TIMEOUT + 0.2), + ] + await run_test(pipeline, frames_to_send=frames_to_send) + + self.assertTrue(timeout_fired) + + +class TestWakePhraseSingleActivation(unittest.IsolatedAsyncioTestCase): + """Test single_activation mode.""" + + async def test_single_activation_resets_on_turn_stop(self): + """With single_activation, reset() transitions back to LISTENING.""" + wake_filter = WakePhraseUserFrameFilter( + phrases=["hey robot"], timeout=WAKE_TIMEOUT, single_activation=True + ) + + # Activate + await wake_filter.process_frame(_make_transcription("hey robot")) + self.assertEqual(wake_filter.state, WakePhraseUserFrameFilter.State.INACTIVE) + + # Simulate turn stop reset + await wake_filter.reset() + self.assertEqual(wake_filter.state, WakePhraseUserFrameFilter.State.LISTENING) + + async def test_default_mode_does_not_reset(self): + """Without single_activation, reset() does not change state.""" + wake_filter = WakePhraseUserFrameFilter(phrases=["hey robot"], timeout=WAKE_TIMEOUT) + + # Activate + await wake_filter.process_frame(_make_transcription("hey robot")) + self.assertEqual(wake_filter.state, WakePhraseUserFrameFilter.State.INACTIVE) + + # Simulate turn stop reset + await wake_filter.reset() + self.assertEqual(wake_filter.state, WakePhraseUserFrameFilter.State.INACTIVE) + + +class TestWakePhraseEvents(unittest.IsolatedAsyncioTestCase): + """Test event handlers.""" + + async def test_detected_event_fires(self): + """on_wake_phrase_detected fires with matched phrase text.""" + wake_filter = WakePhraseUserFrameFilter(phrases=["hey robot"], timeout=WAKE_TIMEOUT) + detected_phrase = None + + @wake_filter.event_handler("on_wake_phrase_detected") + async def on_detected(filter, phrase): + nonlocal detected_phrase + detected_phrase = phrase + + context = LLMContext() + user_aggregator = LLMUserAggregator( + context, + params=LLMUserAggregatorParams(user_frame_filters=[wake_filter]), + ) + pipeline = Pipeline([user_aggregator]) + + frames_to_send = [_make_transcription("hey robot")] + await run_test(pipeline, frames_to_send=frames_to_send) + + self.assertIsNotNone(detected_phrase) + self.assertIn("hey", detected_phrase.lower()) + self.assertIn("robot", detected_phrase.lower()) + + +class TestWakePhrasePipelineIntegration(unittest.IsolatedAsyncioTestCase): + """Test WakePhraseUserFrameFilter in a full pipeline with aggregator.""" + + async def test_transcriptions_blocked_while_listening(self): + """Transcriptions are consumed (not pushed downstream) while LISTENING.""" + wake_filter = WakePhraseUserFrameFilter(phrases=["hey robot"], timeout=WAKE_TIMEOUT) + context = LLMContext() + user_aggregator = LLMUserAggregator( + context, + params=LLMUserAggregatorParams(user_frame_filters=[wake_filter]), + ) + pipeline = Pipeline([user_aggregator]) + + frames_to_send = [ + # These should all be blocked (no wake phrase) + VADUserStartedSpeakingFrame(), + _make_transcription("hello world"), + VADUserStoppedSpeakingFrame(), + ] + # Nothing should come through downstream (transcriptions are consumed, + # VAD frames are blocked by the filter) + expected_down_frames = [] + await run_test( + pipeline, + frames_to_send=frames_to_send, + expected_down_frames=expected_down_frames, + ) + + async def test_full_turn_after_wake_phrase(self): + """After wake phrase, a full user turn completes with context update. + + The VADUserStartedSpeakingFrame sent before the wake phrase is blocked + by the filter (LISTENING state). The transcription containing the wake + phrase passes through and triggers the turn via the default + TranscriptionUserTurnStartStrategy. + """ + wake_filter = WakePhraseUserFrameFilter(phrases=["hey robot"], timeout=1.0) + context = LLMContext() + user_aggregator = LLMUserAggregator( + context, + params=LLMUserAggregatorParams( + user_frame_filters=[wake_filter], + user_turn_strategies=UserTurnStrategies( + stop=[ + SpeechTimeoutUserTurnStopStrategy(user_speech_timeout=TRANSCRIPTION_TIMEOUT) + ], + ), + ), + ) + pipeline = Pipeline([user_aggregator]) + + frames_to_send = [ + # Wake phrase transcription - activates filter AND starts the turn + _make_transcription("hey robot what is the weather"), + # Wait for speech timeout to trigger turn stop + SleepFrame(sleep=TRANSCRIPTION_TIMEOUT + 0.1), + ] + expected_down_frames = [ + # TranscriptionUserTurnStartStrategy triggers turn start + UserStartedSpeakingFrame, + InterruptionFrame, + # SpeechTimeoutUserTurnStopStrategy triggers turn stop + UserStoppedSpeakingFrame, + LLMContextFrame, + ] + await run_test( + pipeline, + frames_to_send=frames_to_send, + expected_down_frames=expected_down_frames, + ) + + # The transcription should be in context + self.assertEqual(len(context.messages), 1) + self.assertEqual(context.messages[0]["role"], "user") + self.assertIn("hey robot", context.messages[0]["content"]) + + async def test_vad_frames_blocked_before_wake(self): + """VAD frames before wake phrase don't trigger turn start.""" + wake_filter = WakePhraseUserFrameFilter(phrases=["hey robot"], timeout=WAKE_TIMEOUT) + context = LLMContext() + + turn_started = False + + user_aggregator = LLMUserAggregator( + context, + params=LLMUserAggregatorParams(user_frame_filters=[wake_filter]), + ) + + @user_aggregator.event_handler("on_user_turn_started") + async def on_user_turn_started(aggregator, strategy): + nonlocal turn_started + turn_started = True + + pipeline = Pipeline([user_aggregator]) + + frames_to_send = [ + # VAD events without wake phrase - should be blocked + VADUserStartedSpeakingFrame(), + _make_transcription("hello world"), + VADUserStoppedSpeakingFrame(), + SleepFrame(sleep=0.3), + ] + await run_test(pipeline, frames_to_send=frames_to_send) + + # Turn should NOT have started because VAD frames were blocked + self.assertFalse(turn_started) + + +if __name__ == "__main__": + unittest.main()