From 84f26ac1cabf63ce0527e7899ed39ed529261688 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Sun, 20 Oct 2024 16:27:17 -0700 Subject: [PATCH] processors: introduce input queues Frame processors can now decide if they should continue processing frames or not, and if so also decide when to continue processing frames. For example, asynchronous TTS services will stop processing frames until they have generated all the audio for an LLM response. --- CHANGELOG.md | 11 +++++ src/pipecat/processors/frame_processor.py | 58 ++++++++++++++++++++++- src/pipecat/services/ai_services.py | 3 ++ src/pipecat/services/cartesia.py | 3 ++ src/pipecat/services/playht.py | 8 +++- 5 files changed, 80 insertions(+), 3 deletions(-) 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