From 115f1c4008f18ad1227d3615fbd70531a9b8c532 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 12 Mar 2026 18:22:02 -0400 Subject: [PATCH] Changes to align with Pipecat norms --- README.md | 26 +- .../07zl-interruptible-smallest-http.py | 122 ------ .../07zl-interruptible-smallest.py | 26 +- pyproject.toml | 2 +- src/pipecat/services/smallest/__init__.py | 13 - src/pipecat/services/smallest/stt.py | 369 ++++++------------ src/pipecat/services/smallest/tts.py | 365 +++++------------ src/pipecat/services/stt_latency.py | 1 - uv.lock | 7 - 9 files changed, 241 insertions(+), 690 deletions(-) delete mode 100644 examples/foundational/07zl-interruptible-smallest-http.py diff --git a/README.md b/README.md index 8af4f942c..631350377 100644 --- a/README.md +++ b/README.md @@ -81,19 +81,19 @@ Catch new features, interviews, and how-tos on our [Pipecat TV](https://www.yout ## 🧩 Available services -| Category | Services | -| ------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| Speech-to-Text | [AssemblyAI](https://docs.pipecat.ai/server/services/stt/assemblyai), [AWS](https://docs.pipecat.ai/server/services/stt/aws), [Azure](https://docs.pipecat.ai/server/services/stt/azure), [Cartesia](https://docs.pipecat.ai/server/services/stt/cartesia), [Deepgram](https://docs.pipecat.ai/server/services/stt/deepgram), [ElevenLabs](https://docs.pipecat.ai/server/services/stt/elevenlabs), [Fal Wizper](https://docs.pipecat.ai/server/services/stt/fal), [Gladia](https://docs.pipecat.ai/server/services/stt/gladia), [Google](https://docs.pipecat.ai/server/services/stt/google), [Gradium](https://docs.pipecat.ai/server/services/stt/gradium), [Groq (Whisper)](https://docs.pipecat.ai/server/services/stt/groq), [NVIDIA Riva](https://docs.pipecat.ai/server/services/stt/riva), [OpenAI (Whisper)](https://docs.pipecat.ai/server/services/stt/openai), [SambaNova (Whisper)](https://docs.pipecat.ai/server/services/stt/sambanova), [Sarvam](https://docs.pipecat.ai/server/services/stt/sarvam), [Soniox](https://docs.pipecat.ai/server/services/stt/soniox), [Speechmatics](https://docs.pipecat.ai/server/services/stt/speechmatics), [Whisper](https://docs.pipecat.ai/server/services/stt/whisper) | -| LLMs | [Anthropic](https://docs.pipecat.ai/server/services/llm/anthropic), [AWS](https://docs.pipecat.ai/server/services/llm/aws), [Azure](https://docs.pipecat.ai/server/services/llm/azure), [Cerebras](https://docs.pipecat.ai/server/services/llm/cerebras), [DeepSeek](https://docs.pipecat.ai/server/services/llm/deepseek), [Fireworks AI](https://docs.pipecat.ai/server/services/llm/fireworks), [Gemini](https://docs.pipecat.ai/server/services/llm/gemini), [Grok](https://docs.pipecat.ai/server/services/llm/grok), [Groq](https://docs.pipecat.ai/server/services/llm/groq), [Mistral](https://docs.pipecat.ai/server/services/llm/mistral), [NVIDIA NIM](https://docs.pipecat.ai/server/services/llm/nim), [Ollama](https://docs.pipecat.ai/server/services/llm/ollama), [OpenAI](https://docs.pipecat.ai/server/services/llm/openai), [OpenRouter](https://docs.pipecat.ai/server/services/llm/openrouter), [Perplexity](https://docs.pipecat.ai/server/services/llm/perplexity), [Qwen](https://docs.pipecat.ai/server/services/llm/qwen), [SambaNova](https://docs.pipecat.ai/server/services/llm/sambanova) [Together AI](https://docs.pipecat.ai/server/services/llm/together) | -| Text-to-Speech | [Async](https://docs.pipecat.ai/server/services/tts/asyncai), [AWS](https://docs.pipecat.ai/server/services/tts/aws), [Azure](https://docs.pipecat.ai/server/services/tts/azure), [Camb AI](https://docs.pipecat.ai/server/services/tts/camb), [Cartesia](https://docs.pipecat.ai/server/services/tts/cartesia), [Deepgram](https://docs.pipecat.ai/server/services/tts/deepgram), [ElevenLabs](https://docs.pipecat.ai/server/services/tts/elevenlabs), [Fish](https://docs.pipecat.ai/server/services/tts/fish), [Google](https://docs.pipecat.ai/server/services/tts/google), [Gradium](https://docs.pipecat.ai/server/services/tts/gradium), [Groq](https://docs.pipecat.ai/server/services/tts/groq), [Hume](https://docs.pipecat.ai/server/services/tts/hume), [Inworld](https://docs.pipecat.ai/server/services/tts/inworld), [LMNT](https://docs.pipecat.ai/server/services/tts/lmnt), [MiniMax](https://docs.pipecat.ai/server/services/tts/minimax), [Neuphonic](https://docs.pipecat.ai/server/services/tts/neuphonic), [NVIDIA Riva](https://docs.pipecat.ai/server/services/tts/riva), [OpenAI](https://docs.pipecat.ai/server/services/tts/openai), [Piper](https://docs.pipecat.ai/server/services/tts/piper), [Resemble](https://docs.pipecat.ai/server/services/tts/resemble), [Rime](https://docs.pipecat.ai/server/services/tts/rime), [Sarvam](https://docs.pipecat.ai/server/services/tts/sarvam), [Speechmatics](https://docs.pipecat.ai/server/services/tts/speechmatics), [XTTS](https://docs.pipecat.ai/server/services/tts/xtts) | -| Speech-to-Speech | [AWS Nova Sonic](https://docs.pipecat.ai/server/services/s2s/aws), [Gemini Multimodal Live](https://docs.pipecat.ai/server/services/s2s/gemini), [Grok Voice Agent](https://docs.pipecat.ai/server/services/s2s/grok), [OpenAI Realtime](https://docs.pipecat.ai/server/services/s2s/openai), [Ultravox](https://docs.pipecat.ai/server/services/s2s/ultravox), | -| Transport | [Daily (WebRTC)](https://docs.pipecat.ai/server/services/transport/daily), [FastAPI Websocket](https://docs.pipecat.ai/server/services/transport/fastapi-websocket), [SmallWebRTCTransport](https://docs.pipecat.ai/server/services/transport/small-webrtc), [WebSocket Server](https://docs.pipecat.ai/server/services/transport/websocket-server), Local | -| Serializers | [Exotel](https://docs.pipecat.ai/server/utilities/serializers/exotel), [Plivo](https://docs.pipecat.ai/server/utilities/serializers/plivo), [Twilio](https://docs.pipecat.ai/server/utilities/serializers/twilio), [Telnyx](https://docs.pipecat.ai/server/utilities/serializers/telnyx), [Vonage](https://docs.pipecat.ai/server/utilities/serializers/vonage) | -| Video | [HeyGen](https://docs.pipecat.ai/server/services/video/heygen), [LemonSlice](https://docs.pipecat.ai/server/services/video/lemonslice), [Tavus](https://docs.pipecat.ai/server/services/video/tavus), [Simli](https://docs.pipecat.ai/server/services/video/simli) | -| Memory | [mem0](https://docs.pipecat.ai/server/services/memory/mem0) | -| Vision & Image | [fal](https://docs.pipecat.ai/server/services/image-generation/fal), [Google Imagen](https://docs.pipecat.ai/server/services/image-generation/google-imagen), [Moondream](https://docs.pipecat.ai/server/services/vision/moondream) | -| Audio Processing | [Silero VAD](https://docs.pipecat.ai/server/utilities/audio/silero-vad-analyzer), [Krisp](https://docs.pipecat.ai/server/utilities/audio/krisp-filter), [Koala](https://docs.pipecat.ai/server/utilities/audio/koala-filter), [ai-coustics](https://docs.pipecat.ai/server/utilities/audio/aic-filter) | -| Analytics & Metrics | [OpenTelemetry](https://docs.pipecat.ai/server/utilities/opentelemetry), [Sentry](https://docs.pipecat.ai/server/services/analytics/sentry) | +| Category | Services | +| ------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| Speech-to-Text | [AssemblyAI](https://docs.pipecat.ai/server/services/stt/assemblyai), [AWS](https://docs.pipecat.ai/server/services/stt/aws), [Azure](https://docs.pipecat.ai/server/services/stt/azure), [Cartesia](https://docs.pipecat.ai/server/services/stt/cartesia), [Deepgram](https://docs.pipecat.ai/server/services/stt/deepgram), [ElevenLabs](https://docs.pipecat.ai/server/services/stt/elevenlabs), [Fal Wizper](https://docs.pipecat.ai/server/services/stt/fal), [Gladia](https://docs.pipecat.ai/server/services/stt/gladia), [Google](https://docs.pipecat.ai/server/services/stt/google), [Gradium](https://docs.pipecat.ai/server/services/stt/gradium), [Groq (Whisper)](https://docs.pipecat.ai/server/services/stt/groq), [NVIDIA Riva](https://docs.pipecat.ai/server/services/stt/riva), [OpenAI (Whisper)](https://docs.pipecat.ai/server/services/stt/openai), [SambaNova (Whisper)](https://docs.pipecat.ai/server/services/stt/sambanova), [Sarvam](https://docs.pipecat.ai/server/services/stt/sarvam), [Smallest](https://docs.pipecat.ai/server/services/stt/smallest), [Soniox](https://docs.pipecat.ai/server/services/stt/soniox), [Speechmatics](https://docs.pipecat.ai/server/services/stt/speechmatics), [Whisper](https://docs.pipecat.ai/server/services/stt/whisper) | +| LLMs | [Anthropic](https://docs.pipecat.ai/server/services/llm/anthropic), [AWS](https://docs.pipecat.ai/server/services/llm/aws), [Azure](https://docs.pipecat.ai/server/services/llm/azure), [Cerebras](https://docs.pipecat.ai/server/services/llm/cerebras), [DeepSeek](https://docs.pipecat.ai/server/services/llm/deepseek), [Fireworks AI](https://docs.pipecat.ai/server/services/llm/fireworks), [Gemini](https://docs.pipecat.ai/server/services/llm/gemini), [Grok](https://docs.pipecat.ai/server/services/llm/grok), [Groq](https://docs.pipecat.ai/server/services/llm/groq), [Mistral](https://docs.pipecat.ai/server/services/llm/mistral), [NVIDIA NIM](https://docs.pipecat.ai/server/services/llm/nim), [Ollama](https://docs.pipecat.ai/server/services/llm/ollama), [OpenAI](https://docs.pipecat.ai/server/services/llm/openai), [OpenRouter](https://docs.pipecat.ai/server/services/llm/openrouter), [Perplexity](https://docs.pipecat.ai/server/services/llm/perplexity), [Qwen](https://docs.pipecat.ai/server/services/llm/qwen), [SambaNova](https://docs.pipecat.ai/server/services/llm/sambanova) [Together AI](https://docs.pipecat.ai/server/services/llm/together) | +| Text-to-Speech | [Async](https://docs.pipecat.ai/server/services/tts/asyncai), [AWS](https://docs.pipecat.ai/server/services/tts/aws), [Azure](https://docs.pipecat.ai/server/services/tts/azure), [Camb AI](https://docs.pipecat.ai/server/services/tts/camb), [Cartesia](https://docs.pipecat.ai/server/services/tts/cartesia), [Deepgram](https://docs.pipecat.ai/server/services/tts/deepgram), [ElevenLabs](https://docs.pipecat.ai/server/services/tts/elevenlabs), [Fish](https://docs.pipecat.ai/server/services/tts/fish), [Google](https://docs.pipecat.ai/server/services/tts/google), [Gradium](https://docs.pipecat.ai/server/services/tts/gradium), [Groq](https://docs.pipecat.ai/server/services/tts/groq), [Hume](https://docs.pipecat.ai/server/services/tts/hume), [Inworld](https://docs.pipecat.ai/server/services/tts/inworld), [LMNT](https://docs.pipecat.ai/server/services/tts/lmnt), [MiniMax](https://docs.pipecat.ai/server/services/tts/minimax), [Neuphonic](https://docs.pipecat.ai/server/services/tts/neuphonic), [NVIDIA Riva](https://docs.pipecat.ai/server/services/tts/riva), [OpenAI](https://docs.pipecat.ai/server/services/tts/openai), [Piper](https://docs.pipecat.ai/server/services/tts/piper), [Resemble](https://docs.pipecat.ai/server/services/tts/resemble), [Rime](https://docs.pipecat.ai/server/services/tts/rime), [Sarvam](https://docs.pipecat.ai/server/services/tts/sarvam), [Smallest](https://docs.pipecat.ai/server/services/tts/smallest), [Speechmatics](https://docs.pipecat.ai/server/services/tts/speechmatics), [XTTS](https://docs.pipecat.ai/server/services/tts/xtts) | +| Speech-to-Speech | [AWS Nova Sonic](https://docs.pipecat.ai/server/services/s2s/aws), [Gemini Multimodal Live](https://docs.pipecat.ai/server/services/s2s/gemini), [Grok Voice Agent](https://docs.pipecat.ai/server/services/s2s/grok), [OpenAI Realtime](https://docs.pipecat.ai/server/services/s2s/openai), [Ultravox](https://docs.pipecat.ai/server/services/s2s/ultravox), | +| Transport | [Daily (WebRTC)](https://docs.pipecat.ai/server/services/transport/daily), [FastAPI Websocket](https://docs.pipecat.ai/server/services/transport/fastapi-websocket), [SmallWebRTCTransport](https://docs.pipecat.ai/server/services/transport/small-webrtc), [WebSocket Server](https://docs.pipecat.ai/server/services/transport/websocket-server), Local | +| Serializers | [Exotel](https://docs.pipecat.ai/server/utilities/serializers/exotel), [Plivo](https://docs.pipecat.ai/server/utilities/serializers/plivo), [Twilio](https://docs.pipecat.ai/server/utilities/serializers/twilio), [Telnyx](https://docs.pipecat.ai/server/utilities/serializers/telnyx), [Vonage](https://docs.pipecat.ai/server/utilities/serializers/vonage) | +| Video | [HeyGen](https://docs.pipecat.ai/server/services/video/heygen), [LemonSlice](https://docs.pipecat.ai/server/services/video/lemonslice), [Tavus](https://docs.pipecat.ai/server/services/video/tavus), [Simli](https://docs.pipecat.ai/server/services/video/simli) | +| Memory | [mem0](https://docs.pipecat.ai/server/services/memory/mem0) | +| Vision & Image | [fal](https://docs.pipecat.ai/server/services/image-generation/fal), [Google Imagen](https://docs.pipecat.ai/server/services/image-generation/google-imagen), [Moondream](https://docs.pipecat.ai/server/services/vision/moondream) | +| Audio Processing | [Silero VAD](https://docs.pipecat.ai/server/utilities/audio/silero-vad-analyzer), [Krisp](https://docs.pipecat.ai/server/utilities/audio/krisp-filter), [Koala](https://docs.pipecat.ai/server/utilities/audio/koala-filter), [ai-coustics](https://docs.pipecat.ai/server/utilities/audio/aic-filter) | +| Analytics & Metrics | [OpenTelemetry](https://docs.pipecat.ai/server/utilities/opentelemetry), [Sentry](https://docs.pipecat.ai/server/services/analytics/sentry) | 📚 [View full services documentation →](https://docs.pipecat.ai/server/services/supported-services) diff --git a/examples/foundational/07zl-interruptible-smallest-http.py b/examples/foundational/07zl-interruptible-smallest-http.py deleted file mode 100644 index 02d6d35c1..000000000 --- a/examples/foundational/07zl-interruptible-smallest-http.py +++ /dev/null @@ -1,122 +0,0 @@ -# -# 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.openai.llm import OpenAILLMService -from pipecat.services.smallest.stt import SmallestSTTService -from pipecat.services.smallest.tts import SmallestHttpTTSService -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 = SmallestSTTService( - api_key=os.getenv("SMALLEST_API_KEY"), - ) - - tts = SmallestHttpTTSService( - api_key=os.getenv("SMALLEST_API_KEY"), - voice_id="sophia", - ) - - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) - - messages = [ - { - "role": "system", - "content": "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(messages) - 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") - messages.append({"role": "system", "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() diff --git a/examples/foundational/07zl-interruptible-smallest.py b/examples/foundational/07zl-interruptible-smallest.py index 3564863b4..6bc9a34b0 100644 --- a/examples/foundational/07zl-interruptible-smallest.py +++ b/examples/foundational/07zl-interruptible-smallest.py @@ -22,7 +22,7 @@ from pipecat.processors.aggregators.llm_response_universal import ( from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport from pipecat.services.openai.llm import OpenAILLMService -from pipecat.services.smallest.stt import SmallestRealtimeSTTService +from pipecat.services.smallest.stt import SmallestSTTService from pipecat.services.smallest.tts import SmallestTTSService from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.daily.transport import DailyParams @@ -50,25 +50,25 @@ transport_params = { async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): logger.info(f"Starting bot") - stt = SmallestRealtimeSTTService( + stt = SmallestSTTService( api_key=os.getenv("SMALLEST_API_KEY"), ) tts = SmallestTTSService( api_key=os.getenv("SMALLEST_API_KEY"), - voice_id="sophia", + settings=SmallestTTSService.Settings( + voice="sophia", + ), ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) + 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.", + ), + ) - messages = [ - { - "role": "system", - "content": "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(messages) + context = LLMContext() user_aggregator, assistant_aggregator = LLMContextAggregatorPair( context, user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()), @@ -97,7 +97,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): @transport.event_handler("on_client_connected") async def on_client_connected(transport, client): logger.info(f"Client connected") - messages.append({"role": "system", "content": "Please introduce yourself to the user."}) + context.add_message({"role": "user", "content": "Please introduce yourself to the user."}) await task.queue_frames([LLMRunFrame()]) @transport.event_handler("on_client_disconnected") diff --git a/pyproject.toml b/pyproject.toml index c01e039e9..1b9acfe96 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -110,7 +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]" ] +smallest = [ "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 index 40098b034..8b1378917 100644 --- a/src/pipecat/services/smallest/__init__.py +++ b/src/pipecat/services/smallest/__init__.py @@ -1,14 +1 @@ -# -# 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 index fb1437d14..a7883881e 100644 --- a/src/pipecat/services/smallest/stt.py +++ b/src/pipecat/services/smallest/stt.py @@ -4,25 +4,22 @@ # SPDX-License-Identifier: BSD 2-Clause License # -"""Smallest AI speech-to-text service implementations. +"""Smallest AI speech-to-text service implementation. -This module provides two STT services using Smallest AI's Waves API: +This module provides a STT service using Smallest AI's Waves API: -- ``SmallestSTTService``: HTTP-based segmented STT. Buffers audio during speech, - sends as a single request once the user stops speaking (VAD-triggered). -- ``SmallestRealtimeSTTService``: WebSocket-based real-time STT. Streams audio +- ``SmallestSTTService``: WebSocket-based real-time STT. Streams audio continuously and receives interim/final transcripts with low latency. """ import asyncio -import io import json +from dataclasses import dataclass, field from enum import Enum -from typing import AsyncGenerator, Optional +from typing import Any, AsyncGenerator, Optional from urllib.parse import urlencode from loguru import logger -from pydantic import BaseModel from pipecat.frames.frames import ( CancelFrame, @@ -36,34 +33,13 @@ from pipecat.frames.frames import ( VADUserStoppedSpeakingFrame, ) from pipecat.processors.frame_processor import FrameDirection -from pipecat.services.settings import STTSettings +from pipecat.services.settings import NOT_GIVEN, STTSettings, _NotGiven from pipecat.services.stt_latency import SMALLEST_TTFS_P99 -from pipecat.services.stt_service import SegmentedSTTService, WebsocketSTTService -from pipecat.transcriptions.language import Language +from pipecat.services.stt_service import WebsocketSTTService +from pipecat.transcriptions.language import Language, resolve_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}") - try: from websockets.asyncio.client import connect as websocket_connect from websockets.protocol import State @@ -73,31 +49,51 @@ except ModuleNotFoundError as e: 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. +def language_to_smallest_stt_language(language: Language) -> str: + """Convert a Language enum to Smallest STT language code. Args: language: The Language enum value to convert. Returns: - The Smallest language code string, or None if unsupported. + The Smallest language code string. """ - BASE_LANGUAGES = { + LANGUAGE_MAP = { + Language.BG: "bg", + Language.BN: "bn", + Language.CS: "cs", + Language.DA: "da", + Language.DE: "de", Language.EN: "en", + Language.ES: "es", + Language.ET: "et", + Language.FI: "fi", + Language.FR: "fr", + Language.GU: "gu", Language.HI: "hi", + Language.HU: "hu", + Language.IT: "it", + Language.KN: "kn", + Language.LT: "lt", + Language.LV: "lv", + Language.ML: "ml", + Language.MR: "mr", + Language.MT: "mt", + Language.NL: "nl", + Language.OR: "or", + Language.PA: "pa", + Language.PL: "pl", + Language.PT: "pt", + Language.RO: "ro", + Language.RU: "ru", + Language.SK: "sk", + Language.SV: "sv", + Language.TA: "ta", + Language.TE: "te", + Language.UK: "uk", } - 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 + return resolve_language(language, LANGUAGE_MAP) class SmallestSTTModel(str, Enum): @@ -106,174 +102,30 @@ class SmallestSTTModel(str, Enum): PULSE = "pulse" -class SmallestSTTService(SegmentedSTTService): - """Smallest AI speech-to-text service using the Waves HTTP API. +@dataclass +class SmallestSTTSettings(STTSettings): + """Settings for SmallestSTTService. - 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. + Parameters: + word_timestamps: Include word-level timestamps. + full_transcript: Include cumulative transcript. + sentence_timestamps: Include sentence-level timestamps. + redact_pii: Redact personally identifiable information. + redact_pci: Redact payment card information. + numerals: Convert spoken numerals to digits. + diarize: Enable speaker diarization. """ - 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.PULSE, - url: str = "https://api.smallest.ai/waves/v1/pulse/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 "pulse". - 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. - """ - params = params or SmallestSTTService.InputParams() - model_str = model.value if isinstance(model, Enum) else model - - super().__init__( - sample_rate=sample_rate, - ttfs_p99_latency=ttfs_p99_latency, - settings=STTSettings(model=model_str, language=params.language), - **kwargs, - ) - - self._api_key = api_key - self._url = url - self._language = params.language - - 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() + word_timestamps: bool | _NotGiven = field(default_factory=lambda: NOT_GIVEN) + full_transcript: bool | _NotGiven = field(default_factory=lambda: NOT_GIVEN) + sentence_timestamps: bool | _NotGiven = field(default_factory=lambda: NOT_GIVEN) + redact_pii: bool | _NotGiven = field(default_factory=lambda: NOT_GIVEN) + redact_pci: bool | _NotGiven = field(default_factory=lambda: NOT_GIVEN) + numerals: str | _NotGiven = field(default_factory=lambda: NOT_GIVEN) + diarize: bool | _NotGiven = field(default_factory=lambda: NOT_GIVEN) -class SmallestRealtimeSTTService(WebsocketSTTService): +class SmallestSTTService(WebsocketSTTService): """Smallest AI real-time speech-to-text service using the Pulse WebSocket API. Streams audio continuously over a WebSocket connection and receives @@ -285,74 +137,67 @@ class SmallestRealtimeSTTService(WebsocketSTTService): Example:: - stt = SmallestRealtimeSTTService( + stt = SmallestSTTService( api_key="your-api-key", - params=SmallestRealtimeSTTService.InputParams( + settings=SmallestSTTService.Settings( language="en", word_timestamps=True, ), ) """ - class InputParams(BaseModel): - """Configuration parameters for Smallest Realtime STT service. - - Parameters: - language: Language code for transcription. Use "multi" for auto-detection. - Defaults to "en". - encoding: Audio encoding format. Defaults to "linear16". - word_timestamps: Include word-level timestamps. Defaults to False. - full_transcript: Include cumulative transcript. Defaults to False. - sentence_timestamps: Include sentence-level timestamps. Defaults to False. - redact_pii: Redact personally identifiable information. Defaults to False. - redact_pci: Redact payment card information. Defaults to False. - numerals: Convert spoken numerals to digits. Defaults to "auto". - diarize: Enable speaker diarization. Defaults to False. - """ - - language: str = "en" - encoding: str = "linear16" - word_timestamps: bool = False - full_transcript: bool = False - sentence_timestamps: bool = False - redact_pii: bool = False - redact_pci: bool = False - numerals: str = "auto" - diarize: bool = False + Settings = SmallestSTTSettings + _settings: Settings def __init__( self, *, api_key: str, base_url: str = "wss://api.smallest.ai", + encoding: str = "linear16", sample_rate: Optional[int] = None, - params: Optional[InputParams] = None, + settings: Optional[Settings] = None, ttfs_p99_latency: Optional[float] = SMALLEST_TTFS_P99, **kwargs, ): - """Initialize the Smallest AI Realtime STT service. + """Initialize the Smallest AI STT service. Args: api_key: Smallest AI API key for authentication. base_url: Base WebSocket URL for the Smallest API. + encoding: Audio encoding format. Defaults to "linear16". sample_rate: Audio sample rate in Hz. If None, uses the pipeline's rate. - params: Configuration parameters for the STT service. + settings: Runtime-updatable settings for the STT service. ttfs_p99_latency: P99 latency from speech end to final transcript in seconds. **kwargs: Additional arguments passed to WebsocketSTTService. """ - self._rt_params = params or SmallestRealtimeSTTService.InputParams() + default_settings = self.Settings( + model=SmallestSTTModel.PULSE.value, + language=language_to_smallest_stt_language(Language.EN), + word_timestamps=False, + full_transcript=False, + sentence_timestamps=False, + redact_pii=False, + redact_pci=False, + numerals="auto", + diarize=False, + ) + + if settings is not None: + default_settings.apply_update(settings) super().__init__( sample_rate=sample_rate, ttfs_p99_latency=ttfs_p99_latency, keepalive_timeout=10, keepalive_interval=5, - settings=STTSettings(model="pulse", language=self._rt_params.language), + settings=default_settings, **kwargs, ) self._api_key = api_key self._base_url = base_url.rstrip("/") + self._encoding = encoding self._receive_task = None self._connected_event = asyncio.Event() self._connected_event.set() @@ -407,10 +252,20 @@ class SmallestRealtimeSTTService(WebsocketSTTService): try: await self._websocket.send(audio) except Exception as e: - yield ErrorFrame(error=f"Smallest Realtime STT error: {e}") + yield ErrorFrame(error=f"Smallest STT error: {e}") yield None + async def _update_settings(self, delta: STTSettings) -> dict[str, Any]: + """Apply a settings delta and reconnect if anything changed.""" + changed = await super()._update_settings(delta) + + if changed: + await self._disconnect() + await self._connect() + + return changed + async def _connect(self): self._connected_event.clear() try: @@ -439,19 +294,19 @@ class SmallestRealtimeSTTService(WebsocketSTTService): if self._websocket and self._websocket.state is State.OPEN: return - logger.debug("Connecting to Smallest Realtime STT") + logger.debug("Connecting to Smallest STT") query_params = { - "language": self._rt_params.language, - "encoding": self._rt_params.encoding, + "language": self._settings.language, + "encoding": self._encoding, "sample_rate": str(self.sample_rate), - "word_timestamps": str(self._rt_params.word_timestamps).lower(), - "full_transcript": str(self._rt_params.full_transcript).lower(), - "sentence_timestamps": str(self._rt_params.sentence_timestamps).lower(), - "redact_pii": str(self._rt_params.redact_pii).lower(), - "redact_pci": str(self._rt_params.redact_pci).lower(), - "numerals": self._rt_params.numerals, - "diarize": str(self._rt_params.diarize).lower(), + "word_timestamps": str(self._settings.word_timestamps).lower(), + "full_transcript": str(self._settings.full_transcript).lower(), + "sentence_timestamps": str(self._settings.sentence_timestamps).lower(), + "redact_pii": str(self._settings.redact_pii).lower(), + "redact_pci": str(self._settings.redact_pci).lower(), + "numerals": self._settings.numerals, + "diarize": str(self._settings.diarize).lower(), } ws_url = f"{self._base_url}/waves/v1/pulse/get_text?{urlencode(query_params)}" @@ -461,11 +316,9 @@ class SmallestRealtimeSTTService(WebsocketSTTService): additional_headers={"Authorization": f"Bearer {self._api_key}"}, ) await self._call_event_handler("on_connected") - logger.debug("Connected to Smallest Realtime STT") + logger.debug("Connected to Smallest STT") except Exception as e: - await self.push_error( - error_msg=f"Smallest Realtime STT connection error: {e}", exception=e - ) + await self.push_error(error_msg=f"Smallest STT connection error: {e}", exception=e) self._websocket = None await self._call_event_handler("on_connection_error", f"{e}") @@ -473,7 +326,7 @@ class SmallestRealtimeSTTService(WebsocketSTTService): """Close the WebSocket connection.""" try: if self._websocket and self._websocket.state is State.OPEN: - logger.debug("Disconnecting from Smallest Realtime STT") + logger.debug("Disconnecting from Smallest STT") await self._websocket.close() except Exception as e: logger.error(f"{self} error closing websocket: {e}") diff --git a/src/pipecat/services/smallest/tts.py b/src/pipecat/services/smallest/tts.py index 1d913f823..ea2d03c66 100644 --- a/src/pipecat/services/smallest/tts.py +++ b/src/pipecat/services/smallest/tts.py @@ -4,34 +4,33 @@ # SPDX-License-Identifier: BSD 2-Clause License # -"""Smallest AI text-to-speech service implementations. +"""Smallest AI text-to-speech service implementation. -This module provides WebSocket-based and HTTP-based integrations with Smallest -AI's Waves API for real-time text-to-speech synthesis. +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 AsyncGenerator, Optional, Union +from typing import Any, AsyncGenerator, Optional 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.settings import TTSSettings -from pipecat.services.tts_service import InterruptibleTTSService, TTSService +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 @@ -89,6 +88,23 @@ def language_to_smallest_tts_language(language: Language) -> Optional[str]: 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. @@ -100,80 +116,61 @@ class SmallestTTSService(InterruptibleTTSService): tts = SmallestTTSService( api_key="your-api-key", - voice_id="sophia", - params=SmallestTTSService.InputParams( - language=Language.EN, + settings=SmallestTTSService.Settings( + voice="sophia", + 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) + Settings = SmallestTTSSettings + _settings: Settings def __init__( self, *, api_key: str, - voice_id: str = "sophia", base_url: str = "wss://waves-api.smallest.ai", - model: str = SmallestTTSModel.LIGHTNING_V3_1, - sample_rate: Optional[int] = 24000, - params: Optional[InputParams] = None, + 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. - voice_id: Voice identifier for synthesis. base_url: Base WebSocket URL for the Smallest API. - model: TTS model to use. Defaults to "lightning-v3.1". - sample_rate: Audio sample rate in Hz. Defaults to 24000. - params: Configuration parameters for the TTS service. + 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. """ - params = params or SmallestTTSService.InputParams() - model_str = model.value if isinstance(model, Enum) else model - lang_str = ( - language_to_smallest_tts_language(params.language) if params.language else "en" + 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=TTSSettings(model=model_str, voice=voice_id, language=lang_str), + settings=default_settings, **kwargs, ) self._api_key = api_key - self._websocket_url = f"{base_url}/api/v1/{model_str}/get_speech/stream" - - self._tts_params = { - "language": lang_str, - "speed": params.speed, - "consistency": params.consistency, - "similarity": params.similarity, - "enhancement": params.enhancement, - } - + 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: @@ -207,18 +204,28 @@ class SmallestTTSService(InterruptibleTTSService): msg = { "text": text, "voice_id": self._settings.voice, - "language": self._tts_params["language"], - "speed": self._tts_params["speed"], - "consistency": self._tts_params["consistency"], - "similarity": self._tts_params["similarity"], - "enhancement": self._tts_params["enhancement"], + "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. @@ -246,6 +253,24 @@ class SmallestTTSService(InterruptibleTTSService): 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() @@ -255,6 +280,9 @@ class SmallestTTSService(InterruptibleTTSService): 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() @@ -263,6 +291,10 @@ class SmallestTTSService(InterruptibleTTSService): 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): @@ -271,16 +303,16 @@ class SmallestTTSService(InterruptibleTTSService): if self._websocket and self._websocket.state is State.OPEN: return - logger.debug("Connecting to Smallest") + logger.debug("Connecting to Smallest TTS") self._websocket = await websocket_connect( - self._websocket_url, + 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 connection error: {e}", exception=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}") @@ -290,7 +322,7 @@ class SmallestTTSService(InterruptibleTTSService): await self.stop_all_metrics() if self._websocket: - logger.debug("Disconnecting from Smallest") + logger.debug("Disconnecting from Smallest TTS") await self._websocket.close() except Exception as e: logger.error(f"{self} error closing websocket: {e}") @@ -312,16 +344,18 @@ class SmallestTTSService(InterruptibleTTSService): return self._websocket raise Exception("Websocket not connected") - async def _handle_interruption(self, frame: InterruptionFrame, direction: FrameDirection): - """Handle an interruption by resetting state. + 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() - 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 _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.""" @@ -371,7 +405,6 @@ class SmallestTTSService(InterruptibleTTSService): await self._connect() try: - await self.start_ttfb_metrics() self._context_id = context_id yield TTSStartedFrame(context_id=context_id) @@ -389,195 +422,3 @@ class SmallestTTSService(InterruptibleTTSService): 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 = "sophia", - model: str = SmallestTTSModel.LIGHTNING_V3_1, - 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-v3.1". - 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. - """ - params = params or SmallestHttpTTSService.InputParams() - model_str = model.value if isinstance(model, Enum) else model - - super().__init__( - sample_rate=sample_rate, - settings=TTSSettings(model=model_str, voice=voice_id, language=params.language), - **kwargs, - ) - - self._api_key = api_key - self._base_url = base_url.rstrip("/") - self._model_url = f"{self._base_url}/api/v1/{model_str}/get_speech" - - self._tts_params = { - "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._settings.voice, - "text": text, - "sample_rate": self.sample_rate, - } - - for key, value in self._tts_params.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) diff --git a/src/pipecat/services/stt_latency.py b/src/pipecat/services/stt_latency.py index ba5d7089e..974fbd9ac 100644 --- a/src/pipecat/services/stt_latency.py +++ b/src/pipecat/services/stt_latency.py @@ -51,4 +51,3 @@ SPEECHMATICS_TTFS_P99: float = 0.74 # These services run locally and should be replaced with measured values NVIDIA_TTFS_P99: float = DEFAULT_TTFS_P99 WHISPER_TTFS_P99: float = DEFAULT_TTFS_P99 - diff --git a/uv.lock b/uv.lock index bda379864..dad52f51f 100644 --- a/uv.lock +++ b/uv.lock @@ -4716,10 +4716,6 @@ simli = [ { name = "simli-ai" }, ] smallest = [ - { name = "httpx" }, - { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, - { name = "numpy", version = "2.4.3", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, - { name = "soundfile" }, { name = "websockets" }, ] soniox = [ @@ -4812,7 +4808,6 @@ requires-dist = [ { name = "google-cloud-texttospeech", marker = "extra == 'google'", specifier = ">=2.31.0,<3" }, { name = "google-genai", marker = "extra == 'google'", specifier = ">=1.57.0,<2" }, { name = "groq", marker = "extra == 'groq'", specifier = ">=0.23.0,<2" }, - { name = "httpx", marker = "extra == 'smallest'", specifier = ">=0.27.0,<1" }, { name = "hume", marker = "extra == 'hume'", specifier = ">=0.11.2,<1" }, { name = "kokoro-onnx", marker = "extra == 'kokoro'", specifier = ">=0.5.0,<1" }, { name = "langchain", marker = "extra == 'langchain'", specifier = "~=0.3.20" }, @@ -4830,7 +4825,6 @@ requires-dist = [ { name = "noisereduce", marker = "extra == 'noisereduce'", specifier = "~=3.0.3" }, { name = "numba", specifier = ">=0.61.2,<1" }, { name = "numpy", specifier = ">=1.26.4,<3" }, - { name = "numpy", marker = "extra == 'smallest'", specifier = ">=1.24.0,<3" }, { name = "nvidia-riva-client", marker = "extra == 'nvidia'", specifier = ">=2.21.1,<3" }, { name = "onnxruntime", specifier = "~=1.23.2" }, { name = "openai", specifier = ">=1.74.0,<3" }, @@ -4843,7 +4837,6 @@ requires-dist = [ { name = "pillow", specifier = ">=11.1.0,<13" }, { name = "pipecat-ai", extras = ["daily"], marker = "extra == 'lemonslice'" }, { name = "pipecat-ai", extras = ["nvidia"], marker = "extra == 'riva'" }, - { name = "pipecat-ai", extras = ["soundfile"], marker = "extra == 'smallest'" }, { name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'assemblyai'" }, { name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'asyncai'" }, { name = "pipecat-ai", extras = ["websockets-base"], marker = "extra == 'aws'" },