wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add Smallest AI TTS service integration

Browse Source 
Adds SmallestTTSService, a WebSocket-based TTS service using Smallest AI's
Lightning v3.1 model. Follows current Pipecat service conventions:

- SmallestTTSSettings dataclass with runtime-updatable settings (voice,
  language, speed, etc.)
- Reconnects on model change; keepalive every 30s to prevent idle timeout
- TTS settings default to None so the API applies its own defaults
- Model enum: SmallestTTSModel.LIGHTNING_V3_1

Includes a foundational example (07zl-interruptible-smallest.py) using
Deepgram STT + Smallest TTS + OpenAI LLM.

STT integration will follow in a separate PR once the hallucination/finalize
behaviour is resolved.

Made-with: Cursor
This commit is contained in:
Harshita Jain
2026-03-20 09:30:16 -07:00
committed by Mark Backman
parent dd45843c42
commit 099814d74a
5 changed files with 549 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

122
examples/foundational/07zl-interruptible-smallest.py Normal file
View File

@@ -0,0 +1,122 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import os
from dotenv import load_dotenv
from loguru import logger
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import LLMRunFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import (
LLMContextAggregatorPair,
LLMUserAggregatorParams,
)
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.services.smallest.tts import SmallestTTSService
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import DailyParams
from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
load_dotenv(override=True)
transport_params = {
"daily": lambda: DailyParams(
audio_in_enabled=True,
audio_out_enabled=True,
),
"twilio": lambda: FastAPIWebsocketParams(
audio_in_enabled=True,
audio_out_enabled=True,
),
"webrtc": lambda: TransportParams(
audio_in_enabled=True,
audio_out_enabled=True,
),
}
async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
logger.info(f"Starting bot")
stt = DeepgramSTTService(
api_key=os.getenv("DEEPGRAM_API_KEY"),
)
tts = SmallestTTSService(
api_key=os.getenv("SMALLEST_API_KEY"),
settings=SmallestTTSService.Settings(
voice="sophia",
),
)
llm = OpenAILLMService(
api_key=os.getenv("OPENAI_API_KEY"),
settings=OpenAILLMService.Settings(
system_instruction="You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be spoken aloud, so avoid special characters that can't easily be spoken, such as emojis or bullet points. Respond to what the user said in a creative and helpful way.",
),
)
context = LLMContext()
user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
context,
user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()),
)
pipeline = Pipeline(
[
transport.input(),
stt,
user_aggregator,
llm,
tts,
transport.output(),
assistant_aggregator,
]
)
task = PipelineTask(
pipeline,
params=PipelineParams(
enable_metrics=True,
enable_usage_metrics=True,
),
)
@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
logger.info(f"Client connected")
context.add_message({"role": "user", "content": "Please introduce yourself to the user."})
await task.queue_frames([LLMRunFrame()])
@transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client):
logger.info(f"Client disconnected")
await task.cancel()
runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
await runner.run(task)
async def bot(runner_args: RunnerArguments):
"""Main bot entry point compatible with Pipecat Cloud."""
transport = await create_transport(runner_args, transport_params)
await run_bot(transport, runner_args)
if __name__ == "__main__":
from pipecat.runner.run import main
main()

1
pyproject.toml
View File

@@ -111,6 +111,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 = [ "pipecat-ai[websockets-base]" ]
sentry = [ "sentry-sdk>=2.28.0,<3" ]
silero = []
simli = [ "simli-ai~=2.0.1"]

1
src/pipecat/services/smallest/__init__.py Normal file
View File

@@ -0,0 +1 @@

424
src/pipecat/services/smallest/tts.py Normal file
View File

@@ -0,0 +1,424 @@
#
# Copyright (c) 2024-2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Smallest AI text-to-speech service implementation.
This module provides a WebSocket-based integration with Smallest AI's
Waves API for real-time text-to-speech synthesis.
"""
import asyncio
import base64
import json
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, AsyncGenerator, Optional
from loguru import logger
from pipecat.frames.frames import (
CancelFrame,
EndFrame,
ErrorFrame,
Frame,
StartFrame,
TTSAudioRawFrame,
TTSStartedFrame,
TTSStoppedFrame,
)
from pipecat.services.settings import NOT_GIVEN, TTSSettings, _NotGiven
from pipecat.services.tts_service import InterruptibleTTSService
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"
LIGHTNING_V3_1 = "lightning-v3.1"
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
@dataclass
class SmallestTTSSettings(TTSSettings):
"""Settings for SmallestTTSService.
Parameters:
speed: Speech speed multiplier.
consistency: Consistency level for voice generation (0-1).
similarity: Similarity level for voice generation (0-1).
enhancement: Enhancement level for voice generation (0-2).
"""
speed: float | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
consistency: float | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
similarity: float | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
enhancement: int | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
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",
settings=SmallestTTSService.Settings(
voice="sophia",
language="en",
speed=1.0,
),
)
"""
Settings = SmallestTTSSettings
_settings: Settings
def __init__(
self,
*,
api_key: str,
base_url: str = "wss://waves-api.smallest.ai",
sample_rate: Optional[int] = None,
settings: Optional[Settings] = None,
**kwargs,
):
"""Initialize the Smallest AI WebSocket TTS service.
Args:
api_key: Smallest AI API key for authentication.
base_url: Base WebSocket URL for the Smallest API.
sample_rate: Audio sample rate in Hz. If None, uses default.
settings: Runtime-updatable settings for the TTS service.
**kwargs: Additional arguments passed to parent InterruptibleTTSService.
"""
default_settings = self.Settings(
model=SmallestTTSModel.LIGHTNING_V3_1.value,
voice="sophia",
language=language_to_smallest_tts_language(Language.EN),
speed=None,
consistency=None,
similarity=None,
enhancement=None,
)
if settings is not None:
default_settings.apply_update(settings)
super().__init__(
aggregate_sentences=True,
push_text_frames=True,
pause_frame_processing=True,
sample_rate=sample_rate,
settings=default_settings,
**kwargs,
)
self._api_key = api_key
self._base_url = base_url.rstrip("/")
self._receive_task = None
self._keepalive_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._settings.voice,
"language": self._settings.language,
"sample_rate": self.sample_rate,
}
if self._settings.speed is not None:
msg["speed"] = self._settings.speed
if self._settings.consistency is not None:
msg["consistency"] = self._settings.consistency
if self._settings.similarity is not None:
msg["similarity"] = self._settings.similarity
if self._settings.enhancement is not None:
msg["enhancement"] = self._settings.enhancement
if self._context_id:
msg["request_id"] = self._context_id
return msg
def _build_websocket_url(self) -> str:
"""Build the WebSocket URL from base URL and model."""
return f"{self._base_url}/api/v1/{self._settings.model}/get_speech/stream"
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 _update_settings(self, delta: TTSSettings) -> dict[str, Any]:
"""Apply a settings delta, reconnecting if model changed.
Per-message fields (speed, consistency, similarity, enhancement, voice,
language) apply automatically on the next ``_build_msg`` call. A model
change requires reconnecting because the model is part of the WebSocket URL.
"""
changed = await super()._update_settings(delta)
if not changed:
return changed
if "model" in changed:
await self._disconnect()
await self._connect()
return changed
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))
if self._websocket and not self._keepalive_task:
self._keepalive_task = self.create_task(self._keepalive_task_handler())
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
if self._keepalive_task:
await self.cancel_task(self._keepalive_task)
self._keepalive_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 TTS")
self._websocket = await websocket_connect(
self._build_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 TTS 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 TTS")
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 _keepalive_task_handler(self):
"""Send periodic keepalive messages to prevent idle timeout."""
KEEPALIVE_INTERVAL = 30
while True:
await asyncio.sleep(KEEPALIVE_INTERVAL)
await self._send_keepalive()
async def _send_keepalive(self):
"""Send a flush message to keep the connection alive."""
if self._websocket and self._websocket.state is State.OPEN:
msg = {"flush": True}
await self._websocket.send(json.dumps(msg))
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:
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}")

2
uv.lock generated
View File

@@ -8480,4 +8480,4 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/c2/38/f249a2050ad1eea0bb364046153942e34abba95dd5520af199aed86fbb49/zstandard-0.25.0-cp314-cp314-win32.whl", hash = "sha256:da469dc041701583e34de852d8634703550348d5822e66a0c827d39b05365b12", size = 444513, upload-time = "2025-09-14T22:18:20.61Z" },
{ url = "https://files.pythonhosted.org/packages/3a/43/241f9615bcf8ba8903b3f0432da069e857fc4fd1783bd26183db53c4804b/zstandard-0.25.0-cp314-cp314-win_amd64.whl", hash = "sha256:c19bcdd826e95671065f8692b5a4aa95c52dc7a02a4c5a0cac46deb879a017a2", size = 516118, upload-time = "2025-09-14T22:18:17.849Z" },
{ url = "https://files.pythonhosted.org/packages/f0/ef/da163ce2450ed4febf6467d77ccb4cd52c4c30ab45624bad26ca0a27260c/zstandard-0.25.0-cp314-cp314-win_arm64.whl", hash = "sha256:d7541afd73985c630bafcd6338d2518ae96060075f9463d7dc14cfb33514383d", size = 476940, upload-time = "2025-09-14T22:18:19.088Z" },
]
]