From dcf7e454f61f3702cfe82f2c4ef20b930a918c3d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 2 Dec 2024 19:00:47 -0800 Subject: [PATCH] services(tts): don't wait for BotStoppedSpeakingFrame to resume input queue --- src/pipecat/services/cartesia.py | 16 +++++++++++----- src/pipecat/services/elevenlabs.py | 14 +++++++++----- src/pipecat/services/playht.py | 16 +++++++++++----- 3 files changed, 31 insertions(+), 15 deletions(-) diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index ad4636a74..b7d4e15c4 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -14,7 +14,6 @@ from loguru import logger from pydantic.main import BaseModel from pipecat.frames.frames import ( - BotStoppedSpeakingFrame, CancelFrame, EndFrame, ErrorFrame, @@ -259,18 +258,25 @@ class CartesiaTTSService(WordTTSService): except Exception as e: logger.error(f"{self} exception: {e}") + async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM): + await super().push_frame(frame, direction) + + # We generate LLMFullResponseEndFrame after we have received all the + # audio from the service which means we can resume processing frames. + if isinstance(frame, LLMFullResponseEndFrame): + await self.resume_processing_frames() + 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 + # If we received a TTSSpeakFrame or 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. + # processing more frames until we have generated LLMFullResponseEndFrame + # (see push_frame()). 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/elevenlabs.py b/src/pipecat/services/elevenlabs.py index 2707df92a..7512d00eb 100644 --- a/src/pipecat/services/elevenlabs.py +++ b/src/pipecat/services/elevenlabs.py @@ -13,7 +13,6 @@ from loguru import logger from pydantic import BaseModel, model_validator from pipecat.frames.frames import ( - BotStoppedSpeakingFrame, CancelFrame, EndFrame, Frame, @@ -262,23 +261,28 @@ class ElevenLabsTTSService(WordTTSService): async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM): await super().push_frame(frame, direction) + if isinstance(frame, (TTSStoppedFrame, StartInterruptionFrame)): self._started = False if isinstance(frame, TTSStoppedFrame): await self.add_word_timestamps([("LLMFullResponseEndFrame", 0), ("Reset", 0)]) + # We generate LLMFullResponseEndFrame after we have received all the + # audio from the service which means we can resume processing frames. + if isinstance(frame, LLMFullResponseEndFrame): + await self.resume_processing_frames() + 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 + # If we received a TTSSpeakFrame or 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. + # processing more frames until we have generated LLMFullResponseEndFrame + # (see push_frame()). 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: diff --git a/src/pipecat/services/playht.py b/src/pipecat/services/playht.py index 9b5890554..6faa32607 100644 --- a/src/pipecat/services/playht.py +++ b/src/pipecat/services/playht.py @@ -17,7 +17,6 @@ from loguru import logger from pydantic.main import BaseModel from pipecat.frames.frames import ( - BotStoppedSpeakingFrame, CancelFrame, EndFrame, ErrorFrame, @@ -235,18 +234,25 @@ class PlayHTTTSService(TTSService): except Exception as e: logger.error(f"{self} exception in receive task: {e}") + async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM): + await super().push_frame(frame, direction) + + # We generate LLMFullResponseEndFrame after we have received all the + # audio from the service which means we can resume processing frames. + if isinstance(frame, LLMFullResponseEndFrame): + await self.resume_processing_frames() + 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 + # If we received a TTSSpeakFrame or 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. + # processing more frames until we have generated LLMFullResponseEndFrame + # (see push_frame()). 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}]")