From b5fe1c9cd87aab562f78f3aece0751022063b6c4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 22 Dec 2025 16:19:25 -0800 Subject: [PATCH 1/3] fix old interruption/speaking strategies docstrings --- .../audio/interruptions/min_words_interruption_strategy.py | 3 ++- src/pipecat/processors/aggregators/llm_response_universal.py | 2 +- src/pipecat/transports/base_transport.py | 4 ++-- .../turns/bot/transcription_bot_turn_start_strategy.py | 2 +- .../turns/user/transcription_user_turn_start_strategy.py | 4 ++-- 5 files changed, 8 insertions(+), 7 deletions(-) diff --git a/src/pipecat/audio/interruptions/min_words_interruption_strategy.py b/src/pipecat/audio/interruptions/min_words_interruption_strategy.py index f89026cc4..1ccdbd49e 100644 --- a/src/pipecat/audio/interruptions/min_words_interruption_strategy.py +++ b/src/pipecat/audio/interruptions/min_words_interruption_strategy.py @@ -42,7 +42,8 @@ class MinWordsInterruptionStrategy(BaseInterruptionStrategy): warnings.simplefilter("always") warnings.warn( "'pipecat.audio.interruptions' is deprecated. " - "Use the new interruption and speaking strategies.", + "Use `pipecat.turns.user.MinWordsUserTurnStartStrategy` with `PipelineTask`'s " + "new `turn_start_strategies` parameter instead.", DeprecationWarning, ) diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index 4e918f440..8be62375b 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/src/pipecat/processors/aggregators/llm_response_universal.py @@ -266,7 +266,7 @@ class LLMUserAggregator(LLMContextAggregator): await self._cleanup() async def reset(self): - """Reset the aggregation state and interruption strategies.""" + """Reset the aggregation state and turn start strategies.""" await super().reset() if self.turn_start_strategies and self.turn_start_strategies.user: diff --git a/src/pipecat/transports/base_transport.py b/src/pipecat/transports/base_transport.py index 27122866c..2b99b75e4 100644 --- a/src/pipecat/transports/base_transport.py +++ b/src/pipecat/transports/base_transport.py @@ -115,8 +115,8 @@ class TransportParams(BaseModel): turn_analyzer: Turn-taking analyzer instance for conversation management. .. deprecated:: 0.0.99 - The `turn_analyzer` parameter is deprecated, use speaking strategies instead. - + The `turn_analyzer` parameter is deprecated, use `PipelineTask`'s + new `turn_start_strategies` parameter instead. """ model_config = ConfigDict(arbitrary_types_allowed=True) 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 b8cfd70e6..60a6f76fc 100644 --- a/src/pipecat/turns/bot/transcription_bot_turn_start_strategy.py +++ b/src/pipecat/turns/bot/transcription_bot_turn_start_strategy.py @@ -4,7 +4,7 @@ # SPDX-License-Identifier: BSD 2-Clause License # -"""Transcription time-based speaking strategy.""" +"""Transcription time-based bot turn start strategy.""" import asyncio from typing import Optional 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 6a5a7c907..7172014d4 100644 --- a/src/pipecat/turns/user/transcription_user_turn_start_strategy.py +++ b/src/pipecat/turns/user/transcription_user_turn_start_strategy.py @@ -21,12 +21,12 @@ class TranscriptionUserTurnStartStrategy(BaseUserTurnStartStrategy): """ def __init__(self): - """Initialize the base interruption strategy.""" + """Initialize transcription-based user turn start strategy.""" super().__init__() self._bot_speaking = False async def reset(self): - """Reset the interruption strategy.""" + """Reset the strategy to its initial state.""" await super().reset() self._bot_speaking = False From 480a9d092c40f8077cca83e1c5bf839a95ed4509 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 22 Dec 2025 16:20:27 -0800 Subject: [PATCH 2/3] TurnAnalyzerBotTurnStartStrategy: make sure to use turn analyzer state --- .../turn_analyzer_bot_turn_start_strategy.py | 36 ++++++++++++------- 1 file changed, 24 insertions(+), 12 deletions(-) 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 b96b347a4..3cf5a18c4 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 @@ -46,6 +46,7 @@ class TurnAnalyzerBotTurnStartStrategy(BaseBotTurnStartStrategy): self._turn_analyzer = turn_analyzer self._timeout = timeout self._text = "" + self._turn_complete = False self._vad_user_speaking = False self._event = asyncio.Event() self._task: Optional[asyncio.Task] = None @@ -54,8 +55,9 @@ class TurnAnalyzerBotTurnStartStrategy(BaseBotTurnStartStrategy): """Reset the strategy to its initial state.""" await super().reset() self._text = "" + self._turn_complete = False self._vad_user_speaking = False - self._event.set() + self._event.clear() async def setup(self, task_manager: BaseTaskManager): """Initialize the strategy with the given task manager. @@ -102,37 +104,44 @@ class TurnAnalyzerBotTurnStartStrategy(BaseBotTurnStartStrategy): 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) + + # If at this point the model says the turn is complete it will be due to + # a timeout, so we mark turn as complete and we trigger the bot turn. + if state == EndOfTurnState.COMPLETE: + self._turn_complete = True + await self._maybe_trigger_bot_turn_started() async def _handle_vad_user_started_speaking(self, _: VADUserStartedSpeakingFrame): """Handle when the VAD indicates the user is speaking.""" + self._turn_complete = False 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) + + # The user stopped speaking and the turn is complete, we now need to + # wait for transcriptions. + self._turn_complete = state == EndOfTurnState.COMPLETE + + # Reset transcription timeout. + self._event.set() async def _handle_transcription(self, frame: TranscriptionFrame): """Handle user transcription.""" # We don't really care about the content. self._text = frame.text + # Reset transcription timeout. self._event.set() async def _handle_interim_transcription(self, frame: InterimTranscriptionFrame): """Handle user interim transcription.""" + # Reset transcription timeout. 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: @@ -151,5 +160,8 @@ class TurnAnalyzerBotTurnStartStrategy(BaseBotTurnStartStrategy): 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() + await self._maybe_trigger_bot_turn_started() + + async def _maybe_trigger_bot_turn_started(self): + if self._text and self._turn_complete: + await self.trigger_bot_turn_started() From 4017bfa769ac293e51623b02ea57aa3915ef9eda Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 22 Dec 2025 21:44:20 -0800 Subject: [PATCH 3/3] LLMUserAggregator: improve turn_analyzer warning --- .../aggregators/llm_response_universal.py | 18 ++++++++++++++---- 1 file changed, 14 insertions(+), 4 deletions(-) diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index 8be62375b..86bbd9905 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/src/pipecat/processors/aggregators/llm_response_universal.py @@ -390,10 +390,20 @@ class LLMUserAggregator(LLMContextAggregator): if not frame.turn_params: return - logger.error( - f"{self}: turn_analyzer in base input transport is deprecated and " - "might result in unexpected behavior. Use PipelineTask's turn_start_strategies with " - "TurnAnalyzerBotTurnStartStrategy instead." + logger.warning( + f"{self}: `turn_analyzer` in base input transport is deprecated and " + "might result in unexpected behavior. Use `PipelineTask`'s `turn_start_strategies` with " + "`TurnAnalyzerBotTurnStartStrategy` instead.:\n\n" + " task = PipelineTask(\n" + " pipeline,\n" + " params=PipelineParams(\n" + " ...,\n" + " turn_start_strategies=TurnStartStrategies(\n" + " bot=[TurnAnalyzerBotTurnStartStrategy(turn_analyzer=LocalSmartTurnAnalyzerV3())]\n" + " ),\n" + " ),\n" + " ...,\n" + " )" ) async def _handle_transcription(self, frame: TranscriptionFrame):