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"}