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)
This commit is contained in:
Mark Backman
2026-03-08 13:16:42 -04:00
parent 764c3c4f32
commit 92b5185165

View File

@@ -22,12 +22,10 @@ from pipecat.frames.frames import (
EndFrame, EndFrame,
ErrorFrame, ErrorFrame,
Frame, Frame,
InterruptionFrame,
LLMFullResponseEndFrame,
StartFrame, StartFrame,
TTSAudioRawFrame, TTSAudioRawFrame,
TTSStoppedFrame,
) )
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.settings import TTSSettings, _warn_deprecated_param from pipecat.services.settings import TTSSettings, _warn_deprecated_param
from pipecat.services.tts_service import TTSService, WebsocketTTSService from pipecat.services.tts_service import TTSService, WebsocketTTSService
from pipecat.utils.tracing.service_decorators import traced_tts from pipecat.utils.tracing.service_decorators import traced_tts
@@ -119,7 +117,7 @@ class DeepgramTTSService(WebsocketTTSService):
super().__init__( super().__init__(
sample_rate=sample_rate, sample_rate=sample_rate,
pause_frame_processing=True, pause_frame_processing=True,
push_stop_frames=True, push_stop_frames=False,
push_start_frame=True, push_start_frame=True,
append_trailing_space=True, append_trailing_space=True,
settings=default_settings, settings=default_settings,
@@ -167,19 +165,6 @@ class DeepgramTTSService(WebsocketTTSService):
await super().cancel(frame) await super().cancel(frame)
await self._disconnect() 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): async def _connect(self):
"""Connect to Deepgram WebSocket and start receive task.""" """Connect to Deepgram WebSocket and start receive task."""
await super()._connect() await super()._connect()
@@ -276,19 +261,19 @@ class DeepgramTTSService(WebsocketTTSService):
return self._websocket return self._websocket
raise Exception("Websocket not connected") 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 The Clear message will clear Deepgram's internal text buffer and stop
sending audio, allowing for a new response to be generated. 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: if self._websocket:
try: 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: except Exception as e:
logger.error(f"{self} error sending Clear message: {e}") logger.error(f"{self} error sending Clear message: {e}")
@@ -297,11 +282,9 @@ class DeepgramTTSService(WebsocketTTSService):
async for message in self._get_websocket(): async for message in self._get_websocket():
if isinstance(message, bytes): if isinstance(message, bytes):
# Binary message contains audio data # 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): elif isinstance(message, str):
# Text message contains metadata or control messages # Text message contains metadata or control messages
try: try:
@@ -312,12 +295,15 @@ class DeepgramTTSService(WebsocketTTSService):
logger.trace(f"Received metadata: {msg}") logger.trace(f"Received metadata: {msg}")
elif msg_type == "Flushed": elif msg_type == "Flushed":
logger.trace(f"Received Flushed: {msg}") 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": elif msg_type == "Cleared":
logger.trace(f"Received Cleared: {msg}") 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": elif msg_type == "Warning":
logger.warning( logger.warning(
f"{self} warning: {msg.get('description', 'Unknown 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: if not self._websocket or self._websocket.state is State.CLOSED:
await self._connect() await self._connect()
await self.start_tts_usage_metrics(text)
# Send text message to Deepgram # Send text message to Deepgram
# Note: We don't send Flush here - that should only be sent when the # Note: We don't send Flush here - that should only be sent when the
# LLM finishes a complete response via flush_audio() # LLM finishes a complete response via flush_audio()