From a4b6db6fb44148709c22c65039f02ef677ad603b Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Tue, 24 Feb 2026 15:06:06 -0500 Subject: [PATCH 1/6] Flatten `LiveOptions` into individual fields on `DeepgramSTTSettings` and `DeepgramSageMakerSTTSettings` for backward-compatible dict-style updates via `STTUpdateSettingsFrame`; during the big service settings refactor, we accidentally got rid of the ability to update individual `LiveOptions` fields with a sparse update --- ...-update-settings-deepgram-sagemaker-stt.py | 7 + .../55a-update-settings-deepgram-stt.py | 7 + src/pipecat/services/deepgram/stt.py | 119 ++++++++++------- .../services/deepgram/stt_sagemaker.py | 121 ++++++++++-------- 4 files changed, 158 insertions(+), 96 deletions(-) diff --git a/examples/foundational/55a-update-settings-deepgram-sagemaker-stt.py b/examples/foundational/55a-update-settings-deepgram-sagemaker-stt.py index 05c92e7e2..fba722648 100644 --- a/examples/foundational/55a-update-settings-deepgram-sagemaker-stt.py +++ b/examples/foundational/55a-update-settings-deepgram-sagemaker-stt.py @@ -112,6 +112,13 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): STTUpdateSettingsFrame(delta=DeepgramSageMakerSTTSettings(language=Language.ES)) ) + # Old-style dict update (for backward-compat testing): + # await asyncio.sleep(10) + # logger.info("Updating Deepgram SageMaker STT settings via dict: punctuate=False, filler_words=True") + # await task.queue_frame( + # STTUpdateSettingsFrame(settings={"punctuate": False, "filler_words": True}) + # ) + @transport.event_handler("on_client_disconnected") async def on_client_disconnected(transport, client): logger.info(f"Client disconnected") diff --git a/examples/foundational/55a-update-settings-deepgram-stt.py b/examples/foundational/55a-update-settings-deepgram-stt.py index 39dde69e9..20068ab75 100644 --- a/examples/foundational/55a-update-settings-deepgram-stt.py +++ b/examples/foundational/55a-update-settings-deepgram-stt.py @@ -106,6 +106,13 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): STTUpdateSettingsFrame(delta=DeepgramSTTSettings(language=Language.ES)) ) + # Old-style dict update (for backward-compat testing): + # await asyncio.sleep(10) + # logger.info("Updating Deepgram STT settings via dict: punctuate=False, filler_words=True") + # await task.queue_frame( + # STTUpdateSettingsFrame(settings={"punctuate": False, "filler_words": True}) + # ) + @transport.event_handler("on_client_disconnected") async def on_client_disconnected(transport, client): logger.info(f"Client disconnected") diff --git a/src/pipecat/services/deepgram/stt.py b/src/pipecat/services/deepgram/stt.py index aa5c2ce8d..768c4d6eb 100644 --- a/src/pipecat/services/deepgram/stt.py +++ b/src/pipecat/services/deepgram/stt.py @@ -6,7 +6,8 @@ """Deepgram speech-to-text service implementation.""" -from dataclasses import dataclass, field +import inspect +from dataclasses import dataclass, field, fields from typing import Any, AsyncGenerator, Dict, Optional from loguru import logger @@ -24,7 +25,7 @@ from pipecat.frames.frames import ( VADUserStoppedSpeakingFrame, ) from pipecat.processors.frame_processor import FrameDirection -from pipecat.services.settings import NOT_GIVEN, STTSettings, _NotGiven, is_given +from pipecat.services.settings import NOT_GIVEN, STTSettings, _NotGiven from pipecat.services.stt_latency import DEEPGRAM_TTFS_P99 from pipecat.services.stt_service import STTService from pipecat.transcriptions.language import Language @@ -51,11 +52,32 @@ except ModuleNotFoundError as e: class DeepgramSTTSettings(STTSettings): """Settings for the Deepgram STT service. + Some commonly used ``LiveOptions`` fields are declared as top-level + fields here so they can be updated individually via + ``STTUpdateSettingsFrame``. Any *additional* ``LiveOptions`` fields + (e.g. ``filler_words``, ``diarize``, ``utterance_end_ms``) can be + passed through the ``extra`` dict — they will be forwarded to + ``LiveOptions`` when the WebSocket connection is (re)established. + This keeps the settings class future-proof: new Deepgram features work + without code changes on the Pipecat side. + Parameters: - live_options: Deepgram ``LiveOptions`` for detailed configuration. + encoding: Audio encoding format (e.g. ``"linear16"``). + channels: Number of audio channels. + interim_results: Whether to return interim transcription results. + smart_format: Whether to enable Deepgram smart formatting. + punctuate: Whether to add punctuation to transcripts. + profanity_filter: Whether to filter profanity from transcripts. + vad_events: Whether to enable Deepgram VAD events (deprecated). """ - live_options: LiveOptions | _NotGiven = field(default_factory=lambda: NOT_GIVEN) + encoding: str | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN) + channels: int | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN) + interim_results: bool | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN) + smart_format: bool | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN) + punctuate: bool | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN) + profanity_filter: bool | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN) + vad_events: bool | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN) class DeepgramSTTService(STTService): @@ -153,22 +175,29 @@ class DeepgramSTTService(STTService): if "language" in merged_options and isinstance(merged_options["language"], Language): merged_options["language"] = merged_options["language"].value - merged_live_options = LiveOptions(**merged_options) + settings_fields = {f.name for f in fields(DeepgramSTTSettings)} + settings_kwargs = {} + extra = {} + for key, value in merged_options.items(): + if key in settings_fields: + settings_kwargs[key] = value + else: + extra[key] = value + + settings = DeepgramSTTSettings(**settings_kwargs) + settings.extra = extra + super().__init__( sample_rate=sample_rate, ttfs_p99_latency=ttfs_p99_latency, - settings=DeepgramSTTSettings( - model=merged_options.get("model"), - language=merged_options.get("language"), - live_options=merged_live_options, - ), + settings=settings, **kwargs, ) self._addons = addons self._should_interrupt = should_interrupt - if merged_live_options.vad_events: + if self._settings.vad_events: import warnings with warnings.catch_warnings(): @@ -199,7 +228,7 @@ class DeepgramSTTService(STTService): Returns: True if VAD events are enabled in the current settings. """ - return self._settings.live_options.vad_events + return self._settings.vad_events def can_generate_metrics(self) -> bool: """Check if this service can generate processing metrics. @@ -210,43 +239,12 @@ class DeepgramSTTService(STTService): return True async def _update_settings(self, delta: STTSettings) -> dict[str, Any]: - """Apply a settings delta, keeping ``live_options`` in sync. - - Top-level ``model`` and ``language`` are the source of truth. When - they are given in *delta* their values are propagated into - ``live_options``. When only ``live_options`` is given, its ``model`` - and ``language`` are propagated *up* to the top-level fields. - - Any change triggers a WebSocket reconnect. - """ - # Determine which top-level fields are explicitly provided. - model_given = isinstance(delta, DeepgramSTTSettings) and is_given( - getattr(delta, "model", NOT_GIVEN) - ) - language_given = isinstance(delta, DeepgramSTTSettings) and is_given( - getattr(delta, "language", NOT_GIVEN) - ) - + """Apply a settings delta and reconnect if anything changed.""" changed = await super()._update_settings(delta) if not changed: return changed - # --- Sync model -------------------------------------------------- - if model_given: - # Top-level model wins → push into live_options. - self._settings.live_options.model = self._settings.model - elif "live_options" in changed and self._settings.live_options.model is not None: - # Only live_options was given → pull model up. - self._settings.model = self._settings.live_options.model - self._sync_model_name_to_metrics() - - # --- Sync language ----------------------------------------------- - if language_given: - self._settings.live_options.language = self._settings.language - elif "live_options" in changed and self._settings.live_options.language is not None: - self._settings.language = self._settings.live_options.language - await self._disconnect() await self._connect() @@ -259,7 +257,6 @@ class DeepgramSTTService(STTService): frame: The start frame containing initialization parameters. """ await super().start(frame) - self._settings.live_options.sample_rate = self.sample_rate await self._connect() async def stop(self, frame: EndFrame): @@ -292,6 +289,36 @@ class DeepgramSTTService(STTService): await self._connection.send(audio) yield None + def _build_live_options(self) -> LiveOptions: + """Build a ``LiveOptions`` from flat settings fields, sample rate, and extras. + + Returns: + A fully-populated ``LiveOptions`` ready for the Deepgram SDK. + """ + valid_kwargs = set(inspect.signature(LiveOptions.__init__).parameters) - {"self"} + + # Start with extras that are valid LiveOptions kwargs. + opts: dict[str, Any] = {k: v for k, v in self._settings.extra.items() if k in valid_kwargs} + + # Override with flat settings fields (these take precedence). + s = self._settings + opts.update( + { + "model": s.model, + "language": s.language, + "encoding": s.encoding, + "channels": s.channels, + "interim_results": s.interim_results, + "smart_format": s.smart_format, + "punctuate": s.punctuate, + "profanity_filter": s.profanity_filter, + "vad_events": s.vad_events, + "sample_rate": self.sample_rate, + } + ) + + return LiveOptions(**opts) + async def _connect(self): logger.debug("Connecting to Deepgram") @@ -313,7 +340,7 @@ class DeepgramSTTService(STTService): ) if not await self._connection.start( - options=self._settings.live_options, addons=self._addons + options=self._build_live_options(), addons=self._addons ): await self.push_error(error_msg=f"Unable to connect to Deepgram") else: diff --git a/src/pipecat/services/deepgram/stt_sagemaker.py b/src/pipecat/services/deepgram/stt_sagemaker.py index bc5eebe37..97309ab93 100644 --- a/src/pipecat/services/deepgram/stt_sagemaker.py +++ b/src/pipecat/services/deepgram/stt_sagemaker.py @@ -13,8 +13,9 @@ languages, and various Deepgram features. """ import asyncio +import inspect import json -from dataclasses import dataclass, field +from dataclasses import dataclass, field, fields from typing import Any, AsyncGenerator, Optional from loguru import logger @@ -32,7 +33,7 @@ from pipecat.frames.frames import ( ) from pipecat.processors.frame_processor import FrameDirection from pipecat.services.aws.sagemaker.bidi_client import SageMakerBidiClient -from pipecat.services.settings import NOT_GIVEN, STTSettings, _NotGiven, is_given +from pipecat.services.settings import NOT_GIVEN, STTSettings, _NotGiven from pipecat.services.stt_latency import DEEPGRAM_SAGEMAKER_TTFS_P99 from pipecat.services.stt_service import STTService from pipecat.transcriptions.language import Language @@ -53,11 +54,32 @@ except ModuleNotFoundError as e: class DeepgramSageMakerSTTSettings(STTSettings): """Settings for the Deepgram SageMaker STT service. + Some commonly used ``LiveOptions`` fields are declared as top-level + fields here so they can be updated individually via + ``STTUpdateSettingsFrame``. Any *additional* ``LiveOptions`` fields + (e.g. ``filler_words``, ``diarize``, ``utterance_end_ms``) can be + passed through the ``extra`` dict — they will be forwarded to + ``LiveOptions`` when the connection is (re)established. This keeps the + settings class future-proof: new Deepgram features work without code + changes on the Pipecat side. + Parameters: - live_options: Deepgram ``LiveOptions`` for detailed configuration. + encoding: Audio encoding format (e.g. ``"linear16"``). + channels: Number of audio channels. + interim_results: Whether to return interim transcription results. + smart_format: Whether to enable Deepgram smart formatting. + punctuate: Whether to add punctuation to transcripts. + profanity_filter: Whether to filter profanity from transcripts. + vad_events: Whether to enable Deepgram VAD events (deprecated). """ - live_options: LiveOptions | _NotGiven = field(default_factory=lambda: NOT_GIVEN) + encoding: str | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN) + channels: int | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN) + interim_results: bool | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN) + smart_format: bool | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN) + punctuate: bool | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN) + profanity_filter: bool | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN) + vad_events: bool | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN) class DeepgramSageMakerSTTService(STTService): @@ -139,15 +161,22 @@ class DeepgramSageMakerSTTService(STTService): if "language" in merged_options and isinstance(merged_options["language"], Language): merged_options["language"] = merged_options["language"].value - merged_live_options = LiveOptions(**merged_options) + settings_fields = {f.name for f in fields(DeepgramSageMakerSTTSettings)} + settings_kwargs = {} + extra = {} + for key, value in merged_options.items(): + if key in settings_fields: + settings_kwargs[key] = value + else: + extra[key] = value + + settings = DeepgramSageMakerSTTSettings(**settings_kwargs) + settings.extra = extra + super().__init__( sample_rate=sample_rate, ttfs_p99_latency=ttfs_p99_latency, - settings=DeepgramSageMakerSTTSettings( - model=merged_options.get("model"), - language=merged_options.get("language"), - live_options=merged_live_options, - ), + settings=settings, **kwargs, ) @@ -167,46 +196,12 @@ class DeepgramSageMakerSTTService(STTService): return True async def _update_settings(self, delta: STTSettings) -> dict[str, Any]: - """Apply a settings delta, keeping ``live_options`` in sync. - - Top-level ``model`` and ``language`` are the source of truth. When - they are given in *delta* their values are propagated into - ``live_options``. When only ``live_options`` is given, its ``model`` - and ``language`` are propagated *up* to the top-level fields. - - Any change triggers a reconnect. - """ - # Determine which top-level fields are explicitly provided. - model_given = isinstance(delta, DeepgramSageMakerSTTSettings) and is_given( - getattr(delta, "model", NOT_GIVEN) - ) - language_given = isinstance(delta, DeepgramSageMakerSTTSettings) and is_given( - getattr(delta, "language", NOT_GIVEN) - ) - + """Apply a settings delta and warn about unhandled changes.""" changed = await super()._update_settings(delta) if not changed: return changed - # --- Sync model -------------------------------------------------- - if model_given: - # Top-level model wins → push into live_options. - self._settings.live_options.model = self._settings.model - elif "live_options" in changed and self._settings.live_options.model is not None: - # Only live_options was given → pull model up. - self._settings.model = self._settings.live_options.model - self._sync_model_name_to_metrics() - - # --- Sync language ----------------------------------------------- - if language_given: - lang = self._settings.language - if isinstance(lang, Language): - lang = lang.value - self._settings.live_options.language = lang - elif "live_options" in changed and self._settings.live_options.language is not None: - self._settings.language = self._settings.live_options.language - # TODO: someday we could reconnect here to apply updated settings. # Code might look something like the below: # await self._disconnect() @@ -223,7 +218,6 @@ class DeepgramSageMakerSTTService(STTService): frame: The start frame containing initialization parameters. """ await super().start(frame) - self._settings.live_options.sample_rate = self.sample_rate await self._connect() async def stop(self, frame: EndFrame): @@ -260,6 +254,36 @@ class DeepgramSageMakerSTTService(STTService): yield ErrorFrame(error=f"Unknown error occurred: {e}") yield None + def _build_live_options(self) -> LiveOptions: + """Build a ``LiveOptions`` from flat settings fields, sample rate, and extras. + + Returns: + A fully-populated ``LiveOptions`` ready for the Deepgram SDK. + """ + valid_kwargs = set(inspect.signature(LiveOptions.__init__).parameters) - {"self"} + + # Start with extras that are valid LiveOptions kwargs. + opts: dict[str, Any] = {k: v for k, v in self._settings.extra.items() if k in valid_kwargs} + + # Override with flat settings fields (these take precedence). + s = self._settings + opts.update( + { + "model": s.model, + "language": s.language, + "encoding": s.encoding, + "channels": s.channels, + "interim_results": s.interim_results, + "smart_format": s.smart_format, + "punctuate": s.punctuate, + "profanity_filter": s.profanity_filter, + "vad_events": s.vad_events, + "sample_rate": self.sample_rate, + } + ) + + return LiveOptions(**opts) + async def _connect(self): """Connect to the SageMaker endpoint and start the BiDi session. @@ -269,12 +293,9 @@ class DeepgramSageMakerSTTService(STTService): """ logger.debug("Connecting to Deepgram on SageMaker...") - # Update sample rate in live_options - self._settings.live_options.sample_rate = self.sample_rate - # Build query string from live_options, converting booleans to strings query_params = {} - for key, value in self._settings.live_options.to_dict().items(): + for key, value in self._build_live_options().to_dict().items(): if value is not None: # Convert boolean values to lowercase strings for Deepgram API if isinstance(value, bool): From 8b6aa4b912958ceaa619192fb779e9c20897b037 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Wed, 25 Feb 2026 18:18:47 -0500 Subject: [PATCH 2/6] Unflatten `LiveOptions` back into a single `live_options` field on `DeepgramSTTSettings` and `DeepgramSageMakerSTTSettings`; add `apply_update` override with delta-merge semantics and `from_mapping` override for backward-compatible dict-style updates --- ...-update-settings-deepgram-sagemaker-stt.py | 11 +- .../55a-update-settings-deepgram-stt.py | 11 +- src/pipecat/services/deepgram/stt.py | 201 +++++++++++------ .../services/deepgram/stt_sagemaker.py | 196 ++++++++++------ tests/test_settings.py | 211 ++++++++++++++++++ 5 files changed, 496 insertions(+), 134 deletions(-) diff --git a/examples/foundational/55a-update-settings-deepgram-sagemaker-stt.py b/examples/foundational/55a-update-settings-deepgram-sagemaker-stt.py index fba722648..e8094183a 100644 --- a/examples/foundational/55a-update-settings-deepgram-sagemaker-stt.py +++ b/examples/foundational/55a-update-settings-deepgram-sagemaker-stt.py @@ -7,6 +7,7 @@ import asyncio import os +from deepgram import LiveOptions from dotenv import load_dotenv from loguru import logger @@ -106,10 +107,16 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): messages.append({"role": "system", "content": "Please introduce yourself to the user."}) await task.queue_frames([LLMRunFrame()]) + # NOTE: after this change, the bot will only respond if you speak Spanish await asyncio.sleep(10) - logger.info("Updating Deepgram SageMaker STT settings: language=es") + logger.info("Updating Deepgram SageMaker STT settings: language=es, punctuate=False") await task.queue_frame( - STTUpdateSettingsFrame(delta=DeepgramSageMakerSTTSettings(language=Language.ES)) + STTUpdateSettingsFrame( + delta=DeepgramSageMakerSTTSettings( + language=Language.ES, + live_options=LiveOptions(punctuate=False), + ) + ) ) # Old-style dict update (for backward-compat testing): diff --git a/examples/foundational/55a-update-settings-deepgram-stt.py b/examples/foundational/55a-update-settings-deepgram-stt.py index 20068ab75..8808f6f4c 100644 --- a/examples/foundational/55a-update-settings-deepgram-stt.py +++ b/examples/foundational/55a-update-settings-deepgram-stt.py @@ -7,6 +7,7 @@ import asyncio import os +from deepgram import LiveOptions from dotenv import load_dotenv from loguru import logger @@ -100,10 +101,16 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): messages.append({"role": "system", "content": "Please introduce yourself to the user."}) await task.queue_frames([LLMRunFrame()]) + # NOTE: after this change, the bot will only respond if you speak Spanish await asyncio.sleep(10) - logger.info("Updating Deepgram STT settings: language=es") + logger.info("Updating Deepgram STT settings: language=es, punctuate=False") await task.queue_frame( - STTUpdateSettingsFrame(delta=DeepgramSTTSettings(language=Language.ES)) + STTUpdateSettingsFrame( + delta=DeepgramSTTSettings( + language=Language.ES, + live_options=LiveOptions(punctuate=False), + ) + ) ) # Old-style dict update (for backward-compat testing): diff --git a/src/pipecat/services/deepgram/stt.py b/src/pipecat/services/deepgram/stt.py index 768c4d6eb..f1c2fd19c 100644 --- a/src/pipecat/services/deepgram/stt.py +++ b/src/pipecat/services/deepgram/stt.py @@ -7,8 +7,8 @@ """Deepgram speech-to-text service implementation.""" import inspect -from dataclasses import dataclass, field, fields -from typing import Any, AsyncGenerator, Dict, Optional +from dataclasses import dataclass, field +from typing import Any, AsyncGenerator, Dict, Mapping, Optional, Type from loguru import logger @@ -25,7 +25,7 @@ from pipecat.frames.frames import ( VADUserStoppedSpeakingFrame, ) from pipecat.processors.frame_processor import FrameDirection -from pipecat.services.settings import NOT_GIVEN, STTSettings, _NotGiven +from pipecat.services.settings import _S, NOT_GIVEN, STTSettings, _NotGiven, is_given from pipecat.services.stt_latency import DEEPGRAM_TTFS_P99 from pipecat.services.stt_service import STTService from pipecat.transcriptions.language import Language @@ -52,32 +52,117 @@ except ModuleNotFoundError as e: class DeepgramSTTSettings(STTSettings): """Settings for the Deepgram STT service. - Some commonly used ``LiveOptions`` fields are declared as top-level - fields here so they can be updated individually via - ``STTUpdateSettingsFrame``. Any *additional* ``LiveOptions`` fields - (e.g. ``filler_words``, ``diarize``, ``utterance_end_ms``) can be - passed through the ``extra`` dict — they will be forwarded to - ``LiveOptions`` when the WebSocket connection is (re)established. - This keeps the settings class future-proof: new Deepgram features work - without code changes on the Pipecat side. + Wraps the Deepgram SDK's ``LiveOptions`` in a single ``live_options`` + field. All Deepgram-specific options (``filler_words``, ``diarize``, + ``utterance_end_ms``, etc.) should be passed directly via + ``LiveOptions``. + + In **delta mode** (i.e. when carried by ``STTUpdateSettingsFrame``), + ``live_options`` is treated as a **delta** — its non-None fields are + merged into the stored ``LiveOptions``, not replaced wholesale. For + example, ``DeepgramSTTSettings(live_options=LiveOptions(punctuate=False))`` + changes only ``punctuate`` and leaves all other options intact. Parameters: - encoding: Audio encoding format (e.g. ``"linear16"``). - channels: Number of audio channels. - interim_results: Whether to return interim transcription results. - smart_format: Whether to enable Deepgram smart formatting. - punctuate: Whether to add punctuation to transcripts. - profanity_filter: Whether to filter profanity from transcripts. - vad_events: Whether to enable Deepgram VAD events (deprecated). + live_options: Deepgram ``LiveOptions`` for STT configuration. + In delta mode only its non-None fields are merged into the + stored options. """ - encoding: str | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN) - channels: int | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN) - interim_results: bool | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN) - smart_format: bool | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN) - punctuate: bool | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN) - profanity_filter: bool | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN) - vad_events: bool | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN) + live_options: LiveOptions | _NotGiven = field(default_factory=lambda: NOT_GIVEN) + + # Valid LiveOptions __init__ parameter names (cached at class level). + _live_options_params: set[str] | None = field(default=None, init=False, repr=False) + + @classmethod + def _get_live_options_params(cls) -> set[str]: + """Return the set of valid ``LiveOptions.__init__`` parameter names.""" + if cls._live_options_params is None: + cls._live_options_params = set(inspect.signature(LiveOptions.__init__).parameters) - { + "self" + } + return cls._live_options_params + + def apply_update(self: _S, delta: _S) -> Dict[str, Any]: + """Merge a delta into this store, with delta-merge for ``live_options``. + + ``live_options`` is merged field-by-field (non-None fields from the + delta overwrite corresponding fields in the stored options) rather + than being replaced wholesale. + + ``model`` and ``language`` are kept in sync bidirectionally between + the top-level settings fields and ``live_options``. + """ + # Pull live_options out of the delta so super() doesn't replace it. + delta_lo = getattr(delta, "live_options", NOT_GIVEN) + if is_given(delta_lo): + delta.live_options = NOT_GIVEN # type: ignore[assignment] + + # Let the base class handle model, language, extra. + changed = super().apply_update(delta) + + # Sync top-level model/language changes into stored live_options. + if "model" in changed: + self.live_options.model = self.model # type: ignore[union-attr] + if "language" in changed: + self.live_options.language = self.language # type: ignore[union-attr] + + # Merge live_options delta. + if is_given(delta_lo): + old_dict = self.live_options.to_dict() # type: ignore[union-attr] + delta_dict = delta_lo.to_dict() + + if delta_dict: + merged = {**old_dict, **delta_dict} + self.live_options = LiveOptions(**merged) + + for key in delta_dict: + old_val = old_dict.get(key, NOT_GIVEN) + if old_val != delta_dict[key]: + changed[key] = old_val + + # Sync model/language from live_options delta to top-level. + if "model" in delta_dict and delta_dict["model"] != self.model: + changed.setdefault("model", self.model) + self.model = delta_dict["model"] + if "language" in delta_dict and delta_dict["language"] != self.language: + changed.setdefault("language", self.language) + self.language = delta_dict["language"] + + return changed + + @classmethod + def from_mapping(cls: Type[_S], settings: Mapping[str, Any]) -> _S: + """Build a delta from a plain dict, routing LiveOptions keys correctly. + + Keys that are valid ``LiveOptions.__init__`` parameters (and not + top-level ``STTSettings`` fields like ``model`` / ``language``) are + collected into a ``LiveOptions`` object. ``model`` and ``language`` + are routed to the top-level settings fields. Truly unknown keys go + to ``extra``. + """ + lo_params = cls._get_live_options_params() + stt_field_names = {"model", "language"} + + kwargs: Dict[str, Any] = {} + lo_kwargs: Dict[str, Any] = {} + extra: Dict[str, Any] = {} + + for key, value in settings.items(): + canonical = cls._aliases.get(key, key) + if canonical in stt_field_names: + kwargs[canonical] = value + elif canonical in lo_params: + lo_kwargs[canonical] = value + else: + extra[key] = value + + if lo_kwargs: + kwargs["live_options"] = LiveOptions(**lo_kwargs) + + instance = cls(**kwargs) + instance.extra = extra + return instance class DeepgramSTTService(STTService): @@ -124,7 +209,9 @@ class DeepgramSTTService(STTService): base_url: Custom Deepgram API base URL. sample_rate: Audio sample rate. If None, uses default or live_options value. - live_options: Deepgram LiveOptions for detailed configuration. + live_options: Deepgram LiveOptions configuration. Treated as a + delta from a set of sensible defaults — only the fields you + set are overridden; all others keep their default values. addons: Additional Deepgram features to enable. should_interrupt: Determine whether the bot should be interrupted when Deepgram VAD events are enabled and the system detects that the user is speaking. @@ -163,29 +250,26 @@ class DeepgramSTTService(STTService): vad_events=False, ) - merged_options = default_options.to_dict() + merged_dict = default_options.to_dict() if live_options: default_model = default_options.model - merged_options.update(live_options.to_dict()) - # NOTE(aleix): Fixes an in deepgram-sdk where `model` is initialized + merged_dict.update(live_options.to_dict()) + # NOTE(aleix): Fixes a bug in deepgram-sdk where `model` is initialized # to the string "None" instead of the value `None`. - if "model" in merged_options and merged_options["model"] == "None": - merged_options["model"] = default_model + if "model" in merged_dict and merged_dict["model"] == "None": + merged_dict["model"] = default_model - if "language" in merged_options and isinstance(merged_options["language"], Language): - merged_options["language"] = merged_options["language"].value + if "language" in merged_dict and isinstance(merged_dict["language"], Language): + merged_dict["language"] = merged_dict["language"].value - settings_fields = {f.name for f in fields(DeepgramSTTSettings)} - settings_kwargs = {} - extra = {} - for key, value in merged_options.items(): - if key in settings_fields: - settings_kwargs[key] = value - else: - extra[key] = value + # Extract model/language for top-level STTSettings fields; everything + # else lives inside LiveOptions. + model = merged_dict.pop("model", None) + language = merged_dict.pop("language", None) - settings = DeepgramSTTSettings(**settings_kwargs) - settings.extra = extra + settings = DeepgramSTTSettings( + model=model, language=language, live_options=LiveOptions(**merged_dict) + ) super().__init__( sample_rate=sample_rate, @@ -197,7 +281,7 @@ class DeepgramSTTService(STTService): self._addons = addons self._should_interrupt = should_interrupt - if self._settings.vad_events: + if self._settings.live_options.vad_events: import warnings with warnings.catch_warnings(): @@ -228,7 +312,7 @@ class DeepgramSTTService(STTService): Returns: True if VAD events are enabled in the current settings. """ - return self._settings.vad_events + return self._settings.live_options.vad_events def can_generate_metrics(self) -> bool: """Check if this service can generate processing metrics. @@ -290,32 +374,17 @@ class DeepgramSTTService(STTService): yield None def _build_live_options(self) -> LiveOptions: - """Build a ``LiveOptions`` from flat settings fields, sample rate, and extras. + """Build a ``LiveOptions`` from stored settings and sample rate. Returns: A fully-populated ``LiveOptions`` ready for the Deepgram SDK. """ - valid_kwargs = set(inspect.signature(LiveOptions.__init__).parameters) - {"self"} + opts: dict[str, Any] = self._settings.live_options.to_dict() - # Start with extras that are valid LiveOptions kwargs. - opts: dict[str, Any] = {k: v for k, v in self._settings.extra.items() if k in valid_kwargs} - - # Override with flat settings fields (these take precedence). - s = self._settings - opts.update( - { - "model": s.model, - "language": s.language, - "encoding": s.encoding, - "channels": s.channels, - "interim_results": s.interim_results, - "smart_format": s.smart_format, - "punctuate": s.punctuate, - "profanity_filter": s.profanity_filter, - "vad_events": s.vad_events, - "sample_rate": self.sample_rate, - } - ) + # Overlay model/language from top-level settings and sample_rate from service. + opts["model"] = self._settings.model + opts["language"] = self._settings.language + opts["sample_rate"] = self.sample_rate return LiveOptions(**opts) diff --git a/src/pipecat/services/deepgram/stt_sagemaker.py b/src/pipecat/services/deepgram/stt_sagemaker.py index 97309ab93..12357a8cd 100644 --- a/src/pipecat/services/deepgram/stt_sagemaker.py +++ b/src/pipecat/services/deepgram/stt_sagemaker.py @@ -15,8 +15,8 @@ languages, and various Deepgram features. import asyncio import inspect import json -from dataclasses import dataclass, field, fields -from typing import Any, AsyncGenerator, Optional +from dataclasses import dataclass, field +from typing import Any, AsyncGenerator, Dict, Mapping, Optional, Type from loguru import logger @@ -33,7 +33,7 @@ from pipecat.frames.frames import ( ) from pipecat.processors.frame_processor import FrameDirection from pipecat.services.aws.sagemaker.bidi_client import SageMakerBidiClient -from pipecat.services.settings import NOT_GIVEN, STTSettings, _NotGiven +from pipecat.services.settings import _S, NOT_GIVEN, STTSettings, _NotGiven, is_given from pipecat.services.stt_latency import DEEPGRAM_SAGEMAKER_TTFS_P99 from pipecat.services.stt_service import STTService from pipecat.transcriptions.language import Language @@ -54,32 +54,117 @@ except ModuleNotFoundError as e: class DeepgramSageMakerSTTSettings(STTSettings): """Settings for the Deepgram SageMaker STT service. - Some commonly used ``LiveOptions`` fields are declared as top-level - fields here so they can be updated individually via - ``STTUpdateSettingsFrame``. Any *additional* ``LiveOptions`` fields - (e.g. ``filler_words``, ``diarize``, ``utterance_end_ms``) can be - passed through the ``extra`` dict — they will be forwarded to - ``LiveOptions`` when the connection is (re)established. This keeps the - settings class future-proof: new Deepgram features work without code - changes on the Pipecat side. + Wraps the Deepgram SDK's ``LiveOptions`` in a single ``live_options`` + field. All Deepgram-specific options (``filler_words``, ``diarize``, + ``utterance_end_ms``, etc.) should be passed directly via + ``LiveOptions``. + + In **delta mode** (i.e. when carried by ``STTUpdateSettingsFrame``), + ``live_options`` is treated as a **delta** — its non-None fields are + merged into the stored ``LiveOptions``, not replaced wholesale. For + example, ``DeepgramSageMakerSTTSettings(live_options=LiveOptions(punctuate=False))`` + changes only ``punctuate`` and leaves all other options intact. Parameters: - encoding: Audio encoding format (e.g. ``"linear16"``). - channels: Number of audio channels. - interim_results: Whether to return interim transcription results. - smart_format: Whether to enable Deepgram smart formatting. - punctuate: Whether to add punctuation to transcripts. - profanity_filter: Whether to filter profanity from transcripts. - vad_events: Whether to enable Deepgram VAD events (deprecated). + live_options: Deepgram ``LiveOptions`` for STT configuration. + In delta mode only its non-None fields are merged into the + stored options. """ - encoding: str | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN) - channels: int | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN) - interim_results: bool | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN) - smart_format: bool | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN) - punctuate: bool | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN) - profanity_filter: bool | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN) - vad_events: bool | None | _NotGiven = field(default_factory=lambda: NOT_GIVEN) + live_options: LiveOptions | _NotGiven = field(default_factory=lambda: NOT_GIVEN) + + # Valid LiveOptions __init__ parameter names (cached at class level). + _live_options_params: set[str] | None = field(default=None, init=False, repr=False) + + @classmethod + def _get_live_options_params(cls) -> set[str]: + """Return the set of valid ``LiveOptions.__init__`` parameter names.""" + if cls._live_options_params is None: + cls._live_options_params = set(inspect.signature(LiveOptions.__init__).parameters) - { + "self" + } + return cls._live_options_params + + def apply_update(self: _S, delta: _S) -> Dict[str, Any]: + """Merge a delta into this store, with delta-merge for ``live_options``. + + ``live_options`` is merged field-by-field (non-None fields from the + delta overwrite corresponding fields in the stored options) rather + than being replaced wholesale. + + ``model`` and ``language`` are kept in sync bidirectionally between + the top-level settings fields and ``live_options``. + """ + # Pull live_options out of the delta so super() doesn't replace it. + delta_lo = getattr(delta, "live_options", NOT_GIVEN) + if is_given(delta_lo): + delta.live_options = NOT_GIVEN # type: ignore[assignment] + + # Let the base class handle model, language, extra. + changed = super().apply_update(delta) + + # Sync top-level model/language changes into stored live_options. + if "model" in changed: + self.live_options.model = self.model # type: ignore[union-attr] + if "language" in changed: + self.live_options.language = self.language # type: ignore[union-attr] + + # Merge live_options delta. + if is_given(delta_lo): + old_dict = self.live_options.to_dict() # type: ignore[union-attr] + delta_dict = delta_lo.to_dict() + + if delta_dict: + merged = {**old_dict, **delta_dict} + self.live_options = LiveOptions(**merged) + + for key in delta_dict: + old_val = old_dict.get(key, NOT_GIVEN) + if old_val != delta_dict[key]: + changed[key] = old_val + + # Sync model/language from live_options delta to top-level. + if "model" in delta_dict and delta_dict["model"] != self.model: + changed.setdefault("model", self.model) + self.model = delta_dict["model"] + if "language" in delta_dict and delta_dict["language"] != self.language: + changed.setdefault("language", self.language) + self.language = delta_dict["language"] + + return changed + + @classmethod + def from_mapping(cls: Type[_S], settings: Mapping[str, Any]) -> _S: + """Build a delta from a plain dict, routing LiveOptions keys correctly. + + Keys that are valid ``LiveOptions.__init__`` parameters (and not + top-level ``STTSettings`` fields like ``model`` / ``language``) are + collected into a ``LiveOptions`` object. ``model`` and ``language`` + are routed to the top-level settings fields. Truly unknown keys go + to ``extra``. + """ + lo_params = cls._get_live_options_params() + stt_field_names = {"model", "language"} + + kwargs: Dict[str, Any] = {} + lo_kwargs: Dict[str, Any] = {} + extra: Dict[str, Any] = {} + + for key, value in settings.items(): + canonical = cls._aliases.get(key, key) + if canonical in stt_field_names: + kwargs[canonical] = value + elif canonical in lo_params: + lo_kwargs[canonical] = value + else: + extra[key] = value + + if lo_kwargs: + kwargs["live_options"] = LiveOptions(**lo_kwargs) + + instance = cls(**kwargs) + instance.extra = extra + return instance class DeepgramSageMakerSTTService(STTService): @@ -130,8 +215,9 @@ class DeepgramSageMakerSTTService(STTService): region: AWS region where the endpoint is deployed (e.g., "us-east-2"). sample_rate: Audio sample rate in Hz. If None, uses value from live_options or defaults to the value from StartFrame. - live_options: Deepgram LiveOptions for detailed configuration. If None, - uses sensible defaults (nova-3 model, English, interim results enabled). + live_options: Deepgram LiveOptions configuration. Treated as a + delta from a set of sensible defaults — only the fields you + set are overridden; all others keep their default values. ttfs_p99_latency: P99 latency from speech end to final transcript in seconds. Override for your deployment. See https://github.com/pipecat-ai/stt-benchmark **kwargs: Additional arguments passed to the parent STTService. @@ -149,29 +235,26 @@ class DeepgramSageMakerSTTService(STTService): ) # Merge with provided options - merged_options = default_options.to_dict() + merged_dict = default_options.to_dict() if live_options: default_model = default_options.model - merged_options.update(live_options.to_dict()) + merged_dict.update(live_options.to_dict()) # Handle the "None" string bug from deepgram-sdk - if "model" in merged_options and merged_options["model"] == "None": - merged_options["model"] = default_model + if "model" in merged_dict and merged_dict["model"] == "None": + merged_dict["model"] = default_model # Convert Language enum to string if needed - if "language" in merged_options and isinstance(merged_options["language"], Language): - merged_options["language"] = merged_options["language"].value + if "language" in merged_dict and isinstance(merged_dict["language"], Language): + merged_dict["language"] = merged_dict["language"].value - settings_fields = {f.name for f in fields(DeepgramSageMakerSTTSettings)} - settings_kwargs = {} - extra = {} - for key, value in merged_options.items(): - if key in settings_fields: - settings_kwargs[key] = value - else: - extra[key] = value + # Extract model/language for top-level STTSettings fields; everything + # else lives inside LiveOptions. + model = merged_dict.pop("model", None) + language = merged_dict.pop("language", None) - settings = DeepgramSageMakerSTTSettings(**settings_kwargs) - settings.extra = extra + settings = DeepgramSageMakerSTTSettings( + model=model, language=language, live_options=LiveOptions(**merged_dict) + ) super().__init__( sample_rate=sample_rate, @@ -255,32 +338,17 @@ class DeepgramSageMakerSTTService(STTService): yield None def _build_live_options(self) -> LiveOptions: - """Build a ``LiveOptions`` from flat settings fields, sample rate, and extras. + """Build a ``LiveOptions`` from stored settings and sample rate. Returns: A fully-populated ``LiveOptions`` ready for the Deepgram SDK. """ - valid_kwargs = set(inspect.signature(LiveOptions.__init__).parameters) - {"self"} + opts: dict[str, Any] = self._settings.live_options.to_dict() - # Start with extras that are valid LiveOptions kwargs. - opts: dict[str, Any] = {k: v for k, v in self._settings.extra.items() if k in valid_kwargs} - - # Override with flat settings fields (these take precedence). - s = self._settings - opts.update( - { - "model": s.model, - "language": s.language, - "encoding": s.encoding, - "channels": s.channels, - "interim_results": s.interim_results, - "smart_format": s.smart_format, - "punctuate": s.punctuate, - "profanity_filter": s.profanity_filter, - "vad_events": s.vad_events, - "sample_rate": self.sample_rate, - } - ) + # Overlay model/language from top-level settings and sample_rate from service. + opts["model"] = self._settings.model + opts["language"] = self._settings.language + opts["sample_rate"] = self.sample_rate return LiveOptions(**opts) diff --git a/tests/test_settings.py b/tests/test_settings.py index 85f89987c..3201e3c24 100644 --- a/tests/test_settings.py +++ b/tests/test_settings.py @@ -7,7 +7,10 @@ """Tests for the typed settings infrastructure in pipecat.services.settings.""" import pytest +from deepgram import LiveOptions +from pipecat.services.deepgram.stt import DeepgramSTTSettings +from pipecat.services.deepgram.stt_sagemaker import DeepgramSageMakerSTTSettings from pipecat.services.settings import ( NOT_GIVEN, LLMSettings, @@ -311,3 +314,211 @@ class TestRoundtrip: assert changed["model"] == "gpt-4o" assert current.model == "gpt-4o-mini" assert current.temperature == 0.9 + + +# --------------------------------------------------------------------------- +# DeepgramSTTSettings: live_options delta merge +# --------------------------------------------------------------------------- + + +class TestDeepgramSTTSettingsApplyUpdate: + def _make_store(self, **lo_kwargs) -> DeepgramSTTSettings: + """Helper to build a store-mode DeepgramSTTSettings.""" + defaults = dict( + encoding="linear16", + channels=1, + interim_results=True, + smart_format=False, + punctuate=True, + profanity_filter=True, + vad_events=False, + ) + defaults.update(lo_kwargs) + s = DeepgramSTTSettings( + model="nova-3-general", + language="en", + live_options=LiveOptions(**defaults), + ) + return s + + def test_apply_update_merges_live_options_as_delta(self): + """Only the given fields in the delta LiveOptions are merged.""" + current = self._make_store() + assert current.live_options.punctuate is True + + delta = DeepgramSTTSettings(live_options=LiveOptions(punctuate=False)) + changed = current.apply_update(delta) + + assert current.live_options.punctuate is False + assert "punctuate" in changed + # Other fields are untouched + assert current.live_options.encoding == "linear16" + assert current.live_options.channels == 1 + + def test_apply_update_syncs_model_from_live_options_to_top_level(self): + """model inside live_options delta should sync to top-level model.""" + current = self._make_store() + assert current.model == "nova-3-general" + + delta = DeepgramSTTSettings(live_options=LiveOptions(model="nova-2")) + changed = current.apply_update(delta) + + assert current.model == "nova-2" + assert "model" in changed + + def test_apply_update_syncs_language_from_live_options_to_top_level(self): + """language inside live_options delta should sync to top-level language.""" + current = self._make_store() + assert current.language == "en" + + delta = DeepgramSTTSettings(live_options=LiveOptions(language="es")) + changed = current.apply_update(delta) + + assert current.language == "es" + assert "language" in changed + + def test_apply_update_syncs_top_level_model_into_live_options(self): + """Top-level model change should propagate into stored live_options.""" + current = self._make_store() + assert current.model == "nova-3-general" + + delta = DeepgramSTTSettings(model="nova-2") + changed = current.apply_update(delta) + + assert current.model == "nova-2" + assert current.live_options.model == "nova-2" + assert "model" in changed + + def test_apply_update_syncs_top_level_language_into_live_options(self): + """Top-level language change should propagate into stored live_options.""" + current = self._make_store() + + delta = DeepgramSTTSettings(language="fr") + changed = current.apply_update(delta) + + assert current.language == "fr" + assert current.live_options.language == "fr" + assert "language" in changed + + def test_apply_update_no_change(self): + """Delta with same values should report no changes.""" + current = self._make_store() + delta = DeepgramSTTSettings(live_options=LiveOptions(punctuate=True)) + changed = current.apply_update(delta) + assert changed == {} + + +class TestDeepgramSTTSettingsFromMapping: + def test_routes_live_options_kwargs(self): + """LiveOptions-valid keys should be collected into live_options.""" + delta = DeepgramSTTSettings.from_mapping({"punctuate": False, "filler_words": True}) + assert is_given(delta.live_options) + assert delta.live_options.punctuate is False + assert delta.live_options.filler_words is True + + def test_routes_model_and_language_to_top_level(self): + """model and language should be top-level fields, not in live_options.""" + delta = DeepgramSTTSettings.from_mapping({"model": "nova-2", "language": "es"}) + assert delta.model == "nova-2" + assert delta.language == "es" + assert not is_given(delta.live_options) + + def test_unknown_keys_go_to_extra(self): + """Keys that aren't LiveOptions params or STT fields go to extra.""" + delta = DeepgramSTTSettings.from_mapping({"unknown_param": 42}) + assert delta.extra == {"unknown_param": 42} + assert not is_given(delta.live_options) + + def test_mixed_keys(self): + """model + LiveOptions keys + unknown keys are routed correctly.""" + delta = DeepgramSTTSettings.from_mapping( + {"model": "nova-2", "punctuate": False, "unknown": "val"} + ) + assert delta.model == "nova-2" + assert delta.live_options.punctuate is False + assert delta.extra == {"unknown": "val"} + + def test_roundtrip_from_mapping_apply_update(self): + """Simulate dict-style update: from_mapping -> apply_update.""" + current = DeepgramSTTSettings( + model="nova-3-general", + language="en", + live_options=LiveOptions( + encoding="linear16", + channels=1, + interim_results=True, + punctuate=True, + profanity_filter=True, + vad_events=False, + ), + ) + + raw = {"punctuate": False, "filler_words": True} + delta = DeepgramSTTSettings.from_mapping(raw) + changed = current.apply_update(delta) + + assert current.live_options.punctuate is False + assert current.live_options.filler_words is True + # Unchanged fields stay put + assert current.live_options.encoding == "linear16" + assert current.model == "nova-3-general" + assert "punctuate" in changed + + def test_roundtrip_model_via_dict(self): + """Dict update with model should change top-level and NOT create live_options.""" + current = DeepgramSTTSettings( + model="nova-3-general", + language="en", + live_options=LiveOptions(encoding="linear16", channels=1), + ) + + raw = {"model": "nova-2"} + delta = DeepgramSTTSettings.from_mapping(raw) + changed = current.apply_update(delta) + + assert current.model == "nova-2" + assert current.live_options.model == "nova-2" + assert "model" in changed + + +# --------------------------------------------------------------------------- +# DeepgramSageMakerSTTSettings: same pattern +# --------------------------------------------------------------------------- + + +class TestDeepgramSageMakerSTTSettingsApplyUpdate: + def _make_store(self, **lo_kwargs) -> DeepgramSageMakerSTTSettings: + defaults = dict( + encoding="linear16", + channels=1, + interim_results=True, + punctuate=True, + ) + defaults.update(lo_kwargs) + return DeepgramSageMakerSTTSettings( + model="nova-3", + language="en", + live_options=LiveOptions(**defaults), + ) + + def test_apply_update_merges_live_options_as_delta(self): + current = self._make_store() + delta = DeepgramSageMakerSTTSettings(live_options=LiveOptions(punctuate=False)) + changed = current.apply_update(delta) + assert current.live_options.punctuate is False + assert "punctuate" in changed + assert current.live_options.encoding == "linear16" + + def test_apply_update_syncs_model_from_live_options(self): + current = self._make_store() + delta = DeepgramSageMakerSTTSettings(live_options=LiveOptions(model="nova-2")) + current.apply_update(delta) + assert current.model == "nova-2" + + def test_from_mapping_routes_correctly(self): + delta = DeepgramSageMakerSTTSettings.from_mapping( + {"model": "nova-2", "punctuate": False, "unknown": "val"} + ) + assert delta.model == "nova-2" + assert delta.live_options.punctuate is False + assert delta.extra == {"unknown": "val"} From e21e8585f07964a6a1de114b259a7fd3a88cba53 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Wed, 25 Feb 2026 18:40:44 -0500 Subject: [PATCH 3/6] Add `deepgram` and `sagemaker` extras to CI test dependencies so Deepgram and Deepgram Sagemaker settings tests can run --- .github/workflows/coverage.yaml | 2 ++ .github/workflows/tests.yaml | 2 ++ 2 files changed, 4 insertions(+) diff --git a/.github/workflows/coverage.yaml b/.github/workflows/coverage.yaml index d65841a7d..26d03861b 100644 --- a/.github/workflows/coverage.yaml +++ b/.github/workflows/coverage.yaml @@ -37,10 +37,12 @@ jobs: uv sync --group dev \ --extra anthropic \ --extra aws \ + --extra deepgram \ --extra google \ --extra langchain \ --extra livekit \ --extra piper \ + --extra sagemaker \ --extra tracing \ --extra websocket diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml index a36a2fbd0..b22d502c4 100644 --- a/.github/workflows/tests.yaml +++ b/.github/workflows/tests.yaml @@ -41,10 +41,12 @@ jobs: uv sync --group dev \ --extra anthropic \ --extra aws \ + --extra deepgram \ --extra google \ --extra langchain \ --extra livekit \ --extra piper \ + --extra sagemaker \ --extra tracing \ --extra websocket From 3c20eda8bf95a8590d907ba72603a2c319fad473 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Thu, 26 Feb 2026 09:32:52 -0500 Subject: [PATCH 4/6] Keep model/language in LiveOptions at construction time so apply_update's bidirectional sync is sufficient; simplify _build_live_options to only add sample_rate --- src/pipecat/services/deepgram/stt.py | 11 +++-------- src/pipecat/services/deepgram/stt_sagemaker.py | 11 +++-------- 2 files changed, 6 insertions(+), 16 deletions(-) diff --git a/src/pipecat/services/deepgram/stt.py b/src/pipecat/services/deepgram/stt.py index f1c2fd19c..b193f899d 100644 --- a/src/pipecat/services/deepgram/stt.py +++ b/src/pipecat/services/deepgram/stt.py @@ -262,10 +262,9 @@ class DeepgramSTTService(STTService): if "language" in merged_dict and isinstance(merged_dict["language"], Language): merged_dict["language"] = merged_dict["language"].value - # Extract model/language for top-level STTSettings fields; everything - # else lives inside LiveOptions. - model = merged_dict.pop("model", None) - language = merged_dict.pop("language", None) + # Sync model/language to top-level STTSettings fields + model = merged_dict.get("model") + language = merged_dict.get("language") settings = DeepgramSTTSettings( model=model, language=language, live_options=LiveOptions(**merged_dict) @@ -380,10 +379,6 @@ class DeepgramSTTService(STTService): A fully-populated ``LiveOptions`` ready for the Deepgram SDK. """ opts: dict[str, Any] = self._settings.live_options.to_dict() - - # Overlay model/language from top-level settings and sample_rate from service. - opts["model"] = self._settings.model - opts["language"] = self._settings.language opts["sample_rate"] = self.sample_rate return LiveOptions(**opts) diff --git a/src/pipecat/services/deepgram/stt_sagemaker.py b/src/pipecat/services/deepgram/stt_sagemaker.py index 12357a8cd..6f91906b7 100644 --- a/src/pipecat/services/deepgram/stt_sagemaker.py +++ b/src/pipecat/services/deepgram/stt_sagemaker.py @@ -247,10 +247,9 @@ class DeepgramSageMakerSTTService(STTService): if "language" in merged_dict and isinstance(merged_dict["language"], Language): merged_dict["language"] = merged_dict["language"].value - # Extract model/language for top-level STTSettings fields; everything - # else lives inside LiveOptions. - model = merged_dict.pop("model", None) - language = merged_dict.pop("language", None) + # Sync model/language to top-level STTSettings fields + model = merged_dict.get("model") + language = merged_dict.get("language") settings = DeepgramSageMakerSTTSettings( model=model, language=language, live_options=LiveOptions(**merged_dict) @@ -344,10 +343,6 @@ class DeepgramSageMakerSTTService(STTService): A fully-populated ``LiveOptions`` ready for the Deepgram SDK. """ opts: dict[str, Any] = self._settings.live_options.to_dict() - - # Overlay model/language from top-level settings and sample_rate from service. - opts["model"] = self._settings.model - opts["language"] = self._settings.language opts["sample_rate"] = self.sample_rate return LiveOptions(**opts) From c184ac09b8323571a4269182e3f9af9fe6d70c1b Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Thu, 26 Feb 2026 09:42:15 -0500 Subject: [PATCH 5/6] Inline `_build_live_options` into `_connect` in `DeepgramSTTService` and `DeepgramSageMakerSTTService` since it's trivial and only called from one place --- src/pipecat/services/deepgram/stt.py | 19 +++++-------------- .../services/deepgram/stt_sagemaker.py | 17 +++++------------ 2 files changed, 10 insertions(+), 26 deletions(-) diff --git a/src/pipecat/services/deepgram/stt.py b/src/pipecat/services/deepgram/stt.py index b193f899d..59107d0f8 100644 --- a/src/pipecat/services/deepgram/stt.py +++ b/src/pipecat/services/deepgram/stt.py @@ -372,17 +372,6 @@ class DeepgramSTTService(STTService): await self._connection.send(audio) yield None - def _build_live_options(self) -> LiveOptions: - """Build a ``LiveOptions`` from stored settings and sample rate. - - Returns: - A fully-populated ``LiveOptions`` ready for the Deepgram SDK. - """ - opts: dict[str, Any] = self._settings.live_options.to_dict() - opts["sample_rate"] = self.sample_rate - - return LiveOptions(**opts) - async def _connect(self): logger.debug("Connecting to Deepgram") @@ -403,9 +392,11 @@ class DeepgramSTTService(STTService): self._on_utterance_end, ) - if not await self._connection.start( - options=self._build_live_options(), addons=self._addons - ): + live_options = LiveOptions( + **{**self._settings.live_options.to_dict(), "sample_rate": self.sample_rate} + ) + + if not await self._connection.start(options=live_options, addons=self._addons): await self.push_error(error_msg=f"Unable to connect to Deepgram") else: headers = { diff --git a/src/pipecat/services/deepgram/stt_sagemaker.py b/src/pipecat/services/deepgram/stt_sagemaker.py index 6f91906b7..ee2121bec 100644 --- a/src/pipecat/services/deepgram/stt_sagemaker.py +++ b/src/pipecat/services/deepgram/stt_sagemaker.py @@ -336,17 +336,6 @@ class DeepgramSageMakerSTTService(STTService): yield ErrorFrame(error=f"Unknown error occurred: {e}") yield None - def _build_live_options(self) -> LiveOptions: - """Build a ``LiveOptions`` from stored settings and sample rate. - - Returns: - A fully-populated ``LiveOptions`` ready for the Deepgram SDK. - """ - opts: dict[str, Any] = self._settings.live_options.to_dict() - opts["sample_rate"] = self.sample_rate - - return LiveOptions(**opts) - async def _connect(self): """Connect to the SageMaker endpoint and start the BiDi session. @@ -356,9 +345,13 @@ class DeepgramSageMakerSTTService(STTService): """ logger.debug("Connecting to Deepgram on SageMaker...") + live_options = LiveOptions( + **{**self._settings.live_options.to_dict(), "sample_rate": self.sample_rate} + ) + # Build query string from live_options, converting booleans to strings query_params = {} - for key, value in self._build_live_options().to_dict().items(): + for key, value in live_options.to_dict().items(): if value is not None: # Convert boolean values to lowercase strings for Deepgram API if isinstance(value, bool): From faed775d9075064af3804be89f94322c36af2c8e Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Thu, 26 Feb 2026 11:02:44 -0500 Subject: [PATCH 6/6] Extract `_DeepgramSTTSettingsBase` with shared `_merge_live_options_delta` to deduplicate LiveOptions merge logic between `__init__` and `apply_update`, and between the Deepgram STT and SageMaker variants; make top-level model/language take precedence over conflicting live_options values in updates; remove unnecessary Language enum-to-string conversion (Language is a StrEnum) --- src/pipecat/services/deepgram/stt.py | 131 ++++++++++------ .../services/deepgram/stt_sagemaker.py | 145 ++---------------- tests/test_settings.py | 72 +++++---- 3 files changed, 134 insertions(+), 214 deletions(-) diff --git a/src/pipecat/services/deepgram/stt.py b/src/pipecat/services/deepgram/stt.py index 59107d0f8..497d6aae1 100644 --- a/src/pipecat/services/deepgram/stt.py +++ b/src/pipecat/services/deepgram/stt.py @@ -49,19 +49,20 @@ except ModuleNotFoundError as e: @dataclass -class DeepgramSTTSettings(STTSettings): - """Settings for the Deepgram STT service. +class _DeepgramSTTSettingsBase(STTSettings): + """Base settings for Deepgram STT services that use ``LiveOptions``. + + Shared by ``DeepgramSTTSettings`` and ``DeepgramSageMakerSTTSettings``. + Not intended for other Deepgram services that don't use ``LiveOptions``. Wraps the Deepgram SDK's ``LiveOptions`` in a single ``live_options`` - field. All Deepgram-specific options (``filler_words``, ``diarize``, - ``utterance_end_ms``, etc.) should be passed directly via - ``LiveOptions``. + field and provides delta-merge semantics: when used as a delta (e.g. + via ``STTUpdateSettingsFrame``), only the non-None fields of + ``live_options`` are merged into the stored options rather than + replacing them wholesale. - In **delta mode** (i.e. when carried by ``STTUpdateSettingsFrame``), - ``live_options`` is treated as a **delta** — its non-None fields are - merged into the stored ``LiveOptions``, not replaced wholesale. For - example, ``DeepgramSTTSettings(live_options=LiveOptions(punctuate=False))`` - changes only ``punctuate`` and leaves all other options intact. + ``model`` and ``language`` are kept in sync bidirectionally between + the top-level settings fields and the nested ``live_options``. Parameters: live_options: Deepgram ``LiveOptions`` for STT configuration. @@ -83,12 +84,56 @@ class DeepgramSTTSettings(STTSettings): } return cls._live_options_params + def _merge_live_options_delta(self, delta: LiveOptions) -> Dict[str, Any]: + """Merge a ``LiveOptions`` delta into the stored ``live_options``. + + Non-None fields from *delta* overwrite corresponding fields in the + stored ``LiveOptions``. ``model`` and ``language`` are synced to + the top-level settings fields when they change. + + Args: + delta: A ``LiveOptions`` whose non-None fields are the desired + overrides. + + Returns: + Dict mapping each changed key to its **previous** value (same + contract as ``apply_update``). + """ + old_dict = self.live_options.to_dict() # type: ignore[union-attr] + delta_dict = delta.to_dict() + + # Deepgram SDK bug: model initialised to the *string* "None". + if delta_dict.get("model") == "None": + del delta_dict["model"] + + if not delta_dict: + return {} + + merged = {**old_dict, **delta_dict} + self.live_options = LiveOptions(**merged) + + # Track what changed. + changed: Dict[str, Any] = {} + for key in delta_dict: + old_val = old_dict.get(key, NOT_GIVEN) + if old_val != delta_dict[key]: + changed[key] = old_val + + # Sync model/language from live_options delta to top-level fields. + if "model" in delta_dict and delta_dict["model"] != self.model: + changed.setdefault("model", self.model) + self.model = delta_dict["model"] + if "language" in delta_dict and delta_dict["language"] != self.language: + changed.setdefault("language", self.language) + self.language = delta_dict["language"] + + return changed + def apply_update(self: _S, delta: _S) -> Dict[str, Any]: """Merge a delta into this store, with delta-merge for ``live_options``. - ``live_options`` is merged field-by-field (non-None fields from the - delta overwrite corresponding fields in the stored options) rather - than being replaced wholesale. + ``live_options`` is merged field-by-field via + ``_merge_live_options_delta`` rather than being replaced wholesale. ``model`` and ``language`` are kept in sync bidirectionally between the top-level settings fields and ``live_options``. @@ -107,27 +152,17 @@ class DeepgramSTTSettings(STTSettings): if "language" in changed: self.live_options.language = self.language # type: ignore[union-attr] - # Merge live_options delta. + # Merge live_options delta. Top-level model/language take precedence + # over conflicting values in live_options, so write them into the + # delta before merging. if is_given(delta_lo): - old_dict = self.live_options.to_dict() # type: ignore[union-attr] - delta_dict = delta_lo.to_dict() + if "model" in changed: + delta_lo.model = self.model + if "language" in changed: + delta_lo.language = self.language - if delta_dict: - merged = {**old_dict, **delta_dict} - self.live_options = LiveOptions(**merged) - - for key in delta_dict: - old_val = old_dict.get(key, NOT_GIVEN) - if old_val != delta_dict[key]: - changed[key] = old_val - - # Sync model/language from live_options delta to top-level. - if "model" in delta_dict and delta_dict["model"] != self.model: - changed.setdefault("model", self.model) - self.model = delta_dict["model"] - if "language" in delta_dict and delta_dict["language"] != self.language: - changed.setdefault("language", self.language) - self.language = delta_dict["language"] + for key, old_val in self._merge_live_options_delta(delta_lo).items(): + changed.setdefault(key, old_val) return changed @@ -165,6 +200,16 @@ class DeepgramSTTSettings(STTSettings): return instance +@dataclass +class DeepgramSTTSettings(_DeepgramSTTSettingsBase): + """Settings for the Deepgram STT service. + + See ``_DeepgramSTTSettingsBase`` for full documentation. + """ + + pass + + class DeepgramSTTService(STTService): """Deepgram speech-to-text service. @@ -250,25 +295,13 @@ class DeepgramSTTService(STTService): vad_events=False, ) - merged_dict = default_options.to_dict() - if live_options: - default_model = default_options.model - merged_dict.update(live_options.to_dict()) - # NOTE(aleix): Fixes a bug in deepgram-sdk where `model` is initialized - # to the string "None" instead of the value `None`. - if "model" in merged_dict and merged_dict["model"] == "None": - merged_dict["model"] = default_model - - if "language" in merged_dict and isinstance(merged_dict["language"], Language): - merged_dict["language"] = merged_dict["language"].value - - # Sync model/language to top-level STTSettings fields - model = merged_dict.get("model") - language = merged_dict.get("language") - settings = DeepgramSTTSettings( - model=model, language=language, live_options=LiveOptions(**merged_dict) + model=default_options.model, + language=default_options.language, + live_options=default_options, ) + if live_options: + settings._merge_live_options_delta(live_options) super().__init__( sample_rate=sample_rate, diff --git a/src/pipecat/services/deepgram/stt_sagemaker.py b/src/pipecat/services/deepgram/stt_sagemaker.py index ee2121bec..ba4b7dfda 100644 --- a/src/pipecat/services/deepgram/stt_sagemaker.py +++ b/src/pipecat/services/deepgram/stt_sagemaker.py @@ -13,10 +13,9 @@ languages, and various Deepgram features. """ import asyncio -import inspect import json -from dataclasses import dataclass, field -from typing import Any, AsyncGenerator, Dict, Mapping, Optional, Type +from dataclasses import dataclass +from typing import Any, AsyncGenerator, Dict, Optional from loguru import logger @@ -33,7 +32,8 @@ from pipecat.frames.frames import ( ) from pipecat.processors.frame_processor import FrameDirection from pipecat.services.aws.sagemaker.bidi_client import SageMakerBidiClient -from pipecat.services.settings import _S, NOT_GIVEN, STTSettings, _NotGiven, is_given +from pipecat.services.deepgram.stt import _DeepgramSTTSettingsBase +from pipecat.services.settings import STTSettings from pipecat.services.stt_latency import DEEPGRAM_SAGEMAKER_TTFS_P99 from pipecat.services.stt_service import STTService from pipecat.transcriptions.language import Language @@ -51,120 +51,13 @@ except ModuleNotFoundError as e: @dataclass -class DeepgramSageMakerSTTSettings(STTSettings): +class DeepgramSageMakerSTTSettings(_DeepgramSTTSettingsBase): """Settings for the Deepgram SageMaker STT service. - Wraps the Deepgram SDK's ``LiveOptions`` in a single ``live_options`` - field. All Deepgram-specific options (``filler_words``, ``diarize``, - ``utterance_end_ms``, etc.) should be passed directly via - ``LiveOptions``. - - In **delta mode** (i.e. when carried by ``STTUpdateSettingsFrame``), - ``live_options`` is treated as a **delta** — its non-None fields are - merged into the stored ``LiveOptions``, not replaced wholesale. For - example, ``DeepgramSageMakerSTTSettings(live_options=LiveOptions(punctuate=False))`` - changes only ``punctuate`` and leaves all other options intact. - - Parameters: - live_options: Deepgram ``LiveOptions`` for STT configuration. - In delta mode only its non-None fields are merged into the - stored options. + See ``_DeepgramSTTSettingsBase`` for full documentation. """ - live_options: LiveOptions | _NotGiven = field(default_factory=lambda: NOT_GIVEN) - - # Valid LiveOptions __init__ parameter names (cached at class level). - _live_options_params: set[str] | None = field(default=None, init=False, repr=False) - - @classmethod - def _get_live_options_params(cls) -> set[str]: - """Return the set of valid ``LiveOptions.__init__`` parameter names.""" - if cls._live_options_params is None: - cls._live_options_params = set(inspect.signature(LiveOptions.__init__).parameters) - { - "self" - } - return cls._live_options_params - - def apply_update(self: _S, delta: _S) -> Dict[str, Any]: - """Merge a delta into this store, with delta-merge for ``live_options``. - - ``live_options`` is merged field-by-field (non-None fields from the - delta overwrite corresponding fields in the stored options) rather - than being replaced wholesale. - - ``model`` and ``language`` are kept in sync bidirectionally between - the top-level settings fields and ``live_options``. - """ - # Pull live_options out of the delta so super() doesn't replace it. - delta_lo = getattr(delta, "live_options", NOT_GIVEN) - if is_given(delta_lo): - delta.live_options = NOT_GIVEN # type: ignore[assignment] - - # Let the base class handle model, language, extra. - changed = super().apply_update(delta) - - # Sync top-level model/language changes into stored live_options. - if "model" in changed: - self.live_options.model = self.model # type: ignore[union-attr] - if "language" in changed: - self.live_options.language = self.language # type: ignore[union-attr] - - # Merge live_options delta. - if is_given(delta_lo): - old_dict = self.live_options.to_dict() # type: ignore[union-attr] - delta_dict = delta_lo.to_dict() - - if delta_dict: - merged = {**old_dict, **delta_dict} - self.live_options = LiveOptions(**merged) - - for key in delta_dict: - old_val = old_dict.get(key, NOT_GIVEN) - if old_val != delta_dict[key]: - changed[key] = old_val - - # Sync model/language from live_options delta to top-level. - if "model" in delta_dict and delta_dict["model"] != self.model: - changed.setdefault("model", self.model) - self.model = delta_dict["model"] - if "language" in delta_dict and delta_dict["language"] != self.language: - changed.setdefault("language", self.language) - self.language = delta_dict["language"] - - return changed - - @classmethod - def from_mapping(cls: Type[_S], settings: Mapping[str, Any]) -> _S: - """Build a delta from a plain dict, routing LiveOptions keys correctly. - - Keys that are valid ``LiveOptions.__init__`` parameters (and not - top-level ``STTSettings`` fields like ``model`` / ``language``) are - collected into a ``LiveOptions`` object. ``model`` and ``language`` - are routed to the top-level settings fields. Truly unknown keys go - to ``extra``. - """ - lo_params = cls._get_live_options_params() - stt_field_names = {"model", "language"} - - kwargs: Dict[str, Any] = {} - lo_kwargs: Dict[str, Any] = {} - extra: Dict[str, Any] = {} - - for key, value in settings.items(): - canonical = cls._aliases.get(key, key) - if canonical in stt_field_names: - kwargs[canonical] = value - elif canonical in lo_params: - lo_kwargs[canonical] = value - else: - extra[key] = value - - if lo_kwargs: - kwargs["live_options"] = LiveOptions(**lo_kwargs) - - instance = cls(**kwargs) - instance.extra = extra - return instance + pass class DeepgramSageMakerSTTService(STTService): @@ -224,7 +117,6 @@ class DeepgramSageMakerSTTService(STTService): """ sample_rate = sample_rate or (live_options.sample_rate if live_options else None) - # Create default options similar to DeepgramSTTService default_options = LiveOptions( encoding="linear16", language=Language.EN, @@ -234,26 +126,13 @@ class DeepgramSageMakerSTTService(STTService): punctuate=True, ) - # Merge with provided options - merged_dict = default_options.to_dict() - if live_options: - default_model = default_options.model - merged_dict.update(live_options.to_dict()) - # Handle the "None" string bug from deepgram-sdk - if "model" in merged_dict and merged_dict["model"] == "None": - merged_dict["model"] = default_model - - # Convert Language enum to string if needed - if "language" in merged_dict and isinstance(merged_dict["language"], Language): - merged_dict["language"] = merged_dict["language"].value - - # Sync model/language to top-level STTSettings fields - model = merged_dict.get("model") - language = merged_dict.get("language") - settings = DeepgramSageMakerSTTSettings( - model=model, language=language, live_options=LiveOptions(**merged_dict) + model=default_options.model, + language=default_options.language, + live_options=default_options, ) + if live_options: + settings._merge_live_options_delta(live_options) super().__init__( sample_rate=sample_rate, diff --git a/tests/test_settings.py b/tests/test_settings.py index 3201e3c24..47cb6e4cf 100644 --- a/tests/test_settings.py +++ b/tests/test_settings.py @@ -407,6 +407,36 @@ class TestDeepgramSTTSettingsApplyUpdate: changed = current.apply_update(delta) assert changed == {} + def test_apply_update_top_level_model_takes_precedence_over_live_options(self): + """When both top-level model and live_options.model are set, top-level wins.""" + current = self._make_store() + assert current.model == "nova-3-general" + + delta = DeepgramSTTSettings( + model="nova-2", + live_options=LiveOptions(model="nova-3"), + ) + changed = current.apply_update(delta) + + assert current.model == "nova-2" + assert current.live_options.model == "nova-2" + assert "model" in changed + + def test_apply_update_top_level_language_takes_precedence_over_live_options(self): + """When both top-level language and live_options.language are set, top-level wins.""" + current = self._make_store() + assert current.language == "en" + + delta = DeepgramSTTSettings( + language="fr", + live_options=LiveOptions(language="es"), + ) + changed = current.apply_update(delta) + + assert current.language == "fr" + assert current.live_options.language == "fr" + assert "language" in changed + class TestDeepgramSTTSettingsFromMapping: def test_routes_live_options_kwargs(self): @@ -482,43 +512,21 @@ class TestDeepgramSTTSettingsFromMapping: # --------------------------------------------------------------------------- -# DeepgramSageMakerSTTSettings: same pattern +# DeepgramSageMakerSTTSettings: smoke test that the shared base is inherited # --------------------------------------------------------------------------- -class TestDeepgramSageMakerSTTSettingsApplyUpdate: - def _make_store(self, **lo_kwargs) -> DeepgramSageMakerSTTSettings: - defaults = dict( - encoding="linear16", - channels=1, - interim_results=True, - punctuate=True, - ) - defaults.update(lo_kwargs) - return DeepgramSageMakerSTTSettings( +class TestDeepgramSageMakerSTTSettings: + def test_inherits_live_options_behavior(self): + """Smoke test: SageMaker settings inherit the shared base correctly.""" + store = DeepgramSageMakerSTTSettings( model="nova-3", language="en", - live_options=LiveOptions(**defaults), + live_options=LiveOptions(encoding="linear16", channels=1, punctuate=True), ) - - def test_apply_update_merges_live_options_as_delta(self): - current = self._make_store() delta = DeepgramSageMakerSTTSettings(live_options=LiveOptions(punctuate=False)) - changed = current.apply_update(delta) - assert current.live_options.punctuate is False + changed = store.apply_update(delta) + + assert store.live_options.punctuate is False + assert store.live_options.encoding == "linear16" assert "punctuate" in changed - assert current.live_options.encoding == "linear16" - - def test_apply_update_syncs_model_from_live_options(self): - current = self._make_store() - delta = DeepgramSageMakerSTTSettings(live_options=LiveOptions(model="nova-2")) - current.apply_update(delta) - assert current.model == "nova-2" - - def test_from_mapping_routes_correctly(self): - delta = DeepgramSageMakerSTTSettings.from_mapping( - {"model": "nova-2", "punctuate": False, "unknown": "val"} - ) - assert delta.model == "nova-2" - assert delta.live_options.punctuate is False - assert delta.extra == {"unknown": "val"}