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 01/12] 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 From 22ca4c5a0259ed5e6b23cbd2663fbe6b725a2dab Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Sun, 20 Oct 2024 22:35:15 -0700 Subject: [PATCH 02/12] processors: cancel input task and empty queue with interruptions --- src/pipecat/processors/frame_processor.py | 32 ++++++++++++++++------- 1 file changed, 23 insertions(+), 9 deletions(-) diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 206e7adc1..2b2069c33 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -63,11 +63,11 @@ 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()). + # 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 + # `resume_processing_frames()`. 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 @@ -134,7 +134,8 @@ class FrameProcessor: await self.stop_processing_metrics() async def cleanup(self): - pass + await self.__cancel_input_task() + await self.__cancel_push_task() def link(self, processor: "FrameProcessor"): self._next = processor @@ -210,13 +211,18 @@ class FrameProcessor: # async def _start_interruption(self): - # Cancel the task. This will stop pushing frames downstream. - self.__push_frame_task.cancel() - await self.__push_frame_task + # Cancel the input task. This will stop processing queued frames. + await self.__cancel_input_task() - # Create a new queue and task. + # Cancel the push frame task. This will stop pushing frames downstream. + await self.__cancel_push_task() + + # Create a new output queue and task. self.__create_push_task() + # Create a new input queue and task. + self.__create_input_task() + async def _stop_interruption(self): # Nothing to do right now. pass @@ -239,6 +245,10 @@ class FrameProcessor: ) self.__input_event = asyncio.Event() + async def __cancel_input_task(self): + self.__input_frame_task.cancel() + await self.__input_frame_task + async def __input_frame_task_handler(self): running = True should_block_frames = False @@ -269,6 +279,10 @@ class FrameProcessor: self.__push_queue = asyncio.Queue() self.__push_frame_task = self.get_event_loop().create_task(self.__push_frame_task_handler()) + async def __cancel_push_task(self): + self.__push_frame_task.cancel() + await self.__push_frame_task + async def __push_frame_task_handler(self): running = True while running: From 2eccb33e737538175127f5c870a940af91ff2842 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Sun, 20 Oct 2024 22:35:46 -0700 Subject: [PATCH 03/12] processors: allow passing a callback when queued frame is processed --- src/pipecat/processors/frame_processor.py | 16 +++++++++++++--- 1 file changed, 13 insertions(+), 3 deletions(-) diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 2b2069c33..007c80c9e 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -8,6 +8,7 @@ import asyncio import inspect from enum import Enum +from typing import Awaitable, Callable, Optional from pipecat.clocks.base_clock import BaseClock from pipecat.frames.frames import ( @@ -155,14 +156,19 @@ class FrameProcessor: return self._clock async def queue_frame( - self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM + self, + frame: Frame, + direction: FrameDirection = FrameDirection.DOWNSTREAM, + callback: Optional[ + Callable[["FrameProcessor", Frame, FrameDirection], Awaitable[None]] + ] = None, ): 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)) + await self.__input_queue.put((frame, direction, callback)) async def resume_processing_frames(self): self.__input_event.set() @@ -259,11 +265,15 @@ class FrameProcessor: self.__input_event.clear() should_block_frames = False - (frame, direction) = await self.__input_queue.get() + (frame, direction, callback) = await self.__input_queue.get() # Process the frame. await self.process_frame(frame, direction) + # If this frame has an associated callback, call it now. + 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): From 6d8b88507140887ba9fe7b7c58546bf50abc2b07 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 6 Nov 2024 11:34:52 -0800 Subject: [PATCH 04/12] transports(base_output): push bot started/stopped frames downstream --- src/pipecat/transports/base_output.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index e6c006145..0ae79b9f0 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -208,12 +208,14 @@ class BaseOutputTransport(FrameProcessor): async def _bot_started_speaking(self): if not self._bot_speaking: logger.debug("Bot started speaking") + await self.push_frame(BotStartedSpeakingFrame()) await self.push_frame(BotStartedSpeakingFrame(), FrameDirection.UPSTREAM) self._bot_speaking = True async def _bot_stopped_speaking(self): if self._bot_speaking: logger.debug("Bot stopped speaking") + await self.push_frame(BotStoppedSpeakingFrame()) await self.push_frame(BotStoppedSpeakingFrame(), FrameDirection.UPSTREAM) self._bot_speaking = False @@ -452,6 +454,7 @@ class BaseOutputTransport(FrameProcessor): # it's actually speaking. if isinstance(frame, TTSAudioRawFrame): await self._bot_started_speaking() + await self.push_frame(BotSpeakingFrame()) await self.push_frame(BotSpeakingFrame(), FrameDirection.UPSTREAM) # Also, push frame downstream in case anyone else needs it. From 49005d02f567d1c4f9864ef4f7e526f885119b01 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 6 Nov 2024 11:36:29 -0800 Subject: [PATCH 05/12] services(tts): use TTSSpeakFrame in say() method --- src/pipecat/services/ai_services.py | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index 63100f908..fd61dd1a7 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -285,11 +285,7 @@ class TTSService(AIService): logger.warning(f"Unknown setting for TTS service: {key}") async def say(self, text: str): - aggregate_sentences = self._aggregate_sentences - self._aggregate_sentences = False - await self.process_frame(TextFrame(text=text), FrameDirection.DOWNSTREAM) - self._aggregate_sentences = aggregate_sentences - await self.flush_audio() + await self.queue_frame(TTSSpeakFrame(text)) async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) From b6f0c16591959d1625e7920dc71c22f0454dfa10 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 6 Nov 2024 12:02:35 -0800 Subject: [PATCH 06/12] examples: restore EndFrame() on 01 and 02 foundational --- examples/foundational/01-say-one-thing.py | 15 +++++------ examples/foundational/01a-local-audio.py | 27 +++++++++---------- examples/foundational/02-llm-say-one-thing.py | 10 +++---- 3 files changed, 22 insertions(+), 30 deletions(-) diff --git a/examples/foundational/01-say-one-thing.py b/examples/foundational/01-say-one-thing.py index 50ae1bad2..f0fec28d8 100644 --- a/examples/foundational/01-say-one-thing.py +++ b/examples/foundational/01-say-one-thing.py @@ -9,11 +9,11 @@ import aiohttp import os import sys -from pipecat.frames.frames import EndFrame, TextFrame +from pipecat.frames.frames import EndFrame, TTSSpeakFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.task import PipelineTask from pipecat.pipeline.runner import PipelineRunner -from pipecat.services.cartesia import CartesiaHttpTTSService +from pipecat.services.cartesia import CartesiaTTSService from pipecat.transports.services.daily import DailyParams, DailyTransport from runner import configure @@ -36,7 +36,7 @@ async def main(): room_url, None, "Say One Thing", DailyParams(audio_out_enabled=True) ) - tts = CartesiaHttpTTSService( + tts = CartesiaTTSService( api_key=os.getenv("CARTESIA_API_KEY"), voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady ) @@ -50,12 +50,9 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): participant_name = participant.get("info", {}).get("userName", "") - await task.queue_frame(TextFrame(f"Hello there, {participant_name}!")) - - # Register an event handler to exit the application when the user leaves. - @transport.event_handler("on_participant_left") - async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.queue_frames( + [TTSSpeakFrame(f"Hello there, {participant_name}!"), EndFrame()] + ) await runner.run(task) diff --git a/examples/foundational/01a-local-audio.py b/examples/foundational/01a-local-audio.py index d39e922d7..e62ab1020 100644 --- a/examples/foundational/01a-local-audio.py +++ b/examples/foundational/01a-local-audio.py @@ -9,7 +9,7 @@ import aiohttp import os import sys -from pipecat.frames.frames import TextFrame +from pipecat.frames.frames import EndFrame, TTSSpeakFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineTask @@ -28,25 +28,24 @@ logger.add(sys.stderr, level="DEBUG") async def main(): - async with aiohttp.ClientSession() as session: - transport = LocalAudioTransport(TransportParams(audio_out_enabled=True)) + transport = LocalAudioTransport(TransportParams(audio_out_enabled=True)) - tts = CartesiaTTSService( - api_key=os.getenv("CARTESIA_API_KEY"), - voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady - ) + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady + ) - pipeline = Pipeline([tts, transport.output()]) + pipeline = Pipeline([tts, transport.output()]) - task = PipelineTask(pipeline) + task = PipelineTask(pipeline) - async def say_something(): - await asyncio.sleep(1) - await task.queue_frame(TextFrame("Hello there!")) + async def say_something(): + await asyncio.sleep(1) + await task.queue_frames([TTSSpeakFrame("Hello there, how is it going!"), EndFrame()]) - runner = PipelineRunner() + runner = PipelineRunner() - await asyncio.gather(runner.run(task), say_something()) + await asyncio.gather(runner.run(task), say_something()) if __name__ == "__main__": diff --git a/examples/foundational/02-llm-say-one-thing.py b/examples/foundational/02-llm-say-one-thing.py index 891628415..ba2be607c 100644 --- a/examples/foundational/02-llm-say-one-thing.py +++ b/examples/foundational/02-llm-say-one-thing.py @@ -13,7 +13,7 @@ from pipecat.frames.frames import EndFrame, LLMMessagesFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineTask -from pipecat.services.cartesia import CartesiaHttpTTSService +from pipecat.services.cartesia import CartesiaTTSService from pipecat.services.openai import OpenAILLMService from pipecat.transports.services.daily import DailyParams, DailyTransport @@ -37,7 +37,7 @@ async def main(): room_url, None, "Say One Thing From an LLM", DailyParams(audio_out_enabled=True) ) - tts = CartesiaHttpTTSService( + tts = CartesiaTTSService( api_key=os.getenv("CARTESIA_API_KEY"), voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady ) @@ -57,11 +57,7 @@ async def main(): @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): - await task.queue_frame(LLMMessagesFrame(messages)) - - @transport.event_handler("on_participant_left") - async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.queue_frames([LLMMessagesFrame(messages), EndFrame()]) await runner.run(task) From a9e565f3550fcdf01652a52d3bfc75bb66414d2d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 6 Nov 2024 13:12:24 -0800 Subject: [PATCH 07/12] processors: fix input queue interruptions --- src/pipecat/processors/frame_processor.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 007c80c9e..04f53f127 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -217,18 +217,18 @@ class FrameProcessor: # async def _start_interruption(self): - # Cancel the input task. This will stop processing queued frames. - await self.__cancel_input_task() - # Cancel the push frame task. This will stop pushing frames downstream. await self.__cancel_push_task() - # Create a new output queue and task. - self.__create_push_task() + # Cancel the input task. This will stop processing queued frames. + await self.__cancel_input_task() # Create a new input queue and task. self.__create_input_task() + # Create a new output queue and task. + self.__create_push_task() + async def _stop_interruption(self): # Nothing to do right now. pass From 5353d131511a4ad1fdc0266f2ccaf28ac648826d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 6 Nov 2024 13:16:58 -0800 Subject: [PATCH 08/12] update CHANGELOG --- CHANGELOG.md | 17 +++++++++-------- 1 file changed, 9 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index cb49ae499..a810258d3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,15 +9,16 @@ 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 +- There's now an input queue in 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.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()`. - Added audio filter `NoisereduceFilter`. From 7071482583125a0b91f9e82bfa5aa035447936de Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 6 Nov 2024 14:18:21 -0800 Subject: [PATCH 09/12] try to use queue_frame() instead of process_frame() --- examples/patient-intake/bot.py | 8 ++++---- src/pipecat/pipeline/parallel_pipeline.py | 4 ++-- src/pipecat/pipeline/pipeline.py | 4 ++-- src/pipecat/pipeline/task.py | 8 +++----- src/pipecat/transports/base_output.py | 4 ++-- src/pipecat/transports/services/daily.py | 4 ++-- src/pipecat/transports/services/livekit.py | 2 +- 7 files changed, 16 insertions(+), 18 deletions(-) diff --git a/examples/patient-intake/bot.py b/examples/patient-intake/bot.py index adb3785e4..c33a2495d 100644 --- a/examples/patient-intake/bot.py +++ b/examples/patient-intake/bot.py @@ -182,7 +182,7 @@ class IntakeProcessor: } ) print(f"!!! about to await llm process frame in start prescrpitions") - await llm.process_frame(OpenAILLMContextFrame(context), FrameDirection.DOWNSTREAM) + await llm.queue_frame(OpenAILLMContextFrame(context), FrameDirection.DOWNSTREAM) print(f"!!! past await process frame in start prescriptions") async def start_allergies(self, function_name, llm, context): @@ -222,7 +222,7 @@ class IntakeProcessor: "content": "Now ask the user if they have any medical conditions the doctor should know about. Once they've answered the question, call the list_conditions function.", } ) - await llm.process_frame(OpenAILLMContextFrame(context), FrameDirection.DOWNSTREAM) + await llm.queue_frame(OpenAILLMContextFrame(context), FrameDirection.DOWNSTREAM) async def start_conditions(self, function_name, llm, context): print("!!! doing start conditions") @@ -261,7 +261,7 @@ class IntakeProcessor: "content": "Finally, ask the user the reason for their doctor visit today. Once they answer, call the list_visit_reasons function.", } ) - await llm.process_frame(OpenAILLMContextFrame(context), FrameDirection.DOWNSTREAM) + await llm.queue_frame(OpenAILLMContextFrame(context), FrameDirection.DOWNSTREAM) async def start_visit_reasons(self, function_name, llm, context): print("!!! doing start visit reasons") @@ -270,7 +270,7 @@ class IntakeProcessor: context.add_message( {"role": "system", "content": "Now, thank the user and end the conversation."} ) - await llm.process_frame(OpenAILLMContextFrame(context), FrameDirection.DOWNSTREAM) + await llm.queue_frame(OpenAILLMContextFrame(context), FrameDirection.DOWNSTREAM) async def save_data(self, function_name, tool_call_id, args, llm, context, result_callback): logger.info(f"!!! Saving data: {args}") diff --git a/src/pipecat/pipeline/parallel_pipeline.py b/src/pipecat/pipeline/parallel_pipeline.py index da68212ff..ad3ebd8dd 100644 --- a/src/pipecat/pipeline/parallel_pipeline.py +++ b/src/pipecat/pipeline/parallel_pipeline.py @@ -110,13 +110,13 @@ class ParallelPipeline(BasePipeline): if direction == FrameDirection.UPSTREAM: # If we get an upstream frame we process it in each sink. - await asyncio.gather(*[s.process_frame(frame, direction) for s in self._sinks]) + await asyncio.gather(*[s.queue_frame(frame, direction) for s in self._sinks]) elif direction == FrameDirection.DOWNSTREAM: # If we get a downstream frame we process it in each source. # TODO(aleix): We are creating task for each frame. For real-time # video/audio this might be too slow. We should use an already # created task instead. - await asyncio.gather(*[s.process_frame(frame, direction) for s in self._sources]) + await asyncio.gather(*[s.queue_frame(frame, direction) for s in self._sources]) # If we get an EndFrame we stop our queue processing tasks and wait on # all the pipelines to finish. diff --git a/src/pipecat/pipeline/pipeline.py b/src/pipecat/pipeline/pipeline.py index a1715570e..703f911fe 100644 --- a/src/pipecat/pipeline/pipeline.py +++ b/src/pipecat/pipeline/pipeline.py @@ -77,9 +77,9 @@ class Pipeline(BasePipeline): await super().process_frame(frame, direction) if direction == FrameDirection.DOWNSTREAM: - await self._source.process_frame(frame, FrameDirection.DOWNSTREAM) + await self._source.queue_frame(frame, FrameDirection.DOWNSTREAM) elif direction == FrameDirection.UPSTREAM: - await self._sink.process_frame(frame, FrameDirection.UPSTREAM) + await self._sink.queue_frame(frame, FrameDirection.UPSTREAM) async def _cleanup_processors(self): for p in self._processors: diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index e34ccae1b..f09013a58 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -160,19 +160,17 @@ class PipelineTask: report_only_initial_ttfb=self._params.report_only_initial_ttfb, clock=self._clock, ) - await self._source.process_frame(start_frame, FrameDirection.DOWNSTREAM) + await self._source.queue_frame(start_frame, FrameDirection.DOWNSTREAM) if self._params.enable_metrics and self._params.send_initial_empty_metrics: - await self._source.process_frame( - self._initial_metrics_frame(), FrameDirection.DOWNSTREAM - ) + await self._source.queue_frame(self._initial_metrics_frame(), FrameDirection.DOWNSTREAM) running = True should_cleanup = True while running: try: frame = await self._push_queue.get() - await self._source.process_frame(frame, FrameDirection.DOWNSTREAM) + await self._source.queue_frame(frame, FrameDirection.DOWNSTREAM) if isinstance(frame, EndFrame): await self._wait_for_endframe() running = not isinstance(frame, (StopTaskFrame, EndFrame)) diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index 0ae79b9f0..afb93ebf8 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -325,7 +325,7 @@ class BaseOutputTransport(FrameProcessor): # async def send_image(self, frame: OutputImageRawFrame | SpriteFrame): - await self.process_frame(frame, FrameDirection.DOWNSTREAM) + await self.queue_frame(frame, FrameDirection.DOWNSTREAM) async def _draw_image(self, frame: OutputImageRawFrame): desired_size = (self._params.camera_out_width, self._params.camera_out_height) @@ -396,7 +396,7 @@ class BaseOutputTransport(FrameProcessor): # async def send_audio(self, frame: OutputAudioRawFrame): - await self.process_frame(frame, FrameDirection.DOWNSTREAM) + await self.queue_frame(frame, FrameDirection.DOWNSTREAM) def _next_audio_frame(self) -> AsyncGenerator[AudioRawFrame, None]: async def without_mixer(vad_stop_secs: float) -> AsyncGenerator[AudioRawFrame, None]: diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index 7f872c1fb..0c83bcc13 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -890,11 +890,11 @@ class DailyTransport(BaseTransport): async def send_image(self, frame: OutputImageRawFrame | SpriteFrame): if self._output: - await self._output.process_frame(frame, FrameDirection.DOWNSTREAM) + await self._output.queue_frame(frame, FrameDirection.DOWNSTREAM) async def send_audio(self, frame: OutputAudioRawFrame): if self._output: - await self._output.process_frame(frame, FrameDirection.DOWNSTREAM) + await self._output.queue_frame(frame, FrameDirection.DOWNSTREAM) def participants(self): return self._client.participants() diff --git a/src/pipecat/transports/services/livekit.py b/src/pipecat/transports/services/livekit.py index 6b6b55c54..5d4fbdd6c 100644 --- a/src/pipecat/transports/services/livekit.py +++ b/src/pipecat/transports/services/livekit.py @@ -495,7 +495,7 @@ class LiveKitTransport(BaseTransport): async def send_audio(self, frame: OutputAudioRawFrame): if self._output: - await self._output.process_frame(frame, FrameDirection.DOWNSTREAM) + await self._output.queue_frame(frame, FrameDirection.DOWNSTREAM) def get_participants(self) -> List[str]: return self._client.get_participants() 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 10/12] 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}]") From ce89bbb16e7c8eab1fc0e9296233482d49382ab9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 6 Nov 2024 14:38:33 -0800 Subject: [PATCH 11/12] tts(elevenlabs): support pausing and resuming frames while speaking --- src/pipecat/services/elevenlabs.py | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/src/pipecat/services/elevenlabs.py b/src/pipecat/services/elevenlabs.py index b4dee5fa3..4a9f80a44 100644 --- a/src/pipecat/services/elevenlabs.py +++ b/src/pipecat/services/elevenlabs.py @@ -13,12 +13,15 @@ from loguru import logger from pydantic import BaseModel, model_validator from pipecat.frames.frames import ( + BotStoppedSpeakingFrame, CancelFrame, EndFrame, Frame, + LLMFullResponseEndFrame, StartFrame, StartInterruptionFrame, TTSAudioRawFrame, + TTSSpeakFrame, TTSStartedFrame, TTSStoppedFrame, ) @@ -285,6 +288,19 @@ class ElevenLabsTTSService(WordTTSService): if isinstance(frame, TTSStoppedFrame): await self.add_word_timestamps([("LLMFullResponseEndFrame", 0)]) + 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._started: + await self.pause_processing_frames() + elif isinstance(frame, BotStoppedSpeakingFrame): + await self.resume_processing_frames() + async def _connect(self): try: voice_id = self._voice_id From beb32711680e518b48a05f65cfb9cdab280b9ed3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 6 Nov 2024 18:54:12 -0800 Subject: [PATCH 12/12] services(tts): make sure word timestamp is reset properly --- src/pipecat/services/ai_services.py | 11 +++++++---- src/pipecat/services/cartesia.py | 2 +- src/pipecat/services/elevenlabs.py | 2 +- 3 files changed, 9 insertions(+), 6 deletions(-) diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index 023cabee8..f7a08802c 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -391,7 +391,6 @@ class WordTTSService(TTSService): def reset_word_timestamps(self): self._initial_word_timestamp = -1 - self._word_timestamps = [] async def add_word_timestamps(self, word_times: List[Tuple[str, float]]): for word, timestamp in word_times: @@ -426,7 +425,10 @@ class WordTTSService(TTSService): while True: try: (word, timestamp) = await self._words_queue.get() - if word == "LLMFullResponseEndFrame" and timestamp == 0: + if word == "Reset" and timestamp == 0: + self.reset_word_timestamps() + frame = None + elif word == "LLMFullResponseEndFrame" and timestamp == 0: frame = LLMFullResponseEndFrame() frame.pts = last_pts elif word == "TTSStoppedFrame" and timestamp == 0: @@ -435,8 +437,9 @@ class WordTTSService(TTSService): else: frame = TextFrame(word) frame.pts = self._initial_word_timestamp + timestamp - last_pts = frame.pts - await self.push_frame(frame) + if frame: + last_pts = frame.pts + await self.push_frame(frame) self._words_queue.task_done() except asyncio.CancelledError: break diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index 2449fede6..ff7b200c1 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -234,7 +234,7 @@ class CartesiaTTSService(WordTTSService): # timestamp to set send context frames. self._context_id = None await self.add_word_timestamps( - [("TTSStoppedFrame", 0), ("LLMFullResponseEndFrame", 0)] + [("TTSStoppedFrame", 0), ("LLMFullResponseEndFrame", 0), ("Reset", 0)] ) elif msg["type"] == "timestamps": await self.add_word_timestamps( diff --git a/src/pipecat/services/elevenlabs.py b/src/pipecat/services/elevenlabs.py index 4a9f80a44..9b1f54351 100644 --- a/src/pipecat/services/elevenlabs.py +++ b/src/pipecat/services/elevenlabs.py @@ -286,7 +286,7 @@ class ElevenLabsTTSService(WordTTSService): if isinstance(frame, (TTSStoppedFrame, StartInterruptionFrame)): self._started = False if isinstance(frame, TTSStoppedFrame): - await self.add_word_timestamps([("LLMFullResponseEndFrame", 0)]) + await self.add_word_timestamps([("LLMFullResponseEndFrame", 0), ("Reset", 0)]) async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction)