send() failed - ConnectionClosed: received 1011 (internal error) Deepgram did not receive audio data or a text message within the timeout window.

[iscarface]

Hello colleagues,
I have a voice assistant app that use DeepGram as STT service. While using it with English - everything is ok. But as soon as I switched to German deepgram start sending me this error. I do use "keepalive" option in connection. I tried both manually and with SDK options.
This error happens randomly, It isn't related to time or anything else (at least I didn't found such relation)

I've tried to reconnect in on_error signal but it's not working.

I'm using Python deepgram-sdk==3.7.4

Here is my code:

    config: ClassVar[DeepgramClientOptions] = DeepgramClientOptions(
        options={"keepalive": "true"}
    )
    client: ClassVar[DeepgramClient] = DeepgramClient(
        os.environ.get("DEEPGRAM_API_KEY"), config
    )
    connection = client.listen.asyncwebsocket.v("1")
    options: LiveOptions = LiveOptions(
          model="nova-2",
          language=language,
          smart_format=True,
          # To get UtteranceEnd, the following must be set:
          interim_results=True,
          utterance_end_ms="5000",  # str(grace_period +1 * 1000),
          vad_events=True,
          # Time in milliseconds of silence to wait for before finalizing speech
          endpointing=300,
      )

      # Inject event listeners
      if not self.event_registered:
          self.stt_service: AsyncDeepgramService = AsyncDeepgramService()
          self.stt_service.connection.on(
              LiveTranscriptionEvents.Transcript, self.on_stt_response
          )
          self.stt_service.connection.on(
              LiveTranscriptionEvents.UtteranceEnd, self.on_utterance_end
          )
          self.stt_service.connection.on(LiveTranscriptionEvents.Error, self.on_deepgram_error)
          self.event_registered = True

      if await self.stt_service.connection.start(self.stt_service.options) is False:
          print("Failed to connect to Deepgram")
          return

Please let me know if you know how to fix this problem or have any ideas what to look at (because I run out of)
Thanks!

[deepgram-community[bot]]

Hey there! It looks like you haven't connected your GitHub account to your Deepgram account. You can do this at https://community.deepgram.com - being verified through this process will allow our team to help you in a much more streamlined fashion.

[iscarface]

Unfortunately can't do that

[deepgram-community[bot]]

It looks like we're missing some important information to help debug your issue. Would you mind providing us with the following details in a reply?

• A request ID that triggered your error or issue.

[iscarface]

Please check if this would be enough.
If not let me know where I can find request ID

backend | [2024-10-22 09:59:22,988.988] ERROR [262567444783136] - 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

[iscarface]

It is important to mention that I'm using not German but Switzerland German (de-CH)

[iscarface]

Friendly reminder.
I'm in production so would be great to have any hints if you have.
Thanks in advance.

[deepgram-community[bot]]

I'm sorry to hear that you're experiencing issues with the Deepgram SDK. From the error message, it seems like Deepgram is not receiving audio data or a text message within the timeout window. This could be due to a variety of reasons such as network issues, incorrect configuration, or issues with the audio source.

Here are a few things you could try:

1. Check your network: Ensure that your server has a stable connection and is able to send data to Deepgram's servers.

2. Check your audio source: Make sure that the audio source is working correctly and is able to send audio data continuously.

3. Check your configuration: From your code, it seems like you're using the "keepalive" option. This option is used to keep the connection alive even if there's no data being sent. However, if Deepgram doesn't receive any data within a certain timeout window, it will close the connection. You could try increasing the timeout window or ensure that there's always data being sent.

4. Reconnect on error: You mentioned that you tried to reconnect in the on_error signal but it's not working. Could you provide more details about this? Are there any error messages when you try to reconnect?

5. Check the Deepgram SDK version: You're using deepgram-sdk==3.7.4. You could try updating to the latest version to see if the issue persists.

If none of these suggestions work, could you provide more details about your setup? For example, what kind of audio source are you using? How are you sending the audio data to Deepgram? This could help us better understand the issue and provide more targeted suggestions.

Naomi

[iscarface]

Hello Naomi,
Thank you for the message.

Check your network: Ensure that your server has a stable connection and is able to send data to Deepgram's servers.

Network is ok, it works smooth, checked couple times. And using the same network English version works perfectly fine.

Check your audio source: Make sure that the audio source is working correctly and is able to send audio data continuously.

My audio source is websocket stream from browser through my fastapi app. I 110% sure app is sent audio chunks to deepgram (logged it and see that chunks are sent)

Check your configuration: From your code, it seems like you're using the "keepalive" option. This option is used to keep the connection alive even if there's no data being sent. However, if Deepgram doesn't receive any data within a certain timeout window, it will close the connection. You could try increasing the timeout window or ensure that there's always data being sent.

The error I describing usually happens after some time of using service (not first phrases).
I tried to send keepalive using library options and manually (to websocket) - result the same. And I'm positive that I'm sending keepAlive almost every second or so.
How can I increase timeout window to keep connection alive?

Reconnect on error: You mentioned that you tried to reconnect in the on_error signal but it's not working. Could you provide more details about this? Are there any error messages when you try to reconnect?

Let me show you how my app works so I can explain why reconnection is not working for me.
I have a FastAPI application that receives audio data through websocket from ReactJS app.
Here is my main endpoint code:

@chat_router.websocket("/listen")
async def websocket_endpoint(
    websocket: WebSocket,
    user: User = Depends(auth_user_by_token),
    db_session: AsyncSession = Depends(get_async_session),
):
    await websocket.accept()
    async with ChatHandler(
        websocket=websocket, user=user, db_session=db_session
    ) as handler:
        await handler.arun()

Here is my method ChatHandler.__aenter__:

async def __aenter__(self) -> Self:
    self.active_language = await self.get_active_language()
    self.grace_period = await self.get_grace_period()

    # Inject event listeners
    if not self.event_registered:
        signals_mapping = {
            "Transcript": self.on_stt_response,
            "UtteranceEnd": self.on_utterance_end,
            "Error": self.on_deepgram_error
        }
        self.stt_service: AsyncDeepgramService = AsyncDeepgramService(
            language=self.active_language.langcode,
            grace_period=self.grace_period,
            user=self.user,
            signals_mapping=signals_mapping
        )
        self.event_registered = True

    if await self.stt_service.connection.start(self.stt_service.options) is False:
        print("Failed to connect to Deepgram")
        return

Here is my AsynDeepgramService:

class AsyncDeepgramService:
    config: ClassVar[DeepgramClientOptions] = DeepgramClientOptions(
        options={}
    )
    client: ClassVar[DeepgramClient] = DeepgramClient(
        os.environ.get("ALFRED_DEEPGRAM_API_KEY"), config
    )
    connection = client.listen.asyncwebsocket.v("1")

    def __init__(self, language, grace_period, user, signals_mapping) -> None:
        self.user = user
        self.signals_mapping = signals_mapping
        self.options: LiveOptions = LiveOptions(
            model="nova-2",
            language=language,
            smart_format=True,
            # To get UtteranceEnd, the following must be set:
            interim_results=True,
            utterance_end_ms="5000",  # str(grace_period +1 * 1000),
            vad_events=True,
            # Time in milliseconds of silence to wait for before finalizing speech
            endpointing=300,
        )
        self.audio_start = time.time()
        try:
            loop = asyncio.get_event_loop()
            self.setup_signals(loop)
        except AsyncDeepgramServiceException as e:
            print(f"Could not get event loop: {e}")
            return

    def setup_signals(self, loop: asyncio.AbstractEventLoop) -> None:
        self.connection.on(LiveTranscriptionEvents.Open, self.on_open)
        self.connection.on(LiveTranscriptionEvents.Metadata, self.on_metadata)
        self.connection.on(
            LiveTranscriptionEvents.SpeechStarted, self.on_speech_started
        )
        self.connection.on(LiveTranscriptionEvents.Close, self.on_close)
        self.connection.on(LiveTranscriptionEvents.Unhandled, self.on_unhandled)

        self.connection.on(LiveTranscriptionEvents.Transcript, self.signals_mapping["Transcript"])
        self.connection.on(LiveTranscriptionEvents.Error, self.signals_mapping["Error"])
        self.connection.on(LiveTranscriptionEvents.UtteranceEnd, self.signals_mapping["UtteranceEnd"])

    async def on_open(self, conn, open, **kwargs):
        pass

    async def on_message(self, conn, result, **kwargs):
        sentence = result.channel.alternatives[0].transcript
        if not sentence:
            return

    async def on_metadata(self, conn, metadata, **kwargs):
        pass

    async def on_speech_started(self, conn, speech_started, **kwargs):
        pass

    async def on_utterance_end(self, conn, utterance_end, **kwargs):
        pass

    async def on_close(self, conn, close, **kwargs):
        await self.track_tokens(TokenType.STT, ServiceName.DEEPGRAM)

    async def on_error(self, conn, error, **kwargs):
        logging.info("Error happened. Connection closed")

    def on_unhandled(self, conn, unhandled, **kwargs):
        print(f"\n\n{unhandled}\n\n")

    async def track_tokens(self, token_type, service_name):
        if self.audio_start is None:
            return
        tokens = time.time() - self.audio_start
        self.audio_start = None
        async with AsyncDBSession() as session:
            await session.execute(
                insert(TokensUsage).values(
                    tokens=tokens,
                    token_type=token_type,
                    service=service_name,
                    user_id=self.user.id,
                )
            )
            await session.commit()
            await session.close()
            logging.info(f"Token usage tracked for {service_name}: {tokens}")
        return

Here is ChatHandler.arun:

    async def arun(self) -> None:
        try:
            while True:
                try:
                    if (
                        self.websocket.client_state != WebSocketState.CONNECTED
                        or self.websocket.application_state != WebSocketState.CONNECTED
                    ):
                        return
                    data = await self.websocket.receive()

                    stt_connected = await self.stt_service.connection.is_connected()
                    if stt_connected:
                        keepalive_sent = await self.stt_service.connection.send(json.dumps({
                            'type': 'KeepAlive'
                        }))
                    else:
                        logging.error("Deepgram connection is not established while sending keep alive")
                        await self.deepgram_connect()
                    if data.get("bytes"):
                        data = data.get("bytes")
                        if stt_connected:
                            await self.stt_service.connection.send(data)
                        else:
                            logging.error("Deepgram connection is not established while sending data")
                            await self.deepgram_connect()
                    elif data.get("json") or data.get("text"):
                        command = data.get("text")
                        await self.process_command(command)
                except RuntimeError as e:
                    logging.error(f"RuntimeError: {e}", exc_info=True)
                    break
                except WebSocketDisconnect as e:
                    logging.error(
                        f"WebSocket disconnected: {e.code} - {e.reason}", exc_info=True
                    )
                    break
                except asyncio.CancelledError:
                    logging.error("Task was cancelled", exc_info=True)
                    break
                except Exception as e:
                    logging.error(f"Unexpected error: {e}", exc_info=True)
                    break
        finally:
            await self.stt_service.connection.finish()

Here is ChatHandler.deepgram_connect:

    async def deepgram_connect(self):
        logging.error("Start reconnecting to Deepgram")
        signals_mapping = {
            "Transcript": self.on_stt_response,
            "UtteranceEnd": self.on_utterance_end,
            "Error": self.on_deepgram_error
        }
        self.stt_service: AsyncDeepgramService = AsyncDeepgramService(
            language=self.active_language.langcode,
            grace_period=self.grace_period,
            user=self.user,
            signals_mapping=signals_mapping
        )

        if await self.stt_service.connection.start(self.stt_service.options) is False:
            print("Failed to connect to Deepgram")
            return
        logging.error("Reconnected to Deepgram successfully")

And error I'm receiving:

[2024-10-23 11:15:44,407.407] ERROR [281473047552032] - 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
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

Even though I have on_error and on_close they are not triggered before this error.
That's why I'm checking if connection is open with deepgram and trying to reconnect.
But even though after reconnection any events 'on_message' or 'on_utterance_end' never fired.

Please let me know if you need any extra information from me.

Check the Deepgram SDK version: You're using deepgram-sdk==3.7.4. You could try updating to the latest version to see if the issue persists.

Version 3.7.5 was released 3 days ago, but I updated and it didn't help.