From c5da3cf2bdaf79e76d3d905803836a30701b4631 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Sat, 7 Mar 2026 08:37:27 -0500 Subject: [PATCH 1/2] Add on-the-fly Configure support for Deepgram Flux STT Wire up the existing settings update infrastructure to send a Configure WebSocket message when keyterm, eot_threshold, eager_eot_threshold, or eot_timeout_ms change mid-stream, avoiding a full reconnect. --- .../55a-update-settings-deepgram-flux-stt.py | 13 +++++ src/pipecat/services/deepgram/flux/stt.py | 51 ++++++++++++++++--- 2 files changed, 58 insertions(+), 6 deletions(-) diff --git a/examples/foundational/55a-update-settings-deepgram-flux-stt.py b/examples/foundational/55a-update-settings-deepgram-flux-stt.py index 72d9b5638..11254127d 100644 --- a/examples/foundational/55a-update-settings-deepgram-flux-stt.py +++ b/examples/foundational/55a-update-settings-deepgram-flux-stt.py @@ -100,6 +100,19 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): context.add_message({"role": "user", "content": "Please introduce yourself to the user."}) await task.queue_frames([LLMRunFrame()]) + # Update configure-able fields mid-stream (sent via Configure message, + # no reconnect needed). + await asyncio.sleep(10) + logger.info("Updating Deepgram Flux STT settings: eot_threshold, keyterm") + await task.queue_frame( + STTUpdateSettingsFrame( + delta=DeepgramFluxSTTSettings( + eot_threshold=0.8, + keyterm=["Pipecat", "Deepgram"], + ) + ) + ) + await asyncio.sleep(10) logger.info("Updating Deepgram Flux STT settings: language=es") await task.queue_frame( diff --git a/src/pipecat/services/deepgram/flux/stt.py b/src/pipecat/services/deepgram/flux/stt.py index 1ed28a749..e87bfc2d7 100644 --- a/src/pipecat/services/deepgram/flux/stt.py +++ b/src/pipecat/services/deepgram/flux/stt.py @@ -53,6 +53,8 @@ class FluxMessageType(str, Enum): RECEIVE_CONNECTED = "Connected" RECEIVE_FATAL_ERROR = "Error" TURN_INFO = "TurnInfo" + CONFIGURE_SUCCESS = "ConfigureSuccess" + CONFIGURE_FAILURE = "ConfigureFailure" class FluxEventType(str, Enum): @@ -114,6 +116,7 @@ class DeepgramFluxSTTService(WebsocketSTTService): """ _settings: DeepgramFluxSTTSettings + _CONFIGURE_FIELDS = {"keyterm", "eot_threshold", "eager_eot_threshold", "eot_timeout_ms"} class InputParams(BaseModel): """Configuration parameters for Deepgram Flux API. @@ -409,6 +412,33 @@ class DeepgramFluxSTTService(WebsocketSTTService): except Exception as e: await self.push_error(error_msg=f"Error sending closeStream: {e}", exception=e) + async def _send_configure(self, fields: set[str]): + """Send a Configure control message to update settings mid-stream. + + Builds a Configure JSON message containing only the fields that changed + and sends it over the existing WebSocket connection. + + Args: + fields: Set of changed field names to include in the message. + """ + message: dict[str, Any] = {"type": "Configure"} + + if "keyterm" in fields: + message["keyterms"] = self._settings.keyterm + + thresholds: dict[str, Any] = {} + if "eot_threshold" in fields: + thresholds["eot_threshold"] = self._settings.eot_threshold + if "eager_eot_threshold" in fields: + thresholds["eager_eot_threshold"] = self._settings.eager_eot_threshold + if "eot_timeout_ms" in fields: + thresholds["eot_timeout_ms"] = self._settings.eot_timeout_ms + if thresholds: + message["thresholds"] = thresholds + + logger.debug(f"{self}: sending Configure message: {message}") + await self._websocket.send(json.dumps(message)) + def can_generate_metrics(self) -> bool: """Check if this service can generate processing metrics. @@ -420,19 +450,20 @@ class DeepgramFluxSTTService(WebsocketSTTService): async def _update_settings(self, delta: DeepgramFluxSTTSettings) -> dict[str, Any]: """Apply a settings delta. - Settings are stored but not applied to the active connection. + Configure-able fields (keyterm, eot_threshold, eager_eot_threshold, + eot_timeout_ms) are sent to Deepgram via a Configure WebSocket message. + Other fields are stored but cannot be applied to the active connection. """ changed = await super()._update_settings(delta) if not changed: return changed - # TODO: someday we could reconnect here to apply updated settings. - # Code might look something like the below: - # await self._disconnect() - # await self._connect() + configure_fields = changed.keys() & self._CONFIGURE_FIELDS + if configure_fields and self._websocket and self._websocket.state is State.OPEN: + await self._send_configure(configure_fields) - self._warn_unhandled_updated_settings(changed) + self._warn_unhandled_updated_settings(changed.keys() - self._CONFIGURE_FIELDS) return changed @@ -629,6 +660,14 @@ class DeepgramFluxSTTService(WebsocketSTTService): await self._handle_fatal_error(data) case FluxMessageType.TURN_INFO: await self._handle_turn_info(data) + case FluxMessageType.CONFIGURE_SUCCESS: + logger.info(f"{self}: Configure accepted: {data}") + case FluxMessageType.CONFIGURE_FAILURE: + error_code = data.get("error_code", "unknown") + description = data.get("description", "no description") + error_msg = f"Configure rejected: [{error_code}] {description}" + logger.warning(f"{self}: {error_msg}") + await self.push_error(error_msg=error_msg) async def _handle_connection_established(self): """Handle successful connection establishment to Deepgram Flux. From 4ebdacdea23ab95949d79e43c79f6064586ac229 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Sat, 7 Mar 2026 08:48:11 -0500 Subject: [PATCH 2/2] Add changelog for #3953 --- changelog/3953.added.md | 1 + 1 file changed, 1 insertion(+) create mode 100644 changelog/3953.added.md diff --git a/changelog/3953.added.md b/changelog/3953.added.md new file mode 100644 index 000000000..d540ff125 --- /dev/null +++ b/changelog/3953.added.md @@ -0,0 +1 @@ +- Added on-the-fly `Configure` support for `DeepgramFluxSTTService`. The `keyterm`, `eot_threshold`, `eager_eot_threshold`, and `eot_timeout_ms` settings can now be updated mid-stream via `STTUpdateSettingsFrame` without requiring a WebSocket reconnect.