e5b60ba095
Move the warning helper into AIService as _warn_init_param_moved_to_settings. It now uses type(self).__name__ to produce messages like "Use settings=AnthropicLLMService.Settings(model=...)" instead of the raw settings class name "AnthropicLLMSettings(model=...)". Callers no longer need to pass the settings class explicitly.
500 lines
18 KiB
Python
500 lines
18 KiB
Python
#
|
|
# Copyright (c) 2024-2026, Daily
|
|
#
|
|
# SPDX-License-Identifier: BSD 2-Clause License
|
|
#
|
|
|
|
"""Deepgram text-to-speech service implementation.
|
|
|
|
This module provides integration with Deepgram's text-to-speech API
|
|
for generating speech from text using various voice models.
|
|
"""
|
|
|
|
import json
|
|
from dataclasses import dataclass
|
|
from typing import Any, AsyncGenerator, Optional
|
|
|
|
import aiohttp
|
|
from loguru import logger
|
|
|
|
from pipecat.frames.frames import (
|
|
CancelFrame,
|
|
EndFrame,
|
|
ErrorFrame,
|
|
Frame,
|
|
StartFrame,
|
|
TTSAudioRawFrame,
|
|
TTSStoppedFrame,
|
|
)
|
|
from pipecat.services.settings import TTSSettings
|
|
from pipecat.services.tts_service import TTSService, WebsocketTTSService
|
|
from pipecat.utils.tracing.service_decorators import traced_tts
|
|
|
|
try:
|
|
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 DeepgramWebsocketTTSService, you need to `pip install pipecat-ai[deepgram]`."
|
|
)
|
|
raise Exception(f"Missing module: {e}")
|
|
|
|
|
|
@dataclass
|
|
class DeepgramTTSSettings(TTSSettings):
|
|
"""Settings for DeepgramTTSService and DeepgramHttpTTSService."""
|
|
|
|
pass
|
|
|
|
|
|
class DeepgramTTSService(WebsocketTTSService):
|
|
"""Deepgram WebSocket-based text-to-speech service.
|
|
|
|
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.
|
|
"""
|
|
|
|
Settings = DeepgramTTSSettings
|
|
_settings: Settings
|
|
|
|
SUPPORTED_ENCODINGS = ("linear16", "mulaw", "alaw")
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
api_key: str,
|
|
voice: Optional[str] = None,
|
|
base_url: str = "wss://api.deepgram.com",
|
|
sample_rate: Optional[int] = None,
|
|
encoding: str = "linear16",
|
|
settings: Optional[Settings] = None,
|
|
**kwargs,
|
|
):
|
|
"""Initialize the Deepgram WebSocket TTS service.
|
|
|
|
Args:
|
|
api_key: Deepgram API key for authentication.
|
|
voice: Voice model to use for synthesis.
|
|
|
|
.. deprecated:: 0.0.105
|
|
Use ``settings=DeepgramTTSService.Settings(voice=...)`` instead.
|
|
|
|
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". Must be one of SUPPORTED_ENCODINGS.
|
|
settings: Runtime-updatable settings. When provided alongside deprecated
|
|
parameters, ``settings`` values take precedence.
|
|
**kwargs: Additional arguments passed to parent InterruptibleTTSService class.
|
|
|
|
Raises:
|
|
ValueError: If encoding is not in SUPPORTED_ENCODINGS.
|
|
"""
|
|
if encoding.lower() not in self.SUPPORTED_ENCODINGS:
|
|
raise ValueError(
|
|
f"Unsupported encoding '{encoding}'. Must be one of {', '.join(self.SUPPORTED_ENCODINGS)} for WebSocket TTS."
|
|
)
|
|
|
|
# 1. Initialize default_settings with hardcoded defaults
|
|
default_settings = self.Settings(
|
|
model=None,
|
|
voice="aura-2-helena-en",
|
|
language=None,
|
|
)
|
|
|
|
# 2. Apply direct init arg overrides (deprecated)
|
|
if voice is not None:
|
|
self._warn_init_param_moved_to_settings("voice", "voice")
|
|
default_settings.model = voice
|
|
default_settings.voice = voice
|
|
|
|
# 3. (No step 3, as there's no params object to apply)
|
|
|
|
# 4. Apply settings delta (canonical API, always wins)
|
|
if settings is not None:
|
|
default_settings.apply_update(settings)
|
|
|
|
super().__init__(
|
|
sample_rate=sample_rate,
|
|
pause_frame_processing=True,
|
|
push_stop_frames=False,
|
|
push_start_frame=True,
|
|
append_trailing_space=True,
|
|
settings=default_settings,
|
|
**kwargs,
|
|
)
|
|
|
|
self._api_key = api_key
|
|
self._base_url = base_url
|
|
self._encoding = encoding
|
|
|
|
self._receive_task = None
|
|
|
|
def can_generate_metrics(self) -> bool:
|
|
"""Check if the service can generate metrics.
|
|
|
|
Returns:
|
|
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 _connect(self):
|
|
"""Connect to Deepgram WebSocket and start receive task."""
|
|
await super()._connect()
|
|
|
|
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."""
|
|
await super()._disconnect()
|
|
|
|
if self._receive_task:
|
|
await self.cancel_task(self._receive_task)
|
|
self._receive_task = None
|
|
|
|
await self._disconnect_websocket()
|
|
|
|
async def _update_settings(self, delta: TTSSettings) -> dict[str, Any]:
|
|
"""Apply a settings delta.
|
|
|
|
Args:
|
|
delta: A :class:`TTSSettings` (or ``DeepgramTTSService.Settings``) delta.
|
|
|
|
Returns:
|
|
Dict mapping changed field names to their previous values.
|
|
"""
|
|
changed = await super()._update_settings(delta)
|
|
|
|
# Deepgram uses voice as the model, so keep them in sync for metrics
|
|
if "voice" in changed:
|
|
self._settings.model = self._settings.voice
|
|
self._sync_model_name_to_metrics()
|
|
|
|
if changed:
|
|
await self._disconnect()
|
|
await self._connect()
|
|
|
|
return changed
|
|
|
|
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._settings.voice}")
|
|
params.append(f"encoding={self._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)
|
|
|
|
headers = {
|
|
k: v for k, v in self._websocket.response.headers.items() if k.startswith("dg-")
|
|
}
|
|
logger.debug(f'{self}: Websocket connection initialized: {{"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 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.
|
|
|
|
Args:
|
|
context_id: The ID of the audio context that was interrupted.
|
|
"""
|
|
await self.stop_all_metrics()
|
|
if self._websocket:
|
|
try:
|
|
await self._websocket.send(json.dumps({"type": "Clear"}))
|
|
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
|
|
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:
|
|
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}")
|
|
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.
|
|
# The on_audio_context_interrupted handler already cleaned up.
|
|
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, context_id: Optional[str] = None):
|
|
"""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, context_id: str) -> AsyncGenerator[Frame, None]:
|
|
"""Generate speech from text using Deepgram's WebSocket TTS API.
|
|
|
|
Args:
|
|
text: The text to synthesize into speech.
|
|
context_id: The context ID for tracking audio frames.
|
|
|
|
Yields:
|
|
Frame: Audio frames containing the synthesized speech, plus start/stop frames.
|
|
"""
|
|
logger.debug(f"{self}: Generating TTS [{text}]")
|
|
|
|
try:
|
|
# Reconnect if the websocket is closed
|
|
if not self._websocket or self._websocket.state is State.CLOSED:
|
|
await self._connect()
|
|
|
|
# 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))
|
|
|
|
# The audio frames will be handled in _receive_messages
|
|
yield None
|
|
|
|
except Exception as e:
|
|
yield ErrorFrame(error=f"Unknown error occurred: {e}")
|
|
|
|
|
|
class DeepgramHttpTTSService(TTSService):
|
|
"""Deepgram HTTP text-to-speech service.
|
|
|
|
Provides text-to-speech synthesis using Deepgram's HTTP TTS API.
|
|
Supports various voice models and audio encoding formats with
|
|
configurable sample rates and quality settings.
|
|
"""
|
|
|
|
Settings = DeepgramTTSSettings
|
|
_settings: Settings
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
api_key: str,
|
|
voice: Optional[str] = None,
|
|
aiohttp_session: aiohttp.ClientSession,
|
|
base_url: str = "https://api.deepgram.com",
|
|
sample_rate: Optional[int] = None,
|
|
encoding: str = "linear16",
|
|
settings: Optional[Settings] = None,
|
|
**kwargs,
|
|
):
|
|
"""Initialize the Deepgram TTS service.
|
|
|
|
Args:
|
|
api_key: Deepgram API key for authentication.
|
|
voice: Voice model to use for synthesis.
|
|
|
|
.. deprecated:: 0.0.105
|
|
Use ``settings=DeepgramHttpTTSService.Settings(voice=...)`` instead.
|
|
|
|
aiohttp_session: Shared aiohttp session for HTTP requests with connection pooling.
|
|
base_url: Custom base URL for Deepgram API. Defaults to "https://api.deepgram.com".
|
|
sample_rate: Audio sample rate in Hz. If None, uses service default.
|
|
encoding: Audio encoding format. Defaults to "linear16".
|
|
settings: Runtime-updatable settings. When provided alongside deprecated
|
|
parameters, ``settings`` values take precedence.
|
|
**kwargs: Additional arguments passed to parent TTSService class.
|
|
"""
|
|
# 1. Initialize default_settings with hardcoded defaults
|
|
default_settings = self.Settings(
|
|
model=None,
|
|
voice="aura-2-helena-en",
|
|
language=None,
|
|
)
|
|
|
|
# 2. Apply direct init arg overrides (deprecated)
|
|
if voice is not None:
|
|
self._warn_init_param_moved_to_settings("voice", "voice")
|
|
default_settings.model = voice
|
|
default_settings.voice = voice
|
|
|
|
# 3. (No step 3, as there's no params object to apply)
|
|
|
|
# 4. Apply settings delta (canonical API, always wins)
|
|
if settings is not None:
|
|
default_settings.apply_update(settings)
|
|
|
|
super().__init__(
|
|
sample_rate=sample_rate,
|
|
push_start_frame=True,
|
|
push_stop_frames=True,
|
|
settings=default_settings,
|
|
**kwargs,
|
|
)
|
|
|
|
self._api_key = api_key
|
|
self._session = aiohttp_session
|
|
self._base_url = base_url
|
|
self._encoding = encoding
|
|
|
|
def can_generate_metrics(self) -> bool:
|
|
"""Check if the service can generate metrics.
|
|
|
|
Returns:
|
|
True, as Deepgram TTS service supports metrics generation.
|
|
"""
|
|
return True
|
|
|
|
@traced_tts
|
|
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
|
|
"""Generate speech from text using Deepgram's TTS API.
|
|
|
|
Args:
|
|
text: The text to synthesize into speech.
|
|
context_id: The context ID for tracking audio frames.
|
|
|
|
Yields:
|
|
Frame: Audio frames containing the synthesized speech, plus start/stop frames.
|
|
"""
|
|
logger.debug(f"{self}: Generating TTS [{text}]")
|
|
|
|
# Build URL with parameters
|
|
url = f"{self._base_url}/v1/speak"
|
|
|
|
headers = {"Authorization": f"Token {self._api_key}", "Content-Type": "application/json"}
|
|
|
|
params = {
|
|
"model": self._settings.voice,
|
|
"encoding": self._encoding,
|
|
"sample_rate": self.sample_rate,
|
|
"container": "none",
|
|
}
|
|
|
|
payload = {
|
|
"text": text,
|
|
}
|
|
|
|
try:
|
|
await self.start_ttfb_metrics()
|
|
|
|
async with self._session.post(
|
|
url, headers=headers, json=payload, params=params
|
|
) as response:
|
|
if response.status != 200:
|
|
error_text = await response.text()
|
|
raise Exception(f"HTTP {response.status}: {error_text}")
|
|
|
|
await self.start_tts_usage_metrics(text)
|
|
|
|
CHUNK_SIZE = self.chunk_size
|
|
|
|
first_chunk = True
|
|
async for chunk in response.content.iter_chunked(CHUNK_SIZE):
|
|
if first_chunk:
|
|
await self.stop_ttfb_metrics()
|
|
first_chunk = False
|
|
|
|
if chunk:
|
|
yield TTSAudioRawFrame(
|
|
audio=chunk,
|
|
sample_rate=self.sample_rate,
|
|
num_channels=1,
|
|
context_id=context_id,
|
|
)
|
|
|
|
except Exception as e:
|
|
yield ErrorFrame(f"Error getting audio: {str(e)}")
|