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

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()

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()

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,
+                "language": self._rt_params.language,
-                "encoding": self._params.encoding,
+                "encoding": self._rt_params.encoding,
                 "sample_rate": str(self.sample_rate),
-                "word_timestamps": str(self._params.word_timestamps).lower(),
+                "word_timestamps": str(self._rt_params.word_timestamps).lower(),
-                "full_transcript": str(self._params.full_transcript).lower(),
+                "full_transcript": str(self._rt_params.full_transcript).lower(),
-                "sentence_timestamps": str(self._params.sentence_timestamps).lower(),
+                "sentence_timestamps": str(self._rt_params.sentence_timestamps).lower(),
-                "redact_pii": str(self._params.redact_pii).lower(),
+                "redact_pii": str(self._rt_params.redact_pii).lower(),
-                "redact_pci": str(self._params.redact_pci).lower(),
+                "redact_pci": str(self._rt_params.redact_pci).lower(),
-                "numerals": self._params.numerals,
+                "numerals": self._rt_params.numerals,
-                "diarize": str(self._params.diarize).lower(),
+                "diarize": str(self._rt_params.diarize).lower(),
             }
 
             ws_url = f"{self._base_url}/waves/v1/pulse/get_text?{urlencode(query_params)}"

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._tts_params = {
-        self.set_voice(voice_id)
+            "language": lang_str,
-
-        self._settings = {
-            "language": language_to_smallest_tts_language(params.language)
-            if params.language
-            else "en",
             "speed": params.speed,
             "consistency": params.consistency,
             "similarity": params.similarity,
@@ -206,12 +206,12 @@ class SmallestTTSService(InterruptibleTTSService):
         """
         msg = {
             "text": text,
-            "voice_id": self._voice_id,
+            "voice_id": self._settings.voice,
-            "language": self._settings["language"],
+            "language": self._tts_params["language"],
-            "speed": self._settings["speed"],
+            "speed": self._tts_params["speed"],
-            "consistency": self._settings["consistency"],
+            "consistency": self._tts_params["consistency"],
-            "similarity": self._settings["similarity"],
+            "similarity": self._tts_params["similarity"],
-            "enhancement": self._settings["enhancement"],
+            "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._tts_params.items():
-            for key, value in self._settings.items():
                 if value is not None:
                     payload[key] = value