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

Add WakePhraseUserTurnStartStrategy (#4064)

Browse Source 
- 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
This commit is contained in:
Mark Backman
2026-03-18 16:47:17 -04:00
committed by GitHub
parent c4be513044
commit bad10177d4
23 changed files with 835 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

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

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

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

24
src/pipecat/turns/types.py Normal file
View File

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

2
src/pipecat/turns/user_start/__init__.py
View File

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

12
src/pipecat/turns/user_start/base_user_turn_start_strategy.py
View File

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

11
src/pipecat/turns/user_start/external_user_turn_start_strategy.py
View File

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

27
src/pipecat/turns/user_start/min_words_user_turn_start_strategy.py
View File

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

12
src/pipecat/turns/user_start/transcription_user_turn_start_strategy.py
View File

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

11
src/pipecat/turns/user_start/vad_user_turn_start_strategy.py
View File

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

281
src/pipecat/turns/user_start/wake_phrase_user_turn_start_strategy.py Normal file
View File

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

7
src/pipecat/turns/user_stop/base_user_turn_stop_strategy.py
View File

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

7
src/pipecat/turns/user_stop/external_user_turn_stop_strategy.py
View File

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

7
src/pipecat/turns/user_stop/speech_timeout_user_turn_stop_strategy.py
View File

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

8
src/pipecat/turns/user_stop/turn_analyzer_user_turn_stop_strategy.py
View File

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

19
src/pipecat/turns/user_turn_controller.py
View File

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

29
src/pipecat/turns/user_turn_strategies.py
View File

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