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

feat: add WakePhraseUserFrameFilter for aggregator-level wake phrase gating

Browse Source 
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
This commit is contained in:
Mark Backman
2026-03-18 09:58:11 -04:00
parent 53388e0426
commit ac0643f82f
7 changed files with 805 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
examples/foundational/10-wake-phrase.py
View File

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

50
src/pipecat/processors/aggregators/llm_response_universal.py
View File

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

16
src/pipecat/processors/filters/wake_check_filter.py
View File

@@ -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 = []

13
src/pipecat/turns/user_filter/__init__.py Normal file
View File

@@ -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",
]

66
src/pipecat/turns/user_filter/base_user_frame_filter.py Normal file
View File

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

250
src/pipecat/turns/user_filter/wake_phrase_user_frame_filter.py Normal file
View File

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

394
tests/test_wake_phrase_user_frame_filter.py Normal file
View File

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