You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Flask+speech_recognition处理2分钟以上音频报Broken pipe错误求解

问题根因

你调用的recognize_google()是谷歌公开的免费无授权语音识别接口,仅支持最长约90秒的音频转写,超出长度的请求会被服务端主动切断,因此抛出连接中断的Broken pipe错误。

可行解决方案

  • 方案1:长音频分片转写(成本最低,无需更换接口)
    先将上传的长音频切分为30~60秒的短片段,分别调用接口转写后拼接结果。需要先安装依赖:
pip install pydub

同时本地需要安装ffmpeg组件(用于音频格式处理)。
修改后的/root/app.py核心逻辑示例:

from flask import Flask, render_template, request, redirect
import speech_recognition as sr
from pydub import AudioSegment
import io
import math

app = Flask(__name__)
# 分片长度,单位毫秒,这里设为45秒
CHUNK_LENGTH = 45 * 1000

@app.route("/", methods=["GET", "POST"])
def index():
    transcript = ""
    if request.method == "POST":
        print("FORM DATA RECEIVED")

        if "file" not in request.files:
            return redirect(request.url)

        file = request.files["file"]
        if file.filename == "":
            return redirect(request.url)

        if file:
            recognizer = sr.Recognizer()
            # 读取音频文件
            audio = AudioSegment.from_file(file)
            # 计算分片数量
            chunks_count = math.ceil(len(audio) / CHUNK_LENGTH)
            # 逐分片转写
            for i in range(chunks_count):
                # 切取当前分片
                chunk = audio[i*CHUNK_LENGTH : (i+1)*CHUNK_LENGTH]
                # 导出为wav格式的字节流
                chunk_bytes = io.BytesIO()
                chunk.export(chunk_bytes, format="wav")
                chunk_bytes.seek(0)
                # 转写当前分片
                with sr.AudioFile(chunk_bytes) as source:
                    data = recognizer.record(source)
                try:
                    chunk_transcript = recognizer.recognize_google(data, key=None)
                    transcript += chunk_transcript + " "
                except sr.RequestError as e:
                    transcript += f"[第{i+1}段转写失败: 网络错误] "
                except sr.UnknownValueError:
                    transcript += f"[第{i+1}段转写失败: 无法识别音频内容] "

    return render_template('index.html', transcript=transcript)


if __name__ == "__main__":
    app.run(debug=True, threaded=True)
  • 方案2:改用离线转写引擎(无网络、长度限制)
    可以替换为Vosk等开源离线语音识别库,无需调用在线接口,支持任意长度的本地音频转写,适合对隐私要求高、网络环境受限的场景。
  • 方案3:替换为商用语音转写服务(稳定性最高)
    如果需要更高的转写准确率、长音频稳定性,可选用百度智能云、阿里云、腾讯云等厂商的商用语音转写接口,官方原生支持数小时级别的长音频转写,仅需按照调用量付费。

额外优化建议

原有代码未做异常捕获,除了长音频问题外,网络波动、音频无有效语音内容都会导致服务报错,建议在调用转写接口的位置统一加异常捕获逻辑,返回前端友好的错误提示。

内容的提问来源于stack exchange,提问作者Dejo

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.10.11 05:18:03