diff --git a/CHANGELOG.md b/CHANGELOG.md index 3f6fcbc1e..a2512d378 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/examples/realtime-ai/bot.py b/examples/realtime-ai/bot.py index 29fb020ea..a2f7320c0 100644 --- a/examples/realtime-ai/bot.py +++ b/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 diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 09c475702..46e2408bc 100644 --- a/src/pipecat/frames/frames.py +++ b/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 diff --git a/src/pipecat/processors/frameworks/realtimeai.py b/src/pipecat/processors/frameworks/realtimeai.py index 4eed08154..3332a0ef7 100644 --- a/src/pipecat/processors/frameworks/realtimeai.py +++ b/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): - self._start_frame = frame - await self._handle_setup(self._setup) - elif isinstance(frame, TranscriptionFrame) or isinstance(frame, InterimTranscriptionFrame): + if 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 diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 828dc3e40..f87772c01 100644 --- a/src/pipecat/transports/base_input.py +++ b/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. - 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() + await self._start_interruption() elif isinstance(frame, UserStoppedSpeakingFrame): logger.debug("User stopped speaking") - await self.push_frame(StopInterruptionFrame()) - await self._internal_push_frame(frame) + await self._stop_interruption() + + 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