diff --git a/src/pipecat/turns/__init__.py b/src/pipecat/turns/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/pipecat/turns/bot/base_bot_turn_start_strategy.py b/src/pipecat/turns/bot/base_bot_turn_start_strategy.py new file mode 100644 index 000000000..bf804ba9c --- /dev/null +++ b/src/pipecat/turns/bot/base_bot_turn_start_strategy.py @@ -0,0 +1,74 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Base turn start strategy for determining when the bot should start speaking.""" + +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 BaseBotTurnStartStrategy(BaseObject): + """Base class for strategies that determine when the bot should start speaking. + + Subclasses should implement logic to detect when the bot should start + speaking. This could be based on analyzing incoming frames (such as + transcriptions), conversation state, or other heuristics. + + Events triggered by bot turn start strategies: + + - `on_push_frame`: Indicates the strategy wants to push a frame. + - `on_bot_turn_started`: Signals that the bot should start speaking. + + """ + + def __init__(self, **kwargs): + """Initialize the base bot turn start strategy.""" + super().__init__(**kwargs) + self._task_manager: Optional[BaseTaskManager] = None + self._register_event_handler("on_push_frame", sync=True) + self._register_event_handler("on_bot_turn_started", sync=True) + + @property + def task_manager(self) -> BaseTaskManager: + """Returns the configured task manager.""" + if not self._task_manager: + raise RuntimeError(f"{self} bot turn start strategy was not properly setup") + return self._task_manager + + async def reset(self): + """Reset the strategy to its initial state.""" + pass + + 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. + """ + self._task_manager = task_manager + + async def cleanup(self): + """Cleanup the strategy.""" + pass + + async def process_frame(self, frame: Frame): + """Process an incoming frame to decide whether the bot should speak. + + Subclasses should override this to implement logic that decides whether + the bot turn has started. + + Args: + frame: The frame to be analyzed. + + """ + pass + + async def trigger_bot_turn_started(self): + """Trigger the `on_bot_turn_started` event.""" + await self._call_event_handler("on_bot_turn_started") diff --git a/src/pipecat/turns/bot/transcription_bot_turn_start_strategy.py b/src/pipecat/turns/bot/transcription_bot_turn_start_strategy.py new file mode 100644 index 000000000..36541f765 --- /dev/null +++ b/src/pipecat/turns/bot/transcription_bot_turn_start_strategy.py @@ -0,0 +1,111 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Transcription time-based speaking strategy.""" + +import asyncio +from typing import Optional + +from pipecat.frames.frames import ( + Frame, + TranscriptionFrame, + VADUserStartedSpeakingFrame, + VADUserStoppedSpeakingFrame, +) +from pipecat.turns.bot.base_bot_turn_start_strategy import BaseBotTurnStartStrategy +from pipecat.utils.asyncio.task_manager import BaseTaskManager + + +class TranscriptionBotTurnStartStrategy(BaseBotTurnStartStrategy): + """Bot turn start strategy based on transcriptions. + + This strategy assumes the bot should start speaking once a transcription + has been received and the user is not actively speaking. It handles + multiple or delayed transcription frames gracefully. + """ + + def __init__(self, *, timeout: float = 0.5): + """Initialize the transcription-based bot turn start strategy. + + Args: + timeout: A short delay used internally to handle consecutive or + slightly delayed transcriptions. + """ + super().__init__() + self._timeout = timeout + self._text = "" + self._vad_user_speaking = False + self._event = asyncio.Event() + self._task: Optional[asyncio.Task] = None + + async def reset(self): + """Reset the strategy to its initial state.""" + await super().reset() + self._text = "" + self._vad_user_speaking = False + self._event.clear() + + 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) + self._task = task_manager.create_task(self._task_handler(), f"{self}::_task_handler") + + async def cleanup(self): + """Cleanup the strategy.""" + await super().cleanup() + if self._task: + await self.task_manager.cancel_task(self._task) + self._task = None + + async def process_frame(self, frame: Frame): + """Process an incoming frame to update strategy state. + + Updates internal transcription text and VAD state. The bot turn will be + triggered when appropriate based on the collected frames. + + Args: + frame: The frame to be analyzed. + + """ + if isinstance(frame, VADUserStartedSpeakingFrame): + await self._handle_vad_user_started_speaking(frame) + elif isinstance(frame, VADUserStoppedSpeakingFrame): + await self._handle_vad_user_stopped_speaking(frame) + elif isinstance(frame, TranscriptionFrame): + await self._handle_transcription(frame) + + async def _handle_vad_user_started_speaking(self, _: VADUserStartedSpeakingFrame): + """Handle when the VAD indicates the user is speaking.""" + self._vad_user_speaking = True + + async def _handle_vad_user_stopped_speaking(self, _: VADUserStoppedSpeakingFrame): + """Handle when the VAD indicates the user has stopped speaking.""" + self._vad_user_speaking = False + + async def _handle_transcription(self, frame: TranscriptionFrame): + """Handle user transcription.""" + self._text += frame.text + self._event.set() + + async def _task_handler(self): + """Asynchronously monitor transcriptions and trigger bot turn when ready. + + If transcription text exists and the user is not currently speaking, + triggers the bot turn. Handles multiple or delayed transcriptions + gracefully. + + """ + while True: + try: + await asyncio.wait_for(self._event.wait(), timeout=self._timeout) + self._event.clear() + except asyncio.TimeoutError: + if self._text and not self._vad_user_speaking: + await self.trigger_bot_turn_started() diff --git a/src/pipecat/turns/bot/turn_analyzer_bot_turn_start_strategy.py b/src/pipecat/turns/bot/turn_analyzer_bot_turn_start_strategy.py new file mode 100644 index 000000000..4e09338fb --- /dev/null +++ b/src/pipecat/turns/bot/turn_analyzer_bot_turn_start_strategy.py @@ -0,0 +1,152 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Bot turn start strategy based on turn detection analyzers.""" + +import asyncio +from typing import Optional + +from pipecat.audio.turn.base_turn_analyzer import BaseTurnAnalyzer, EndOfTurnState +from pipecat.frames.frames import ( + Frame, + InputAudioRawFrame, + InterimTranscriptionFrame, + MetricsFrame, + StartFrame, + TranscriptionFrame, + VADUserStartedSpeakingFrame, + VADUserStoppedSpeakingFrame, +) +from pipecat.metrics.metrics import MetricsData +from pipecat.processors.frame_processor import FrameDirection +from pipecat.turns.bot.base_bot_turn_start_strategy import BaseBotTurnStartStrategy +from pipecat.utils.asyncio.task_manager import BaseTaskManager + + +class TurnAnalyzerBotTurnStartStrategy(BaseBotTurnStartStrategy): + """Bot turn start strategy using a turn detection model to detect end of user turn. + + This strategy uses the turn detection models to determine when the user has + finished speaking, combining audio, VAD, and transcription frames. Once the + turn is considered complete, the bot turn is triggered. + + """ + + def __init__(self, *, turn_analyzer: BaseTurnAnalyzer, timeout: float = 0.5): + """Initialize the bot turn start strategy. + + Args: + turn_analyzer: The turn detection analyzer instance to detect end of user turn. + timeout: Short delay used internally to handle frame timing and event triggering. + """ + super().__init__() + self._turn_analyzer = turn_analyzer + self._timeout = timeout + self._text = "" + self._vad_user_speaking = False + self._event = asyncio.Event() + self._task: Optional[asyncio.Task] = None + + async def reset(self): + """Reset the strategy to its initial state.""" + await super().reset() + self._text = "" + self._vad_user_speaking = False + self._event.set() + + 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) + self._task = task_manager.create_task(self._task_handler(), f"{self}::_task_handler") + + async def cleanup(self): + """Cleanup the strategy.""" + await super().cleanup() + if self._task: + await self.task_manager.cancel_task(self._task) + self._task = None + + async def process_frame(self, frame: Frame): + """Process an incoming frame to update the turn analyzer and strategy state. + + Args: + frame: The frame to be analyzed. + """ + await super().process_frame(frame) + + if isinstance(frame, StartFrame): + await self._start(frame) + elif isinstance(frame, VADUserStartedSpeakingFrame): + await self._handle_vad_user_started_speaking(frame) + elif isinstance(frame, VADUserStoppedSpeakingFrame): + await self._handle_vad_user_stopped_speaking(frame) + elif isinstance(frame, InputAudioRawFrame): + await self._handle_input_audio(frame) + elif isinstance(frame, (TranscriptionFrame, InterimTranscriptionFrame)): + await self._handle_transcription(frame) + + 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) + + async def _handle_input_audio(self, frame: InputAudioRawFrame): + """Handle input audio to check if the turn is completed.""" + state = self._turn_analyzer.append_audio(frame.audio, self._vad_user_speaking) + await self._handle_end_of_turn(state) + + async def _handle_vad_user_started_speaking(self, _: VADUserStartedSpeakingFrame): + """Handle when the VAD indicates the user is speaking.""" + self._vad_user_speaking = True + self._event.set() + + async def _handle_vad_user_stopped_speaking(self, _: VADUserStoppedSpeakingFrame): + """Handle when the VAD indicates the user has stopped speaking.""" + self._vad_user_speaking = False + self._event.set() + + state, prediction = await self._turn_analyzer.analyze_end_of_turn() + await self._handle_prediction_result(prediction) + await self._handle_end_of_turn(state) + + async def _handle_transcription(self, frame: TranscriptionFrame | InterimTranscriptionFrame): + """Handle user transcription.""" + # We don't really care about the content. + self._text = frame.text + self._event.set() + + async def _handle_end_of_turn(self, state: EndOfTurnState): + """Handle completion of end-of-turn analysis.""" + if state == EndOfTurnState.COMPLETE: + self._event.set() + + async def _handle_prediction_result(self, result: Optional[MetricsData]): + """Handle a prediction result event from the turn analyzer.""" + if result: + await self._call_event_handler( + "on_push_frame", + MetricsFrame(data=[result]), + FrameDirection.DOWNSTREAM, + ) + + async def _task_handler(self): + """Asynchronously monitor events and trigger bot turn when appropriate. + + If we have not received a transcription in the specified amount of time + (and we initially received one) and the turn analyzer said the turn is + done, then the bot is ready to speak. + + """ + while True: + try: + await asyncio.wait_for(self._event.wait(), timeout=self._timeout) + self._event.clear() + except asyncio.TimeoutError: + if self._text: + await self.trigger_bot_turn_started() diff --git a/src/pipecat/turns/user/__init__.py b/src/pipecat/turns/user/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/pipecat/turns/user/base_user_turn_start_strategy.py b/src/pipecat/turns/user/base_user_turn_start_strategy.py new file mode 100644 index 000000000..216932f45 --- /dev/null +++ b/src/pipecat/turns/user/base_user_turn_start_strategy.py @@ -0,0 +1,73 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Base turn start strategy for determining when the user starts speaking.""" + +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 BaseUserTurnStartStrategy(BaseObject): + """Base class for strategies that determine when a user starts speaking. + + Subclasses should implement logic to detect the start of a user's turn. + This could be based on voice activity, number of words spoken, or other + heuristics. + + Events triggered by user turn start strategies: + + - `on_push_frame`: Indicates the strategy wants to push a frame. + - `on_user_turn_started`: Signals that a user turn has started. + """ + + def __init__(self, **kwargs): + """Initialize the base user turn start strategy.""" + super().__init__(**kwargs) + self._task_manager: Optional[BaseTaskManager] = None + self._register_event_handler("on_push_frame", sync=True) + self._register_event_handler("on_user_turn_started", sync=True) + + @property + def task_manager(self) -> BaseTaskManager: + """Returns the configured task manager.""" + if not self._task_manager: + raise RuntimeError(f"{self} user turn start strategy was not properly setup") + return self._task_manager + + async def reset(self): + """Reset the strategy to its initial state.""" + pass + + 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. + """ + self._task_manager = task_manager + + async def cleanup(self): + """Cleanup the strategy.""" + pass + + async def process_frame(self, frame: Frame): + """Process an incoming frame. + + Subclasses should override this to implement logic that decides whether + the user turn has started. + + Args: + frame: The frame to be processed. + + """ + pass + + async def trigger_user_turn_started(self): + """Trigger the `on_user_turn_started` event.""" + await self._call_event_handler("on_user_turn_started") diff --git a/src/pipecat/turns/user/min_words_user_turn_start_strategy.py b/src/pipecat/turns/user/min_words_user_turn_start_strategy.py new file mode 100644 index 000000000..61204aec8 --- /dev/null +++ b/src/pipecat/turns/user/min_words_user_turn_start_strategy.py @@ -0,0 +1,91 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""User turn start strategy based on a minimum number of words spoken by the user.""" + +from loguru import logger + +from pipecat.frames.frames import Frame, InterimTranscriptionFrame, TranscriptionFrame +from pipecat.turns.user.base_user_turn_start_strategy import BaseUserTurnStartStrategy + + +class MinWordsUserTurnStartStrategy(BaseUserTurnStartStrategy): + """User turn start strategy based on a minimum number of words spoken by the user. + + This strategy signals the start of a user turn once the user has spoken at + least a specified number of words, as determined from transcription frames. + Optionally, interim transcriptions can be used for earlier detection. + + """ + + def __init__(self, *, min_words: int, use_interim: bool = True): + """Initialize the minimum words bot turn start strategy. + + Args: + min_words: Minimum number of spoken words required to trigger the + start of a user turn. + use_interim: Whether to consider interim transcription frames for + earlier detection. + """ + super().__init__() + self._min_words = min_words + self._use_interim = use_interim + self._text = "" + + async def reset(self): + """Reset the strategy to its initial state.""" + await super().reset() + self._text = "" + + async def process_frame(self, frame: Frame): + """Process an incoming frame to detect the start of a user turn. + + This method updates internal state based on transcription frames and + triggers the user turn once the minimum word count is reached. + + Args: + frame: The frame to be analyzed. + """ + await super().process_frame(frame) + + if isinstance(frame, TranscriptionFrame): + await self._handle_transcription(frame) + elif isinstance(frame, InterimTranscriptionFrame) and self._use_interim: + await self._handle_interim_transcription(frame) + + async def _handle_transcription(self, frame: TranscriptionFrame): + """Handle a completed transcription frame and check word count. + + Args: + frame: The transcription frame to be processed. + """ + self._text += frame.text + + word_count = len(self._text.split()) + should_trigger = word_count >= self._min_words + + logger.debug( + f"{self} should_trigger={should_trigger} num_spoken_words={word_count} min_words={self._min_words}" + ) + + if should_trigger: + await self.trigger_user_turn_started() + + async def _handle_interim_transcription(self, frame: InterimTranscriptionFrame): + """Handle an interim transcription frame and check word count. + + Args: + frame: The interim transcription frame to be processed. + """ + word_count = len(frame.text.split()) + should_trigger = word_count >= self._min_words + + logger.debug( + f"{self} interim=True should_trigger={should_trigger} num_spoken_words={word_count} min_words={self._min_words}" + ) + + if should_trigger: + await self.trigger_user_turn_started() diff --git a/src/pipecat/turns/user/vad_user_turn_start_strategy.py b/src/pipecat/turns/user/vad_user_turn_start_strategy.py new file mode 100644 index 000000000..3a0e491c2 --- /dev/null +++ b/src/pipecat/turns/user/vad_user_turn_start_strategy.py @@ -0,0 +1,30 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""User turn start strategy based on VAD events.""" + +from pipecat.frames.frames import Frame, VADUserStartedSpeakingFrame +from pipecat.turns.user.base_user_turn_start_strategy import BaseUserTurnStartStrategy + + +class VADUserTurnStartStrategy(BaseUserTurnStartStrategy): + """User turn start strategy based on VAD (Voice Activity Detection). + + This strategy assumes the user turn starts as soon as a VAD frame indicates + that the user has started speaking. + + """ + + async def process_frame(self, frame: Frame): + """Process an incoming frame to detect user turn start. + + Args: + frame: The frame to be analyzed. + """ + await super().process_frame(frame) + + if isinstance(frame, VADUserStartedSpeakingFrame): + await self.trigger_user_turn_started()