From 0c779b4c3d1f64e6ee114aa76880b8766642a4af Mon Sep 17 00:00:00 2001 From: filipi87 Date: Wed, 6 May 2026 11:01:58 -0300 Subject: [PATCH] Implementing dynamic watchdog timeout for Deepgram Flux STT --- src/pipecat/services/deepgram/flux/base.py | 9 ++++++--- src/pipecat/services/deepgram/flux/sagemaker/stt.py | 1 + src/pipecat/services/deepgram/flux/stt.py | 1 + 3 files changed, 8 insertions(+), 3 deletions(-) diff --git a/src/pipecat/services/deepgram/flux/base.py b/src/pipecat/services/deepgram/flux/base.py index d2a8a24cd..b69b60749 100644 --- a/src/pipecat/services/deepgram/flux/base.py +++ b/src/pipecat/services/deepgram/flux/base.py @@ -191,6 +191,7 @@ class DeepgramFluxSTTBase(STTService): self._last_stt_time: float | None = None self._watchdog_task: asyncio.Task | None = None self._user_is_speaking = False + self._last_audio_chunk_duration: float = 0.0 # Flux event handlers self._register_event_handler("on_start_of_turn") @@ -291,9 +292,11 @@ class DeepgramFluxSTTBase(STTService): """ while self._transport_is_active(): now = time.monotonic() - # More than 500 ms without sending new audio to Flux - if self._user_is_speaking and self._last_stt_time and now - self._last_stt_time > 0.5: - logger.warning("Sending silence to Flux to prevent dangling task") + # Send silence if we go more than 500 ms or twice the chunk size + # without sending new audio to Flux. + threshold = max(self._last_audio_chunk_duration * 2, 0.5) + if self._user_is_speaking and self._last_stt_time and now - self._last_stt_time > threshold: + logger.warning(f"No audio received for {threshold * 1000:.0f} ms. Sending silence to Flux to prevent a dangling task") try: await self._send_silence() except Exception as e: diff --git a/src/pipecat/services/deepgram/flux/sagemaker/stt.py b/src/pipecat/services/deepgram/flux/sagemaker/stt.py index 0f7a6fded..177e44063 100644 --- a/src/pipecat/services/deepgram/flux/sagemaker/stt.py +++ b/src/pipecat/services/deepgram/flux/sagemaker/stt.py @@ -245,6 +245,7 @@ class DeepgramFluxSageMakerSTTService(DeepgramFluxSTTBase): if self._client and self._client.is_active: try: self._last_stt_time = time.monotonic() + self._last_audio_chunk_duration = len(audio) / (self.sample_rate * 2) await self._client.send_audio_chunk(audio) except Exception as e: yield ErrorFrame(error=f"Unknown error occurred: {e}") diff --git a/src/pipecat/services/deepgram/flux/stt.py b/src/pipecat/services/deepgram/flux/stt.py index f89b37f0b..2499201cc 100644 --- a/src/pipecat/services/deepgram/flux/stt.py +++ b/src/pipecat/services/deepgram/flux/stt.py @@ -390,6 +390,7 @@ class DeepgramFluxSTTService(DeepgramFluxSTTBase, WebsocketService): try: self._last_stt_time = time.monotonic() + self._last_audio_chunk_duration = len(audio) / (self.sample_rate * 2) await self.send_with_retry(audio, self._report_error) except Exception as e: yield ErrorFrame(error=f"Unknown error occurred: {e}")