如何将FastAPI WebSocket接收的字节数据转为临时音频文件用于语音识别

要处理WebSocket接收的字节音频数据并传给speech_recognition，可以用Python内置的tempfile模块创建临时文件，同时注意异步环境下调用同步语音识别函数的问题，以下是完整修改代码：

import tempfile
import asyncio
import os
import speech_recognition as sr
from fastapi import WebSocket, Cookie, Annotated

@router.websocket('/change-voice/{speakerId}/{bahasaYangDigunakan}')
async def change_voice(
    websocket: WebSocket,
    speakerId: int,
    BahasaYangDigunakan: int,
    access_token: Annotated[str| None, Cookie()] = None,
):
    await websocket.accept()
    r = sr.Recognizer()  # 初始化语音识别器
    while True:
        data = await websocket.receive_bytes()
        
        # 创建带.wav后缀的临时文件，写入音频字节数据
        with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as temp_audio:
            temp_audio.write(data)
            temp_audio_path = temp_audio.name
        
        try:
            # 用线程执行同步识别逻辑，避免阻塞FastAPI事件循环
            transcript = await asyncio.to_thread(
                _run_recognition, temp_audio_path, r, str(BahasaYangDigunakan)
            )
            # 可选：将识别结果回传给WebSocket客户端
            await websocket.send_text(transcript)
        finally:
            # 清理临时文件
            os.unlink(temp_audio_path)

def _run_recognition(audio_path, recognizer, language):
    """封装同步语音识别逻辑"""
    with sr.AudioFile(audio_path) as source:
        audio_data = recognizer.record(source)
        return recognizer.recognize_google(audio_data, language=language)

关键说明 #

• 临时文件处理：通过tempfile.NamedTemporaryFile创建指定后缀的临时文件，写入字节数据后记录路径，使用完成后手动删除文件避免占用磁盘空间。
• 异步兼容：speech_recognition的核心API是同步实现，直接在异步函数中调用会阻塞FastAPI的事件循环，因此用asyncio.to_thread将识别逻辑放到单独线程执行。
• 音频格式要求：确保前端传入的是WAV格式音频，如果是MP3等其他格式，需要先转码（可使用pydub库：from pydub import AudioSegment; AudioSegment.from_file(temp_audio_path).export(temp_audio_path, format="wav")）。

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