【Bug已解决】[Feature Proposal] Add FunASR ASR Distributed Inference Examples 解决方案一、现象长什么样用FunASR阿里巴巴开源的语音识别工具箱做批量 / 长音频推理时单张 GPU 成了明显瓶颈几千个音频文件排队单卡串行跑要数小时GPU 利用率还不到 50%因为前后处理是 CPU 密集。想「多卡并行」却没有官方分布式推理示例于是用户各显神通有的用multiprocessing起 N 个进程各加载一份模型VRAM 爆了有的手动把音频列表按 rank 切分但聚合结果错位第 3 卡的转写混到第 1 卡的输出里。最容易踩的「非崩溃但错」现象# 多进程各加载模型结果文件互相覆盖 / 顺序错乱 rank0 写出 result.jsonl (含它负责的 1000 条) rank1 也写出 result.jsonl (同名覆盖或 append 乱序)本质FunASR 官方示例几乎都是单卡model.generate()缺少「如何把一批音频正确、省显存地分到多卡上推理并聚合」的分布式范式。用户缺范例就只能自己 hack于是要么显存爆、要么结果错位。二、背景FunASR 的推理入口是AutoModelmodel.generate(audio_in...)。单卡模式下你把所有音频塞进一个列表模型逐个或分批generate。问题在规模计算 vs 前后处理不匹配ASR 的 VAD语音活动检测、特征提取、后处理很多在 CPU/模型外单卡 GPU 在等这些时就空闲。多卡能把「不同音频」分到不同卡整体吞吐上去。没有分布式范例用户不知道该「按音频分片、每卡一份模型副本、各自写各自的结果分片、最后合并」还是「一份模型靠 tensor parallel」。对 ASR 这种数据并行友好的任务前者按音频分片最简单高效但没人示范。于是出现两类错误实践每进程全量加载模型N 个进程各自AutoModel(...)每个吃一份完整 VRAMN 张卡显存都爆。正确做法是「每卡一个进程、一份模型」本来就该这样但用户常误以为要「主进程加载、子进程共享」在 FunASR 里共享 CUDA 上下文很麻烦反而写错。结果聚合错位分片推理后各 rank 把结果写同一文件顺序/覆盖混乱。正确做法是「各 rank 写独立分片文件最后按全局索引合并」。这个 Feature Proposal 就是要补上「用 Accelerate / torch.distributed 做 FunASR 数据并行推理」的范例。三、根因能力缺口分析把这个缺口当 bug 分析根因是FunASR 缺分布式推理范式导致用户错用多进程显存爆或错聚合结果错位三层第一层主因缺「按音频数据并行的标准范式」。用户不知道应该rank process_index; 负责 audio_list[rank::world_size]于是要么全量加载爆显存要么手动切分得不对。范例缺失是直接原因。第二层结果聚合无规范。多卡各自产出结果怎么汇成一份有序输出没有示范用户用「同文件名写」导致覆盖/乱序。这是数据并行最常见的坑但缺范例时必踩。第三层模型加载与 rank 生命周期不匹配。用户常在「fork 出的子进程里加载模型」但 FunASR/CUDA 要求模型在目标 rank 的进程里、init_process_group 之后才加载。顺序错了会报CUDA error或上下文错乱。范例应明确「先 init、后加载」的顺序。一句话缺分布式推理范例 → 用户错用多进程显存爆/错聚合结果错位/错加载顺序CUDA 上下文乱。四、最小可运行复现下面用纯 Python 模拟「错误聚合同文件覆盖vs 正确聚合分片文件后合并」的差异不需要 GPUfrom dataclasses import dataclass from typing import List, Dict dataclass class Audio: gid: int # 全局索引 path: str def shard(audios: List[Audio], rank: int, world: int) - List[Audio]: return audios[rank::world] def wrong_aggregate(rank_results: Dict[int, List[str]], out_file: str): # 错误所有 rank 写同一个文件后写覆盖先写 with open(out_file, w) as f: f.write(\n.join(rank_results[max(rank_results)])) def correct_aggregate(rank_results: Dict[int, Dict[int, str]], out_file: str): # 正确按全局 gid 合并各 rank 分片 merged {} for res in rank_results.values(): merged.update(res) with open(out_file, w) as f: for gid in sorted(merged): f.write(f{gid}: {merged[gid]}\n) def main(): audios [Audio(i, fa{i}.wav) for i in range(8)] world 2 r0 {a.gid: ftext{a.gid} for a in shard(audios, 0, world)} r1 {a.gid: ftext{a.gid} for a in shard(audios, 1, world)} # 正确聚合合并后应有 8 条、顺序正确 correct_aggregate({0: r0, 1: r1}, result.txt) print(正确聚合后条目数:, len(r0) len(r1)) # 8 if __name__ __main__: main()跑出来正确聚合得到 8 条有序结果——演示了「分片推理 按 gid 合并」为何能对。错误聚合则会丢一半覆盖。五、解决方案第一层最小直接修复最省事的落地用 Accelerate 做「按音频数据并行」的标准范式每 rank 加载一份模型、负责 1/N 音频、写独立分片、最后合并from accelerate import Accelerator from funasr import AutoModel accelerator Accelerator() # 先 initAccelerator 已帮做再在每 rank 加载模型 —— 顺序关键 model AutoModel( modelparaformer-zh, deviceaccelerator.device, disable_updateTrue, ) # 读取完整音频列表按 rank 分片 all_audios load_audio_manifest(manifest.txt) # 全局有序列表 my_audios all_audios[accelerator.process_index::accelerator.num_processes] results {} for a in my_audios: text model.generate(inputa.path, batch_size_s300)[0][text] results[a.gid] text # 用全局 gid 作 key # 每 rank 写独立分片绝不共用文件名 with open(fresult_rank{accelerator.process_index}.jsonl, w) as f: for gid, text in results.items(): f.write(f{gid}\t{text}\n)最后在所有 rank 结束后用一个小脚本按gid合并各result_rank*.jsonl。这是避免覆盖/乱序的关键。六、解决方案第二层结构性改进第一层是「手动范式」第二层是「封装一个 FunASR 分布式推理工具把分片/加载/聚合都固化」从设计上消灭错用from dataclasses import dataclass from typing import List, Dict, Callable dataclass class DistASRConfig: model_name: str manifest: str out_dir: str class FunASRDistributor: FunASR 数据并行推理的规范范式单一事实来源。 def __init__(self, accelerator, model_builder: Callable): self.acc accelerator # 关键init 之后、在当前 rank 进程里加载模型 self.model model_builder(deviceaccelerator.device) def run(self, audios, infer_fn: Callable) - Dict[int, str]: # 1) 按 rank 分片 my audios[self.acc.process_index::self.acc.num_processes] # 2) 各自推理 out: Dict[int, str] {} for a in my: out[a.gid] infer_fn(self.model, a.path) # 3) 写独立分片 self._dump_shard(out) return out def _dump_shard(self, out: Dict[int, str]): rank self.acc.process_index path f{self.out_dir}/result_rank{rank}.jsonl with open(path, w) as f: for gid, text in out.items(): f.write(f{gid}\t{text}\n) staticmethod def merge(out_dir: str, total: int) - List[str]: # 4) 合并按全局 gid 排序 merged: Dict[int, str] {} import glob for fp in glob.glob(f{out_dir}/result_rank*.jsonl): for line in open(fp): gid, text line.rstrip(\n).split(\t, 1) merged[int(gid)] text return [merged[i] for i in range(total)] # 用法 dist FunASRDistributor(accelerator, lambda device: AutoModel(modelparaformer-zh, devicedevice)) audios [Audio(i, fa{i}.wav) for i in range(8000)] dist.run(audios, lambda m, p: m.generate(inputp)[0][text]) # 合并任意一 rank 或单独脚本 if accelerator.is_main_process: final FunASRDistributor.merge(out, total8000)这样分片、加载顺序、独立分片写、合并都固化用户不会再踩显存爆/结果错位。七、解决方案第三层断言 / CI 守护把「分片不重叠不遗漏」「独立分片写不覆盖」「合并有序」固化成测试import pytest def test_shard_no_overlap_no_missing(): audios [Audio(i, fa{i}) for i in range(8)] world 2 s0 shard(audios, 0, world) s1 shard(audios, 1, world) gids {a.gid for a in s0} | {a.gid for a in s1} assert gids set(range(8)) # 不遗漏 assert not ({a.gid for a in s0} {a.gid for a in s1}) # 不重叠 def test_merge_ordered(): r0 {0: t0, 2: t2, 4: t4, 6: t6} r1 {1: t1, 3: t3, 5: t5, 7: t7} out FunASRDistributor.merge_from_dicts([r0, r1], total8) assert out [ft{i} for i in range(8)] def test_one_model_per_rank(): # 每 rank 只加载一次模型避免 N 进程 N 份全量加载 loads [] def builder(device): loads.append(device) return model acc FakeAccelerator(process_index0, num_processes4) d FunASRDistributor(acc, builder) assert len(loads) 1 # 仅一次加载 def test_no_shared_output_file(): # 每 rank 写独立文件不会同文件名覆盖 paths [fresult_rank{r}.jsonl for r in range(4)] assert len(set(paths)) 4 def test_full_pipeline_distributed(): audios [Audio(i, fa{i}) for i in range(16)] world 4 parts [shard(audios, r, world) for r in range(world)] all_gids set() for p in parts: all_gids | {a.gid for a in p} assert all_gids set(range(16))八、排查清单看多卡 ASR 是否显存爆每进程全量加载或结果文件覆盖/乱序 → 是缺分布式范式。确认是否「先 init_process_group / Accelerator 再加载模型」顺序错会 CUDA 上下文乱。临时救火用 Accelerator 按process_index::num_processes分片音频每 rank 一份模型、独立分片文件。检查合并脚本是否按全局 gid 排序避免覆盖各 rank 写不同文件名。长期方向采用封装好的FunASRDistributor范式分片/加载/写/合并一体化。升级 FunASR / 参考合了该范例的版本并跑上面的「分片不重叠」「合并有序」用例。若音频极长单条就超 VRAM应在分片基础上再做「按 chunk 切分单条音频」而非单纯加卡。九、小结FunASR 分布式推理范例缺口不是工具不能用而是缺「按音频数据并行」的标准范式导致用户错用多进程显存爆、错聚合结果覆盖/乱序、错加载顺序CUDA 上下文乱。最小修复是用 Accelerate 按 rank 分片音频、每 rank 一份模型、写独立分片文件、最后按 gid 合并结构性方向是封装FunASRDistributor把分片/加载/写/合并固化最后用 pytest 把「分片不重叠不遗漏」「独立分片不覆盖」「合并有序」锁死。抓住「ASR 是数据并行友好的任务、按音频分片 独立分片写 全局索引合并」这条所有批量 ASR 的分布式化都能照此落地。