Merge pull request #3374 from pipecat-ai/filipi/external_turn_controllers_interruptions

External turn controllers improvements

changelog/3374.added.md

@@ -0,0 +1 @@
+- Added `should_interrupt` property to `DeepgramFluxSTTService`, `DeepgramSTTService`, and `SpeechmaticsSTTService` to configure whether the bot should be interrupted when the external service detects user speech.

src/pipecat/services/deepgram/flux/stt.py

@@ -116,6 +116,7 @@ class DeepgramFluxSTTService(WebsocketSTTService):
         model: str = "flux-general-en",
         flux_encoding: str = "linear16",
         params: Optional[InputParams] = None,
+        should_interrupt: bool = True,
         **kwargs,
     ):
         """Initialize the Deepgram Flux STT service.
@@ -129,6 +130,7 @@ class DeepgramFluxSTTService(WebsocketSTTService):
                 Raw signed little-endian 16-bit PCM encoding.
             params: InputParams instance containing detailed API configuration options.
                 If None, default parameters will be used.
+            should_interrupt: Determine whether the bot should be interrupted when Flux detects that the user is speaking.
             **kwargs: Additional arguments passed to the parent WebsocketSTTService class.
 
         Examples:
@@ -166,6 +168,7 @@ class DeepgramFluxSTTService(WebsocketSTTService):
         self._url = url
         self._model = model
         self._params = params or DeepgramFluxSTTService.InputParams()
+        self._should_interrupt = should_interrupt
         self._flux_encoding = flux_encoding
         # This is the currently only supported language
         self._language = Language.EN
@@ -592,7 +595,8 @@ class DeepgramFluxSTTService(WebsocketSTTService):
         logger.debug("User started speaking")
         self._user_is_speaking = True
         await self.broadcast_frame(UserStartedSpeakingFrame)
-        await self.push_interruption_task_frame_and_wait()
+        if self._should_interrupt:
+            await self.push_interruption_task_frame_and_wait()
         await self.start_metrics()
         await self._call_event_handler("on_start_of_turn", transcript)
         if transcript:

src/pipecat/services/deepgram/stt.py

@@ -61,6 +61,7 @@ class DeepgramSTTService(STTService):
         sample_rate: Optional[int] = None,
         live_options: Optional[LiveOptions] = None,
         addons: Optional[Dict] = None,
+        should_interrupt: bool = True,
         **kwargs,
     ):
         """Initialize the Deepgram STT service.
@@ -76,6 +77,7 @@ class DeepgramSTTService(STTService):
             sample_rate: Audio sample rate. If None, uses default or live_options value.
             live_options: Deepgram LiveOptions for detailed configuration.
             addons: Additional Deepgram features to enable.
+            should_interrupt: Determine whether the bot should be interrupted when Deepgram VAD events are enabled and the system detects that the user is speaking.
             **kwargs: Additional arguments passed to the parent STTService.
         """
         sample_rate = sample_rate or (live_options.sample_rate if live_options else None)
@@ -119,6 +121,7 @@ class DeepgramSTTService(STTService):
         self.set_model_name(merged_options["model"])
         self._settings = merged_options
         self._addons = addons
+        self._should_interrupt = should_interrupt
 
         self._client = DeepgramClient(
             api_key,
@@ -274,7 +277,8 @@ class DeepgramSTTService(STTService):
         await self.start_metrics()
         await self._call_event_handler("on_speech_started", *args, **kwargs)
         await self.broadcast_frame(UserStartedSpeakingFrame)
-        await self.push_interruption_task_frame_and_wait()
+        if self._should_interrupt:
+            await self.push_interruption_task_frame_and_wait()
 
     async def _on_utterance_end(self, *args, **kwargs):
         await self._call_event_handler("on_utterance_end", *args, **kwargs)

src/pipecat/services/speechmatics/stt.py

@@ -287,6 +287,7 @@ class SpeechmaticsSTTService(STTService):
         base_url: str | None = None,
         sample_rate: int | None = None,
         params: InputParams | None = None,
+        should_interrupt: bool = True,
         **kwargs,
     ):
         """Initialize the Speechmatics STT service.
@@ -298,6 +299,7 @@ class SpeechmaticsSTTService(STTService):
                 or defaults to `wss://eu2.rt.speechmatics.com/v2`.
             sample_rate: Optional audio sample rate in Hz.
             params: Optional[InputParams]: Input parameters for the service.
+            should_interrupt: Determine whether the bot should be interrupted when Speechmatics turn_detection_mode is configured to detect user speech.
             **kwargs: Additional arguments passed to STTService.
         """
         super().__init__(sample_rate=sample_rate, **kwargs)
@@ -316,6 +318,7 @@ class SpeechmaticsSTTService(STTService):
 
         # Default params
         params = params or SpeechmaticsSTTService.InputParams()
+        self._should_interrupt = should_interrupt
 
         # Deprecation check
         self._check_deprecated_args(kwargs, params)
@@ -623,7 +626,8 @@ class SpeechmaticsSTTService(STTService):
         logger.debug(f"{self} StartOfTurn received")
         # await self.start_processing_metrics()
         await self.broadcast_frame(UserStartedSpeakingFrame)
-        await self.push_interruption_task_frame_and_wait()
+        if self._should_interrupt:
+            await self.push_interruption_task_frame_and_wait()
 
     async def _handle_end_of_turn(self, message: dict[str, Any]) -> None:
         """Handle EndOfTurn events.