Merge pull request #3286 from pipecat-ai/aleix/improve-turn-analyzer-bot-turn-start-strategy

improve turn analyzer bot turn start strategy

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

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

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)

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

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

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