FastAPI WebSocket+Whisper实时音频转写异常:固定错误结果与白噪声问题排查
FastAPI WebSocket+Whisper实时音频转写异常:固定错误结果与白噪声问题排查
看起来你遇到了音频转写里的典型格式不匹配问题,我帮你梳理下核心问题点和解决方案:
核心问题诊断
你遇到的两个现象其实是同一个根源:后端接收的音频数据格式完全不符合Whisper模型的要求:
- 固定输出"QUIET MUSIC...":这是Whisper模型收到无效/无法识别的音频数据时,输出的默认占位文本(类似模型对“空音频/噪声”的默认响应)
- 保存成WAV是白噪声:说明你对音频字节的解析、格式转换逻辑错误,导致原始音频数据被乱码处理
具体代码问题分析
采样率与声道不匹配
Whisper模型默认要求输入是16kHz采样率、单声道的音频,但你的代码直接假设输入是16kHz,没有处理前端可能传来的其他采样率(比如常见的44.1kHz、48kHz)或立体声数据。自定义字节转换逻辑的隐患
你的bytes_to_float32_array函数硬编码了sample_width=2,但前端传来的音频位深可能和这个值不匹配;同时完全没处理声道数,立体声数据直接转成单声道数组会导致音频错乱。WebSocket接收逻辑的小错误
在WebSocketDisconnect异常块里,你尝试发送Final transcription error: {e},但这个块里并没有定义变量e,会导致额外报错。
分步修复方案
第一步:统一前后端音频格式要求
先让前端输出符合Whisper要求的音频格式,以React为例,录制时指定参数:
// 前端录制配置示例 const stream = await navigator.mediaDevices.getUserMedia({ audio: true }); // 强制设置为16kHz采样率、单声道、16位PCM const mediaRecorder = new MediaRecorder(stream, { mimeType: 'audio/wav', audioBitsPerSecond: 16000 * 16, }); // 额外确保单声道(部分浏览器需要手动设置) const audioTrack = stream.getAudioTracks()[0]; const monoTrack = audioTrack.clone(); monoTrack.applyConstraints({ channelCount: 1 });
第二步:替换自定义转换逻辑,用成熟库处理音频格式
放弃手动写字节转换,用pydub或Whisper原生工具来处理格式转换,避免出错:
from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.middleware.cors import CORSMiddleware import numpy as np import whisper import io from pydub import AudioSegment app = FastAPI() # CORS配置保留 app.add_middleware( CORSMiddleware, allow_origins=["http://localhost:3000"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) model = whisper.load_model("base") def transcribe_from_bytes(audio_bytes): try: # 用pydub解析音频,统一转换为Whisper要求的格式 audio = AudioSegment.from_file(io.BytesIO(audio_bytes)) # 转换为16kHz、单声道、16位深度 audio = audio.set_frame_rate(16000).set_channels(1).set_sample_width(2) # 转成Whisper需要的float32数组(归一化到[-1.0, 1.0]) audio_array = np.array(audio.get_array_of_samples()).astype(np.float32) / 32768.0 result = model.transcribe( audio_array, language="ru", fp16=False ) return result["text"] except Exception as e: print(f"Ошибка при транскрипции из байтов: {e}") return {"text": "", "error": str(e)} @app.websocket("/voice") async def websocket_endpoint(websocket: WebSocket): await websocket.accept() print("WebSocket connection established") audio_buffer = bytearray() try: while True: data = await websocket.receive_bytes() audio_buffer.extend(data) # 可以考虑累积一定长度再转写,避免实时转写的性能问题 transcription = transcribe_from_bytes(audio_buffer) print(transcription) await websocket.send_text(f"Transcription: {transcription}") except WebSocketDisconnect: print("WebSocket disconnected") except Exception as e: print(f"Error: {e}") await websocket.send_text(f"Final transcription error: {str(e)}") await websocket.close()
第三步:验证音频数据正确性
在转写前,可以先把接收的音频字节保存成WAV文件,确认是否能正常播放:
# 在transcribe_from_bytes函数里添加临时保存代码 with open("test_audio.wav", "wb") as f: f.write(audio_bytes)
如果保存的文件能正常播放,说明格式没问题,再排查模型转写的问题。
关键注意点
- Whisper对输入音频的格式要求非常严格,必须是16kHz单声道的浮点数组或归一化后的16位PCM数据
- 尽量避免手动处理音频字节转换,用成熟的音频处理库(pydub、librosa)来做格式转换,减少出错概率
- 实时转写时可以考虑设置音频片段的最小长度,避免频繁调用模型导致性能问题
内容来源于stack exchange




