From 40c36f8a2ac39343952d66c5e963a3fac1e1cf85 Mon Sep 17 00:00:00 2001 From: Harshita Jain Date: Fri, 27 Feb 2026 16:10:49 -0800 Subject: [PATCH] Update Smallest AI services to use ServiceSettings and add examples Migrate STT/TTS services from deprecated set_model_name()/set_voice() to the new ServiceSettings pattern (STTSettings/TTSSettings). Add default voice_id ("sophia") for TTS services, fix voice references, and include two foundational example scripts showing WebSocket and HTTP usage. Made-with: Cursor --- .../07zl-interruptible-smallest-http.py | 122 ++++++++++++++++++ .../07zl-interruptible-smallest.py | 122 ++++++++++++++++++ src/pipecat/services/smallest/stt.py | 34 ++--- src/pipecat/services/smallest/tts.py | 59 +++++---- 4 files changed, 290 insertions(+), 47 deletions(-) create mode 100644 examples/foundational/07zl-interruptible-smallest-http.py create mode 100644 examples/foundational/07zl-interruptible-smallest.py diff --git a/examples/foundational/07zl-interruptible-smallest-http.py b/examples/foundational/07zl-interruptible-smallest-http.py new file mode 100644 index 000000000..02d6d35c1 --- /dev/null +++ b/examples/foundational/07zl-interruptible-smallest-http.py @@ -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.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 new file mode 100644 index 000000000..3564863b4 --- /dev/null +++ b/examples/foundational/07zl-interruptible-smallest.py @@ -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.openai.llm import OpenAILLMService +from pipecat.services.smallest.stt import SmallestRealtimeSTTService +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 = SmallestRealtimeSTTService( + api_key=os.getenv("SMALLEST_API_KEY"), + ) + + tts = SmallestTTSService( + 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/src/pipecat/services/smallest/stt.py b/src/pipecat/services/smallest/stt.py index 3046ed855..fb1437d14 100644 --- a/src/pipecat/services/smallest/stt.py +++ b/src/pipecat/services/smallest/stt.py @@ -36,6 +36,7 @@ from pipecat.frames.frames import ( VADUserStoppedSpeakingFrame, ) from pipecat.processors.frame_processor import FrameDirection +from pipecat.services.settings import STTSettings from pipecat.services.stt_latency import SMALLEST_TTFS_P99 from pipecat.services.stt_service import SegmentedSTTService, WebsocketSTTService from pipecat.transcriptions.language import Language @@ -154,21 +155,20 @@ class SmallestSTTService(SegmentedSTTService): 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, ) - 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}", @@ -340,23 +340,23 @@ class SmallestRealtimeSTTService(WebsocketSTTService): 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() + 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), **kwargs, ) self._api_key = api_key self._base_url = base_url.rstrip("/") - self._params = params or SmallestRealtimeSTTService.InputParams() self._receive_task = None self._connected_event = asyncio.Event() self._connected_event.set() - self.set_model_name("pulse") - def can_generate_metrics(self) -> bool: """Check if this service can generate processing metrics.""" return True @@ -442,16 +442,16 @@ class SmallestRealtimeSTTService(WebsocketSTTService): logger.debug("Connecting to Smallest Realtime STT") query_params = { - "language": self._params.language, - "encoding": self._params.encoding, + "language": self._rt_params.language, + "encoding": self._rt_params.encoding, "sample_rate": str(self.sample_rate), - "word_timestamps": str(self._params.word_timestamps).lower(), - "full_transcript": str(self._params.full_transcript).lower(), - "sentence_timestamps": str(self._params.sentence_timestamps).lower(), - "redact_pii": str(self._params.redact_pii).lower(), - "redact_pci": str(self._params.redact_pci).lower(), - "numerals": self._params.numerals, - "diarize": str(self._params.diarize).lower(), + "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(), } ws_url = f"{self._base_url}/waves/v1/pulse/get_text?{urlencode(query_params)}" diff --git a/src/pipecat/services/smallest/tts.py b/src/pipecat/services/smallest/tts.py index 866c39cd6..1d913f823 100644 --- a/src/pipecat/services/smallest/tts.py +++ b/src/pipecat/services/smallest/tts.py @@ -30,6 +30,7 @@ from pipecat.frames.frames import ( TTSStoppedFrame, ) from pipecat.processors.frame_processor import FrameDirection +from pipecat.services.settings import TTSSettings from pipecat.services.tts_service import InterruptibleTTSService, TTSService from pipecat.transcriptions.language import Language from pipecat.utils.tracing.service_decorators import traced_tts @@ -99,7 +100,7 @@ class SmallestTTSService(InterruptibleTTSService): tts = SmallestTTSService( api_key="your-api-key", - voice_id="emily", + voice_id="sophia", params=SmallestTTSService.InputParams( language=Language.EN, speed=1.0, @@ -128,7 +129,7 @@ class SmallestTTSService(InterruptibleTTSService): self, *, api_key: str, - voice_id: str, + voice_id: str = "sophia", base_url: str = "wss://waves-api.smallest.ai", model: str = SmallestTTSModel.LIGHTNING_V3_1, sample_rate: Optional[int] = 24000, @@ -146,27 +147,26 @@ class SmallestTTSService(InterruptibleTTSService): params: Configuration parameters 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" + ) + 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), **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", + self._tts_params = { + "language": lang_str, "speed": params.speed, "consistency": params.consistency, "similarity": params.similarity, @@ -206,12 +206,12 @@ class SmallestTTSService(InterruptibleTTSService): """ 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"], + "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"], } if self._context_id: @@ -431,7 +431,7 @@ class SmallestHttpTTSService(TTSService): self, *, api_key: str, - voice_id: str, + voice_id: str = "sophia", model: str = SmallestTTSModel.LIGHTNING_V3_1, base_url: str = "https://waves-api.smallest.ai", sample_rate: Optional[int] = None, @@ -449,20 +449,20 @@ class SmallestHttpTTSService(TTSService): 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() + 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("/") - - 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 = { + self._tts_params = { "language": params.language, "speed": params.speed, "consistency": params.consistency, @@ -539,13 +539,12 @@ class SmallestHttpTTSService(TTSService): await self.start_ttfb_metrics() payload = { - "voice_id": self._voice_id, + "voice_id": self._settings.voice, "text": text, "sample_rate": self.sample_rate, } - # Only include non-None settings - for key, value in self._settings.items(): + for key, value in self._tts_params.items(): if value is not None: payload[key] = value