Merge pull request #3958 from pipecat-ai/mb/deepgram-tts-audio-context

Route Deepgram WebSocket TTS audio through audio context queue

changelog/3958.changed.md

@@ -0,0 +1 @@
+- Changed `DeepgramTTSService` to send a Clear message on interruption instead of disconnecting and reconnecting the WebSocket, allowing the connection to persist throughout the session.

changelog/3958.fixed.md

@@ -0,0 +1 @@
+- Fixed `TTSService` audio context queue getting blocked when `append_to_audio_context()` was called with a `None` context ID, which prevented subsequent audio from being delivered.

src/pipecat/services/deepgram/tts.py

@@ -22,12 +22,10 @@ from pipecat.frames.frames import (
     EndFrame,
     ErrorFrame,
     Frame,
-    InterruptionFrame,
-    LLMFullResponseEndFrame,
     StartFrame,
     TTSAudioRawFrame,
+    TTSStoppedFrame,
 )
-from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.settings import TTSSettings, _warn_deprecated_param
 from pipecat.services.tts_service import TTSService, WebsocketTTSService
 from pipecat.utils.tracing.service_decorators import traced_tts
@@ -120,7 +118,7 @@ class DeepgramTTSService(WebsocketTTSService):
         super().__init__(
             sample_rate=sample_rate,
             pause_frame_processing=True,
-            push_stop_frames=True,
+            push_stop_frames=False,
             push_start_frame=True,
             append_trailing_space=True,
             settings=default_settings,
@@ -168,19 +166,6 @@ class DeepgramTTSService(WebsocketTTSService):
         await super().cancel(frame)
         await self._disconnect()
 
-    async def process_frame(self, frame: Frame, direction: FrameDirection):
-        """Process frames with special handling for LLM response end.
-
-        Args:
-            frame: The frame to process.
-            direction: The direction of frame processing.
-        """
-        await super().process_frame(frame, direction)
-
-        # When the LLM finishes responding, flush any remaining text in Deepgram's buffer
-        if isinstance(frame, (LLMFullResponseEndFrame, EndFrame)):
-            await self.flush_audio()
-
     async def _connect(self):
         """Connect to Deepgram WebSocket and start receive task."""
         await super()._connect()
@@ -277,19 +262,19 @@ class DeepgramTTSService(WebsocketTTSService):
             return self._websocket
         raise Exception("Websocket not connected")
 
-    async def _handle_interruption(self, frame: InterruptionFrame, direction: FrameDirection):
+    async def on_audio_context_interrupted(self, context_id: str):
-        """Handle interruption by sending Clear message to Deepgram.
+        """Send Clear message to Deepgram when an audio context is interrupted.
 
         The Clear message will clear Deepgram's internal text buffer and stop
         sending audio, allowing for a new response to be generated.
-        """
-        await super()._handle_interruption(frame, direction)
 
-        # Send Clear message to stop current audio generation
+        Args:
+            context_id: The ID of the audio context that was interrupted.
+        """
+        await self.stop_all_metrics()
         if self._websocket:
             try:
-                clear_msg = {"type": "Clear"}
+                await self._websocket.send(json.dumps({"type": "Clear"}))
-                await self._websocket.send(json.dumps(clear_msg))
             except Exception as e:
                 logger.error(f"{self} error sending Clear message: {e}")
 
@@ -298,11 +283,9 @@ class DeepgramTTSService(WebsocketTTSService):
         async for message in self._get_websocket():
             if isinstance(message, bytes):
                 # Binary message contains audio data
-                await self.stop_ttfb_metrics()
+                ctx_id = self.get_active_audio_context_id()
-                frame = TTSAudioRawFrame(
+                frame = TTSAudioRawFrame(message, self.sample_rate, 1, context_id=ctx_id)
-                    message, self.sample_rate, 1, context_id=self.get_active_audio_context_id()
+                await self.append_to_audio_context(ctx_id, frame)
-                )
-                await self.push_frame(frame)
             elif isinstance(message, str):
                 # Text message contains metadata or control messages
                 try:
@@ -313,12 +296,15 @@ class DeepgramTTSService(WebsocketTTSService):
                         logger.trace(f"Received metadata: {msg}")
                     elif msg_type == "Flushed":
                         logger.trace(f"Received Flushed: {msg}")
-                        # Flushed indicates the end of audio generation for the current buffer
+                        ctx_id = self.get_active_audio_context_id()
-                        # This happens after flush_audio() is called
+                        await self.append_to_audio_context(
+                            ctx_id, TTSStoppedFrame(context_id=ctx_id)
+                        )
+                        await self.remove_audio_context(ctx_id)
                     elif msg_type == "Cleared":
                         logger.trace(f"Received Cleared: {msg}")
-                        # Buffer has been cleared after interruption
+                        # Buffer has been cleared after interruption.
-                        # TTSStoppedFrame will be sent by the interruption handler
+                        # The on_audio_context_interrupted handler already cleaned up.
                     elif msg_type == "Warning":
                         logger.warning(
                             f"{self} warning: {msg.get('description', 'Unknown warning')}"
@@ -359,8 +345,6 @@ class DeepgramTTSService(WebsocketTTSService):
             if not self._websocket or self._websocket.state is State.CLOSED:
                 await self._connect()
 
-            await self.start_tts_usage_metrics(text)
-
             # Send text message to Deepgram
             # Note: We don't send Flush here - that should only be sent when the
             # LLM finishes a complete response via flush_audio()

src/pipecat/services/tts_service.py

@@ -1179,6 +1179,9 @@ class TTSService(AIService):
             context_id: The context to append audio to.
             frame: The audio or control frame to append.
         """
+        if not context_id:
+            logger.debug(f"{self} unable to append audio to context: no context ID provided")
+            return
         if self.audio_context_available(context_id):
             logger.trace(f"{self} appending audio {frame} to audio context {context_id}")
             await self._audio_contexts[context_id].put(frame)