diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index c0ceaeb70..485ae85a4 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -20,12 +20,9 @@ from pipecat.frames.frames import ( StartFrame, StartInterruptionFrame, TTSSpeakFrame, - TTSStartedFrame, - TTSStoppedFrame, TTSVoiceUpdateFrame, TextFrame, - VisionImageRawFrame, - FunctionCallResultFrame + VisionImageRawFrame ) from pipecat.processors.async_frame_processor import AsyncFrameProcessor from pipecat.processors.frame_processor import FrameDirection, FrameProcessor @@ -178,12 +175,12 @@ class TTSService(AIService): self._push_text_frames: bool = push_text_frames self._current_sentence: str = "" - @ abstractmethod + @abstractmethod async def set_voice(self, voice: str): pass # Converts the text to audio. - @ abstractmethod + @abstractmethod async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: pass @@ -212,11 +209,9 @@ class TTSService(AIService): if not text: return - await self.push_frame(TTSStartedFrame()) await self.start_processing_metrics() await self.process_generator(self.run_tts(text)) await self.stop_processing_metrics() - await self.push_frame(TTSStoppedFrame()) if self._push_text_frames: # We send the original text after the audio. This way, if we are # interrupted, the text is not added to the assistant context. @@ -269,7 +264,7 @@ class STTService(AIService): self._smoothing_factor = 0.2 self._prev_volume = 0 - @ abstractmethod + @abstractmethod async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]: """Returns transcript as a string""" pass @@ -335,7 +330,7 @@ class ImageGenService(AIService): super().__init__(**kwargs) # Renders the image. Returns an Image object. - @ abstractmethod + @abstractmethod async def run_image_gen(self, prompt: str) -> AsyncGenerator[Frame, None]: pass @@ -358,7 +353,7 @@ class VisionService(AIService): super().__init__(**kwargs) self._describe_text = None - @ abstractmethod + @abstractmethod async def run_vision(self, frame: VisionImageRawFrame) -> AsyncGenerator[Frame, None]: pass diff --git a/src/pipecat/services/azure.py b/src/pipecat/services/azure.py index aa3b5831e..feb355b86 100644 --- a/src/pipecat/services/azure.py +++ b/src/pipecat/services/azure.py @@ -18,9 +18,10 @@ from pipecat.frames.frames import ( EndFrame, ErrorFrame, Frame, - MetricsFrame, StartFrame, SystemFrame, + TTSStartedFrame, + TTSStoppedFrame, TranscriptionFrame, URLImageRawFrame) from pipecat.processors.frame_processor import FrameDirection @@ -106,8 +107,10 @@ class AzureTTSService(TTSService): if result.reason == ResultReason.SynthesizingAudioCompleted: await self.start_tts_usage_metrics(text) await self.stop_ttfb_metrics() + await self.push_frame(TTSStartedFrame()) # Azure always sends a 44-byte header. Strip it off. yield AudioRawFrame(audio=result.audio_data[44:], sample_rate=16000, num_channels=1) + await self.push_frame(TTSStoppedFrame()) elif result.reason == ResultReason.Canceled: cancellation_details = result.cancellation_details logger.warning(f"Speech synthesis canceled: {cancellation_details.reason}") diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index c0f9d0e08..735d12b5b 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -21,8 +21,9 @@ from pipecat.frames.frames import ( StartInterruptionFrame, StartFrame, EndFrame, + TTSStartedFrame, + TTSStoppedFrame, TextFrame, - MetricsFrame, LLMFullResponseEndFrame ) from pipecat.services.ai_services import TTSService @@ -154,6 +155,7 @@ class CartesiaTTSService(TTSService): continue if msg["type"] == "done": await self.stop_ttfb_metrics() + await self.push_frame(TTSStoppedFrame()) # Unset _context_id but not the _context_id_start_timestamp # because we are likely still playing out audio and need the # timestamp to set send context frames. @@ -176,6 +178,7 @@ class CartesiaTTSService(TTSService): await self.push_frame(frame) elif msg["type"] == "error": logger.error(f"{self} error: {msg}") + await self.push_frame(TTSStoppedFrame()) await self.stop_all_metrics() await self.push_error(ErrorFrame(f'{self} error: {msg["error"]}')) else: @@ -214,6 +217,7 @@ class CartesiaTTSService(TTSService): await self._connect() if not self._context_id: + await self.push_frame(TTSStartedFrame()) await self.start_ttfb_metrics() self._context_id = str(uuid.uuid4()) @@ -234,7 +238,8 @@ class CartesiaTTSService(TTSService): await self._websocket.send(json.dumps(msg)) await self.start_tts_usage_metrics(text) except Exception as e: - logger.exception(f"{self} error sending message: {e}") + logger.error(f"{self} error sending message: {e}") + await self.push_frame(TTSStoppedFrame()) await self._disconnect() await self._connect() return diff --git a/src/pipecat/services/deepgram.py b/src/pipecat/services/deepgram.py index 8d58def56..035fdd25c 100644 --- a/src/pipecat/services/deepgram.py +++ b/src/pipecat/services/deepgram.py @@ -15,9 +15,10 @@ from pipecat.frames.frames import ( ErrorFrame, Frame, InterimTranscriptionFrame, - MetricsFrame, StartFrame, SystemFrame, + TTSStartedFrame, + TTSStoppedFrame, TranscriptionFrame) from pipecat.processors.frame_processor import FrameDirection from pipecat.services.ai_services import AsyncAIService, TTSService @@ -96,10 +97,12 @@ class DeepgramTTSService(TTSService): await self.start_tts_usage_metrics(text) + await self.push_frame(TTSStartedFrame()) async for data in r.content: await self.stop_ttfb_metrics() frame = AudioRawFrame(audio=data, sample_rate=self._sample_rate, num_channels=1) yield frame + await self.push_frame(TTSStoppedFrame()) except Exception as e: logger.exception(f"{self} exception: {e}") diff --git a/src/pipecat/services/elevenlabs.py b/src/pipecat/services/elevenlabs.py index 0de629034..974619ea8 100644 --- a/src/pipecat/services/elevenlabs.py +++ b/src/pipecat/services/elevenlabs.py @@ -9,7 +9,7 @@ import aiohttp from typing import AsyncGenerator, Literal from pydantic import BaseModel -from pipecat.frames.frames import AudioRawFrame, ErrorFrame, Frame, MetricsFrame +from pipecat.frames.frames import AudioRawFrame, ErrorFrame, Frame, TTSStartedFrame, TTSStoppedFrame from pipecat.services.ai_services import TTSService from loguru import logger @@ -70,8 +70,10 @@ class ElevenLabsTTSService(TTSService): await self.start_tts_usage_metrics(text) + await self.push_frame(TTSStartedFrame()) async for chunk in r.content: if len(chunk) > 0: await self.stop_ttfb_metrics() frame = AudioRawFrame(chunk, 16000, 1) yield frame + await self.push_frame(TTSStoppedFrame()) diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py index 8e6ed83f9..12708647c 100644 --- a/src/pipecat/services/openai.py +++ b/src/pipecat/services/openai.py @@ -24,6 +24,8 @@ from pipecat.frames.frames import ( LLMFullResponseStartFrame, LLMMessagesFrame, LLMModelUpdateFrame, + TTSStartedFrame, + TTSStoppedFrame, TextFrame, URLImageRawFrame, VisionImageRawFrame, @@ -342,11 +344,13 @@ class OpenAITTSService(TTSService): await self.start_tts_usage_metrics(text) + await self.push_frame(TTSStartedFrame()) async for chunk in r.iter_bytes(8192): if len(chunk) > 0: await self.stop_ttfb_metrics() frame = AudioRawFrame(chunk, 24_000, 1) yield frame + await self.push_frame(TTSStoppedFrame()) except BadRequestError as e: logger.exception(f"{self} error generating TTS: {e}") diff --git a/src/pipecat/services/playht.py b/src/pipecat/services/playht.py index b738040b6..2f4ae9851 100644 --- a/src/pipecat/services/playht.py +++ b/src/pipecat/services/playht.py @@ -9,7 +9,7 @@ import struct from typing import AsyncGenerator -from pipecat.frames.frames import AudioRawFrame, Frame, MetricsFrame +from pipecat.frames.frames import AudioRawFrame, Frame, TTSStartedFrame, TTSStoppedFrame from pipecat.services.ai_services import TTSService from loguru import logger @@ -62,6 +62,7 @@ class PlayHTTTSService(TTSService): await self.start_tts_usage_metrics(text) + await self.push_frame(TTSStartedFrame()) async for chunk in playht_gen: # skip the RIFF header. if in_header: @@ -81,5 +82,6 @@ class PlayHTTTSService(TTSService): await self.stop_ttfb_metrics() frame = AudioRawFrame(chunk, 16000, 1) yield frame + await self.push_frame(TTSStoppedFrame()) except Exception as e: logger.exception(f"{self} error generating TTS: {e}") diff --git a/src/pipecat/services/xtts.py b/src/pipecat/services/xtts.py index a4b144b9a..38f0f9a64 100644 --- a/src/pipecat/services/xtts.py +++ b/src/pipecat/services/xtts.py @@ -8,7 +8,13 @@ import aiohttp from typing import Any, AsyncGenerator, Dict -from pipecat.frames.frames import AudioRawFrame, ErrorFrame, Frame, MetricsFrame, StartFrame +from pipecat.frames.frames import ( + AudioRawFrame, + ErrorFrame, + Frame, + StartFrame, + TTSStartedFrame, + TTSStoppedFrame) from pipecat.services.ai_services import TTSService from loguru import logger @@ -99,8 +105,9 @@ class XTTSService(TTSService): await self.start_tts_usage_metrics(text) - buffer = bytearray() + await self.push_frame(TTSStartedFrame()) + buffer = bytearray() async for chunk in r.content.iter_chunked(1024): if len(chunk) > 0: await self.stop_ttfb_metrics() @@ -131,3 +138,5 @@ class XTTSService(TTSService): resampled_audio_bytes = resampled_audio.astype(np.int16).tobytes() frame = AudioRawFrame(resampled_audio_bytes, 16000, 1) yield frame + + await self.push_frame(TTSStoppedFrame()) diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index 6d619cd41..31cf2bff2 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -56,6 +56,11 @@ class BaseOutputTransport(FrameProcessor): self._stopped_event = asyncio.Event() + # Indicates if the bot is currently speaking. This is useful when we + # have an interruption since all the queued messages will be thrown + # away and we would lose the TTSStoppedFrame. + self._bot_speaking = False + # Create sink frame task. This is the task that will actually write # audio or video frames. We write audio/video in a task so we can keep # generating frames upstream while, for example, the audio is playing. @@ -169,6 +174,9 @@ class BaseOutputTransport(FrameProcessor): self._push_frame_task.cancel() await self._push_frame_task self._create_push_task() + # Let's send a bot stopped speaking if we have to. + if self._bot_speaking: + await self._bot_stopped_speaking() async def _handle_audio(self, frame: AudioRawFrame): if not self._params.audio_out_enabled: @@ -214,10 +222,10 @@ class BaseOutputTransport(FrameProcessor): elif isinstance(frame, TransportMessageFrame): await self.send_message(frame) elif isinstance(frame, TTSStartedFrame): - await self._internal_push_frame(BotStartedSpeakingFrame(), FrameDirection.UPSTREAM) + await self._bot_started_speaking() await self._internal_push_frame(frame) elif isinstance(frame, TTSStoppedFrame): - await self._internal_push_frame(BotStoppedSpeakingFrame(), FrameDirection.UPSTREAM) + await self._bot_stopped_speaking() await self._internal_push_frame(frame) else: await self._internal_push_frame(frame) @@ -230,6 +238,14 @@ class BaseOutputTransport(FrameProcessor): except Exception as e: logger.exception(f"{self} error processing sink queue: {e}") + async def _bot_started_speaking(self): + self._bot_speaking = True + await self._internal_push_frame(BotStartedSpeakingFrame(), FrameDirection.UPSTREAM) + + async def _bot_stopped_speaking(self): + self._bot_speaking = False + await self._internal_push_frame(BotStoppedSpeakingFrame(), FrameDirection.UPSTREAM) + # # Push frames task #