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..86bbd9905 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: @@ -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): 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/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() 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