diff --git a/CHANGELOG.md b/CHANGELOG.md index 01a0d3fa9..cb49ae499 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- There's now an input queue for each frame processor. When you call + `FrameProcessor.push_frame()` this will internally call + `FrameProcessor.queue_frame()` on the next processor (upstream or + downstream). Then, the queue frames will get processed. With this input queue + it is now possible for FrameProcessors to block processing more frames via a + list of blocking frames. For example, some TTS services now block processing + more frames if they see a `TTSSpeakFrame` or a `LLMFullResponseEndFrame` until + the bot has stopped speaking. This makes sure we don't mix audio from + different sentences. The way to resume processing frames is by calling + `FrameProcessor.resume_processing_frames()`. + - Added audio filter `NoisereduceFilter`. - Introduce input transport audio filters (`BaseAudioFilter`). Audio filters can diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index f458f43ff..206e7adc1 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -36,6 +36,7 @@ class FrameProcessor: self, *, name: str | None = None, + block_on_frames: tuple = (), metrics: FrameProcessorMetrics | None = None, loop: asyncio.AbstractEventLoop | None = None, **kwargs, @@ -62,6 +63,13 @@ class FrameProcessor: self._metrics = metrics or FrameProcessorMetrics() self._metrics.set_processor_name(self.name) + # Processors have an input queue. The input queue can be processed + # immediately (default) or it can be processed when the processor asks + # to process a new frame (via process_next_frame()). + self._block_on_frames = block_on_frames + self._should_queue_input_frames = False + self.__create_input_task() + # Every processor in Pipecat should only output frames from a single # task. This avoid problems like audio overlapping. System frames are # the exception to this rule. This create this task. @@ -145,6 +153,19 @@ class FrameProcessor: def get_clock(self) -> BaseClock: return self._clock + async def queue_frame( + self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM + ): + if isinstance(frame, SystemFrame): + # We don't want to queue system frames. + await self.process_frame(frame, direction) + else: + # We queue everything else. + await self.__input_queue.put((frame, direction)) + + async def resume_processing_frames(self): + self.__input_event.set() + async def process_frame(self, frame: Frame, direction: FrameDirection): if isinstance(frame, StartFrame): self._clock = frame.clock @@ -204,13 +225,46 @@ class FrameProcessor: try: if direction == FrameDirection.DOWNSTREAM and self._next: logger.trace(f"Pushing {frame} from {self} to {self._next}") - await self._next.process_frame(frame, direction) + await self._next.queue_frame(frame, direction) elif direction == FrameDirection.UPSTREAM and self._prev: logger.trace(f"Pushing {frame} upstream from {self} to {self._prev}") - await self._prev.process_frame(frame, direction) + await self._prev.queue_frame(frame, direction) except Exception as e: logger.exception(f"Uncaught exception in {self}: {e}") + def __create_input_task(self): + self.__input_queue = asyncio.Queue() + self.__input_frame_task = self.get_event_loop().create_task( + self.__input_frame_task_handler() + ) + self.__input_event = asyncio.Event() + + async def __input_frame_task_handler(self): + running = True + should_block_frames = False + while running: + try: + if should_block_frames: + await self.__input_event.wait() + self.__input_event.clear() + should_block_frames = False + + (frame, direction) = await self.__input_queue.get() + + # Process the frame. + await self.process_frame(frame, direction) + + # Check if we should block incoming frames from now on (until we + # resume processing). + if isinstance(frame, self._block_on_frames): + should_block_frames = True + + running = not isinstance(frame, EndFrame) + + self.__input_queue.task_done() + except asyncio.CancelledError: + break + def __create_push_task(self): self.__push_queue = asyncio.Queue() self.__push_frame_task = self.get_event_loop().create_task(self.__push_frame_task_handler()) diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index 62d52f76e..63100f908 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -15,6 +15,7 @@ from loguru import logger from pipecat.audio.utils import calculate_audio_volume, exp_smoothing from pipecat.frames.frames import ( AudioRawFrame, + BotStoppedSpeakingFrame, CancelFrame, EndFrame, ErrorFrame, @@ -297,6 +298,8 @@ class TTSService(AIService): await self._process_text_frame(frame) elif isinstance(frame, StartInterruptionFrame): await self._handle_interruption(frame, direction) + elif isinstance(frame, BotStoppedSpeakingFrame): + await self.resume_processing_frames() elif isinstance(frame, (LLMFullResponseEndFrame, EndFrame)): sentence = self._current_sentence self._current_sentence = "" diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index 35f91b4cd..4e9918950 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -18,9 +18,11 @@ from pipecat.frames.frames import ( EndFrame, ErrorFrame, Frame, + LLMFullResponseEndFrame, StartFrame, StartInterruptionFrame, TTSAudioRawFrame, + TTSSpeakFrame, TTSStartedFrame, TTSStoppedFrame, ) @@ -100,6 +102,7 @@ class CartesiaTTSService(WordTTSService): aggregate_sentences=True, push_text_frames=False, sample_rate=sample_rate, + block_on_frames=(TTSSpeakFrame, LLMFullResponseEndFrame), **kwargs, ) diff --git a/src/pipecat/services/playht.py b/src/pipecat/services/playht.py index 4837c058f..64179b587 100644 --- a/src/pipecat/services/playht.py +++ b/src/pipecat/services/playht.py @@ -21,9 +21,11 @@ from pipecat.frames.frames import ( EndFrame, ErrorFrame, Frame, + LLMFullResponseEndFrame, StartFrame, StartInterruptionFrame, TTSAudioRawFrame, + TTSSpeakFrame, TTSStartedFrame, TTSStoppedFrame, ) @@ -121,7 +123,11 @@ class PlayHTTTSService(TTSService): params: InputParams = InputParams(), **kwargs, ): - super().__init__(sample_rate=sample_rate, **kwargs) + super().__init__( + sample_rate=sample_rate, + block_on_frames=(TTSSpeakFrame, LLMFullResponseEndFrame), + **kwargs, + ) self._api_key = api_key self._user_id = user_id