From 865768039b70cd630d1195450dbc65491daa504f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 6 Nov 2024 14:20:25 -0800 Subject: [PATCH] processors: remove block_on_frames and add pause_processing_frames() instead --- CHANGELOG.md | 9 +++------ src/pipecat/processors/frame_processor.py | 20 ++++++++------------ src/pipecat/services/ai_services.py | 3 --- src/pipecat/services/cartesia.py | 15 ++++++++++++++- src/pipecat/services/playht.py | 15 ++++++++++++++- 5 files changed, 39 insertions(+), 23 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a810258d3..b6761a5b7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,12 +14,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 `FrameProcessor.queue_frame()` on the next processor (upstream or downstream) and the frame will be internally queued (except system frames). Then, the queued frames will get processed. With this input queue it is also 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()`. + for FrameProcessors to block processing more frames by calling + `FrameProcessor.pause_processing_frames()`. The way to resume processing + frames is by calling `FrameProcessor.resume_processing_frames()`. - Added audio filter `NoisereduceFilter`. diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 04f53f127..9f614ba85 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -37,7 +37,6 @@ class FrameProcessor: self, *, name: str | None = None, - block_on_frames: tuple = (), metrics: FrameProcessorMetrics | None = None, loop: asyncio.AbstractEventLoop | None = None, **kwargs, @@ -65,10 +64,10 @@ class FrameProcessor: self._metrics.set_processor_name(self.name) # Processors have an input queue. The input queue will be processed - # immediately (default) or it will block if one of the `block_on_frames` - # is found. To resume processing frames we need to call + # immediately (default) or it will block if `pause_processing_frames()` + # is called. To resume processing frames we need to call # `resume_processing_frames()`. - self._block_on_frames = block_on_frames + self.__should_block_frames = False self.__create_input_task() # Every processor in Pipecat should only output frames from a single @@ -170,8 +169,12 @@ class FrameProcessor: # We queue everything else. await self.__input_queue.put((frame, direction, callback)) + async def pause_processing_frames(self): + self.__should_block_frames = True + async def resume_processing_frames(self): self.__input_event.set() + self.__should_block_frames = False async def process_frame(self, frame: Frame, direction: FrameDirection): if isinstance(frame, StartFrame): @@ -257,13 +260,11 @@ class FrameProcessor: async def __input_frame_task_handler(self): running = True - should_block_frames = False while running: try: - if should_block_frames: + if self.__should_block_frames: await self.__input_event.wait() self.__input_event.clear() - should_block_frames = False (frame, direction, callback) = await self.__input_queue.get() @@ -274,11 +275,6 @@ class FrameProcessor: if callback: await callback(self, 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() diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index fd61dd1a7..023cabee8 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -15,7 +15,6 @@ from loguru import logger from pipecat.audio.utils import calculate_audio_volume, exp_smoothing from pipecat.frames.frames import ( AudioRawFrame, - BotStoppedSpeakingFrame, CancelFrame, EndFrame, ErrorFrame, @@ -294,8 +293,6 @@ 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 4e9918950..2449fede6 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -14,6 +14,7 @@ from loguru import logger from pydantic.main import BaseModel from pipecat.frames.frames import ( + BotStoppedSpeakingFrame, CancelFrame, EndFrame, ErrorFrame, @@ -102,7 +103,6 @@ class CartesiaTTSService(WordTTSService): aggregate_sentences=True, push_text_frames=False, sample_rate=sample_rate, - block_on_frames=(TTSSpeakFrame, LLMFullResponseEndFrame), **kwargs, ) @@ -261,6 +261,19 @@ class CartesiaTTSService(WordTTSService): except Exception as e: logger.error(f"{self} exception: {e}") + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + # If we received a TTSSpeakFrame and the LLM response included text (it + # might be that it's only a function calling response) we pause + # processing more frames until we receive a BotStoppedSpeakingFrame. + if isinstance(frame, TTSSpeakFrame): + await self.pause_processing_frames() + elif isinstance(frame, LLMFullResponseEndFrame) and self._context_id: + await self.pause_processing_frames() + elif isinstance(frame, BotStoppedSpeakingFrame): + await self.resume_processing_frames() + async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: logger.debug(f"Generating TTS: [{text}]") diff --git a/src/pipecat/services/playht.py b/src/pipecat/services/playht.py index 64179b587..167181586 100644 --- a/src/pipecat/services/playht.py +++ b/src/pipecat/services/playht.py @@ -17,6 +17,7 @@ from loguru import logger from pydantic.main import BaseModel from pipecat.frames.frames import ( + BotStoppedSpeakingFrame, CancelFrame, EndFrame, ErrorFrame, @@ -125,7 +126,6 @@ class PlayHTTTSService(TTSService): ): super().__init__( sample_rate=sample_rate, - block_on_frames=(TTSSpeakFrame, LLMFullResponseEndFrame), **kwargs, ) @@ -266,6 +266,19 @@ class PlayHTTTSService(TTSService): except Exception as e: logger.error(f"{self} exception in receive task: {e}") + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + # If we received a TTSSpeakFrame and the LLM response included text (it + # might be that it's only a function calling response) we pause + # processing more frames until we receive a BotStoppedSpeakingFrame. + if isinstance(frame, TTSSpeakFrame): + await self.pause_processing_frames() + elif isinstance(frame, LLMFullResponseEndFrame) and self._request_id: + await self.pause_processing_frames() + elif isinstance(frame, BotStoppedSpeakingFrame): + await self.resume_processing_frames() + async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: logger.debug(f"Generating TTS: [{text}]")