首页 → 博客 → 如何为音频存档搭建批量转录流水线

如何为音频存档搭建批量转录流水线

发布于 2026-08-05 · 7 分钟阅读

一个文件是一行代码。一万个文件是另一个问题：中途一定会有东西失败，进程会被杀掉，一半文件已经处理完，而你需要知道是哪一半。下面是一条能扛住这些情况的批量流水线的形状。

从一个文件到一万个文件，会坏在哪里

遍历一个目录的循环，在存档不大时工作得很好。规模上来之后有四件事会出问题：

下面每一节都是为了修掉其中一条。

第一步：让进度可持久

最重要的一个决定：已完成的记录必须比进程活得久。一个结果目录就够了——输出文件的存在本身就是状态。

import pathlib

SRC = pathlib.Path("recordings")
OUT = pathlib.Path("transcripts")
OUT.mkdir(exist_ok=True)

def pending():
    for p in sorted(SRC.glob("*.mp3")):
        if not (OUT / (p.stem + ".txt")).exists():
            yield p

这样任务就是幂等的：跑起来、杀掉、再跑一次——它会精确地从停下的地方继续，绝不会为同一个文件付两次钱。存档更大时，用一张带状态列的 SQLite 表同样有效，还能做查询；原理不变。

输出要原子地写，否则进程在写入中途被杀，会留下一个看起来已完成的截断文件：

tmp = out.with_suffix(".part")
tmp.write_text(text, encoding="utf-8")
tmp.rename(out)          # 原子操作：要么没有新文件，要么是完整的

第二步：同时处理多个文件

转录是等待网络，不是本地计算，所以线程是对的工具，一个线程池就够：

import os
from concurrent.futures import ThreadPoolExecutor
from openai import OpenAI

client = OpenAI(base_url="https://voicesscribe.com/v1", api_key=os.environ["VS_KEY"], max_retries=0)

def transcribe(path):
    with path.open("rb") as f:
        return client.audio.transcriptions.create(model="whisper-1", file=f).text

with ThreadPoolExecutor(max_workers=4) as pool:
    for path, text in zip(pending(), pool.map(transcribe, pending())):
        (OUT / (path.stem + ".txt")).write_text(text, encoding="utf-8")

max_workers 要按套餐的每秒请求数限制来定，而不是按 CPU 核数。开得比限制更宽不会更快——只会把成功的请求变成 429。

第三步：只重试失败的，不重试成功的

两类错误需要完全相反的处理。5xx 或超时值得重试；损坏文件上的 400 永远会以同样的方式失败。

import time
from openai import APIStatusError, APIConnectionError

def with_retry(path, attempts=3):
    for n in range(attempts):
        try:
            return transcribe(path)
        except APIConnectionError:
            pass                                  # 网络抖动，重试
        except APIStatusError as e:
            if e.status_code == 429 or e.status_code >= 500:
                pass                              # 限流或服务端故障，重试
            else:
                raise                             # 400/401，重试没有意义
        time.sleep(2 ** n)                        # 1 秒、2 秒、4 秒
    raise RuntimeError(f"failed after {attempts}: {path}")

指数退避对 429 尤其重要：立刻重试只会让你继续超过限制。注意上面客户端上的 max_retries=0——否则 SDK 也在重试，两套策略互相打架。

不要让一个坏文件停掉整轮任务。记下来，继续跑，最后统一处理失败清单——通常是几千个里的两三个。

第四步：发送前先处理音频

账单实际上在这一步就决定了。两条规则解决大部分问题：

# 一条命令转换整个存档
find recordings -name '*.wav' -print0 |
  xargs -0 -P4 -I{} ffmpeg -loglevel error -i {} -ac 1 -ar 16000 -b:a 48k {}.mp3

超过 25 MB 的文件无论如何都要切分；而按单声道 48 kbit/s 计算，这个门槛大约是一小时音频，所以多数存档转换之后根本不再需要切分。

第五步：开跑之前先知道成本

按音频分钟计费，因此总额是可以预先算出来的——去测量存档，而不是猜：

ffprobe -v error -show_entries format=duration -of csv=p=0 file.mp3
import subprocess

def minutes(path):
    out = subprocess.run(["ffprobe", "-v", "error", "-show_entries",
                          "format=duration", "-of", "csv=p=0", str(path)],
                         capture_output=True, text=True).stdout
    return float(out) / 60

total = sum(minutes(p) for p in SRC.glob("*.mp3"))
print(f"{total:.0f} min, {total * RATE:.2f} at the current rate")

在流水线之前跑，不是之后。它只要一分钟，不花一分钱，却是「计划内支出」和「意外账单」之间的差别。

第六步：盯住这一轮

长时间批处理需要三个可见的数字：完成数、失败数、速度。每一百个文件打印一次计数就够：

done = failed = 0
for path in pending():
    try:
        save(path, with_retry(path))
        done += 1
    except Exception as e:
        failed += 1
        print("FAIL", path.name, e)
    if (done + failed) % 100 == 0:
        print(f"{done} done, {failed} failed")

客户后台从另一侧展示同一轮任务——消耗的分钟数、请求数和逐条历史——这是确认流水线确实在做你以为的事情的最快方式。

不需要造的东西

用你自己的录音试一试。 注册只需一分钟，免费额度足够判断识别质量。

免费获取 API 密钥

常见问题

一次可以发多少个文件？

并发受套餐的每秒请求数限制约束，而不是受服务限制。把线程池设得略低于该限制；开得更宽只会产生 429 响应。

进程中途挂掉会怎样？

只要进度是从输出文件推导出来的，就什么也不会丢：下一轮会跳过所有已有转录的文件，从那里继续。

重试的文件要重复付费吗？

返回错误的失败请求不计音频分钟；成功的转录才计费。所以跳过已完成的文件很重要——重复做成功过的工作才是花钱的地方。

发送前需要转换音频吗？

对存档来说需要，而且好处是双份的：16 kHz 单声道上传快得多，文件也不再越过 25 MB 限制，省掉了切分。

怎样估算大型存档的成本？

用 ffprobe 把时长加总，再乘以你的每分钟单价。按音频分钟计费，所以开跑前算出来的估值就是你实际要付的数字。

相关阅读