From 98d3f697f1a0a0644ddd1f6da2cbdedb14435351 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Wed, 18 Mar 2026 16:47:17 -0400 Subject: [PATCH] Add WakePhraseUserTurnStartStrategy (#4064) - Add WakePhraseUserTurnStartStrategy for gating interaction behind wake phrase detection, with timeout and single_activation modes - Add default_user_turn_start_strategies() and default_user_turn_stop_strategies() helper functions - Deprecate WakeCheckFilter in favor of the new strategy - Extend ProcessFrameResult to stop strategies for short-circuit evaluation - Fix MinWordsUserTurnStartStrategy including filtered text in output --- changelog/4064.added.2.md | 1 + changelog/4064.added.md | 1 + changelog/4064.changed.md | 1 + changelog/4064.deprecated.md | 1 + changelog/4064.fixed.md | 1 + examples/foundational/10-wake-phrase.py | 39 +- .../aggregators/llm_response_universal.py | 9 + .../processors/filters/wake_check_filter.py | 18 + src/pipecat/turns/types.py | 24 ++ src/pipecat/turns/user_start/__init__.py | 2 + .../base_user_turn_start_strategy.py | 12 +- .../external_user_turn_start_strategy.py | 11 +- .../min_words_user_turn_start_strategy.py | 27 +- .../transcription_user_turn_start_strategy.py | 12 +- .../vad_user_turn_start_strategy.py | 11 +- .../wake_phrase_user_turn_start_strategy.py | 281 ++++++++++++++ .../user_stop/base_user_turn_stop_strategy.py | 7 +- .../external_user_turn_stop_strategy.py | 7 +- .../speech_timeout_user_turn_stop_strategy.py | 7 +- .../turn_analyzer_user_turn_stop_strategy.py | 8 +- src/pipecat/turns/user_turn_controller.py | 19 +- src/pipecat/turns/user_turn_strategies.py | 29 +- ...st_wake_phrase_user_turn_start_strategy.py | 346 ++++++++++++++++++ 23 files changed, 835 insertions(+), 39 deletions(-) create mode 100644 changelog/4064.added.2.md create mode 100644 changelog/4064.added.md create mode 100644 changelog/4064.changed.md create mode 100644 changelog/4064.deprecated.md create mode 100644 changelog/4064.fixed.md create mode 100644 src/pipecat/turns/types.py create mode 100644 src/pipecat/turns/user_start/wake_phrase_user_turn_start_strategy.py create mode 100644 tests/test_wake_phrase_user_turn_start_strategy.py diff --git a/changelog/4064.added.2.md b/changelog/4064.added.2.md new file mode 100644 index 000000000..f2dc79a16 --- /dev/null +++ b/changelog/4064.added.2.md @@ -0,0 +1 @@ +- Added `default_user_turn_start_strategies()` and `default_user_turn_stop_strategies()` helper functions for composing custom strategy lists. diff --git a/changelog/4064.added.md b/changelog/4064.added.md new file mode 100644 index 000000000..5c28bde94 --- /dev/null +++ b/changelog/4064.added.md @@ -0,0 +1 @@ +- Added `WakePhraseUserTurnStartStrategy` for triggering user turns based on wake phrases, with support for `single_activation` mode. Deprecates `WakeCheckFilter`. diff --git a/changelog/4064.changed.md b/changelog/4064.changed.md new file mode 100644 index 000000000..c66263fbe --- /dev/null +++ b/changelog/4064.changed.md @@ -0,0 +1 @@ +- Extended `ProcessFrameResult` to stop strategies, allowing a stop strategy to short-circuit evaluation of subsequent strategies by returning `STOP`. diff --git a/changelog/4064.deprecated.md b/changelog/4064.deprecated.md new file mode 100644 index 000000000..b5f42c215 --- /dev/null +++ b/changelog/4064.deprecated.md @@ -0,0 +1 @@ +- Deprecated `WakeCheckFilter` in favor of `WakePhraseUserTurnStartStrategy`. diff --git a/changelog/4064.fixed.md b/changelog/4064.fixed.md new file mode 100644 index 000000000..6824ce26f --- /dev/null +++ b/changelog/4064.fixed.md @@ -0,0 +1 @@ +- Fixed `MinWordsUserTurnStartStrategy` including text below the word threshold in the output by resetting aggregation when the minimum word count is not met. diff --git a/examples/foundational/10-wake-phrase.py b/examples/foundational/10-wake-phrase.py index 0a4244c33..5b93efb0e 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,11 @@ 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_start import WakePhraseUserTurnStartStrategy +from pipecat.turns.user_turn_strategies import ( + UserTurnStrategies, + default_user_turn_start_strategies, +) load_dotenv(override=True) @@ -52,7 +56,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 +77,28 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): ), ) - hey_robot_filter = WakeCheckFilter(["hey robot", "hey, robot"]) - context = LLMContext() user_aggregator, assistant_aggregator = LLMContextAggregatorPair( context, - user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()), + user_params=LLMUserAggregatorParams( + user_turn_strategies=UserTurnStrategies( + start=[ + WakePhraseUserTurnStartStrategy( + phrases=["pipecat"], + # Timeout before wake phrase must be spoken again + timeout=5.0, + ), + *default_user_turn_start_strategies(), + ] + ), + vad_analyzer=SileroVADAnalyzer(), + ), ) pipeline = Pipeline( [ transport.input(), # Transport user input - stt, # STT - hey_robot_filter, # Filter out speech not directed at the robot + stt, user_aggregator, # User responses llm, # LLM tts, # TTS @@ -102,12 +120,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): async def on_client_connected(transport, client): logger.info(f"Client connected") # Kick off the conversation. - context.add_message( - { - "role": "user", - "content": "Please introduce yourself. Tell the user they should say 'Hey Robot' before talking to you.", - } - ) + context.add_message({"role": "user", "content": "Please introduce yourself to the user."}) await task.queue_frames([LLMRunFrame()]) @transport.event_handler("on_client_disconnected") diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index b8a7061d8..605db31f6 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/src/pipecat/processors/aggregators/llm_response_universal.py @@ -447,6 +447,9 @@ class LLMUserAggregator(LLMContextAggregator): self._user_turn_controller.add_event_handler( "on_user_turn_stop_timeout", self._on_user_turn_stop_timeout ) + self._user_turn_controller.add_event_handler( + "on_reset_aggregation", self._on_reset_aggregation + ) self._user_idle_controller = UserIdleController( user_idle_timeout=self._params.user_idle_timeout @@ -748,6 +751,12 @@ class LLMUserAggregator(LLMContextAggregator): await self._maybe_emit_user_turn_stopped(strategy) + async def _on_reset_aggregation( + self, controller: UserTurnController, strategy: BaseUserTurnStartStrategy + ): + logger.debug(f"{self}: Resetting aggregation (strategy: {strategy})") + await self.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..6a9e524e6 100644 --- a/src/pipecat/processors/filters/wake_check_filter.py +++ b/src/pipecat/processors/filters/wake_check_filter.py @@ -6,6 +6,9 @@ """Wake phrase detection filter for Pipecat transcription processing. +.. deprecated:: 0.0.106 + Use :class:`~pipecat.turns.user_start.WakePhraseUserTurnStartStrategy` instead. + 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 +16,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 +29,11 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor class WakeCheckFilter(FrameProcessor): """Frame processor that filters transcription frames based on wake phrase detection. + .. deprecated:: 0.0.106 + Use :class:`~pipecat.turns.user_start.WakePhraseUserTurnStartStrategy` instead, + which integrates with the user turn strategy system and supports configurable + timeouts and single-activation mode. + 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. @@ -65,12 +74,21 @@ class WakeCheckFilter(FrameProcessor): def __init__(self, wake_phrases: List[str], keepalive_timeout: float = 3): """Initialize the wake phrase filter. + .. deprecated:: 0.0.106 + Use :class:`~pipecat.turns.user_start.WakePhraseUserTurnStartStrategy` instead. + Args: wake_phrases: List of wake phrases to detect in transcriptions. keepalive_timeout: Duration in seconds to keep passing frames after wake detection. Defaults to 3 seconds. """ super().__init__() + warnings.warn( + "WakeCheckFilter is deprecated since v0.0.106. " + "Use WakePhraseUserTurnStartStrategy instead.", + DeprecationWarning, + stacklevel=2, + ) self._participant_states = {} self._keepalive_timeout = keepalive_timeout self._wake_patterns = [] diff --git a/src/pipecat/turns/types.py b/src/pipecat/turns/types.py new file mode 100644 index 000000000..5ebdf20a3 --- /dev/null +++ b/src/pipecat/turns/types.py @@ -0,0 +1,24 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Shared result type for user turn strategy frame processing.""" + +from enum import Enum + + +class ProcessFrameResult(Enum): + """Result of processing a frame in a user turn strategy. + + Controls whether the strategy loop in the controller continues to the + next strategy or stops early. + + Attributes: + CONTINUE: Continue to the next strategy in the loop. + STOP: Stop evaluating further strategies for this frame. + """ + + CONTINUE = "continue" + STOP = "stop" diff --git a/src/pipecat/turns/user_start/__init__.py b/src/pipecat/turns/user_start/__init__.py index fb84f62ed..94d12708d 100644 --- a/src/pipecat/turns/user_start/__init__.py +++ b/src/pipecat/turns/user_start/__init__.py @@ -9,6 +9,7 @@ from .external_user_turn_start_strategy import ExternalUserTurnStartStrategy from .min_words_user_turn_start_strategy import MinWordsUserTurnStartStrategy from .transcription_user_turn_start_strategy import TranscriptionUserTurnStartStrategy from .vad_user_turn_start_strategy import VADUserTurnStartStrategy +from .wake_phrase_user_turn_start_strategy import WakePhraseUserTurnStartStrategy __all__ = [ "BaseUserTurnStartStrategy", @@ -17,4 +18,5 @@ __all__ = [ "TranscriptionUserTurnStartStrategy", "UserTurnStartedParams", "VADUserTurnStartStrategy", + "WakePhraseUserTurnStartStrategy", ] diff --git a/src/pipecat/turns/user_start/base_user_turn_start_strategy.py b/src/pipecat/turns/user_start/base_user_turn_start_strategy.py index 25f5c8303..fd928a8a7 100644 --- a/src/pipecat/turns/user_start/base_user_turn_start_strategy.py +++ b/src/pipecat/turns/user_start/base_user_turn_start_strategy.py @@ -11,6 +11,7 @@ from typing import Optional, Type from pipecat.frames.frames import Frame from pipecat.processors.frame_processor import FrameDirection +from pipecat.turns.types import ProcessFrameResult from pipecat.utils.asyncio.task_manager import BaseTaskManager from pipecat.utils.base_object import BaseObject @@ -76,6 +77,7 @@ class BaseUserTurnStartStrategy(BaseObject): self._register_event_handler("on_push_frame", sync=True) self._register_event_handler("on_broadcast_frame", sync=True) self._register_event_handler("on_user_turn_started", sync=True) + self._register_event_handler("on_reset_aggregation", sync=True) @property def task_manager(self) -> BaseTaskManager: @@ -100,7 +102,7 @@ class BaseUserTurnStartStrategy(BaseObject): """Reset the strategy to its initial state.""" pass - async def process_frame(self, frame: Frame): + async def process_frame(self, frame: Frame) -> ProcessFrameResult: """Process an incoming frame. Subclasses should override this to implement logic that decides whether @@ -108,6 +110,10 @@ class BaseUserTurnStartStrategy(BaseObject): Args: frame: The frame to be processed. + + Returns: + A ProcessFrameResult indicating the outcome. Subclasses that return + None are treated as CONTINUE for backward compatibility. """ pass @@ -138,3 +144,7 @@ class BaseUserTurnStartStrategy(BaseObject): enable_user_speaking_frames=self._enable_user_speaking_frames, ), ) + + async def trigger_reset_aggregation(self): + """Trigger the `on_reset_aggregation` event.""" + await self._call_event_handler("on_reset_aggregation") diff --git a/src/pipecat/turns/user_start/external_user_turn_start_strategy.py b/src/pipecat/turns/user_start/external_user_turn_start_strategy.py index 9a85cad6d..1031a6b46 100644 --- a/src/pipecat/turns/user_start/external_user_turn_start_strategy.py +++ b/src/pipecat/turns/user_start/external_user_turn_start_strategy.py @@ -7,6 +7,7 @@ """User turn start strategy triggered by externally emitted frames.""" from pipecat.frames.frames import Frame, UserStartedSpeakingFrame +from pipecat.turns.types import ProcessFrameResult from pipecat.turns.user_start.base_user_turn_start_strategy import BaseUserTurnStartStrategy @@ -27,13 +28,17 @@ class ExternalUserTurnStartStrategy(BaseUserTurnStartStrategy): """ super().__init__(enable_interruptions=False, enable_user_speaking_frames=False, **kwargs) - async def process_frame(self, frame: Frame): + async def process_frame(self, frame: Frame) -> ProcessFrameResult: """Process an incoming frame to detect user turn start. Args: frame: The frame to be analyzed. - """ - await super().process_frame(frame) + Returns: + STOP if a user started speaking frame was received, CONTINUE otherwise. + """ if isinstance(frame, UserStartedSpeakingFrame): await self.trigger_user_turn_started() + return ProcessFrameResult.STOP + + return ProcessFrameResult.CONTINUE diff --git a/src/pipecat/turns/user_start/min_words_user_turn_start_strategy.py b/src/pipecat/turns/user_start/min_words_user_turn_start_strategy.py index 57a0ca0d8..10bcfcdad 100644 --- a/src/pipecat/turns/user_start/min_words_user_turn_start_strategy.py +++ b/src/pipecat/turns/user_start/min_words_user_turn_start_strategy.py @@ -15,6 +15,7 @@ from pipecat.frames.frames import ( InterimTranscriptionFrame, TranscriptionFrame, ) +from pipecat.turns.types import ProcessFrameResult from pipecat.turns.user_start.base_user_turn_start_strategy import BaseUserTurnStartStrategy @@ -47,7 +48,7 @@ class MinWordsUserTurnStartStrategy(BaseUserTurnStartStrategy): await super().reset() self._bot_speaking = False - async def process_frame(self, frame: Frame): + async def process_frame(self, frame: Frame) -> ProcessFrameResult: """Process an incoming frame to detect the start of a user turn. This method updates internal state based on transcription frames and @@ -55,17 +56,20 @@ class MinWordsUserTurnStartStrategy(BaseUserTurnStartStrategy): Args: frame: The frame to be analyzed. - """ - await super().process_frame(frame) + Returns: + STOP if the minimum word count was reached, CONTINUE otherwise. + """ if isinstance(frame, BotStartedSpeakingFrame): await self._handle_bot_started_speaking(frame) elif isinstance(frame, BotStoppedSpeakingFrame): await self._handle_bot_stopped_speaking(frame) elif isinstance(frame, TranscriptionFrame): - await self._handle_transcription(frame) + return await self._handle_transcription(frame) elif isinstance(frame, InterimTranscriptionFrame) and self._use_interim: - await self._handle_transcription(frame) + return await self._handle_transcription(frame) + + return ProcessFrameResult.CONTINUE async def _handle_bot_started_speaking(self, frame: BotStartedSpeakingFrame): """Handle bot started speaking frame. @@ -87,11 +91,16 @@ class MinWordsUserTurnStartStrategy(BaseUserTurnStartStrategy): """ self._bot_speaking = False - async def _handle_transcription(self, frame: TranscriptionFrame | InterimTranscriptionFrame): - """Handle a completed transcription frame and check word count. + async def _handle_transcription( + self, frame: TranscriptionFrame | InterimTranscriptionFrame + ) -> ProcessFrameResult: + """Handle a transcription frame and check word count. Args: frame: The transcription frame to be processed. + + Returns: + STOP if the minimum word count was reached, CONTINUE otherwise. """ min_words = self._min_words if self._bot_speaking else 1 @@ -106,3 +115,7 @@ class MinWordsUserTurnStartStrategy(BaseUserTurnStartStrategy): if should_trigger: await self.trigger_user_turn_started() + return ProcessFrameResult.STOP + await self.trigger_reset_aggregation() + + return ProcessFrameResult.CONTINUE diff --git a/src/pipecat/turns/user_start/transcription_user_turn_start_strategy.py b/src/pipecat/turns/user_start/transcription_user_turn_start_strategy.py index b69b127ea..34a3e83e6 100644 --- a/src/pipecat/turns/user_start/transcription_user_turn_start_strategy.py +++ b/src/pipecat/turns/user_start/transcription_user_turn_start_strategy.py @@ -7,6 +7,7 @@ """User turn start strategy based on transcriptions.""" from pipecat.frames.frames import Frame, InterimTranscriptionFrame, TranscriptionFrame +from pipecat.turns.types import ProcessFrameResult from pipecat.turns.user_start.base_user_turn_start_strategy import BaseUserTurnStartStrategy @@ -25,15 +26,20 @@ class TranscriptionUserTurnStartStrategy(BaseUserTurnStartStrategy): super().__init__(**kwargs) self._use_interim = use_interim - async def process_frame(self, frame: Frame): + async def process_frame(self, frame: Frame) -> ProcessFrameResult: """Process an incoming frame to detect the start of a user turn. Args: frame: The frame to be processed. - """ - await super().process_frame(frame) + Returns: + STOP if a transcription was received, CONTINUE otherwise. + """ if isinstance(frame, InterimTranscriptionFrame) and self._use_interim: await self.trigger_user_turn_started() + return ProcessFrameResult.STOP elif isinstance(frame, TranscriptionFrame): await self.trigger_user_turn_started() + return ProcessFrameResult.STOP + + return ProcessFrameResult.CONTINUE diff --git a/src/pipecat/turns/user_start/vad_user_turn_start_strategy.py b/src/pipecat/turns/user_start/vad_user_turn_start_strategy.py index 4bdf48594..2bc3875e0 100644 --- a/src/pipecat/turns/user_start/vad_user_turn_start_strategy.py +++ b/src/pipecat/turns/user_start/vad_user_turn_start_strategy.py @@ -7,6 +7,7 @@ """User turn start strategy based on VAD events.""" from pipecat.frames.frames import Frame, VADUserStartedSpeakingFrame +from pipecat.turns.types import ProcessFrameResult from pipecat.turns.user_start.base_user_turn_start_strategy import BaseUserTurnStartStrategy @@ -18,13 +19,17 @@ class VADUserTurnStartStrategy(BaseUserTurnStartStrategy): """ - async def process_frame(self, frame: Frame): + async def process_frame(self, frame: Frame) -> ProcessFrameResult: """Process an incoming frame to detect user turn start. Args: frame: The frame to be analyzed. - """ - await super().process_frame(frame) + Returns: + STOP if the user started speaking, CONTINUE otherwise. + """ if isinstance(frame, VADUserStartedSpeakingFrame): await self.trigger_user_turn_started() + return ProcessFrameResult.STOP + + return ProcessFrameResult.CONTINUE diff --git a/src/pipecat/turns/user_start/wake_phrase_user_turn_start_strategy.py b/src/pipecat/turns/user_start/wake_phrase_user_turn_start_strategy.py new file mode 100644 index 000000000..bcc069ad7 --- /dev/null +++ b/src/pipecat/turns/user_start/wake_phrase_user_turn_start_strategy.py @@ -0,0 +1,281 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""User turn start strategy that gates interaction behind wake phrase detection.""" + +import asyncio +import enum +import re +from typing import List, Optional + +from loguru import logger + +from pipecat.frames.frames import ( + BotSpeakingFrame, + Frame, + TranscriptionFrame, + UserSpeakingFrame, + VADUserStartedSpeakingFrame, +) +from pipecat.turns.types import ProcessFrameResult +from pipecat.turns.user_start.base_user_turn_start_strategy import BaseUserTurnStartStrategy +from pipecat.utils.asyncio.task_manager import BaseTaskManager + + +class _WakeState(enum.Enum): + """Internal state for wake phrase detection.""" + + IDLE = "idle" + AWAKE = "awake" + + +class WakePhraseUserTurnStartStrategy(BaseUserTurnStartStrategy): + """User turn start strategy that requires a wake phrase before interaction. + + Blocks subsequent strategies until a wake phrase is detected in a final + transcription. After detection, allows interaction for a configurable + timeout period before requiring the wake phrase again. Use + ``single_activation=True`` to require the wake phrase before every turn. + + This strategy should be placed first in the start strategies list. + + Event handlers available: + + - on_wake_phrase_detected: Called when a wake phrase is matched. + - on_wake_phrase_timeout: Called when the inactivity timeout expires + (timeout mode only). + + Example:: + + # Timeout mode (default): wake phrase unlocks interaction for 10s + strategy = WakePhraseUserTurnStartStrategy( + phrases=["hey pipecat", "ok pipecat"], + timeout=10.0, + ) + + # Single activation: wake phrase required before every turn + strategy = WakePhraseUserTurnStartStrategy( + phrases=["hey pipecat"], + single_activation=True, + ) + + @strategy.event_handler("on_wake_phrase_detected") + async def on_wake_phrase_detected(strategy, phrase): + ... + + @strategy.event_handler("on_wake_phrase_timeout") + async def on_wake_phrase_timeout(strategy): + ... + + Args: + phrases: List of wake phrases to detect. + timeout: Inactivity timeout in seconds before returning to IDLE. + In timeout mode, the timer resets on activity (user, bot speech). + In single activation mode, acts as a keepalive window — the strategy + stays AWAKE for this duration after wake phrase detection, allowing + the current turn to complete before returning to IDLE. + single_activation: If True, the wake phrase is required before every + turn. The strategy returns to IDLE after each turn completes. + **kwargs: Additional keyword arguments passed to parent. + """ + + def __init__( + self, + *, + phrases: List[str], + timeout: float = 10.0, + single_activation: bool = False, + **kwargs, + ): + """Initialize the wake phrase user turn start strategy. + + Args: + phrases: List of wake phrases to detect. + timeout: Inactivity timeout in seconds before returning to IDLE. + In timeout mode, the timer resets on activity. In single activation + mode, acts as a keepalive window after wake phrase detection. + single_activation: If True, the wake phrase is required before every + turn. The strategy returns to IDLE after each turn completes. + **kwargs: Additional keyword arguments passed to parent. + """ + super().__init__(**kwargs) + self._phrases = phrases + self._timeout = timeout + self._single_activation = single_activation + + self._patterns: List[re.Pattern] = [] + 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._state = _WakeState.IDLE + self._accumulated_text = "" + + self._timeout_event = asyncio.Event() + self._timeout_task: Optional[asyncio.Task] = None + + self._register_event_handler("on_wake_phrase_detected") + self._register_event_handler("on_wake_phrase_timeout") + + @property + def state(self) -> _WakeState: + """Returns the current wake state.""" + return self._state + + async def setup(self, task_manager: BaseTaskManager): + """Initialize the strategy with the given task manager. + + 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 strategy.""" + await super().cleanup() + if self._timeout_task: + await self.task_manager.cancel_task(self._timeout_task) + self._timeout_task = None + + async def reset(self): + """Reset the strategy. + + In timeout mode, preserves state and refreshes timeout since reset + means a turn started (activity). In single activation mode, does + nothing — the keepalive timeout (started when the wake phrase was + detected) handles the transition back to IDLE. + """ + await super().reset() + if self._state == _WakeState.AWAKE: + if not self._single_activation: + self._refresh_timeout() + + async def process_frame(self, frame: Frame) -> ProcessFrameResult: + """Process an incoming frame for wake phrase detection or passthrough. + + Args: + frame: The frame to be processed. + + Returns: + STOP when the wake phrase is detected or when in IDLE state + (blocks subsequent strategies), CONTINUE when in AWAKE state + (allows subsequent strategies to proceed). + """ + await super().process_frame(frame) + + if self._state == _WakeState.IDLE: + return await self._process_idle(frame) + else: + return await self._process_awake(frame) + + async def _process_idle(self, frame: Frame) -> ProcessFrameResult: + """Process a frame while in IDLE state. + + Only final ``TranscriptionFrame`` instances are checked for wake phrase + matches. When a match is found, a user turn start is triggered. + Transcription frames that don't match have their text cleared so that + pre-wake-phrase speech is not added to the LLM context. All frames + return STOP to block subsequent strategies. + """ + if isinstance(frame, TranscriptionFrame): + if self._check_wake_phrase(frame.text): + await self.trigger_user_turn_started() + return ProcessFrameResult.STOP + await self.trigger_reset_aggregation() + + return ProcessFrameResult.STOP + + async def _process_awake(self, frame: Frame) -> ProcessFrameResult: + """Process a frame while in AWAKE state. + + Refreshes the timeout on activity frames (timeout mode only). Returns + CONTINUE so subsequent strategies can process the frame. + """ + if not self._single_activation: + if isinstance(frame, (UserSpeakingFrame, BotSpeakingFrame)): + self._refresh_timeout() + elif isinstance(frame, TranscriptionFrame): + self._refresh_timeout() + elif isinstance(frame, VADUserStartedSpeakingFrame): + self._refresh_timeout() + + return ProcessFrameResult.CONTINUE + + @staticmethod + def _strip_punctuation(text: str) -> str: + """Strip punctuation from text, keeping only letters, digits, and whitespace.""" + return re.sub(r"[^\w\s]", "", text) + + def _check_wake_phrase(self, text: str) -> bool: + """Check if the accumulated text contains a wake phrase. + + Punctuation is stripped before matching so that STT output like + "Hey, Pipecat!" still matches the phrase "hey pipecat". + + Args: + text: New transcription text to append and check. + + Returns: + True if a wake phrase was found, False otherwise. + """ + self._accumulated_text += " " + self._strip_punctuation(text) + # Cap accumulated text to prevent unbounded growth. + if len(self._accumulated_text) > 250: + self._accumulated_text = self._accumulated_text[-250:] + + for i, pattern in enumerate(self._patterns): + if pattern.search(self._accumulated_text): + phrase = self._phrases[i] + logger.debug(f"{self} wake phrase detected: {phrase!r}") + self._transition_to_awake(phrase) + return True + + return False + + def _transition_to_awake(self, phrase: str): + """Transition from IDLE to AWAKE state.""" + self._state = _WakeState.AWAKE + self._accumulated_text = "" + self._refresh_timeout() + self.task_manager.create_task( + self._call_event_handler("on_wake_phrase_detected", phrase), + f"{self}::on_wake_phrase_detected", + ) + + def _transition_to_idle(self): + """Transition from AWAKE to IDLE state.""" + logger.debug(f"{self} wake phrase timeout, returning to IDLE") + self._state = _WakeState.IDLE + self._accumulated_text = "" + self.task_manager.create_task( + self._call_event_handler("on_wake_phrase_timeout"), + f"{self}::on_wake_phrase_timeout", + ) + + def _refresh_timeout(self): + """Refresh the inactivity timeout.""" + self._timeout_event.set() + + async def _timeout_task_handler(self): + """Background task that monitors inactivity timeout.""" + while True: + try: + await asyncio.wait_for( + self._timeout_event.wait(), + timeout=self._timeout, + ) + self._timeout_event.clear() + except asyncio.TimeoutError: + if self._state == _WakeState.AWAKE: + self._transition_to_idle() diff --git a/src/pipecat/turns/user_stop/base_user_turn_stop_strategy.py b/src/pipecat/turns/user_stop/base_user_turn_stop_strategy.py index c0042f902..7c493475e 100644 --- a/src/pipecat/turns/user_stop/base_user_turn_stop_strategy.py +++ b/src/pipecat/turns/user_stop/base_user_turn_stop_strategy.py @@ -11,6 +11,7 @@ from typing import Optional, Type from pipecat.frames.frames import Frame from pipecat.processors.frame_processor import FrameDirection +from pipecat.turns.types import ProcessFrameResult from pipecat.utils.asyncio.task_manager import BaseTaskManager from pipecat.utils.base_object import BaseObject @@ -89,7 +90,7 @@ class BaseUserTurnStopStrategy(BaseObject): """Reset the strategy to its initial state.""" pass - async def process_frame(self, frame: Frame): + async def process_frame(self, frame: Frame) -> ProcessFrameResult: """Process an incoming frame to decide whether the user stopped speaking. Subclasses should override this to implement logic that decides whether @@ -97,6 +98,10 @@ class BaseUserTurnStopStrategy(BaseObject): Args: frame: The frame to be analyzed. + + Returns: + A ProcessFrameResult indicating the outcome. Subclasses that return + None are treated as CONTINUE for backward compatibility. """ pass diff --git a/src/pipecat/turns/user_stop/external_user_turn_stop_strategy.py b/src/pipecat/turns/user_stop/external_user_turn_stop_strategy.py index 58e037fbf..4ffa3360b 100644 --- a/src/pipecat/turns/user_stop/external_user_turn_stop_strategy.py +++ b/src/pipecat/turns/user_stop/external_user_turn_stop_strategy.py @@ -16,6 +16,7 @@ from pipecat.frames.frames import ( UserStartedSpeakingFrame, UserStoppedSpeakingFrame, ) +from pipecat.turns.types import ProcessFrameResult from pipecat.turns.user_stop.base_user_turn_stop_strategy import BaseUserTurnStopStrategy from pipecat.utils.asyncio.task_manager import BaseTaskManager @@ -69,7 +70,7 @@ class ExternalUserTurnStopStrategy(BaseUserTurnStopStrategy): await self.task_manager.cancel_task(self._task) self._task = None - async def process_frame(self, frame: Frame): + async def process_frame(self, frame: Frame) -> ProcessFrameResult: """Process an incoming frame to update strategy state. Updates internal transcription text and VAD state. The user end turn @@ -78,6 +79,8 @@ class ExternalUserTurnStopStrategy(BaseUserTurnStopStrategy): Args: frame: The frame to be analyzed. + Returns: + Always returns CONTINUE so subsequent stop strategies are evaluated. """ if isinstance(frame, UserStartedSpeakingFrame): await self._handle_user_started_speaking(frame) @@ -88,6 +91,8 @@ class ExternalUserTurnStopStrategy(BaseUserTurnStopStrategy): elif isinstance(frame, TranscriptionFrame): await self._handle_transcription(frame) + return ProcessFrameResult.CONTINUE + async def _handle_user_started_speaking(self, _: UserStartedSpeakingFrame): """Handle when the external service indicates the user is speaking.""" self._user_speaking = True diff --git a/src/pipecat/turns/user_stop/speech_timeout_user_turn_stop_strategy.py b/src/pipecat/turns/user_stop/speech_timeout_user_turn_stop_strategy.py index ec94d9176..341a4c1e3 100644 --- a/src/pipecat/turns/user_stop/speech_timeout_user_turn_stop_strategy.py +++ b/src/pipecat/turns/user_stop/speech_timeout_user_turn_stop_strategy.py @@ -17,6 +17,7 @@ from pipecat.frames.frames import ( VADUserStartedSpeakingFrame, VADUserStoppedSpeakingFrame, ) +from pipecat.turns.types import ProcessFrameResult from pipecat.turns.user_stop.base_user_turn_stop_strategy import BaseUserTurnStopStrategy from pipecat.utils.asyncio.task_manager import BaseTaskManager @@ -83,7 +84,7 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy): await self.task_manager.cancel_task(self._timeout_task) self._timeout_task = None - async def process_frame(self, frame: Frame): + async def process_frame(self, frame: Frame) -> ProcessFrameResult: """Process an incoming frame to update strategy state. Updates internal transcription text and VAD state. The user end turn @@ -92,6 +93,8 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy): Args: frame: The frame to be analyzed. + Returns: + Always returns CONTINUE so subsequent stop strategies are evaluated. """ if isinstance(frame, STTMetadataFrame): self._stt_timeout = frame.ttfs_p99_latency @@ -102,6 +105,8 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy): elif isinstance(frame, TranscriptionFrame): await self._handle_transcription(frame) + return ProcessFrameResult.CONTINUE + async def _handle_vad_user_started_speaking(self, _: VADUserStartedSpeakingFrame): """Handle when the VAD indicates the user is speaking.""" self._vad_user_speaking = True diff --git a/src/pipecat/turns/user_stop/turn_analyzer_user_turn_stop_strategy.py b/src/pipecat/turns/user_stop/turn_analyzer_user_turn_stop_strategy.py index 232bde223..a0df2efbb 100644 --- a/src/pipecat/turns/user_stop/turn_analyzer_user_turn_stop_strategy.py +++ b/src/pipecat/turns/user_stop/turn_analyzer_user_turn_stop_strategy.py @@ -22,6 +22,7 @@ from pipecat.frames.frames import ( VADUserStoppedSpeakingFrame, ) from pipecat.metrics.metrics import MetricsData +from pipecat.turns.types import ProcessFrameResult from pipecat.turns.user_stop.base_user_turn_stop_strategy import BaseUserTurnStopStrategy from pipecat.utils.asyncio.task_manager import BaseTaskManager @@ -88,11 +89,14 @@ class TurnAnalyzerUserTurnStopStrategy(BaseUserTurnStopStrategy): await self.task_manager.cancel_task(self._timeout_task) self._timeout_task = None - async def process_frame(self, frame: Frame): + async def process_frame(self, frame: Frame) -> ProcessFrameResult: """Process an incoming frame to update the turn analyzer and strategy state. Args: frame: The frame to be analyzed. + + Returns: + Always returns CONTINUE so subsequent stop strategies are evaluated. """ await super().process_frame(frame) @@ -109,6 +113,8 @@ class TurnAnalyzerUserTurnStopStrategy(BaseUserTurnStopStrategy): elif isinstance(frame, TranscriptionFrame): await self._handle_transcription(frame) + return ProcessFrameResult.CONTINUE + async def _start(self, frame: StartFrame): """Process the start frame to configure the turn analyzer.""" self._turn_analyzer.set_sample_rate(frame.audio_in_sample_rate) diff --git a/src/pipecat/turns/user_turn_controller.py b/src/pipecat/turns/user_turn_controller.py index a064fc47b..9abed3932 100644 --- a/src/pipecat/turns/user_turn_controller.py +++ b/src/pipecat/turns/user_turn_controller.py @@ -19,7 +19,11 @@ from pipecat.frames.frames import ( VADUserStoppedSpeakingFrame, ) from pipecat.processors.frame_processor import FrameDirection -from pipecat.turns.user_start import BaseUserTurnStartStrategy, UserTurnStartedParams +from pipecat.turns.types import ProcessFrameResult +from pipecat.turns.user_start import ( + BaseUserTurnStartStrategy, + UserTurnStartedParams, +) from pipecat.turns.user_stop import BaseUserTurnStopStrategy, UserTurnStoppedParams from pipecat.turns.user_turn_strategies import UserTurnStrategies from pipecat.utils.asyncio.task_manager import BaseTaskManager @@ -94,6 +98,7 @@ class UserTurnController(BaseObject): self._register_event_handler("on_user_turn_started", sync=True) self._register_event_handler("on_user_turn_stopped", sync=True) self._register_event_handler("on_user_turn_stop_timeout", sync=True) + self._register_event_handler("on_reset_aggregation", sync=True) @property def task_manager(self) -> BaseTaskManager: @@ -161,10 +166,14 @@ class UserTurnController(BaseObject): await self._handle_transcription(frame) for strategy in self._user_turn_strategies.start or []: - await strategy.process_frame(frame) + result = await strategy.process_frame(frame) + if result == ProcessFrameResult.STOP: + break for strategy in self._user_turn_strategies.stop or []: - await strategy.process_frame(frame) + result = await strategy.process_frame(frame) + if result == ProcessFrameResult.STOP: + break async def _setup_strategies(self): for s in self._user_turn_strategies.start or []: @@ -172,6 +181,7 @@ class UserTurnController(BaseObject): s.add_event_handler("on_push_frame", self._on_push_frame) s.add_event_handler("on_broadcast_frame", self._on_broadcast_frame) s.add_event_handler("on_user_turn_started", self._on_user_turn_started) + s.add_event_handler("on_reset_aggregation", self._on_reset_aggregation) for s in self._user_turn_strategies.stop or []: await s.setup(self.task_manager) @@ -242,6 +252,9 @@ class UserTurnController(BaseObject): ): await self._trigger_user_turn_stop(strategy, params) + async def _on_reset_aggregation(self, strategy: BaseUserTurnStartStrategy): + await self._call_event_handler("on_reset_aggregation", strategy) + async def _trigger_user_turn_start( self, strategy: Optional[BaseUserTurnStartStrategy], params: UserTurnStartedParams ): diff --git a/src/pipecat/turns/user_turn_strategies.py b/src/pipecat/turns/user_turn_strategies.py index 0435f141c..5789c6328 100644 --- a/src/pipecat/turns/user_turn_strategies.py +++ b/src/pipecat/turns/user_turn_strategies.py @@ -23,6 +23,31 @@ from pipecat.turns.user_stop import ( ) +def default_user_turn_start_strategies() -> List[BaseUserTurnStartStrategy]: + """Return the default user turn start strategies. + + Returns ``[VADUserTurnStartStrategy, TranscriptionUserTurnStartStrategy]``. + Useful when building a custom strategy list that extends the defaults. + + Example:: + + start_strategies = [ + WakePhraseUserTurnStartStrategy(phrases=["hey pipecat"]), + *default_user_turn_start_strategies(), + ] + """ + return [VADUserTurnStartStrategy(), TranscriptionUserTurnStartStrategy()] + + +def default_user_turn_stop_strategies() -> List[BaseUserTurnStopStrategy]: + """Return the default user turn stop strategies. + + Returns ``[TurnAnalyzerUserTurnStopStrategy(LocalSmartTurnAnalyzerV3)]``. + Useful when building a custom strategy list that extends the defaults. + """ + return [TurnAnalyzerUserTurnStopStrategy(turn_analyzer=LocalSmartTurnAnalyzerV3())] + + @dataclass class UserTurnStrategies: """Container for user turn start and stop strategies. @@ -45,9 +70,9 @@ class UserTurnStrategies: def __post_init__(self): if not self.start: - self.start = [VADUserTurnStartStrategy(), TranscriptionUserTurnStartStrategy()] + self.start = default_user_turn_start_strategies() if not self.stop: - self.stop = [TurnAnalyzerUserTurnStopStrategy(turn_analyzer=LocalSmartTurnAnalyzerV3())] + self.stop = default_user_turn_stop_strategies() @dataclass diff --git a/tests/test_wake_phrase_user_turn_start_strategy.py b/tests/test_wake_phrase_user_turn_start_strategy.py new file mode 100644 index 000000000..e79b9ab99 --- /dev/null +++ b/tests/test_wake_phrase_user_turn_start_strategy.py @@ -0,0 +1,346 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import unittest + +from pipecat.frames.frames import ( + BotSpeakingFrame, + InterimTranscriptionFrame, + TranscriptionFrame, + UserSpeakingFrame, + VADUserStartedSpeakingFrame, +) +from pipecat.turns.types import ProcessFrameResult +from pipecat.turns.user_start.wake_phrase_user_turn_start_strategy import ( + WakePhraseUserTurnStartStrategy, + _WakeState, +) +from pipecat.utils.asyncio.task_manager import TaskManager, TaskManagerParams + + +class TestWakePhraseUserTurnStartStrategy(unittest.IsolatedAsyncioTestCase): + def _create_strategy(self, **kwargs) -> WakePhraseUserTurnStartStrategy: + kwargs.setdefault("phrases", ["hey pipecat"]) + kwargs.setdefault("timeout", 10.0) + return WakePhraseUserTurnStartStrategy(**kwargs) + + async def _setup_strategy(self, strategy: WakePhraseUserTurnStartStrategy): + task_manager = TaskManager() + loop = asyncio.get_running_loop() + task_manager.setup(TaskManagerParams(loop=loop)) + await strategy.setup(task_manager) + return task_manager + + async def test_wake_phrase_in_final_transcription(self): + strategy = self._create_strategy() + await self._setup_strategy(strategy) + + result = await strategy.process_frame( + TranscriptionFrame(text="hey pipecat", user_id="user1", timestamp="") + ) + self.assertEqual(result, ProcessFrameResult.STOP) + self.assertEqual(strategy.state, _WakeState.AWAKE) + + await strategy.cleanup() + + async def test_interim_transcription_ignored(self): + """Interim transcriptions are never used for wake phrase matching.""" + strategy = self._create_strategy() + await self._setup_strategy(strategy) + + result = await strategy.process_frame( + InterimTranscriptionFrame(text="hey pipecat", user_id="user1", timestamp="") + ) + self.assertEqual(result, ProcessFrameResult.STOP) + self.assertEqual(strategy.state, _WakeState.IDLE) + + await strategy.cleanup() + + async def test_no_wake_phrase_returns_stop(self): + strategy = self._create_strategy() + await self._setup_strategy(strategy) + + result = await strategy.process_frame( + TranscriptionFrame(text="hello world", user_id="user1", timestamp="") + ) + self.assertEqual(result, ProcessFrameResult.STOP) + self.assertEqual(strategy.state, _WakeState.IDLE) + + await strategy.cleanup() + + async def test_non_matching_text_resets_aggregation(self): + """Non-matching transcription triggers aggregation reset to prevent LLM context pollution.""" + strategy = self._create_strategy() + await self._setup_strategy(strategy) + + reset_called = False + + @strategy.event_handler("on_reset_aggregation") + async def on_reset_aggregation(strategy): + nonlocal reset_called + reset_called = True + + await strategy.process_frame( + TranscriptionFrame(text="hello world", user_id="user1", timestamp="") + ) + self.assertTrue(reset_called) + + await strategy.cleanup() + + async def test_vad_frame_returns_stop_in_listening(self): + strategy = self._create_strategy() + await self._setup_strategy(strategy) + + result = await strategy.process_frame(VADUserStartedSpeakingFrame()) + self.assertEqual(result, ProcessFrameResult.STOP) + self.assertEqual(strategy.state, _WakeState.IDLE) + + await strategy.cleanup() + + async def test_inactive_returns_continue(self): + strategy = self._create_strategy() + await self._setup_strategy(strategy) + + # Trigger wake phrase first. + await strategy.process_frame( + TranscriptionFrame(text="hey pipecat", user_id="user1", timestamp="") + ) + self.assertEqual(strategy.state, _WakeState.AWAKE) + + # Subsequent frames should return CONTINUE. + result = await strategy.process_frame(VADUserStartedSpeakingFrame()) + self.assertEqual(result, ProcessFrameResult.CONTINUE) + + result = await strategy.process_frame( + TranscriptionFrame(text="what is the weather", user_id="user1", timestamp="") + ) + self.assertEqual(result, ProcessFrameResult.CONTINUE) + + await strategy.cleanup() + + async def test_accumulation_across_frames(self): + strategy = self._create_strategy() + await self._setup_strategy(strategy) + + result = await strategy.process_frame( + TranscriptionFrame(text="hey", user_id="user1", timestamp="") + ) + self.assertEqual(result, ProcessFrameResult.STOP) + self.assertEqual(strategy.state, _WakeState.IDLE) + + result = await strategy.process_frame( + TranscriptionFrame(text="pipecat", user_id="user1", timestamp="") + ) + self.assertEqual(result, ProcessFrameResult.STOP) + self.assertEqual(strategy.state, _WakeState.AWAKE) + + await strategy.cleanup() + + async def test_multiple_phrases(self): + strategy = self._create_strategy(phrases=["hey pipecat", "ok computer"]) + await self._setup_strategy(strategy) + + result = await strategy.process_frame( + TranscriptionFrame(text="ok computer", user_id="user1", timestamp="") + ) + self.assertEqual(result, ProcessFrameResult.STOP) + self.assertEqual(strategy.state, _WakeState.AWAKE) + + await strategy.cleanup() + + async def test_punctuation_stripped(self): + """STT punctuation like 'Hey, Pipecat!' should still match.""" + strategy = self._create_strategy() + await self._setup_strategy(strategy) + + result = await strategy.process_frame( + TranscriptionFrame(text="Hey, Pipecat!", user_id="user1", timestamp="") + ) + self.assertEqual(result, ProcessFrameResult.STOP) + self.assertEqual(strategy.state, _WakeState.AWAKE) + + await strategy.cleanup() + + async def test_reset_preserves_inactive_state(self): + strategy = self._create_strategy() + await self._setup_strategy(strategy) + + await strategy.process_frame( + TranscriptionFrame(text="hey pipecat", user_id="user1", timestamp="") + ) + self.assertEqual(strategy.state, _WakeState.AWAKE) + + await strategy.reset() + self.assertEqual(strategy.state, _WakeState.AWAKE) + + await strategy.cleanup() + + async def test_timeout_returns_to_listening(self): + strategy = self._create_strategy(timeout=0.1) + await self._setup_strategy(strategy) + + # Trigger wake phrase. + await strategy.process_frame( + TranscriptionFrame(text="hey pipecat", user_id="user1", timestamp="") + ) + self.assertEqual(strategy.state, _WakeState.AWAKE) + + # Wait for timeout to expire. + await asyncio.sleep(0.3) + + self.assertEqual(strategy.state, _WakeState.IDLE) + + await strategy.cleanup() + + async def test_activity_refreshes_timeout(self): + strategy = self._create_strategy(timeout=0.2) + await self._setup_strategy(strategy) + + # Trigger wake phrase. + await strategy.process_frame( + TranscriptionFrame(text="hey pipecat", user_id="user1", timestamp="") + ) + self.assertEqual(strategy.state, _WakeState.AWAKE) + + # Send activity before timeout. + await asyncio.sleep(0.1) + await strategy.process_frame(UserSpeakingFrame()) + self.assertEqual(strategy.state, _WakeState.AWAKE) + + # Send more activity. + await asyncio.sleep(0.1) + await strategy.process_frame(BotSpeakingFrame()) + self.assertEqual(strategy.state, _WakeState.AWAKE) + + # Wait for timeout to expire after last activity. + await asyncio.sleep(0.3) + self.assertEqual(strategy.state, _WakeState.IDLE) + + await strategy.cleanup() + + async def test_wake_phrase_detected_event(self): + strategy = self._create_strategy() + await self._setup_strategy(strategy) + + detected_phrase = None + + @strategy.event_handler("on_wake_phrase_detected") + async def on_wake_phrase_detected(strategy, phrase): + nonlocal detected_phrase + detected_phrase = phrase + + await strategy.process_frame( + TranscriptionFrame(text="hey pipecat", user_id="user1", timestamp="") + ) + + # Event fires in a background task, give it a moment. + await asyncio.sleep(0.05) + self.assertEqual(detected_phrase, "hey pipecat") + + await strategy.cleanup() + + async def test_wake_phrase_timeout_event(self): + strategy = self._create_strategy(timeout=0.1) + await self._setup_strategy(strategy) + + timeout_fired = False + + @strategy.event_handler("on_wake_phrase_timeout") + async def on_wake_phrase_timeout(strategy): + nonlocal timeout_fired + timeout_fired = True + + await strategy.process_frame( + TranscriptionFrame(text="hey pipecat", user_id="user1", timestamp="") + ) + + # Wait for timeout. + await asyncio.sleep(0.3) + self.assertTrue(timeout_fired) + + await strategy.cleanup() + + async def test_single_activation_stays_inactive_after_reset(self): + """In single activation mode, reset() keeps INACTIVE so the current turn can finish.""" + strategy = self._create_strategy(single_activation=True, timeout=0.5) + await self._setup_strategy(strategy) + + # Trigger wake phrase. + result = await strategy.process_frame( + TranscriptionFrame(text="hey pipecat", user_id="user1", timestamp="") + ) + self.assertEqual(result, ProcessFrameResult.STOP) + self.assertEqual(strategy.state, _WakeState.AWAKE) + + # Simulate turn start (controller calls reset on all start strategies). + await strategy.reset() + # State remains INACTIVE so frames continue to flow. + self.assertEqual(strategy.state, _WakeState.AWAKE) + + # Subsequent frames should pass through (CONTINUE). + result = await strategy.process_frame(VADUserStartedSpeakingFrame()) + self.assertEqual(result, ProcessFrameResult.CONTINUE) + + result = await strategy.process_frame( + TranscriptionFrame(text="what is the weather", user_id="user1", timestamp="") + ) + self.assertEqual(result, ProcessFrameResult.CONTINUE) + + await strategy.cleanup() + + async def test_single_activation_timeout_returns_to_listening(self): + """In single activation mode, the keepalive timeout returns to LISTENING.""" + strategy = self._create_strategy(single_activation=True, timeout=0.1) + await self._setup_strategy(strategy) + + # Trigger wake phrase. + await strategy.process_frame( + TranscriptionFrame(text="hey pipecat", user_id="user1", timestamp="") + ) + self.assertEqual(strategy.state, _WakeState.AWAKE) + + # Wait for keepalive timeout to expire. + await asyncio.sleep(0.3) + self.assertEqual(strategy.state, _WakeState.IDLE) + + # Frames should now be blocked again. + result = await strategy.process_frame(VADUserStartedSpeakingFrame()) + self.assertEqual(result, ProcessFrameResult.STOP) + + await strategy.cleanup() + + async def test_single_activation_requires_wake_phrase_after_timeout(self): + """Single activation mode requires wake phrase again after keepalive timeout.""" + strategy = self._create_strategy(single_activation=True, timeout=0.1) + await self._setup_strategy(strategy) + + # First turn: wake phrase -> INACTIVE -> timeout -> LISTENING. + await strategy.process_frame( + TranscriptionFrame(text="hey pipecat", user_id="user1", timestamp="") + ) + self.assertEqual(strategy.state, _WakeState.AWAKE) + await asyncio.sleep(0.3) + self.assertEqual(strategy.state, _WakeState.IDLE) + + # Without wake phrase, frames are blocked. + result = await strategy.process_frame( + TranscriptionFrame(text="what is the weather", user_id="user1", timestamp="") + ) + self.assertEqual(result, ProcessFrameResult.STOP) + + # Second turn: wake phrase again. + result = await strategy.process_frame( + TranscriptionFrame(text="hey pipecat", user_id="user1", timestamp="") + ) + self.assertEqual(result, ProcessFrameResult.STOP) + self.assertEqual(strategy.state, _WakeState.AWAKE) + + await strategy.cleanup() + + +if __name__ == "__main__": + unittest.main()