diff --git a/pyproject.toml b/pyproject.toml index 48226b14f..c01e039e9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -110,6 +110,7 @@ runner = [ "python-dotenv>=1.0.0,<2.0.0", "uvicorn>=0.32.0,<1.0.0", "fastapi>=0. sagemaker = ["aws_sdk_sagemaker_runtime_http2; python_version>='3.12'"] sambanova = [] sarvam = [ "sarvamai==0.1.26", "pipecat-ai[websockets-base]" ] +smallest = [ "httpx>=0.27.0,<1", "numpy>=1.24.0,<3", "pipecat-ai[soundfile]", "pipecat-ai[websockets-base]" ] sentry = [ "sentry-sdk>=2.28.0,<3" ] silero = [] simli = [ "simli-ai~=2.0.1"] diff --git a/src/pipecat/services/smallest/__init__.py b/src/pipecat/services/smallest/__init__.py new file mode 100644 index 000000000..40098b034 --- /dev/null +++ b/src/pipecat/services/smallest/__init__.py @@ -0,0 +1,14 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import sys + +from pipecat.services import DeprecatedModuleProxy + +from .stt import * +from .tts import * + +sys.modules[__name__] = DeprecatedModuleProxy(globals(), "smallest", "smallest.[stt,tts]") diff --git a/src/pipecat/services/smallest/stt.py b/src/pipecat/services/smallest/stt.py new file mode 100644 index 000000000..b23055370 --- /dev/null +++ b/src/pipecat/services/smallest/stt.py @@ -0,0 +1,252 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Smallest AI speech-to-text service implementation. + +This module provides a segmented (HTTP-based) Speech-to-Text service using +Smallest AI's Waves API. Audio is buffered during speech, then sent as a single +request once the user stops speaking (VAD-triggered). +""" + +import io +from enum import Enum +from typing import AsyncGenerator, Optional + +from loguru import logger +from pydantic import BaseModel + +from pipecat.frames.frames import ( + ErrorFrame, + Frame, + TranscriptionFrame, +) +from pipecat.services.stt_latency import SMALLEST_TTFS_P99 +from pipecat.services.stt_service import SegmentedSTTService +from pipecat.transcriptions.language import Language +from pipecat.utils.time import time_now_iso8601 +from pipecat.utils.tracing.service_decorators import traced_stt + +try: + import httpx +except ModuleNotFoundError as e: + logger.error(f"Exception: {e}") + logger.error("In order to use Smallest, you need to `pip install pipecat-ai[smallest]`.") + raise Exception(f"Missing module: {e}") + +try: + import numpy as np +except ModuleNotFoundError as e: + logger.error(f"Exception: {e}") + logger.error("In order to use Smallest, you need to `pip install pipecat-ai[smallest]`.") + raise Exception(f"Missing module: {e}") + +try: + import soundfile as sf +except ModuleNotFoundError as e: + logger.error(f"Exception: {e}") + logger.error("In order to use Smallest, you need to `pip install pipecat-ai[smallest]`.") + raise Exception(f"Missing module: {e}") + + +def language_to_smallest_language(language: Language) -> Optional[str]: + """Convert a Language enum to Smallest's language code format. + + Smallest AI currently supports English and Hindi. Falls back to extracting + the base language code if the exact Language enum isn't mapped. + + Args: + language: The Language enum value to convert. + + Returns: + The Smallest language code string, or None if unsupported. + """ + BASE_LANGUAGES = { + Language.EN: "en", + Language.HI: "hi", + } + + result = BASE_LANGUAGES.get(language) + + if not result: + lang_str = str(language.value) + base_code = lang_str.split("-")[0].lower() + result = base_code if base_code in BASE_LANGUAGES.values() else None + + return result + + +class SmallestSTTModel(str, Enum): + """Available Smallest AI STT models.""" + + LIGHTNING = "lightning" + + +class SmallestSTTService(SegmentedSTTService): + """Smallest AI speech-to-text service using the Waves HTTP API. + + This is a segmented STT service that buffers audio while the user speaks + (using VAD) and sends the complete audio segment to Smallest AI's HTTP + endpoint for transcription once the user stops speaking. + + Requires VAD to be enabled in the pipeline. + """ + + class InputParams(BaseModel): + """Configuration parameters for Smallest STT service. + + Parameters: + language: Language code for transcription. Defaults to "en". + age_detection: Enable age detection. Defaults to False. + emotion_detection: Enable emotion detection. Defaults to False. + gender_detection: Enable gender detection. Defaults to False. + """ + + language: str = "en" + age_detection: bool = False + emotion_detection: bool = False + gender_detection: bool = False + + def __init__( + self, + *, + api_key: str, + model: str = SmallestSTTModel.LIGHTNING, + url: str = "https://waves-api.smallest.ai/api/v1/lightning/get_text", + sample_rate: Optional[int] = None, + params: Optional[InputParams] = None, + ttfs_p99_latency: Optional[float] = SMALLEST_TTFS_P99, + **kwargs, + ): + """Initialize the Smallest AI STT service. + + Args: + api_key: Smallest AI API key for authentication. + model: Model to use for transcription. Defaults to "lightning". + url: API endpoint URL. Defaults to the Smallest Waves API endpoint. + sample_rate: Audio sample rate. If None, will be determined from the + start frame. + params: Configuration parameters for the STT service. + ttfs_p99_latency: P99 latency from speech end to final transcript in seconds. + Override for your deployment. + **kwargs: Additional arguments passed to the parent SegmentedSTTService. + """ + super().__init__( + sample_rate=sample_rate, + ttfs_p99_latency=ttfs_p99_latency, + **kwargs, + ) + + params = params or SmallestSTTService.InputParams() + + self._api_key = api_key + self._url = url + self._language = params.language + + model_str = model.value if isinstance(model, Enum) else model + self.set_model_name(model_str) + + self._client = httpx.AsyncClient() + self._headers = { + "Authorization": f"Bearer {self._api_key}", + } + self._payload = { + "model": model_str, + "age_detection": "true" if params.age_detection else "false", + "gender_detection": "true" if params.gender_detection else "false", + "emotion_detection": "true" if params.emotion_detection else "false", + "language": params.language, + } + + def can_generate_metrics(self) -> bool: + """Check if this service can generate processing metrics. + + Returns: + True, as Smallest STT supports metrics generation. + """ + return True + + @traced_stt + async def _handle_transcription( + self, transcript: str, is_final: bool, language: Optional[Language] = None + ): + """Handle a transcription result with tracing. + + This method is decorated with @traced_stt for observability. + The actual work (pushing frames) is done in run_stt; this method + exists solely as a tracing hook. + """ + pass + + def _audio_bytes_to_wav_buffer(self, audio: bytes) -> io.BytesIO: + """Convert raw PCM16 audio bytes to a WAV-formatted buffer. + + The Smallest API expects WAV-formatted audio. This converts raw signed + 16-bit PCM audio bytes into a WAV buffer with proper headers. + + Args: + audio: Raw PCM16 audio bytes. + + Returns: + A BytesIO buffer containing WAV-formatted audio data. + """ + audio_float = np.frombuffer(audio, dtype=np.int16).astype(np.float32) / 32768.0 + wav_buffer = io.BytesIO() + sf.write(wav_buffer, audio_float, self.sample_rate, format="WAV", subtype="PCM_16") + wav_buffer.seek(0) + return wav_buffer + + async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]: + """Transcribe audio using the Smallest AI HTTP API. + + Called by the base SegmentedSTTService when the user stops speaking. + The audio parameter contains the complete WAV-encoded speech segment. + + Args: + audio: WAV-encoded audio bytes from the speech segment. + + Yields: + TranscriptionFrame on success, ErrorFrame on failure. + """ + wav_buffer = self._audio_bytes_to_wav_buffer(audio) + + await self.start_processing_metrics() + await self.start_ttfb_metrics() + + try: + response = await self._client.post( + self._url, + headers=self._headers, + content=wav_buffer.getvalue(), + params=self._payload, + ) + response.raise_for_status() + result = response.json() + text: str = result.get("transcription", "").strip() + except httpx.HTTPStatusError as e: + logger.error(f"{self} API error: {e.response.status_code} - {e.response.text}") + yield ErrorFrame(error=f"Smallest API error: {e.response.status_code}", exception=e) + return + except Exception as e: + logger.exception(f"{self} transcription error: {type(e).__name__}: {e}") + yield ErrorFrame(error=f"Smallest transcription error: {type(e).__name__}: {e}") + return + + await self.stop_ttfb_metrics() + await self.stop_processing_metrics() + + if text: + logger.debug(f"Transcription: [{text}]") + await self._handle_transcription(text, True, self._language) + yield TranscriptionFrame( + text, + self._user_id, + time_now_iso8601(), + ) + + async def cleanup(self): + """Clean up resources used by the Smallest STT service.""" + await super().cleanup() + await self._client.aclose() diff --git a/src/pipecat/services/smallest/tts.py b/src/pipecat/services/smallest/tts.py new file mode 100644 index 000000000..d21e472e2 --- /dev/null +++ b/src/pipecat/services/smallest/tts.py @@ -0,0 +1,589 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Smallest AI text-to-speech service implementations. + +This module provides WebSocket-based and HTTP-based integrations with Smallest +AI's Waves API for real-time text-to-speech synthesis. +""" + +import base64 +import json +from enum import Enum +from typing import AsyncGenerator, Optional, Union + +from loguru import logger +from pydantic import BaseModel, Field + +from pipecat.frames.frames import ( + CancelFrame, + EndFrame, + ErrorFrame, + Frame, + InterruptionFrame, + StartFrame, + TTSAudioRawFrame, + TTSStartedFrame, + TTSStoppedFrame, +) +from pipecat.processors.frame_processor import FrameDirection +from pipecat.services.tts_service import InterruptibleTTSService, TTSService +from pipecat.transcriptions.language import Language +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 Smallest, you need to `pip install pipecat-ai[smallest]`.") + raise Exception(f"Missing module: {e}") + + +class SmallestTTSModel(str, Enum): + """Available Smallest AI TTS models.""" + + LIGHTNING_V2 = "lightning-v2" + + +def language_to_smallest_tts_language(language: Language) -> Optional[str]: + """Convert a Language enum to a Smallest TTS language string. + + Args: + language: The Language enum value to convert. + + Returns: + The Smallest language code string, or None if unsupported. + """ + BASE_LANGUAGES = { + Language.AR: "ar", + Language.BN: "bn", + Language.DE: "de", + Language.EN: "en", + Language.ES: "es", + Language.FR: "fr", + Language.GU: "gu", + Language.HE: "he", + Language.HI: "hi", + Language.IT: "it", + Language.KN: "kn", + Language.MR: "mr", + Language.NL: "nl", + Language.PL: "pl", + Language.RU: "ru", + Language.TA: "ta", + } + + result = BASE_LANGUAGES.get(language) + + if not result: + lang_str = str(language.value) + base_code = lang_str.split("-")[0].lower() + result = base_code if base_code in BASE_LANGUAGES.values() else None + + return result + + +class SmallestTTSService(InterruptibleTTSService): + """Smallest AI real-time text-to-speech service using WebSocket streaming. + + Provides real-time text-to-speech synthesis using Smallest AI's WebSocket API. + Supports streaming audio generation with configurable voice parameters and + language settings. Handles interruptions by reconnecting the WebSocket. + + Example:: + + tts = SmallestTTSService( + api_key="your-api-key", + voice_id="emily", + params=SmallestTTSService.InputParams( + language=Language.EN, + speed=1.0, + ), + ) + """ + + class InputParams(BaseModel): + """Configuration parameters for Smallest TTS service. + + Parameters: + language: Language for synthesis. Defaults to English. + speed: Speech speed multiplier. Defaults to 1.0. + consistency: Consistency level for voice generation (0-1). Defaults to 0.5. + similarity: Similarity level for voice generation (0-1). Defaults to 0. + enhancement: Enhancement level for voice generation (0-2). Defaults to 1. + """ + + language: Optional[Language] = Language.EN + speed: Optional[Union[str, float]] = 1.0 + consistency: Optional[float] = Field(default=0.5, ge=0, le=1) + similarity: Optional[float] = Field(default=0, ge=0, le=1) + enhancement: Optional[int] = Field(default=1, ge=0, le=2) + + def __init__( + self, + *, + api_key: str, + voice_id: str, + base_url: str = "wss://waves-api.smallest.ai", + model: str = SmallestTTSModel.LIGHTNING_V2, + sample_rate: Optional[int] = 24000, + params: Optional[InputParams] = None, + **kwargs, + ): + """Initialize the Smallest AI WebSocket TTS service. + + Args: + api_key: Smallest AI API key for authentication. + voice_id: Voice identifier for synthesis. + base_url: Base WebSocket URL for the Smallest API. + model: TTS model to use. Defaults to "lightning-v2". + sample_rate: Audio sample rate in Hz. Defaults to 24000. + params: Configuration parameters for the TTS service. + **kwargs: Additional arguments passed to parent InterruptibleTTSService. + """ + super().__init__( + aggregate_sentences=True, + push_text_frames=True, + pause_frame_processing=True, + sample_rate=sample_rate, + **kwargs, + ) + + params = params or SmallestTTSService.InputParams() + + self._api_key = api_key + model_str = model.value if isinstance(model, Enum) else model + self._websocket_url = f"{base_url}/api/v1/{model_str}/get_speech/stream" + + self.set_model_name(model_str) + self.set_voice(voice_id) + + self._settings = { + "language": language_to_smallest_tts_language(params.language) + if params.language + else "en", + "speed": params.speed, + "consistency": params.consistency, + "similarity": params.similarity, + "enhancement": params.enhancement, + } + + self._receive_task = None + self._context_id: Optional[str] = None + + def can_generate_metrics(self) -> bool: + """Check if this service can generate processing metrics. + + Returns: + True, as Smallest service supports metrics generation. + """ + return True + + def language_to_service_language(self, language: Language) -> Optional[str]: + """Convert a Language enum to Smallest service language format. + + Args: + language: The language to convert. + + Returns: + The Smallest-specific language code, or None if not supported. + """ + return language_to_smallest_tts_language(language) + + def _build_msg(self, text: str) -> dict: + """Build a WebSocket message for the Smallest API. + + Args: + text: The text to synthesize. + + Returns: + Dictionary with the API message payload. + """ + msg = { + "text": text, + "voice_id": self._voice_id, + "language": self._settings["language"], + "speed": self._settings["speed"], + "consistency": self._settings["consistency"], + "similarity": self._settings["similarity"], + "enhancement": self._settings["enhancement"], + } + + if self._context_id: + msg["request_id"] = self._context_id + + return msg + + async def start(self, frame: StartFrame): + """Start the Smallest 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 Smallest TTS service. + + Args: + frame: The end frame. + """ + await super().stop(frame) + await self._disconnect() + + async def cancel(self, frame: CancelFrame): + """Cancel the Smallest TTS service. + + Args: + frame: The cancel frame. + """ + await super().cancel(frame) + await self._disconnect() + + async def _connect(self): + """Connect to Smallest 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 Smallest 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 _connect_websocket(self): + """Establish WebSocket connection to the Smallest API.""" + try: + if self._websocket and self._websocket.state is State.OPEN: + return + + logger.debug("Connecting to Smallest") + + self._websocket = await websocket_connect( + self._websocket_url, + additional_headers={"Authorization": f"Bearer {self._api_key}"}, + ) + + await self._call_event_handler("on_connected") + except Exception as e: + await self.push_error(error_msg=f"Smallest connection error: {e}", exception=e) + self._websocket = None + await self._call_event_handler("on_connection_error", f"{e}") + + async def _disconnect_websocket(self): + """Close the WebSocket connection and clean up state.""" + try: + await self.stop_all_metrics() + + if self._websocket: + logger.debug("Disconnecting from Smallest") + await self._websocket.close() + except Exception as e: + logger.error(f"{self} error closing websocket: {e}") + finally: + self._context_id = None + self._websocket = None + await self._call_event_handler("on_disconnected") + + def _get_websocket(self): + """Get the WebSocket connection if available. + + Returns: + The active WebSocket connection. + + Raises: + Exception: If no WebSocket connection is available. + """ + if self._websocket: + return self._websocket + raise Exception("Websocket not connected") + + async def _handle_interruption(self, frame: InterruptionFrame, direction: FrameDirection): + """Handle an interruption by resetting state. + + Args: + frame: The interruption frame. + direction: The direction of frame processing. + """ + await super()._handle_interruption(frame, direction) + await self.stop_all_metrics() + self._context_id = None + + async def _receive_messages(self): + """Receive and process messages from the Smallest WebSocket API.""" + async for message in self._get_websocket(): + msg = json.loads(message) + status = msg.get("status") + + if status == "complete": + msg_request_id = msg.get("request_id") + if ( + self._context_id + and msg_request_id + and msg_request_id == self._context_id + ): + await self.stop_all_metrics() + await self.push_frame(TTSStoppedFrame(context_id=self._context_id)) + self._context_id = None + elif status == "chunk": + await self.stop_ttfb_metrics() + frame = TTSAudioRawFrame( + audio=base64.b64decode(msg["data"]["audio"]), + sample_rate=self.sample_rate, + num_channels=1, + context_id=self._context_id, + ) + await self.push_frame(frame) + elif status == "error": + logger.error(f"{self} error: {msg}") + await self.push_frame(TTSStoppedFrame(context_id=self._context_id)) + await self.stop_all_metrics() + await self.push_error(error_msg=f"Smallest TTS error: {msg.get('error', msg)}") + self._context_id = None + else: + logger.warning(f"{self} unknown message status: {msg}") + + @traced_tts + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: + """Generate speech from text using Smallest's WebSocket streaming API. + + Args: + text: The text to synthesize into speech. + context_id: Unique identifier for this TTS context. + + Yields: + Frame: TTSStartedFrame to signal start; audio arrives via WebSocket. + """ + logger.debug(f"{self}: Generating TTS [{text}]") + + try: + if not self._websocket or self._websocket.state is State.CLOSED: + await self._connect() + + try: + await self.start_ttfb_metrics() + self._context_id = context_id + yield TTSStartedFrame(context_id=context_id) + + msg = self._build_msg(text=text) + await self._get_websocket().send(json.dumps(msg)) + await self.start_tts_usage_metrics(text) + except Exception as e: + logger.error(f"{self} error sending message: {e}") + yield ErrorFrame(error=f"Smallest TTS send error: {e}") + yield TTSStoppedFrame(context_id=context_id) + await self._disconnect() + await self._connect() + return + yield None + except Exception as e: + logger.error(f"{self} exception: {e}") + yield ErrorFrame(error=f"Smallest TTS error: {e}") + + +class SmallestHttpTTSService(TTSService): + """Smallest AI text-to-speech service using the HTTP API. + + Provides text-to-speech synthesis using Smallest AI's HTTP REST API. + Suitable for applications that prefer simpler HTTP-based communication + over WebSocket connections. + + Example:: + + tts = SmallestHttpTTSService( + api_key="your-api-key", + voice_id="anushka", + params=SmallestHttpTTSService.InputParams( + language=Language.HI, + speed=1.2, + ), + ) + """ + + class InputParams(BaseModel): + """Configuration parameters for Smallest HTTP TTS service. + + Parameters: + language: Language code for synthesis. Defaults to "en". + speed: Speech speed multiplier. Defaults to 1.0. + consistency: Consistency level for voice generation. + similarity: Similarity level for voice generation. + enhancement: Enhancement level for voice generation. + """ + + language: str = "en" + speed: float = 1.0 + consistency: Optional[float] = None + similarity: Optional[float] = None + enhancement: Optional[float] = None + + def __init__( + self, + *, + api_key: str, + voice_id: str, + model: str = SmallestTTSModel.LIGHTNING_V2, + base_url: str = "https://waves-api.smallest.ai", + sample_rate: Optional[int] = None, + params: Optional[InputParams] = None, + **kwargs, + ): + """Initialize the Smallest AI HTTP TTS service. + + Args: + api_key: Smallest AI API key for authentication. + voice_id: Voice identifier for synthesis. + model: TTS model to use. Defaults to "lightning-v2". + base_url: Base URL for the Smallest API. + sample_rate: Audio sample rate in Hz. + params: Configuration parameters for the TTS service. + **kwargs: Additional arguments passed to parent TTSService. + """ + super().__init__(sample_rate=sample_rate, **kwargs) + + params = params or SmallestHttpTTSService.InputParams() + + self._api_key = api_key + self._base_url = base_url.rstrip("/") + + model_str = model.value if isinstance(model, Enum) else model + self.set_model_name(model_str) + self.set_voice(voice_id) + + self._model_url = f"{self._base_url}/api/v1/{model_str}/get_speech" + + self._settings = { + "language": params.language, + "speed": params.speed, + "consistency": params.consistency, + "similarity": params.similarity, + "enhancement": params.enhancement, + } + + self._session = None + + def can_generate_metrics(self) -> bool: + """Check if this service can generate processing metrics. + + Returns: + True, as Smallest HTTP service supports metrics generation. + """ + return True + + async def start(self, frame: StartFrame): + """Start the Smallest HTTP TTS service. + + Args: + frame: The start frame containing initialization parameters. + """ + await super().start(frame) + try: + import aiohttp + + self._session = aiohttp.ClientSession() + except ModuleNotFoundError as e: + logger.error(f"Exception: {e}") + logger.error( + "In order to use Smallest HTTP TTS, you need to `pip install aiohttp`." + ) + raise Exception(f"Missing module: {e}") + + async def stop(self, frame: EndFrame): + """Stop the Smallest HTTP TTS service. + + Args: + frame: The end frame. + """ + await super().stop(frame) + if self._session: + await self._session.close() + self._session = None + + async def cancel(self, frame: CancelFrame): + """Cancel the Smallest HTTP TTS service. + + Args: + frame: The cancel frame. + """ + await super().cancel(frame) + if self._session: + await self._session.close() + self._session = None + + @traced_tts + async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: + """Generate speech from text using the Smallest HTTP API. + + Args: + text: The text to synthesize into speech. + context_id: Unique identifier for this TTS context. + + Yields: + Frame: TTSStartedFrame, TTSAudioRawFrame chunks, and TTSStoppedFrame. + """ + logger.debug(f"{self}: Generating TTS [{text}]") + + if not self._session: + yield ErrorFrame(error="Smallest HTTP TTS session not initialized") + return + + try: + await self.start_ttfb_metrics() + + payload = { + "voice_id": self._voice_id, + "text": text, + "sample_rate": self.sample_rate, + } + + # Only include non-None settings + for key, value in self._settings.items(): + if value is not None: + payload[key] = value + + headers = { + "Authorization": f"Bearer {self._api_key}", + "Content-Type": "application/json", + } + + yield TTSStartedFrame(context_id=context_id) + + async with self._session.post( + self._model_url, json=payload, headers=headers + ) as response: + if response.status != 200: + error_text = await response.text() + logger.error(f"{self} API error: {error_text}") + yield ErrorFrame(error=f"Smallest API error: {error_text}") + return + + result = await response.read() + + await self.stop_ttfb_metrics() + await self.start_tts_usage_metrics(text) + + yield TTSAudioRawFrame( + audio=result, + sample_rate=self.sample_rate, + num_channels=1, + context_id=context_id, + ) + + except Exception as e: + logger.error(f"{self} exception: {e}") + yield ErrorFrame(error=f"Smallest TTS error: {e}") + finally: + yield TTSStoppedFrame(context_id=context_id)