processors(realtime-ai): add support for interrupting the bot

CHANGELOG.md

@@ -11,6 +11,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Added `RealtimeAIProcessor`...
 
+- Added `BotInterruptionFrame` which allows interrupting the bot while talking.
+
 - Added `LLMMessagesAppendFrame` which allows appending messages to the current
   LLM context.
 

examples/realtime-ai/bot.py

@@ -11,7 +11,12 @@ import os
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.task import PipelineParams, PipelineTask
 from pipecat.pipeline.runner import PipelineRunner
-from pipecat.processors.frameworks.realtimeai import RealtimeAIConfig, RealtimeAILLMConfig, RealtimeAIProcessor, RealtimeAISetup, RealtimeAITTSConfig
+from pipecat.processors.frameworks.realtimeai import (
+    RealtimeAIConfig,
+    RealtimeAILLMConfig,
+    RealtimeAIProcessor,
+    RealtimeAISetup,
+    RealtimeAITTSConfig)
 from pipecat.transports.services.daily import DailyParams, DailyTransport
 from pipecat.vad.silero import SileroVADAnalyzer
 

src/pipecat/frames/frames.py

@@ -268,6 +268,16 @@ class StopInterruptionFrame(SystemFrame):
     pass
 
 
+@dataclass
+class BotInterruptionFrame(SystemFrame):
+    """Emitted by when the bot should be interrupted. This will mainly cause the
+    same actions as if the user interrupted except that the
+    UserStartedSpeakingFrame and UserStoppedSpeakingFrame won't be generated.
+
+    """
+    pass
+
+
 @dataclass
 class BotSpeakingFrame(SystemFrame):
     """Emitted by transport outputs while the bot is still speaking. This can be

src/pipecat/processors/frameworks/realtimeai.py

@@ -4,18 +4,21 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
+import asyncio
 import dataclasses
 
 from typing import List, Literal, Optional, Type
 from pydantic import BaseModel, ValidationError
 
 from pipecat.frames.frames import (
+    BotInterruptionFrame,
     Frame,
     InterimTranscriptionFrame,
     LLMMessagesAppendFrame,
     LLMMessagesUpdateFrame,
     LLMModelUpdateFrame,
     StartFrame,
+    SystemFrame,
     TTSSpeakFrame,
     TTSVoiceUpdateFrame,
     TranscriptionFrame,
@@ -67,6 +70,7 @@ class RealtimeAILLMMessageData(BaseModel):
 
 class RealtimeAITTSMessageData(BaseModel):
     text: str
+    interrupt: Optional[bool] = False
 
 
 class RealtimeAIMessageData(BaseModel):
@@ -152,24 +156,49 @@ class RealtimeAIProcessor(FrameProcessor):
         self._tts: FrameProcessor | None = None
         self._pipeline: FrameProcessor | None = None
 
+        self._frame_handler_task = self.get_event_loop().create_task(self._frame_handler())
+        self._frame_queue = asyncio.Queue()
+
     async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, SystemFrame):
+            await self.push_frame(frame, direction)
+        else:
+            await self._frame_queue.put((frame, direction))
+
+        if isinstance(frame, StartFrame):
+            self._start_frame = frame
+            await self._handle_setup(self._setup)
+
+    async def cleanup(self):
+        self._frame_handler_task.cancel()
+        await self._frame_handler_task
+
+    async def _frame_handler(self):
+        while True:
+            try:
+                (frame, direction) = await self._frame_queue.get()
+                await self._handle_frame(frame, direction)
+            except asyncio.CancelledError:
+                break
+
+    async def _handle_frame(self, frame: Frame, direction: FrameDirection):
         if isinstance(frame, TransportMessageFrame):
             await self._handle_message(frame)
         else:
             await self.push_frame(frame, direction)
 
-        if isinstance(frame, StartFrame):
+        if isinstance(frame, TranscriptionFrame) or isinstance(frame, InterimTranscriptionFrame):
-            self._start_frame = frame
-            await self._handle_setup(self._setup)
-        elif isinstance(frame, TranscriptionFrame) or isinstance(frame, InterimTranscriptionFrame):
             await self._handle_transcriptions(frame)
         elif isinstance(frame, UserStartedSpeakingFrame) or isinstance(frame, UserStoppedSpeakingFrame):
             await self._handle_interruptions(frame)
 
-    # TODO(aleix): Once we add support for using custom piplines, the STTs will
-    # be in the pipeline after this processor. This means the STT will have to
-    # push transcriptions upstream as well.
     async def _handle_transcriptions(self, frame: Frame):
+        # TODO(aleix): Once we add support for using custom piplines, the STTs will
+        # be in the pipeline after this processor. This means the STT will have to
+        # push transcriptions upstream as well.
+
         message = None
         if isinstance(frame, TranscriptionFrame):
             message = RealtimeAITranscriptionMessage(
@@ -221,6 +250,8 @@ class RealtimeAIProcessor(FrameProcessor):
                     await self._handle_llm_update_context(message.data.llm)
                 case "tts-speak":
                     await self._handle_tts_speak(message.data.tts)
+                case "tts-interrupt":
+                    await self._handle_tts_interrupt()
 
             # Send a message to indicate we successfully executed the command.
             await self._send_response(message.type, True)
@@ -300,9 +331,14 @@ class RealtimeAIProcessor(FrameProcessor):
 
     async def _handle_tts_speak(self, data: RealtimeAITTSMessageData):
         if data and data.text:
+            if data.interrupt:
+                await self._handle_tts_interrupt()
             frame = TTSSpeakFrame(text=data.text)
             await self.push_frame(frame)
 
+    async def _handle_tts_interrupt(self):
+        await self.push_frame(BotInterruptionFrame(), FrameDirection.UPSTREAM)
+
     async def _send_response(self, type: str, success: bool, error: str | None = None):
         # TODO(aleix): This is a bit hacky, but we might get invalid
         # configuration or something might going wrong during setup and we would

src/pipecat/transports/base_input.py

@@ -11,6 +11,7 @@ from concurrent.futures import ThreadPoolExecutor
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.frames.frames import (
     AudioRawFrame,
+    BotInterruptionFrame,
     CancelFrame,
     StartFrame,
     EndFrame,
@@ -78,6 +79,8 @@ class BaseInputTransport(FrameProcessor):
         elif isinstance(frame, EndFrame):
             await self._internal_push_frame(frame, direction)
             await self.stop()
+        elif isinstance(frame, BotInterruptionFrame):
+            await self._handle_interruptions(frame, False)
         else:
             await self._internal_push_frame(frame, direction)
 
@@ -108,24 +111,35 @@ class BaseInputTransport(FrameProcessor):
     # Handle interruptions
     #
 
-    async def _handle_interruptions(self, frame: Frame):
+    async def _start_interruption(self):
+        # Cancel the task. This will stop pushing frames downstream.
+        self._push_frame_task.cancel()
+        await self._push_frame_task
+        # Push an out-of-band frame (i.e. not using the ordered push
+        # frame task) to stop everything, specially at the output
+        # transport.
+        await self.push_frame(StartInterruptionFrame())
+        # Create a new queue and task.
+        self._create_push_task()
+
+    async def _stop_interruption(self):
+        await self.push_frame(StopInterruptionFrame())
+
+    async def _handle_interruptions(self, frame: Frame, push_frame: bool):
         if self.interruptions_allowed:
             # Make sure we notify about interruptions quickly out-of-band
-            if isinstance(frame, UserStartedSpeakingFrame):
+            if isinstance(frame, BotInterruptionFrame):
+                logger.debug("Bot interruption")
+                await self._start_interruption()
+            elif isinstance(frame, UserStartedSpeakingFrame):
                 logger.debug("User started speaking")
-                # Cancel the task. This will stop pushing frames downstream.
+                await self._start_interruption()
-                self._push_frame_task.cancel()
-                await self._push_frame_task
-                # Push an out-of-band frame (i.e. not using the ordered push
-                # frame task) to stop everything, specially at the output
-                # transport.
-                await self.push_frame(StartInterruptionFrame())
-                # Create a new queue and task.
-                self._create_push_task()
             elif isinstance(frame, UserStoppedSpeakingFrame):
                 logger.debug("User stopped speaking")
-                await self.push_frame(StopInterruptionFrame())
+                await self._stop_interruption()
-        await self._internal_push_frame(frame)
+
+        if push_frame:
+            await self._internal_push_frame(frame)
 
     #
     # Audio input
@@ -149,7 +163,7 @@ class BaseInputTransport(FrameProcessor):
                 frame = UserStoppedSpeakingFrame()
 
             if frame:
-                await self._handle_interruptions(frame)
+                await self._handle_interruptions(frame, True)
 
             vad_state = new_vad_state
         return vad_state