From a4b6db6fb44148709c22c65039f02ef677ad603b Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Tue, 24 Feb 2026 15:06:06 -0500 Subject: [PATCH] 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):