From 92b5185165fded60cb4a3b6c9ca02df79997dd0b Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Sun, 8 Mar 2026 13:16:42 -0400 Subject: [PATCH 1/3] Route Deepgram WebSocket TTS audio through audio context queue The Deepgram TTS service was bypassing pipecats audio context management system, pushing audio frames directly via push_frame() instead of routing them through append_to_audio_context(). This caused stale audio to leak into the pipeline after interruptions and missed ordered playback guarantees. - Route audio frames through append_to_audio_context() with context availability checks to discard stale post-interruption frames - Handle Flushed responses by appending TTSStoppedFrame and removing the audio context to signal completion - Replace _handle_interruption override with on_audio_context_interrupted hook (the recommended pattern used by ElevenLabs and Cartesia) - Remove redundant process_frame override that caused double-flush (base class already flushes via on_turn_context_completed) - Remove redundant start_tts_usage_metrics call (base class handles aggregated usage metrics) --- src/pipecat/services/deepgram/tts.py | 54 ++++++++++------------------ 1 file changed, 19 insertions(+), 35 deletions(-) diff --git a/src/pipecat/services/deepgram/tts.py b/src/pipecat/services/deepgram/tts.py index 5d6e5ffdc..ae064fffd 100644 --- a/src/pipecat/services/deepgram/tts.py +++ b/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 @@ -119,7 +117,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, @@ -167,19 +165,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() @@ -276,19 +261,19 @@ class DeepgramTTSService(WebsocketTTSService): return self._websocket raise Exception("Websocket not connected") - async def _handle_interruption(self, frame: InterruptionFrame, direction: FrameDirection): - """Handle interruption by sending Clear message to Deepgram. + async def on_audio_context_interrupted(self, context_id: str): + """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(clear_msg)) + await self._websocket.send(json.dumps({"type": "Clear"})) except Exception as e: logger.error(f"{self} error sending Clear message: {e}") @@ -297,11 +282,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() - frame = TTSAudioRawFrame( - message, self.sample_rate, 1, context_id=self.get_active_audio_context_id() - ) - await self.push_frame(frame) + ctx_id = self.get_active_audio_context_id() + frame = TTSAudioRawFrame(message, self.sample_rate, 1, context_id=ctx_id) + await self.append_to_audio_context(ctx_id, frame) elif isinstance(message, str): # Text message contains metadata or control messages try: @@ -312,12 +295,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 - # This happens after flush_audio() is called + ctx_id = self.get_active_audio_context_id() + 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 - # TTSStoppedFrame will be sent by the interruption handler + # Buffer has been cleared after interruption. + # The on_audio_context_interrupted handler already cleaned up. elif msg_type == "Warning": logger.warning( f"{self} warning: {msg.get('description', 'Unknown warning')}" @@ -358,8 +344,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() From d5c0789ab585efc43b836a25f55b5216a9c60c17 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Sun, 8 Mar 2026 13:18:02 -0400 Subject: [PATCH 2/3] Add changelog for #3958 --- changelog/3958.changed.md | 1 + changelog/3958.fixed.md | 1 + 2 files changed, 2 insertions(+) create mode 100644 changelog/3958.changed.md create mode 100644 changelog/3958.fixed.md diff --git a/changelog/3958.changed.md b/changelog/3958.changed.md new file mode 100644 index 000000000..dbeabd0f2 --- /dev/null +++ b/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. diff --git a/changelog/3958.fixed.md b/changelog/3958.fixed.md new file mode 100644 index 000000000..8bc21dffe --- /dev/null +++ b/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. From 50cc01a578c0f66f478eaf1a7c2f97dad315df25 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Tue, 10 Mar 2026 11:25:59 -0400 Subject: [PATCH 3/3] Guard against None context ID in append_to_audio_context After interruption, both _playing_context_id and _turn_context_id are None. If a subclass calls append_to_audio_context(None, frame), the recovery path matches (None == None) and creates a bogus audio context that blocks the handler from ever processing the real context. Early-return when context_id is falsy to prevent this. --- src/pipecat/services/tts_service.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/pipecat/services/tts_service.py b/src/pipecat/services/tts_service.py index db695fd52..3c0161fa5 100644 --- a/src/pipecat/services/tts_service.py +++ b/src/pipecat/services/tts_service.py @@ -1182,6 +1182,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)