开篇先聊一个现实问题当你的任务是把 800 本有声书、累计约 8000 小时的音频切成一句一句并对应到官方电子书文本时第一反应是什么很多人首先想到的是“让大模型来搞定”。但真实算一笔账就会发现纯用 LLM 做对齐成本、延迟和结果稳定性都很难接受。本文记录了一套完全不依赖 LLM 参与循环的对齐流水线用传统 ASR 转写 文本匹配算法在 6 天内完成了这个规模的任务。这篇文章适合正在做音频语料治理、字幕对齐、TTS 数据集构建的开发者读完你可以得到一套可直接落地的工程方案。1. 项目背景800 本有声书对齐为什么不用 LLM1.1 什么是有声书与文本对齐有声书与文本对齐指的是把音频内容中的每一个词、每一句话映射到电子书文本中的对应位置并给出准确的时间戳。它的典型输出是一个带时间范围的结构化文本比如下面这样第 1 句: It was the best of times start12.30s end15.80s 第 2 句: it was the worst of times start16.10s end19.40s这种对齐结果的应用场景很广泛给有声书制作同步字幕、构建语音合成训练集、做音频检索、生成章节级导航、给盲人阅读器提供逐句跟读能力等等。无论哪一种场景前提都是音频和文本能精确落到同一个时间轴上。1.2 为什么选择 no LLM in the loop看到这个任务很多人的第一方案是把音频转成文本再丢给 LLM 去“理解”并匹配。这在几十个小时的数据上是可行的但扩大到 8000 小时就有几个绕不开的问题Token 成本极高。8000 小时音频如果按每分钟约 150 个英文单词估算大约是 7200 万词转换成 token 后接近 1 亿量级。逐段调用 LLM 的成本非常夸张。延迟不可控。即便并行调用让 LLM 去逐段匹配并纠错整体耗时很容易超过音频本身的长度。生成式模型不稳定。LLM 在“补全”或“润色”过程中可能修改原词导致对齐结果变成“再创作”而不是原始文本。可复现性差。同样输入在不同调用下可能得到不同结果这在生产级数据治理中是不可接受的。所以更合理的路线是把语音转写和文本匹配拆成两个确定性较强的步骤用 ASR 负责“听到什么”用动态规划算法负责“文本怎么对应到时间戳”全程不需要 LLM 参与循环。1.3 处理规模与目标这次任务的基本盘是800 本有声书。总时长约 8000 小时平均每本 10 小时。目标是在 6 天144 小时内完成全量处理。对齐粒度至少到句子级最好能到词级。最终输出 JSONL 格式的结构化标注方便下游任务使用。这是一次典型的“大规模离线批处理”任务。它考验的不是单个算法的精度而是整条流水线的吞吐能力、健壮性和可恢复性。下面我们逐步拆解。2. 整体架构与流程设计在动手写代码之前先规划整条流水线。我把它分成六个阶段音频预处理统一格式、采样率、声道必要时做切分。ASR 转录用 Whisper 类模型生成带词级时间戳的转写结果。电子书文本清洗去除目录、页码、脚注等噪音。文本规范化统一大小写、标点、数字和缩写。时间戳对齐把清洗后的参考文本与 ASR token 序列做匹配。结果校验与导出计算覆盖率、检查时间单调性输出 JSONL。整体数据流可以用下面这个简单的图来表示原始音频 ──► 音频预处理 ──► ASR 转录 ──┐ ├──► 文本对齐 ──► 结果校验 ──► JSONL 电子书文本 ─► 文本清洗 ──► 文本规范化 ─┘从这个流程可以看到ASR 转录和文本清洗互不依赖可以并行执行。只有到最后对齐阶段两条流才汇合。这为并行调度提供了很大便利。在选型上ASR 使用faster-whisper它基于 CTranslate2 推理相比原始 Open AI Whisper Python 包在 GPU 上的吞吐更高且原生支持词级时间戳。对齐算法使用difflib.SequenceMatcher做动态规划匹配虽然看起来简单但在英文有声书这类独白为主的场景下效果已经足够稳定。3. 环境准备与项目结构3.1 运行环境本文示例以 Linux 环境为主你可以在 Ubuntu 20.04/22.04、Debian 等系统上直接运行。主要依赖如下Python 3.10 或更高版本。CUDA 11.8 或 12.xNVIDIA 显卡驱动已正确安装。FFmpeg用于音频格式转换和切分。PyTorchfaster-whisper 会依赖 CTranslate2不一定需要完整 PyTorch但安装 torch 可避免部分环境冲突。faster-whisper、numpy、pydub或直接命令行调用 FFmpeg。这里有两点需要说明。第一版本号建议根据你实际环境确认特别是 CUDA 和显卡驱动版本不同版本组合可能会有兼容问题。第二本文所有命令和代码都为核心逻辑演示实际生产环境请根据你的数据规模和资源情况调整。3.2 项目目录设计我建议把任务拆分成模块每个模块只负责一件事。下面是一个可用的目录结构audiobook-align/ ├── config.py # 全局配置 ├── preprocess.py # 音频预处理 ├── transcribe.py # ASR 转录 ├── normalize.py # 文本规范化 ├── align.py # 时间戳对齐 ├── pipeline.py # 主流程调度 ├── requirements.txt └── run.shconfig.py保存路径、模型名、切分长度等参数其他地方统一从这里读取方便多本书批量切换。# config.py import os BASE_DIR os.path.dirname(os.path.abspath(__file__)) RAW_AUDIO_DIR os.path.join(BASE_DIR, data/raw_audio) RAW_TEXT_DIR os.path.join(BASE_DIR, data/raw_text) OUTPUT_DIR os.path.join(BASE_DIR, output) AUDIO_SAMPLE_RATE 16000 AUDIO_CHANNELS 1 SEGMENT_SECONDS 1800 # 30 分钟一段 ASR_MODEL_SIZE large-v3 ASR_BEAM_SIZE 5 ASR_LANGUAGE en ASR_DEVICE cuda ASR_COMPUTE_TYPE float16 # 并行进程数根据机器 CPU 核数调整 MAX_WORKERS 43.3 安装依赖requirements.txt可以这样写faster-whisper1.0.0 numpy1.24.0安装命令pip install -r requirements.txt另外确保系统安装了 FFmpegsudo apt update sudo apt install -y ffmpeg4. 音频预处理与文本清洗4.1 音频切分与格式统一8000 小时音频直接喂给 ASR 并不现实主要有两个原因。一是 Whisper 虽然有长音频处理能力但实际显存占用和推理稳定性会随着片段变长而下降。二是如果某个片段在中间位置出错切得越细重试成本越低。我采用固定时长切分每段 30 分钟。同时把所有音频统一转成 16kHz 单声道 WAV这个采样率对语音识别足够而且可以显著减少 I/O 和显存压力。ffmpeg -i input.m4a -f segment -segment_time 1800 \ -ar 16000 -ac 1 -c:a pcm_s16le book_part_%04d.wav如果原音频是 MP3也可以直接复制音频流切分速度更快ffmpeg -i input.mp3 -f segment -segment_time 1800 -c copy book_part_%04d.mp3但要注意-c copy对 MP3 通常没问题对 m4a/AAC 可能存在采样率或格式兼容问题。最稳妥的方式还是转成 WAV。除了切分音频预处理阶段还可以做响度归一化。部分有声书在章节切换时响度差异很大可能影响 ASR 的稳定性。使用 FFmpeg 的loudnorm滤镜可以统一响度ffmpeg -i input.wav -af loudnormI-16:TP-1.5:LRA11 output_norm.wav这里I-16表示目标整体响度约为 -16 LUFSTP-1.5是真实峰值上限LRA11是响度范围。具体参数可以根据实际音频情况调整不要照搬。4.2 电子书文本清洗图书文本不像音频那样“连贯”。它包含封面、版权页、目录、章节标题、页码、脚注、插图说明等。这些内容在朗读时不会出现如果不清理会让后续对齐产生大量未匹配文本。清洗规则通常包括去掉目录部分尤其是带页码的目录。去掉页眉页脚、页码行。去掉脚注标记和脚注内容。保留章节标题但单独作为标题元素处理。去掉连续空行把文本整理成段落序列。下面是一个简单的清洗示意# normalize.py import re def clean_book_text(raw_text: str) - list[str]: # 去掉常见的页码行例如 Page 12 text re.sub(r(?m)^\s*Page\s\d\s*$, , raw_text) # 去掉连续空白 text re.sub(r\n{3,}, \n\n, text) paragraphs [p.strip() for p in text.split(\n\n) if p.strip()] return paragraphs注意这不是一个万能规则。不同的电子书来源格式差异很大。更推荐的做法是先对一批样本做统计看哪些行符合“页码/目录/版权信息”的特征再写针对性规则。规则宁可保守也不要过度清理因为漏掉一些噪音最多让对齐覆盖率下降但清理过度会直接把正文段落删掉那是更严重的错误。4.3 文本规范化清洗后的文本通常包含大小写、标点、数字、缩写等。而 ASR 的输出则是“自然语音”的文本。举个例子书中写的是“Dr. Smith”ASR 可能识别为“Doctor Smith”书中写“102 years”ASR 可能是“one hundred two years”。这两者不做规范化分词后直接匹配会出现大量 false negative。文本规范化的目标是让参考文本和 ASR 文本在 token 层面尽量接近。常用的策略包括def normalize_for_align(text: str) - str: text text.lower() # 数字转英文 text number_to_words(text) # 常见缩写展开 text text.replace(dr., doctor) text text.replace(mr., mister) text text.replace(mrs., missus) text text.replace(st., saint) # 去掉对匹配无意义的标点 text re.sub(r[^a-z0-9\s], , text) text re.sub(r\s, , text) return text.strip()数字转英文可以借助现成库比如num2words也可以写一个覆盖常见数字规律的模块。在项目里我建议只处理 0 到 9999 的数字以及常用的日期、年份其他复杂表达式交给规则兜底即可。4.4 命名规范与中间产物大规模任务中中间产物命名是否规范直接决定后续能不能断点续跑。我使用的目录结构如下output/ ├── book_001/ │ ├── segments/ │ │ ├── part_0000.json │ │ ├── part_0001.json │ │ └── ... │ ├── aligned.jsonl │ └── report.json每个音频分段对应一个part_xxxx.json里面保存该分段的 ASR token 列表。如果某一段处理失败只需要重新生成这一段的 JSON不需要整个书重跑。5. 核心实现ASR 转录5.1 使用 faster-whisper 获取词级时间戳faster-whisper 的 API 与 openai-whisper 类似但对大模型推理做了优化。核心代码如下# transcribe.py import json from faster_whisper import WhisperModel from config import ( ASR_MODEL_SIZE, ASR_BEAM_SIZE, ASR_LANGUAGE, ASR_DEVICE, ASR_COMPUTE_TYPE, ) def load_model(): return WhisperModel( ASR_MODEL_SIZE, deviceASR_DEVICE, compute_typeASR_COMPUTE_TYPE, ) def transcribe_audio(model, audio_path: str) - list[dict]: segments, info model.transcribe( audio_path, languageASR_LANGUAGE, beam_sizeASR_BEAM_SIZE, word_timestampsTrue, vad_filterTrue, ) words [] for segment in segments: if not segment.words: continue for word in segment.words: words.append({ token: word.word.strip(), start: round(word.start, 3), end: round(word.end, 3), }) return words几个关键点word_timestampsTrue是获得词级时间戳的关键参数。vad_filterTrue可以过滤掉静音片段减少无效计算。beam_size5是精度和速度之间的折中如果追求更高速度可以降到 1但准确率可能下降。info里包含检测到的语言、音频时长等信息可以写到日志里用于审计。5.2 分批转录与容错由于整个任务有上万个音频分段转录模块必须有“失败重试”和“跳过已完成”的能力。下面是一个单本书的转录逻辑import os import json def transcribe_book(model, book_id: str, segment_dir: str, out_dir: str) - int: os.makedirs(out_dir, exist_okTrue) wav_files sorted([ f for f in os.listdir(segment_dir) if f.endswith(.wav) or f.endswith(.mp3) ]) done_count 0 for wav_name in wav_files: json_path os.path.join(out_dir, wav_name.replace(.wav, .json)) if os.path.exists(json_path): done_count 1 continue audio_path os.path.join(segment_dir, wav_name) try: words transcribe_audio(model, audio_path) with open(json_path, w, encodingutf-8) as f: json.dump({words: words}, f, ensure_asciiFalse, indent2) done_count 1 except Exception as exc: print(f[error] {wav_name}: {exc}) return done_count这样即使某个分段因为音频损坏或显存波动导致失败也不会影响已经完成的部分。等任务重新执行时会直接跳过已生成的 JSON 文件这在大规模批处理里非常重要。5.3 ASR 时间开销估算在规划 6 天完成 8000 小时任务时需要先估算单卡吞吐。通常用 RTFReal-Time Factor来衡量RTF 处理音频所需时间 / 音频时长如果一张 A100 上运行large-v3、beam_size5实测 RTF 大致在 0.02 到 0.05 之间。为了计算方便取一个中间值 0.03单卡处理 1 小时音频需要 0.03 小时即 108 秒。单卡 6 天144 小时可处理约 144 / 0.03 4800 小时音频。8000 小时音频至少需要 8000 / 4800 ≈ 1.7 张卡。考虑到切分、对齐、异常重试等开销准备 2 到 4 张卡是比较充裕的选择。当然这里的 RTF 只是估算实际数值取决于 GPU 型号、显存、beam size、batch size、音频长度分布等建议先用小样本测试后再确认设备数量。如果只有一张消费级显卡比如 RTX 3090/4090也能跑但 6 天可能比较紧张。此时可以考虑把模型换成medium或small或者降低 beam size。6. 核心实现文本对齐算法6.1 对齐问题建模音频和文本对齐本质上是一个序列匹配问题。我们有两组序列ASR token 序列每个 token 带有时间戳。参考文本 token 序列来自电子书经过规范化。对齐的目标就是找到这两个序列之间的映射关系。理想情况下ASR 的每个 token 都能在参考文本里找到完全一致的 token但实际场景会有三种偏差ASR 识别错误导致 token 不一致。ASR 多识别了原文没有的词比如口头语、重复、静音幻觉。ASR 漏识别了原文有的词比如快读、生僻词。因此不能追求“完全一致”而要允许插入、删除、替换。这正是动态规划算法擅长的问题。Python 标准库中的difflib.SequenceMatcher正好提供了最长匹配子序列的能力可以帮我们完成这一步。6.2 词级时间戳映射先写一个核心函数给定 ASR 词列表和参考文本 token 列表返回参考文本中每个 token 是否匹配到 ASR 词以及对应的起止时间。# align.py from difflib import SequenceMatcher from typing import Optional def align_tokens( asr_tokens: list[dict], ref_tokens: list[str], ) - list[dict]: asr_words [t[token].lower() for t in asr_tokens] ref_words [t.lower() for t in ref_tokens] matcher SequenceMatcher(None, asr_words, ref_words, autojunkFalse) result [] asr_pos 0 for tag, i1, i2, j1, j2 in matcher.get_opcodes(): if tag equal: # ASR 和参考文本完全匹配的片段 for offset in range(j2 - j1): src asr_tokens[asr_pos offset] result.append({ token: ref_tokens[j1 offset], start: src[start], end: src[end], matched: True, }) asr_pos (i2 - i1) elif tag delete: # ASR 中有但参考文本中没有的词 asr_pos (i2 - i1) elif tag insert: # 参考文本中有但 ASR 没有识别出来的词 for offset in range(j2 - j1): result.append({ token: ref_tokens[j1 offset], start: None, end: None, matched: False, }) elif tag replace: # ASR 和参考文本在该区间存在差异 asr_pos (i2 - i1) for offset in range(j2 - j1): result.append({ token: ref_tokens[j1 offset], start: None, end: None, matched: False, }) return result这段代码的关键在于维护asr_pos指针。SequenceMatcher 返回的 opcodes 是按顺序处理的所以asr_pos始终指向当前尚未消费的 ASR token。equal块中参考文本 token 与 ASR token 一一对应时间戳可以直接赋值。delete表示 ASR 产生了多余内容跳过即可。insert和replace表示参考文本中有些词没有对应的 ASR 时间只能标记为matchedFalse。6.3 从词级时间戳生成句子级时间戳得到词级映射后下一步是把它汇聚成句子级时间戳。参考文本可以先按句子切分然后在每个句子内部找到第一个匹配 token 的start和最后一个匹配 token 的end。def merge_sentence_timestamps( aligned_tokens: list[dict], sentences: list[str], ) - list[dict]: # 这里假设 aligned_tokens 的顺序就是句子拼接后的顺序 # 实际项目中需要把句子切分后的 token 索引也保留下来 sentence_results [] idx 0 for sentence in sentences: tokens sentence.split() count len(tokens) window aligned_tokens[idx: idx count] idx count matched [w for w in window if w[matched]] if matched: start matched[0][start] end matched[-1][end] else: start None end None sentence_results.append({ text: sentence, start: start, end: end, matched_token_ratio: len(matched) / count if count else 0, }) return sentence_results这个函数的输入依赖一句话的 token 数量刚好与aligned_tokens的一段切片对应所以实际实现时最好在切句阶段就保留每个句子对应的 token 下标范围。为了演示这里做了简化。句子级时间戳的好处是更稳定。词级时间戳容易受到 Whisper 语音端点检测的微小偏移影响而句子级取首尾词时间戳可以把局部误差“包住”。6.4 对齐质量校验对齐完成之后不能直接把结果交出去。至少要做三方面校验。第一是覆盖率检查。一个句子的matched_token_ratio如果低于 0.6说明这句话要么 ASR 错误严重要么参考文本与音频内容不一致。这类句子需要单独标记在后续人工抽检中优先排查。第二是时间单调性检查。正常情况下句子时间戳应该是单调递增的。如果出现前一句的end大于后一句的start说明切分或映射出了问题要触发告警。第三是随机人工抽检。虽然整体流程是自动化的但仍然需要从结果中随机抽取一部分句子人工听一遍确认时间戳是否落在正确位置。这个环节最好由熟悉项目需求的人来做而不是完全交给机器。7. 并行调度与整体运行7.1 基于多进程的任务调度由于不同书之间完全独立任务天然可以横向扩展。最简单的方式是使用 Python 的concurrent.futures.ProcessPoolExecutor做多进程并行# pipeline.py import os from concurrent.futures import ProcessPoolExecutor, as_completed from config import ( RAW_AUDIO_DIR, RAW_TEXT_DIR, OUTPUT_DIR, MAX_WORKERS, ) def process_one_book(book_id: str): out_dir os.path.join(OUTPUT_DIR, book_id) aligned_path os.path.join(out_dir, aligned.jsonl) if os.path.exists(aligned_path): print(f[skip] {book_id}) return book_id # 1. 检查音频分段是否已生成没有则预处理 # 2. 加载 ASR 模型转录所有分段 # 3. 清洗参考文本并规范化 # 4. 执行对齐 # 5. 导出 JSONL # 这里省略详细调用实际代码模块化后按序调用即可 return book_id def main(): book_ids [fbook_{i:03d} for i in range(800)] with ProcessPoolExecutor(max_workersMAX_WORKERS) as pool: futures [pool.submit(process_one_book, bid) for bid in book_ids] for future in as_completed(futures): print(f[done] {future.result()}) if __name__ __main__: main()这里有一个问题每本书都需要加载一次 ASR 模型如果MAX_WORKERS4则会有 4 个进程各自持有一个大模型显存占用会翻倍。更稳妥的方式是让每个进程在处理完一本书后不退出模型而是继续处理下一本直到整个 worker 退出。上面的方式其实每次submit都会分配到不同进程进程内模型不会反复加载但要注意多个进程同时加载大模型时显存是否足够。如果机器显存不够可以改成单进程串行处理多本书或者用消息队列做任务分发把模型加载集中在少量 worker 上。这里需要根据实际硬件来定。7.2 断点续跑断点续跑是 6 天跑完 8000 小时的关键。每一本书在完成最终 JSONL 后下次运行会直接跳过。除此之外音频分段、ASR 分段 JSON 都是中间产物可以随时恢复。我建议每隔 1 小时或每处理完 5 本书就打印一次进度统计方便判断当前速度是否达到预期。completed sum(1 for f in os.listdir(OUTPUT_DIR) if os.path.exists(os.path.join(OUTPUT_DIR, f, aligned.jsonl))) remaining 800 - completed print(f[progress] completed{completed} remaining{remaining})7.3 输出格式示例最终每个句子对应一条 JSONL{book_id: book_001, chapter: chapter_01, sentence_index: 1, start: 12.3, end: 15.8, text: It was the best of times, matched_token_ratio: 1.0}这种格式方便下游直接加载。无论你是要生成字幕还是要把音频片段截取出来做 TTS 训练集都很方便。8. 常见问题与排查思路在整个流程中有一些问题出现的频率非常高。我把它们整理成了表格方便快速定位。问题现象常见原因解决思路ASR 转写速度很慢GPU 驱动或 CTranslate2 未正确使用 GPU检查nvidia-smi确认devicecuda且compute_typefloat16显存不足 OOM同时加载多个大模型或音频分段过长降低进程数改用large-v2或medium模型缩短分段时长对齐覆盖率低参考文本包含大量目录、脚注等噪音加强文本清洗规则先统计未匹配文本的分布时间戳错位严重ASR 与参考文本文本规范不一致检查数字、缩写、标点是否统一规范化某本书中途失败音频文件损坏或格式不标准增加异常捕获记录错误分段路径修复后重跑输出句子顺序混乱句子切分维度与 ASR 分段维度没有统一在切句时保留全局索引不要依赖分段内相对位置进程间显存冲突ProcessPoolExecutor多个 worker 各自加载模型限制 worker 数量或改成任务队列模式排查时我推荐按这个顺序来先确认 ASR 分词是否合理再检查参考文本规范化和清洗是否到位最后看对齐算法中的asr_pos指针是否在长文本下累积了误差。多数问题都出在前两步而不是动态规划本身。9. 最佳实践与优化建议9.1 把流水线做成可审计的大规模数据治理最怕的是“跑完了但不知道结果对不对”。建议每一本书最终输出一个report.json至少包含以下字段音频总时长。ASR 识别的总词数。参考文本总词数。匹配词数。匹配率。平均置信度如果能拿到 ASR 置信度的话。处理耗时。有了这个审计报告即使某个书有问题也能快速定位到阶段。9.2 优先用确定性规则而不是模型在“no LLM in the loop”的约束下仍然有很多可以用规则解决的问题。比如数字展开、缩写替换、标点清理这些都可以用规则实现不需要引入额外的模型。只有遇到规则写不清楚、且需要语义理解的场景才值得考虑更复杂的模型。而对于对齐任务传统动态规划已经足够好完全可以用确定性的方法获得稳定结果。9.3 预留人工抽检闭环完全自动化的流程始终存在风险。建议在最终交付前从每本书中随机抽取 20 到 30 个句子导出成 CSV 或在线表格让人工听一段音频并标记时间戳是否准确。这个抽检比例不需要高但能有效发现系统性问题。9.4 关于扩展性如果你后续要把这套流程应用到更大规模比如数万小时音频建议引入消息队列如 Redis Streams 或 RabbitMQ 做任务分发把“模型加载”和“任务获取”解耦。调度层记录每个任务的执行状态失败任务自动重试。这样可以平滑扩展更多 worker。另外如果你想在更多语言上使用faster-whisper 本身支持多语言但参考文本的规范化规则需要针对语言单独维护。以中文为例分词、数字读法、标点处理都和英文差异很大不能直接复用。做这类大规模音频对齐任务最核心的其实不是某个算法的准确率提升多少而是把任务拆分得足够细让每个环节都可以独立验证、独立重跑。音频切分、ASR 转写、文本清洗、动态规划对齐每一步都简单可靠组合起来才能在有限的资源里完成 800 本有声书、8000 小时音频的批量处理。如果你也正在搭建类似的语料流水线可以参考这套思路先跑通一个小规模样本再逐步扩展到全量数据。