Connection Issues with Deepgram Service

[jonnyjohnson1]

Occasionally deepgram disconnects. This is rare. Like once every 50 tries.

My settings:
pipecat-ai = "^0.0.53"
deepgram-sdk = "^3.8.0"

I see the keepalive setting is set to True in the DeepgramSTT class:

 self._client = DeepgramClient(
            api_key,
            config=DeepgramClientOptions(
                url=url,
                options={"keepalive": "true"},  # verbose=logging.DEBUG
            ),
        )

Here is the error that occurs:

ConnectionClosed in AbstractAsyncWebSocketClient._listening with code 1011: Deepgram did not receive audio data or a text message within the timeout window. See https://dpgr.am/net0001
send() failed - ConnectionClosed: received 1011 (internal error) Deepgram did not receive audio data or a text message within the timeout window. See https://dpgr.am/net0001; then sent 1011 (internal error) Deepgram did not receive audio data or a text message within the timeout window. See https://dpgr.am/net0001
ConnectionClosed in AbstractAsyncWebSocketClient._listening with code 1011: Deepgram did not receive audio data or a text message within the timeout window. See https://dpgr.am/net0001
send() failed - ConnectionClosed: received 1011 (internal error) Deepgram did not receive audio data or a text message within the timeout window. See https://dpgr.am/net0001; then sent 1011 (internal error) Deepgram did not receive audio data or a text message within the timeout window. See https://dpgr.am/net0001

[jonnyjohnson1]

It looks like we could add error handling to the connection through the deepgram-sdk per their readme:

def on_error(self, error, **kwargs):
    print(f"\n\n{error}\n\n")

def on_close(self, close, **kwargs):
    print(f"\n\n{close}\n\n")

dg_connection.on(LiveTranscriptionEvents.Error, on_error)
dg_connection.on(LiveTranscriptionEvents.Close, on_close)
'''

In the deepgram pipecat class we'd write:

self._connection.on(LiveTranscriptionEvents.Error, self._on_error) # add to line 102

Then add this method:  

async def _on_error(self, error, **kwargs):
logger.error(f"{self}: {error}")
# TODO begin adding error handling logic here. Particularly this ConnectionClosed error that is being thrown.

[jonnyjohnson1]

In an attempt to set the keepalive time directly from 10s to 7s, I'm finding this recommendation with GPT to be able to adjust the timeout setting on the deepgram client by setting up the websocket connection directly.

import asyncio
import deepgram
import websockets

class DeepgramSTTService(STTService):
    def __init__(self, api_key: str):
        self._client = deepgram.Client(api_key)
        self._connection = None
        self._connection: AsyncListenWebSocketClient = self.start_stream()

    async def start_stream(self):
        ws_url = self._client.listen.asyncwebsocket.v("1")._url  # Extract WebSocket URL
        timeout = 7.0  # Set 7-second timeout

        try:
            async with websockets.connect(ws_url, timeout=timeout) as ws:
                self._connection = ws
                print("WebSocket connected successfully!")

        except asyncio.TimeoutError:
            print("WebSocket connection timed out!")
    
     async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
             if self._connection:
                  await self._connection.send(audio)
                  yield None