使用Deepgram API语音转文本遇1011保活超时错误的排查求助

问题描述 #

我正在尝试使用Deepgram API实现语音转文本功能，参考官方FastAPI实时转写教程编写了后端代码，但服务运行5秒后出现错误：ConnectionClosedError(None, Close(code=1011, reason='keepalive ping timeout'))，需要排查问题根源并找到解决方法。

后端代码如下：

from fastapi import WebSocket, APIRouter
from deepgram import Deepgram
import os
from typing import Dict, Callable

router = APIRouter()

@router.websocket('/socket')
async def speech_to_text_websocket_endpoint(websocket: WebSocket):
    await websocket.accept()

    try:
        deepgram_socket = await process_audio(websocket) 
        while True:
            data = await websocket.receive_bytes()
            deepgram_socket.send(data)
    except Exception as e:
        raise Exception(f'Could not process audio: {e}')
    finally:
        await websocket.close()

async def process_audio(fast_socket: WebSocket):
    async def get_transcript(data: Dict) -> None:
        if 'channel' in data:
            transcript = data['channel']['alternatives'][0]['transcript']
    
            if transcript:
                await fast_socket.send_text(transcript)

    deepgram_socket = await connect_to_deepgram(get_transcript)

    return deepgram_socket

async def connect_to_deepgram(transcript_received_handler: Callable[[Dict], None]):
    dg_client = Deepgram(os.getenv('DEEPGRAM_API_KEY'))
    options = {'punctuate': True, 'interim_results': False}

    try:
        socket = await dg_client.transcription.live(options)
        socket.registerHandler(socket.event.CLOSE, lambda c: print(f'Connection closed with code {c}.'))
        socket.registerHandler(socket.event.TRANSCRIPT_RECEIVED, transcript_received_handler)
    
        return socket
    except Exception as e:
        raise Exception(f'Could not open socket: {e}')

问题根源分析 #

这个错误大概率是后端代码未处理WebSocket心跳机制导致的，而非Deepgram服务本身的问题。Deepgram的实时转写WebSocket连接会定期发送ping帧检测连接活性，如果你的服务在规定时间内没有返回pong响应，Deepgram就会触发超时并关闭连接（1011属于服务器超时类关闭码）。

解决方法 #

1. 自动回复Deepgram的Ping帧 #

修改connect_to_deepgram函数，添加Ping事件处理逻辑，自动返回Pong响应维持连接：

async def connect_to_deepgram(transcript_received_handler: Callable[[Dict], None]):
    dg_client = Deepgram(os.getenv('DEEPGRAM_API_KEY'))
    options = {'punctuate': True, 'interim_results': False}

    try:
        socket = await dg_client.transcription.live(options)
        socket.registerHandler(socket.event.CLOSE, lambda c: print(f'Connection closed with code {c}.'))
        socket.registerHandler(socket.event.TRANSCRIPT_RECEIVED, transcript_received_handler)
        # 添加Ping事件处理，自动回复Pong
        socket.registerHandler(socket.event.PING, lambda _: socket.pong())
    
        return socket
    except Exception as e:
        raise Exception(f'Could not open socket: {e}')

2. 避免主循环阻塞 #

当前代码的主循环会一直等待客户端发送音频数据，若客户端暂停发送，循环会阻塞，无法处理Deepgram的Ping帧。使用asyncio.gather并发处理客户端消息接收和Deepgram连接的活性检测：

import asyncio

@router.websocket('/socket')
async def speech_to_text_websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    deepgram_socket = None
    try:
        deepgram_socket = await process_audio(websocket) 
        
        async def listen_client():
            while True:
                data = await websocket.receive_bytes()
                deepgram_socket.send(data)
        
        async def keep_deepgram_alive():
            while deepgram_socket.is_active():
                await asyncio.sleep(0.1)
        
        await asyncio.gather(listen_client(), keep_deepgram_alive())
    except Exception as e:
        raise Exception(f'Could not process audio: {e}')
    finally:
        if deepgram_socket and deepgram_socket.is_active():
            await deepgram_socket.finish()
        await websocket.close()

3. 处理音频中断场景 #

如果客户端会出现长时间无音频发送的情况，有两种处理方式：

• 客户端暂停时发送静音音频数据，维持连接活性；
• 暂时停止转写时，主动调用deepgram_socket.finish()关闭连接，后续需要时重新建立。

内容的提问来源于stack exchange，提问作者Hanie Asemi