使用Google Speech-to-Text流式识别时遇音频超时错误求助

核心问题排查方向 #

1. 音频编码格式不匹配 #

Google Speech-to-Text流式API仅支持特定音频编码（如LINEAR16、FLAC等），前端浏览器默认录制的音频通常是WebM/opus格式，直接转发会导致API无法解析，触发超时。

解决：前端录制时指定兼容的编码，或后端实时转码后再转发给Google API。

2. 音频分块转发延迟 #

Google流式API要求音频分块间隔不超过10秒（默认超时阈值），如果后端先缓存多个分块再批量转发，会导致间隔超时。

解决：后端接收WebSocket分块后立即转发，不要缓存。

3. WebSocket字节处理错误 #

前端发送Blob时，后端若未正确解析为原始音频字节（比如误处理为字符串而非二进制），会导致发送给Google的音频数据损坏。

解决：后端确保接收并转发二进制字节流。

4. Google API配置错误 #

未正确设置流式请求的encoding、sample_rate_hertz等参数，导致API无法识别音频格式，触发超时。

解决：严格匹配实际音频的编码和采样率配置。

代码修改示例 #

前端JS修改（指定LINEAR16编码） #

async function startRecording() {
  const stream = await navigator.mediaDevices.getUserMedia({ audio: true });
  const mediaRecorder = new MediaRecorder(stream, {
    mimeType: 'audio/wav; codecs=1' // 指定LINEAR16编码的WAV格式
  });
  const socket = new WebSocket('ws://localhost:8000/ws');

  mediaRecorder.ondataavailable = (event) => {
    if (event.data.size > 0) {
      socket.send(event.data); // 直接发送二进制Blob
    }
  };

  mediaRecorder.start(100); // 每100ms发送一个分块，避免间隔过长
}

后端FastAPI修改（实时转发分块） #

from fastapi import FastAPI, WebSocket
from google.cloud import speech_v1p1beta1 as speech
import asyncio
from asyncio import Queue

app = FastAPI()

async def stream_to_google(audio_queue: Queue):
    client = speech.SpeechClient()
    config = speech.RecognitionConfig(
        encoding=speech.RecognitionConfig.AudioEncoding.LINEAR16,
        sample_rate_hertz=16000, # 匹配前端录制的采样率
        language_code="zh-CN",
    )
    streaming_config = speech.StreamingRecognitionConfig(
        config=config,
        interim_results=True
    )

    # 生成流式请求
    async def request_generator():
        while True:
            chunk = await audio_queue.get()
            if chunk is None:
                break
            yield speech.StreamingRecognizeRequest(audio_content=chunk)
            audio_queue.task_done()

    responses = await client.streaming_recognize(
        requests=request_generator(),
        config=streaming_config
    )

    # 处理识别结果
    async for response in responses:
        for result in response.results:
            print(f"识别结果: {result.alternatives[0].transcript}")

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    audio_queue = Queue(maxsize=10)
    # 启动Google流式识别任务
    recognize_task = asyncio.create_task(stream_to_google(audio_queue))
    try:
        while True:
            data = await websocket.receive_bytes() # 接收二进制字节
            await audio_queue.put(data)
    except Exception as e:
        print(f"错误: {e}")
    finally:
        await audio_queue.put(None) # 通知识别任务结束
        await recognize_task
        await websocket.close()

关键验证步骤 #

1. 确认前端录制的音频编码和采样率与Google API配置完全一致。
2. 检查后端WebSocket接收的字节是否与前端录制的原始音频字节一致（可保存为文件验证）。
3. 确保后端分块转发的间隔不超过10秒，建议每100-500ms发送一个分块。
4. 查看Google Cloud控制台的API日志，确认超时错误的具体触发原因（是格式错误还是无数据输入）。

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