From ffbbb1b3f562cf8800c9847969f065d1c8a717ac Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 29 Dec 2025 15:43:26 -0800 Subject: [PATCH] turns(user): add support for enabling/disabling interruptions --- .../aggregators/llm_response_universal.py | 5 +++- .../bot/external_bot_turn_start_strategy.py | 5 ++-- .../transcription_bot_turn_start_strategy.py | 5 ++-- .../turn_analyzer_bot_turn_start_strategy.py | 5 ++-- .../user/base_user_turn_start_strategy.py | 27 +++++++++++++----- .../user/external_user_turn_start_strategy.py | 10 +++++-- .../min_words_user_turn_start_strategy.py | 5 ++-- .../transcription_user_turn_start_strategy.py | 26 +++++------------ tests/test_bot_turn_start_strategy.py | 28 +++++++++---------- tests/test_user_turn_start_strategy.py | 21 ++++++-------- 10 files changed, 73 insertions(+), 64 deletions(-) diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index 9feb23199..0bf41b8c7 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/src/pipecat/processors/aggregators/llm_response_universal.py @@ -561,8 +561,11 @@ class LLMUserAggregator(LLMContextAggregator): await s.reset() if params.enable_user_speaking_frames: - # TODO(aleix): These frames should really come from the top of the pipeline. + # TODO(aleix): This frame should really come from the top of the pipeline. await self.broadcast_frame(UserStartedSpeakingFrame) + + if params.enable_interruptions: + # TODO(aleix): This frame should really come from the top of the pipeline. await self.broadcast_frame(InterruptionFrame) await self._call_event_handler("on_user_turn_started", strategy) diff --git a/src/pipecat/turns/bot/external_bot_turn_start_strategy.py b/src/pipecat/turns/bot/external_bot_turn_start_strategy.py index ddc6fb888..8557c11ed 100644 --- a/src/pipecat/turns/bot/external_bot_turn_start_strategy.py +++ b/src/pipecat/turns/bot/external_bot_turn_start_strategy.py @@ -29,14 +29,15 @@ class ExternalBotTurnStartStrategy(BaseBotTurnStartStrategy): """ - def __init__(self, *, timeout: float = 0.5): + def __init__(self, *, timeout: float = 0.5, **kwargs): """Initialize the external bot turn start strategy. Args: timeout: A short delay used internally to handle consecutive or slightly delayed transcriptions. + **kwargs: Additional keyword arguments. """ - super().__init__(enable_user_speaking_frames=False) + super().__init__(enable_user_speaking_frames=False, **kwargs) self._timeout = timeout self._text = "" self._user_speaking = False diff --git a/src/pipecat/turns/bot/transcription_bot_turn_start_strategy.py b/src/pipecat/turns/bot/transcription_bot_turn_start_strategy.py index 60a6f76fc..6a91e93d1 100644 --- a/src/pipecat/turns/bot/transcription_bot_turn_start_strategy.py +++ b/src/pipecat/turns/bot/transcription_bot_turn_start_strategy.py @@ -28,14 +28,15 @@ class TranscriptionBotTurnStartStrategy(BaseBotTurnStartStrategy): multiple or delayed transcription frames gracefully. """ - def __init__(self, *, timeout: float = 0.5): + def __init__(self, *, timeout: float = 0.5, **kwargs): """Initialize the transcription-based bot turn start strategy. Args: timeout: A short delay used internally to handle consecutive or slightly delayed transcriptions. + **kwargs: Additional keyword arguments. """ - super().__init__() + super().__init__(**kwargs) self._timeout = timeout self._text = "" self._vad_user_speaking = False 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 index 3cf5a18c4..cbfe9a22a 100644 --- a/src/pipecat/turns/bot/turn_analyzer_bot_turn_start_strategy.py +++ b/src/pipecat/turns/bot/turn_analyzer_bot_turn_start_strategy.py @@ -35,14 +35,15 @@ class TurnAnalyzerBotTurnStartStrategy(BaseBotTurnStartStrategy): """ - def __init__(self, *, turn_analyzer: BaseTurnAnalyzer, timeout: float = 0.5): + def __init__(self, *, turn_analyzer: BaseTurnAnalyzer, timeout: float = 0.5, **kwargs): """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. + **kwargs: Additional keyword arguments. """ - super().__init__() + super().__init__(**kwargs) self._turn_analyzer = turn_analyzer self._timeout = timeout self._text = "" diff --git a/src/pipecat/turns/user/base_user_turn_start_strategy.py b/src/pipecat/turns/user/base_user_turn_start_strategy.py index bfe4197bb..d13a991b6 100644 --- a/src/pipecat/turns/user/base_user_turn_start_strategy.py +++ b/src/pipecat/turns/user/base_user_turn_start_strategy.py @@ -32,6 +32,7 @@ class UserTurnStartedParams: """ + enable_interruptions: bool enable_user_speaking_frames: bool @@ -49,18 +50,27 @@ class BaseUserTurnStartStrategy(BaseObject): - `on_user_turn_started`: Signals that a user turn has started. """ - def __init__(self, *, enable_user_speaking_frames: bool = True, **kwargs): + def __init__( + self, + *, + enable_interruptions: bool = True, + enable_user_speaking_frames: bool = True, + **kwargs, + ): """Initialize the base user turn start strategy. Args: - enable_user_speaking_frames: If True, the aggregator will emit frames - indicating when the user starts speaking, as well as interruption - frames. This is enabled by default, but you may want to disable it - if another component (e.g., an STT service) is already generating - these frames. + enable_interruptions: If True, the user aggregator will emit an + interruption frame when the user turn starts. + enable_user_speaking_frames: If True, the user aggregator will emit + frames indicating when the user starts speaking, as well as + interruption frames. This is enabled by default, but you may want + to disable it if another component (e.g., an STT service) is + already generating these frames. **kwargs: Additional keyword arguments. """ super().__init__(**kwargs) + self._enable_interruptions = enable_interruptions self._enable_user_speaking_frames = enable_user_speaking_frames self._task_manager: Optional[BaseTaskManager] = None self._register_event_handler("on_push_frame", sync=True) @@ -123,5 +133,8 @@ class BaseUserTurnStartStrategy(BaseObject): """Trigger the `on_user_turn_started` event.""" await self._call_event_handler( "on_user_turn_started", - UserTurnStartedParams(enable_user_speaking_frames=self._enable_user_speaking_frames), + UserTurnStartedParams( + enable_interruptions=self._enable_interruptions, + enable_user_speaking_frames=self._enable_user_speaking_frames, + ), ) diff --git a/src/pipecat/turns/user/external_user_turn_start_strategy.py b/src/pipecat/turns/user/external_user_turn_start_strategy.py index 60f07d023..03ed2fd1b 100644 --- a/src/pipecat/turns/user/external_user_turn_start_strategy.py +++ b/src/pipecat/turns/user/external_user_turn_start_strategy.py @@ -19,9 +19,13 @@ class ExternalUserTurnStartStrategy(BaseUserTurnStartStrategy): """ - def __init__(self): - """Initialize the external user turn start strategy.""" - super().__init__(enable_user_speaking_frames=False) + def __init__(self, **kwargs): + """Initialize the external user turn start strategy. + + Args: + **kwargs: Additional keyword arguments. + """ + super().__init__(enable_user_speaking_frames=False, **kwargs) async def process_frame(self, frame: Frame): """Process an incoming frame to detect user turn start. 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 index 64b01ae66..1412a4cd9 100644 --- a/src/pipecat/turns/user/min_words_user_turn_start_strategy.py +++ b/src/pipecat/turns/user/min_words_user_turn_start_strategy.py @@ -27,7 +27,7 @@ class MinWordsUserTurnStartStrategy(BaseUserTurnStartStrategy): """ - def __init__(self, *, min_words: int, use_interim: bool = True): + def __init__(self, *, min_words: int, use_interim: bool = True, **kwargs): """Initialize the minimum words bot turn start strategy. Args: @@ -35,8 +35,9 @@ class MinWordsUserTurnStartStrategy(BaseUserTurnStartStrategy): start of a user turn. use_interim: Whether to consider interim transcription frames for earlier detection. + **kwargs: Additional keyword arguments. """ - super().__init__() + super().__init__(**kwargs) self._min_words = min_words self._use_interim = use_interim self._bot_speaking = False diff --git a/src/pipecat/turns/user/transcription_user_turn_start_strategy.py b/src/pipecat/turns/user/transcription_user_turn_start_strategy.py index 7172014d4..ed04c62d7 100644 --- a/src/pipecat/turns/user/transcription_user_turn_start_strategy.py +++ b/src/pipecat/turns/user/transcription_user_turn_start_strategy.py @@ -6,7 +6,7 @@ """User turn start strategy based on transcriptions.""" -from pipecat.frames.frames import BotStartedSpeakingFrame, Frame, TranscriptionFrame +from pipecat.frames.frames import Frame, InterimTranscriptionFrame, TranscriptionFrame from pipecat.turns.user.base_user_turn_start_strategy import BaseUserTurnStartStrategy @@ -20,15 +20,10 @@ class TranscriptionUserTurnStartStrategy(BaseUserTurnStartStrategy): """ - def __init__(self): + def __init__(self, *, use_interim: bool = True, **kwargs): """Initialize transcription-based user turn start strategy.""" - super().__init__() - self._bot_speaking = False - - async def reset(self): - """Reset the strategy to its initial state.""" - await super().reset() - self._bot_speaking = False + super().__init__(**kwargs) + self._use_interim = use_interim async def process_frame(self, frame: Frame): """Process an incoming frame to detect the start of a user turn. @@ -38,14 +33,7 @@ class TranscriptionUserTurnStartStrategy(BaseUserTurnStartStrategy): """ await super().process_frame(frame) - if isinstance(frame, BotStartedSpeakingFrame): - await self._handle_bot_started_speaking(frame) - elif isinstance(frame, TranscriptionFrame): - await self._handle_transcription(frame) - - async def _handle_bot_started_speaking(self, _: BotStartedSpeakingFrame): - self._bot_speaking = True - - async def _handle_transcription(self, _: TranscriptionFrame): - if self._bot_speaking: + if isinstance(frame, InterimTranscriptionFrame) and self._use_interim: + await self.trigger_user_turn_started() + elif isinstance(frame, TranscriptionFrame): await self.trigger_user_turn_started() diff --git a/tests/test_bot_turn_start_strategy.py b/tests/test_bot_turn_start_strategy.py index 8e4396fc2..d3867b746 100644 --- a/tests/test_bot_turn_start_strategy.py +++ b/tests/test_bot_turn_start_strategy.py @@ -33,7 +33,7 @@ class TestTranscriptionBotTurnStartStrategy(unittest.IsolatedAsyncioTestCase): should_start = None @strategy.event_handler("on_bot_turn_started") - async def on_bot_turn_started(strategy, enable_user_speaking_frames): + async def on_bot_turn_started(strategy, params): nonlocal should_start should_start = True @@ -58,7 +58,7 @@ class TestTranscriptionBotTurnStartStrategy(unittest.IsolatedAsyncioTestCase): should_start = None @strategy.event_handler("on_bot_turn_started") - async def on_bot_turn_started(strategy, enable_user_speaking_frames): + async def on_bot_turn_started(strategy, params): nonlocal should_start should_start = True @@ -89,7 +89,7 @@ class TestTranscriptionBotTurnStartStrategy(unittest.IsolatedAsyncioTestCase): should_start = None @strategy.event_handler("on_bot_turn_started") - async def on_bot_turn_started(strategy, enable_user_speaking_frames): + async def on_bot_turn_started(strategy, params): nonlocal should_start should_start = True @@ -136,7 +136,7 @@ class TestTranscriptionBotTurnStartStrategy(unittest.IsolatedAsyncioTestCase): should_start = None @strategy.event_handler("on_bot_turn_started") - async def on_bot_turn_started(strategy, enable_user_speaking_frames): + async def on_bot_turn_started(strategy, params): nonlocal should_start should_start = True @@ -170,7 +170,7 @@ class TestTranscriptionBotTurnStartStrategy(unittest.IsolatedAsyncioTestCase): should_start = None @strategy.event_handler("on_bot_turn_started") - async def on_bot_turn_started(strategy, enable_user_speaking_frames): + async def on_bot_turn_started(strategy, params): nonlocal should_start should_start = True @@ -212,7 +212,7 @@ class TestTranscriptionBotTurnStartStrategy(unittest.IsolatedAsyncioTestCase): should_start = None @strategy.event_handler("on_bot_turn_started") - async def on_bot_turn_started(strategy, enable_user_speaking_frames): + async def on_bot_turn_started(strategy, params): nonlocal should_start should_start = True @@ -242,7 +242,7 @@ class TestTranscriptionBotTurnStartStrategy(unittest.IsolatedAsyncioTestCase): should_start = None @strategy.event_handler("on_bot_turn_started") - async def on_bot_turn_started(strategy, enable_user_speaking_frames): + async def on_bot_turn_started(strategy, params): nonlocal should_start should_start = True @@ -278,7 +278,7 @@ class TestTranscriptionBotTurnStartStrategy(unittest.IsolatedAsyncioTestCase): should_start = None @strategy.event_handler("on_bot_turn_started") - async def on_bot_turn_started(strategy, enable_user_speaking_frames): + async def on_bot_turn_started(strategy, params): nonlocal should_start should_start = True @@ -316,7 +316,7 @@ class TestTranscriptionBotTurnStartStrategy(unittest.IsolatedAsyncioTestCase): should_start = None @strategy.event_handler("on_bot_turn_started") - async def on_bot_turn_started(strategy, enable_user_speaking_frames): + async def on_bot_turn_started(strategy, params): nonlocal should_start should_start = True @@ -350,7 +350,7 @@ class TestTranscriptionBotTurnStartStrategy(unittest.IsolatedAsyncioTestCase): should_start = None @strategy.event_handler("on_bot_turn_started") - async def on_bot_turn_started(strategy, enable_user_speaking_frames): + async def on_bot_turn_started(strategy, params): nonlocal should_start should_start = True @@ -395,7 +395,7 @@ class TestTranscriptionBotTurnStartStrategy(unittest.IsolatedAsyncioTestCase): should_start = None @strategy.event_handler("on_bot_turn_started") - async def on_bot_turn_started(strategy, enable_user_speaking_frames): + async def on_bot_turn_started(strategy, params): nonlocal should_start should_start = True @@ -415,7 +415,7 @@ class TestTranscriptionBotTurnStartStrategy(unittest.IsolatedAsyncioTestCase): should_start = None @strategy.event_handler("on_bot_turn_started") - async def on_bot_turn_started(strategy, enable_user_speaking_frames): + async def on_bot_turn_started(strategy, params): nonlocal should_start should_start = True @@ -440,7 +440,7 @@ class TestTranscriptionBotTurnStartStrategy(unittest.IsolatedAsyncioTestCase): should_start = None @strategy.event_handler("on_bot_turn_started") - async def on_bot_turn_started(strategy, enable_user_speaking_frames): + async def on_bot_turn_started(strategy, params): nonlocal should_start should_start = True @@ -484,7 +484,7 @@ class TestExternalBotTurnStartStrategy(unittest.IsolatedAsyncioTestCase): should_start = None @strategy.event_handler("on_bot_turn_started") - async def on_bot_turn_started(strategy, enable_user_speaking_frames): + async def on_bot_turn_started(strategy, params): nonlocal should_start should_start = True diff --git a/tests/test_user_turn_start_strategy.py b/tests/test_user_turn_start_strategy.py index 1212e075a..dcc9c0731 100644 --- a/tests/test_user_turn_start_strategy.py +++ b/tests/test_user_turn_start_strategy.py @@ -29,7 +29,7 @@ class TestMinWordsInterruptionStrategy(unittest.IsolatedAsyncioTestCase): should_start = None @strategy.event_handler("on_user_turn_started") - async def on_user_turn_started(strategy, enable_user_speaking_frames): + async def on_user_turn_started(strategy, params): nonlocal should_start should_start = True @@ -61,7 +61,7 @@ class TestMinWordsInterruptionStrategy(unittest.IsolatedAsyncioTestCase): should_start = None @strategy.event_handler("on_user_turn_started") - async def on_user_turn_started(strategy, enable_user_speaking_frames): + async def on_user_turn_started(strategy, params): nonlocal should_start should_start = True @@ -83,7 +83,7 @@ class TestMinWordsInterruptionStrategy(unittest.IsolatedAsyncioTestCase): should_start = None @strategy.event_handler("on_user_turn_started") - async def on_user_turn_started(strategy, enable_user_speaking_frames): + async def on_user_turn_started(strategy, params): nonlocal should_start should_start = True @@ -104,7 +104,7 @@ class TestMinWordsInterruptionStrategy(unittest.IsolatedAsyncioTestCase): should_start = None @strategy.event_handler("on_user_turn_started") - async def on_user_turn_started(strategy, enable_user_speaking_frames): + async def on_user_turn_started(strategy, params): nonlocal should_start should_start = True @@ -117,7 +117,7 @@ class TestMinWordsInterruptionStrategy(unittest.IsolatedAsyncioTestCase): should_start = None @strategy.event_handler("on_user_turn_started") - async def on_user_turn_started(strategy, enable_user_speaking_frames): + async def on_user_turn_started(strategy, params): nonlocal should_start should_start = True @@ -134,7 +134,7 @@ class TestVADUserTurnStartStrategy(unittest.IsolatedAsyncioTestCase): should_start = None @strategy.event_handler("on_user_turn_started") - async def on_user_turn_started(strategy, enable_user_speaking_frames): + async def on_user_turn_started(strategy, params): nonlocal should_start should_start = True @@ -152,14 +152,11 @@ class TestTranscriptionUserTurnStartStrategy(unittest.IsolatedAsyncioTestCase): should_start = None @strategy.event_handler("on_user_turn_started") - async def on_user_turn_started(strategy, enable_user_speaking_frames): + async def on_user_turn_started(strategy, params): nonlocal should_start should_start = True - await strategy.process_frame(TranscriptionFrame(text="Hello!", user_id="", timestamp="now")) - self.assertFalse(should_start) - - await strategy.process_frame(BotStartedSpeakingFrame()) + await strategy.process_frame(VADUserStartedSpeakingFrame()) self.assertFalse(should_start) await strategy.process_frame(TranscriptionFrame(text="Hello!", user_id="", timestamp="now")) @@ -173,7 +170,7 @@ class TestExternalUserTurnStartStrategy(unittest.IsolatedAsyncioTestCase): should_start = None @strategy.event_handler("on_user_turn_started") - async def on_user_turn_started(strategy, enable_user_speaking_frames): + async def on_user_turn_started(strategy, params): nonlocal should_start should_start = True