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)
This commit is contained in:
Paul Kompfner
@@ -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,
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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"}
|
||||
|
||||
Reference in New Issue
Block a user