From c926063d743c14eba99ce87f01b3e507359c5e14 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= <aleix@daily.co>
Date: Tue, 18 Feb 2025 09:39:13 -0800
Subject: [PATCH] deepgram: handle error event and reconnect

---
 CHANGELOG.md                     |  4 ++++
 src/pipecat/services/deepgram.py | 33 ++++++++++++++++++++++++++++----
 2 files changed, 33 insertions(+), 4 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 373a03ebd..7d4de0b74 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Fixed an issue that would cause `DeepgramSTTService` to stop working after an
+  error occurred (e.g. sudden network loss). If the network recovered we would
+  not reconnect.
+
 - Fixed a `STTMuteFilter` issue that would not mute user audio frames causing
   transcriptions to be generated by the STT service.
 
diff --git a/src/pipecat/services/deepgram.py b/src/pipecat/services/deepgram.py
index b9d39fa63..43782e4eb 100644
--- a/src/pipecat/services/deepgram.py
+++ b/src/pipecat/services/deepgram.py
@@ -34,6 +34,7 @@ try:
         AsyncListenWebSocketClient,
         DeepgramClient,
         DeepgramClientOptions,
+        ErrorResponse,
         LiveOptions,
         LiveResultResponse,
         LiveTranscriptionEvents,
@@ -155,13 +156,10 @@ class DeepgramSTTService(STTService):
                 options={"keepalive": "true"},  # verbose=logging.DEBUG
             ),
         )
-        self._connection: AsyncListenWebSocketClient = self._client.listen.asyncwebsocket.v("1")
-        self._connection.on(LiveTranscriptionEvents.Transcript, self._on_message)
+
         if self.vad_enabled:
             self._register_event_handler("on_speech_started")
             self._register_event_handler("on_utterance_end")
-            self._connection.on(LiveTranscriptionEvents.SpeechStarted, self._on_speech_started)
-            self._connection.on(LiveTranscriptionEvents.UtteranceEnd, self._on_utterance_end)
 
     @property
     def vad_enabled(self):
@@ -202,6 +200,24 @@ class DeepgramSTTService(STTService):
 
     async def _connect(self):
         logger.debug("Connecting to Deepgram")
+
+        self._connection: AsyncListenWebSocketClient = self._client.listen.asyncwebsocket.v("1")
+
+        self._connection.on(
+            LiveTranscriptionEvents(LiveTranscriptionEvents.Transcript), self._on_message
+        )
+        self._connection.on(LiveTranscriptionEvents(LiveTranscriptionEvents.Error), self._on_error)
+
+        if self.vad_enabled:
+            self._connection.on(
+                LiveTranscriptionEvents(LiveTranscriptionEvents.SpeechStarted),
+                self._on_speech_started,
+            )
+            self._connection.on(
+                LiveTranscriptionEvents(LiveTranscriptionEvents.UtteranceEnd),
+                self._on_utterance_end,
+            )
+
         if not await self._connection.start(self._settings):
             logger.error(f"{self}: unable to connect to Deepgram")
 
@@ -214,6 +230,15 @@ class DeepgramSTTService(STTService):
         await self.start_ttfb_metrics()
         await self.start_processing_metrics()
 
+    async def _on_error(self, *args, **kwargs):
+        error: ErrorResponse = kwargs["error"]
+        logger.warning(f"{self} connection error, will retry: {error}")
+        await self.stop_all_metrics()
+        # NOTE(aleix): we don't disconnect (i.e. call finish on the connection)
+        # because this triggers more errors internally in the Deepgram SDK. So,
+        # we just forget about the previous connection and create a new one.
+        await self._connect()
+
     async def _on_speech_started(self, *args, **kwargs):
         await self.start_metrics()
         await self._call_event_handler("on_speech_started", *args, **kwargs)