introduce new user and bot turn start strategies

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

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

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

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

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

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