services: send TTSStartFrame/TTSStopFrame when really needed

src/pipecat/services/ai_services.py

@@ -20,12 +20,9 @@ from pipecat.frames.frames import (
     StartFrame,
     StartInterruptionFrame,
     TTSSpeakFrame,
-    TTSStartedFrame,
-    TTSStoppedFrame,
     TTSVoiceUpdateFrame,
     TextFrame,
-    VisionImageRawFrame,
+    VisionImageRawFrame
-    FunctionCallResultFrame
 )
 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
 

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}")

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

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}")
 

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())

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}")
 

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}")

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())

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
     #