diff --git a/CHANGELOG.md b/CHANGELOG.md index a398330e7..a3adbdc38 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,7 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added -- Added `cache_read_input_tokens`, `cache_creation_input_tokens` and +- Added `cache_read_input_tokens`, `cache_creation_input_tokens` and `reasoning_tokens` to OTel spans for LLM call - Added `LiveKitRESTHelper` utility class for managing LiveKit rooms via REST API. @@ -95,6 +95,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Updated `FishAudioTTSService` default model to `s1`. +- Updated `DeepgramTTSService` to use Deepgram's TTS websocket API. ⚠️ This is + a potential breaking change, which only affects you if you're self-hosting + `DeepgramTTSService`. The new service uses Websockets and improves TTFB + latency. + - Updated `daily-python` to 0.22.0. - `BaseTextAggregator` changes: diff --git a/pyproject.toml b/pyproject.toml index f3e7350c4..078d8e6d7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -55,7 +55,7 @@ azure = [ "azure-cognitiveservices-speech~=1.42.0"] cartesia = [ "cartesia~=2.0.3", "pipecat-ai[websockets-base]" ] cerebras = [] daily = [ "daily-python~=0.22.0" ] -deepgram = [ "deepgram-sdk~=4.7.0" ] +deepgram = [ "deepgram-sdk~=4.7.0", "pipecat-ai[websockets-base]" ] deepseek = [] elevenlabs = [ "pipecat-ai[websockets-base]" ] fal = [ "fal-client~=0.5.9" ] diff --git a/src/pipecat/services/deepgram/tts.py b/src/pipecat/services/deepgram/tts.py index f75d40b09..08e0ab2c9 100644 --- a/src/pipecat/services/deepgram/tts.py +++ b/src/pipecat/services/deepgram/tts.py @@ -10,35 +10,45 @@ This module provides integration with Deepgram's text-to-speech API for generating speech from text using various voice models. """ +import json from typing import AsyncGenerator, Optional import aiohttp from loguru import logger from pipecat.frames.frames import ( + CancelFrame, + EndFrame, ErrorFrame, Frame, + InterruptionFrame, + LLMFullResponseEndFrame, + StartFrame, TTSAudioRawFrame, TTSStartedFrame, TTSStoppedFrame, ) -from pipecat.services.tts_service import TTSService +from pipecat.processors.frame_processor import FrameDirection +from pipecat.services.tts_service import TTSService, WebsocketTTSService from pipecat.utils.tracing.service_decorators import traced_tts try: - from deepgram import DeepgramClient, DeepgramClientOptions, SpeakOptions + from websockets.asyncio.client import connect as websocket_connect + from websockets.protocol import State except ModuleNotFoundError as e: logger.error(f"Exception: {e}") - logger.error("In order to use Deepgram, you need to `pip install pipecat-ai[deepgram]`.") + logger.error( + "In order to use DeepgramWebsocketTTSService, you need to `pip install pipecat-ai[deepgram]`." + ) raise Exception(f"Missing module: {e}") -class DeepgramTTSService(TTSService): - """Deepgram text-to-speech service. +class DeepgramTTSService(WebsocketTTSService): + """Deepgram WebSocket-based text-to-speech service. - Provides text-to-speech synthesis using Deepgram's streaming API. - Supports various voice models and audio encoding formats with - configurable sample rates and quality settings. + Provides real-time text-to-speech synthesis using Deepgram's WebSocket API. + Supports streaming audio generation with interruption handling via the Clear + message for conversational AI use cases. """ def __init__( @@ -46,42 +56,211 @@ class DeepgramTTSService(TTSService): *, api_key: str, voice: str = "aura-2-helena-en", - base_url: str = "", + base_url: str = "wss://api.deepgram.com", sample_rate: Optional[int] = None, encoding: str = "linear16", **kwargs, ): - """Initialize the Deepgram TTS service. + """Initialize the Deepgram WebSocket TTS service. Args: api_key: Deepgram API key for authentication. voice: Voice model to use for synthesis. Defaults to "aura-2-helena-en". - base_url: Custom base URL for Deepgram API. Uses default if empty. + base_url: WebSocket base URL for Deepgram API. Defaults to "wss://api.deepgram.com". sample_rate: Audio sample rate in Hz. If None, uses service default. encoding: Audio encoding format. Defaults to "linear16". - **kwargs: Additional arguments passed to parent TTSService class. + **kwargs: Additional arguments passed to parent InterruptibleTTSService class. """ super().__init__(sample_rate=sample_rate, **kwargs) + self._api_key = api_key + self._base_url = base_url self._settings = { "encoding": encoding, } self.set_voice(voice) - client_options = DeepgramClientOptions(url=base_url) - self._deepgram_client = DeepgramClient(api_key, config=client_options) + self._receive_task = None def can_generate_metrics(self) -> bool: """Check if the service can generate metrics. Returns: - True, as Deepgram TTS service supports metrics generation. + True, as Deepgram WebSocket TTS service supports metrics generation. """ return True + async def start(self, frame: StartFrame): + """Start the Deepgram WebSocket TTS service. + + Args: + frame: The start frame containing initialization parameters. + """ + await super().start(frame) + await self._connect() + + async def stop(self, frame: EndFrame): + """Stop the Deepgram WebSocket TTS service. + + Args: + frame: The end frame. + """ + await super().stop(frame) + await self._disconnect() + + async def cancel(self, frame: CancelFrame): + """Cancel the Deepgram WebSocket TTS service. + + Args: + frame: The cancel frame. + """ + 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 self._connect_websocket() + + if self._websocket and not self._receive_task: + self._receive_task = self.create_task(self._receive_task_handler(self._report_error)) + + async def _disconnect(self): + """Disconnect from Deepgram WebSocket and clean up tasks.""" + if self._receive_task: + await self.cancel_task(self._receive_task) + self._receive_task = None + + await self._disconnect_websocket() + + async def _connect_websocket(self): + """Connect to Deepgram WebSocket API with configured settings.""" + try: + if self._websocket and self._websocket.state is State.OPEN: + return + + logger.debug("Connecting to Deepgram WebSocket") + + # Build WebSocket URL with query parameters + params = [] + params.append(f"model={self._voice_id}") + params.append(f"encoding={self._settings['encoding']}") + params.append(f"sample_rate={self.sample_rate}") + + url = f"{self._base_url}/v1/speak?{'&'.join(params)}" + + headers = {"Authorization": f"Token {self._api_key}"} + + self._websocket = await websocket_connect(url, additional_headers=headers) + + await self._call_event_handler("on_connected") + except Exception as e: + logger.error(f"{self} exception: {e}") + await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + self._websocket = None + await self._call_event_handler("on_connection_error", f"{e}") + + async def _disconnect_websocket(self): + """Close WebSocket connection and reset state.""" + try: + await self.stop_all_metrics() + + if self._websocket: + logger.debug("Disconnecting from Deepgram WebSocket") + # Send Close message to gracefully close the connection + await self._websocket.send(json.dumps({"type": "Close"})) + await self._websocket.close() + except Exception as e: + logger.error(f"{self} exception: {e}") + await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + finally: + self._websocket = None + await self._call_event_handler("on_disconnected") + + def _get_websocket(self): + """Get active websocket connection or raise exception.""" + if self._websocket: + 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. + + 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 + if self._websocket: + try: + clear_msg = {"type": "Clear"} + await self._websocket.send(json.dumps(clear_msg)) + except Exception as e: + logger.error(f"{self} error sending Clear message: {e}") + + async def _receive_messages(self): + """Receive and process messages from Deepgram WebSocket.""" + 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) + await self.push_frame(frame) + elif isinstance(message, str): + # Text message contains metadata or control messages + try: + msg = json.loads(message) + msg_type = msg.get("type") + + if msg_type == "Metadata": + 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 + await self.push_frame(TTSStoppedFrame()) + elif msg_type == "Cleared": + logger.trace(f"Received Cleared: {msg}") + # Buffer has been cleared after interruption + # TTSStoppedFrame will be sent by the interruption handler + elif msg_type == "Warning": + logger.warning( + f"{self} warning: {msg.get('description', 'Unknown warning')}" + ) + else: + logger.debug(f"Received unknown message type: {msg}") + except json.JSONDecodeError: + logger.error(f"Invalid JSON message: {message}") + + async def flush_audio(self): + """Flush any pending audio synthesis by sending Flush command. + + This should be called when the LLM finishes a complete response to force + generation of audio from Deepgram's internal text buffer. + """ + if self._websocket: + try: + flush_msg = {"type": "Flush"} + await self._websocket.send(json.dumps(flush_msg)) + except Exception as e: + logger.error(f"{self} error sending Flush message: {e}") + @traced_tts async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: - """Generate speech from text using Deepgram's TTS API. + """Generate speech from text using Deepgram's WebSocket TTS API. Args: text: The text to synthesize into speech. @@ -91,29 +270,24 @@ class DeepgramTTSService(TTSService): """ logger.debug(f"{self}: Generating TTS [{text}]") - options = SpeakOptions( - model=self._voice_id, - encoding=self._settings["encoding"], - sample_rate=self.sample_rate, - container="none", - ) - try: + # Reconnect if the websocket is closed + if not self._websocket or self._websocket.state is State.CLOSED: + await self._connect() + await self.start_ttfb_metrics() - - response = await self._deepgram_client.speak.asyncrest.v("1").stream_raw( - {"text": text}, options - ) - await self.start_tts_usage_metrics(text) + yield TTSStartedFrame() - async for data in response.aiter_bytes(): - await self.stop_ttfb_metrics() - if data: - yield TTSAudioRawFrame(audio=data, sample_rate=self.sample_rate, num_channels=1) + # 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() + speak_msg = {"type": "Speak", "text": text} + await self._get_websocket().send(json.dumps(speak_msg)) - yield TTSStoppedFrame() + # The actual audio frames will be handled in _receive_messages + yield None except Exception as e: logger.error(f"{self} exception: {e}") diff --git a/uv.lock b/uv.lock index 3db38f2a9..35ae93faf 100644 --- a/uv.lock +++ b/uv.lock @@ -4475,6 +4475,7 @@ daily = [ ] deepgram = [ { name = "deepgram-sdk" }, + { name = "websockets" }, ] elevenlabs = [ { name = "websockets" }, @@ -4720,6 +4721,7 @@ requires-dist = [ { name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'asyncai'" }, { name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'aws'" }, { name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'cartesia'" }, + { name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'deepgram'" }, { name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'elevenlabs'" }, { name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'fish'" }, { name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'gladia'" },