Update Deepgram Flux with the new service settings pattern

examples/foundational/55a-update-settings-deepgram-flux-stt.py

@@ -0,0 +1,128 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import os
+
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import LLMRunFrame, STTUpdateSettingsFrame
+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.cartesia.tts import CartesiaTTSService
+from pipecat.services.deepgram.flux.stt import DeepgramFluxSTTService, DeepgramFluxSTTSettings
+from pipecat.services.openai.llm import OpenAILLMService
+from pipecat.transcriptions.language import Language
+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 = DeepgramFluxSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+
+    tts = CartesiaTTSService(
+        api_key=os.getenv("CARTESIA_API_KEY"),
+        voice_id="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
+    )
+
+    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,
+        ),
+        idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
+    )
+
+    @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()])
+
+        await asyncio.sleep(10)
+        logger.info("Updating Deepgram Flux STT settings: language=es")
+        await task.queue_frame(
+            STTUpdateSettingsFrame(update=DeepgramFluxSTTSettings(language=Language.ES))
+        )
+
+    @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/deepgram/flux/stt.py

@@ -9,6 +9,7 @@
 import asyncio
 import json
 import time
+from dataclasses import dataclass, field
 from enum import Enum
 from typing import Any, AsyncGenerator, Dict, Optional
 from urllib.parse import urlencode
@@ -27,7 +28,7 @@ from pipecat.frames.frames import (
     UserStartedSpeakingFrame,
     UserStoppedSpeakingFrame,
 )
-from pipecat.services.settings import STTSettings
+from pipecat.services.settings import NOT_GIVEN, STTSettings, _NotGiven
 from pipecat.services.stt_service import WebsocketSTTService
 from pipecat.transcriptions.language import Language
 from pipecat.utils.time import time_now_iso8601
@@ -68,6 +69,34 @@ class FluxEventType(str, Enum):
     UPDATE = "Update"
 
 
+@dataclass
+class DeepgramFluxSTTSettings(STTSettings):
+    """Settings for the Deepgram Flux STT service.
+
+    Parameters:
+        eager_eot_threshold: EagerEndOfTurn/TurnResumed threshold. Off by default.
+            Lower values = more aggressive (faster response, more LLM calls).
+            Higher values = more conservative (slower response, fewer LLM calls).
+        eot_threshold: End-of-turn confidence required to finish a turn (default 0.7).
+        eot_timeout_ms: Time in ms after speech to finish a turn regardless of EOT
+            confidence (default 5000).
+        keyterm: Keyterms to boost recognition accuracy for specialized terminology.
+        mip_opt_out: Opt out of the Deepgram Model Improvement Program (default False).
+        tag: Tags to label requests for identification during usage reporting.
+        min_confidence: Minimum confidence required to create a TranscriptionFrame.
+        encoding: Audio encoding format (e.g. ``"linear16"``).
+    """
+
+    eager_eot_threshold: float | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
+    eot_threshold: float | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
+    eot_timeout_ms: int | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
+    keyterm: list | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
+    mip_opt_out: bool | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
+    tag: list | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
+    min_confidence: float | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
+    encoding: str | _NotGiven = field(default_factory=lambda: NOT_GIVEN)
+
+
 class DeepgramFluxSTTService(WebsocketSTTService):
     """Deepgram Flux speech-to-text service.
 
@@ -76,6 +105,8 @@ class DeepgramFluxSTTService(WebsocketSTTService):
     including advanced turn detection and EagerEndOfTurn events for improved conversational AI performance.
     """
 
+    _settings: DeepgramFluxSTTSettings
+
     class InputParams(BaseModel):
         """Configuration parameters for Deepgram Flux API.
 
@@ -168,14 +199,23 @@ class DeepgramFluxSTTService(WebsocketSTTService):
             **kwargs,
         )
 
+        params = params or DeepgramFluxSTTService.InputParams()
+        self._settings = DeepgramFluxSTTSettings(
+            model=model,
+            language=Language.EN,
+            encoding=flux_encoding,
+            eager_eot_threshold=params.eager_eot_threshold,
+            eot_threshold=params.eot_threshold,
+            eot_timeout_ms=params.eot_timeout_ms,
+            keyterm=params.keyterm or [],
+            mip_opt_out=params.mip_opt_out,
+            tag=params.tag or [],
+            min_confidence=params.min_confidence,
+        )
+        self.set_model_name(model)
         self._api_key = api_key
         self._url = url
-        self._model = model
-        self._params = params or DeepgramFluxSTTService.InputParams()
         self._should_interrupt = should_interrupt
-        self._flux_encoding = flux_encoding
-        # This is the currently only supported language
-        self._language = Language.EN
         self._websocket_url = None
         self._receive_task = None
         # Flux event handlers
@@ -330,7 +370,7 @@ class DeepgramFluxSTTService(WebsocketSTTService):
         """
         return True
 
-    async def _update_settings(self, update: STTSettings) -> dict[str, Any]:
+    async def _update_settings(self, update: DeepgramFluxSTTSettings) -> dict[str, Any]:
         """Apply a settings update.
 
         Settings are stored but not applied to the active connection.
@@ -361,29 +401,29 @@ class DeepgramFluxSTTService(WebsocketSTTService):
         await super().start(frame)
 
         url_params = [
-            f"model={self._model}",
+            f"model={self._settings.model}",
             f"sample_rate={self.sample_rate}",
-            f"encoding={self._flux_encoding}",
+            f"encoding={self._settings.encoding}",
         ]
 
-        if self._params.eager_eot_threshold is not None:
+        if self._settings.eager_eot_threshold is not None:
-            url_params.append(f"eager_eot_threshold={self._params.eager_eot_threshold}")
+            url_params.append(f"eager_eot_threshold={self._settings.eager_eot_threshold}")
 
-        if self._params.eot_threshold is not None:
+        if self._settings.eot_threshold is not None:
-            url_params.append(f"eot_threshold={self._params.eot_threshold}")
+            url_params.append(f"eot_threshold={self._settings.eot_threshold}")
 
-        if self._params.eot_timeout_ms is not None:
+        if self._settings.eot_timeout_ms is not None:
-            url_params.append(f"eot_timeout_ms={self._params.eot_timeout_ms}")
+            url_params.append(f"eot_timeout_ms={self._settings.eot_timeout_ms}")
 
-        if self._params.mip_opt_out is not None:
+        if self._settings.mip_opt_out is not None:
-            url_params.append(f"mip_opt_out={str(self._params.mip_opt_out).lower()}")
+            url_params.append(f"mip_opt_out={str(self._settings.mip_opt_out).lower()}")
 
         # Add keyterm parameters (can have multiple)
-        for keyterm in self._params.keyterm:
+        for keyterm in self._settings.keyterm:
             url_params.append(urlencode({"keyterm": keyterm}))
 
         # Add tag parameters (can have multiple)
-        for tag_value in self._params.tag:
+        for tag_value in self._settings.tag:
             url_params.append(urlencode({"tag": tag_value}))
 
         self._websocket_url = f"{self._url}?{'&'.join(url_params)}"
@@ -682,7 +722,7 @@ class DeepgramFluxSTTService(WebsocketSTTService):
         # Compute the average confidence
         average_confidence = self._calculate_average_confidence(data)
 
-        if not self._params.min_confidence or average_confidence > self._params.min_confidence:
+        if not self._settings.min_confidence or average_confidence > self._settings.min_confidence:
             # EndOfTurn means Flux has determined the turn is complete,
             # so this TranscriptionFrame is always finalized
             await self.push_frame(
@@ -690,7 +730,7 @@ class DeepgramFluxSTTService(WebsocketSTTService):
                     transcript,
                     self._user_id,
                     time_now_iso8601(),
-                    self._language,
+                    self._settings.language,
                     result=data,
                     finalized=True,
                 )
@@ -700,7 +740,7 @@ class DeepgramFluxSTTService(WebsocketSTTService):
                 f"Transcription confidence below min_confidence threshold: {average_confidence}"
             )
 
-        await self._handle_transcription(transcript, True, self._language)
+        await self._handle_transcription(transcript, True, self._settings.language)
         await self.stop_processing_metrics()
         await self.broadcast_frame(UserStoppedSpeakingFrame)
         await self._call_event_handler("on_end_of_turn", transcript)
@@ -744,7 +784,7 @@ class DeepgramFluxSTTService(WebsocketSTTService):
                 transcript,
                 self._user_id,
                 time_now_iso8601(),
-                self._language,
+                self._settings.language,
                 result=data,
             )
         )