Paraformer-large批量处理音频文件：自动化脚本部署教程

Paraformer-large批量处理音频文件：自动化脚本部署教程

1. 为什么需要批量处理？单次上传太慢了

你有没有遇到过这样的情况：手头有几十个会议录音、上百条客服对话、或者一整个培训课程的音频片段，却只能一个一个点开Gradio界面上传、等待识别、复制结果——光是点鼠标就累得手腕发酸，更别说中间还要反复检查格式、重试失败任务、手动整理输出文本。

Paraformer-large本身识别质量高、支持长音频、带VAD和标点预测，但默认的Gradio界面是为“交互式单次使用”设计的，不是为“工程化批量生产”准备的。真正的效率提升，不在于模型多快，而在于能不能让模型自己动起来。

这篇教程不讲怎么调参、不聊模型结构，只聚焦一件事：
把你硬盘里成堆的.wav/.mp3文件，一键喂给Paraformer-large
自动完成格式转换、分段切片、并发识别、结果归档
最终生成带时间戳的SRT字幕、纯文本TXT、或结构化JSON供下游系统读取

全程离线运行，不依赖网络，不上传隐私音频，所有逻辑都在你自己的GPU服务器上闭环执行。

2. 批量脚本核心设计思路：绕过Gradio，直连FunASR引擎

Gradio是个好用的演示工具，但它本质是Web服务层，加了一层HTTP请求、UI渲染、状态管理的开销。批量处理要的是零交互、低延迟、可编程、可监控——所以我们不走Web接口，而是直接复用镜像里已预装的FunASR底层能力。

关键认知转变：

• ❌ 不再用curl http://localhost:6006/...模拟点击上传
• 直接调用model.generate()接口，把本地路径当参数传进去
• 用Python标准库+concurrent.futures做文件遍历与并发控制
• 用ffmpeg-python统一转码，避免“识别失败：采样率不匹配”这类低级报错

整个脚本只有1个依赖：你已经在镜像里装好的funasr。不需要额外pip install，不改环境，不碰conda配置。

3. 三步部署：从空目录到全自动转写流水线

3.1 创建项目结构（5秒完成）

在你的工作目录下（比如/root/workspace），新建以下结构：

mkdir -p batch_asr/{input,outputs/logs,outputs/text,outputs/srt} touch batch_asr/run_batch.py

目录说明：

• input/：你把所有待识别的音频文件（.wav,.mp3,.flac）丢进去
• outputs/text/：每条音频生成的纯文本（如meeting_01.txt）
• outputs/srt/：带时间戳的字幕文件（如meeting_01.srt）
• outputs/logs/：运行日志和错误记录（排查问题全靠它）

注意：不要把音频放在/root/workspace根目录，脚本默认只扫描batch_asr/input/子目录，避免误处理其他文件。

3.2 复制粘贴这个run_batch.py（已实测可用）

# batch_asr/run_batch.py import os import time import json import logging from pathlib import Path from concurrent.futures import ThreadPoolExecutor, as_completed from funasr import AutoModel # ========== 配置区（只需改这里）========== INPUT_DIR = "/root/workspace/batch_asr/input" OUTPUT_TEXT_DIR = "/root/workspace/batch_asr/outputs/text" OUTPUT_SRT_DIR = "/root/workspace/batch_asr/outputs/srt" OUTPUT_LOG_DIR = "/root/workspace/batch_asr/outputs/logs" MAX_WORKERS = 4 # 并发数，根据GPU显存调整：4090D建议设3-4，3090建议设2 # ======================================== # 日志初始化 os.makedirs(OUTPUT_LOG_DIR, exist_ok=True) logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s", handlers=[ logging.FileHandler(f"{OUTPUT_LOG_DIR}/batch_run_{int(time.time())}.log", encoding="utf-8"), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) # 加载模型（全局只加载一次，避免重复初始化） logger.info("正在加载Paraformer-large模型...") model = AutoModel( model="iic/speech_paraformer-large-vad-punc_asr_nat-zh-cn-16k-common-vocab8404-pytorch", model_revision="v2.0.4", device="cuda:0" ) logger.info("模型加载成功 ") def process_single_audio(audio_path: Path): """处理单个音频文件，返回 (文件名, 文本结果, SRT内容, 错误信息)""" start_time = time.time() filename = audio_path.stem logger.info(f"开始处理：{filename}{audio_path.suffix}") try: # FunASR原生支持直接传路径，自动处理格式/采样率 res = model.generate( input=str(audio_path), batch_size_s=300, # 控制显存占用，长音频更稳 hotword="阿里巴巴;达摩院;语音识别" # 可选：加入领域热词提升准确率 ) if not res or len(res) == 0: raise RuntimeError("模型返回空结果") text_result = res[0]["text"] srt_content = generate_srt_from_res(res[0]) # 保存文本 text_path = Path(OUTPUT_TEXT_DIR) / f"{filename}.txt" with open(text_path, "w", encoding="utf-8") as f: f.write(text_result) # 保存SRT srt_path = Path(OUTPUT_SRT_DIR) / f"{filename}.srt" with open(srt_path, "w", encoding="utf-8") as f: f.write(srt_content) duration = time.time() - start_time logger.info(f" {filename} 处理完成 | 耗时 {duration:.1f}s | 字数 {len(text_result)}") return filename, text_result, srt_content, None except Exception as e: error_msg = f"❌ {filename} 处理失败：{str(e)}" logger.error(error_msg) return filename, "", "", str(e) def generate_srt_from_res(res_dict): """将FunASR返回的segments转成标准SRT格式（含时间戳）""" if "sentences" not in res_dict: return "" srt_lines = [] for i, seg in enumerate(res_dict["sentences"], 1): start = int(seg["start"] * 1000) # 转毫秒 end = int(seg["end"] * 1000) text = seg.get("text", "").strip() if not text: continue # SRT时间格式：HH:MM:SS,mmm --> HH:MM:SS,mmm def ms_to_srt(ms): h, ms = divmod(ms, 3600000) m, ms = divmod(ms, 60000) s, ms = divmod(ms, 1000) return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}" srt_lines.extend([ str(i), f"{ms_to_srt(start)} --> {ms_to_srt(end)}", text, "" ]) return "\n".join(srt_lines) def main(): input_path = Path(INPUT_DIR) audio_files = [ f for f in input_path.iterdir() if f.is_file() and f.suffix.lower() in {".wav", ".mp3", ".flac", ".m4a"} ] if not audio_files: logger.warning(" input目录为空，请先放入音频文件") return logger.info(f"发现 {len(audio_files)} 个待处理音频文件") os.makedirs(OUTPUT_TEXT_DIR, exist_ok=True) os.makedirs(OUTPUT_SRT_DIR, exist_ok=True) # 并发执行 results = [] with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor: future_to_file = { executor.submit(process_single_audio, f): f for f in audio_files } for future in as_completed(future_to_file): results.append(future.result()) # 统计汇总 success_count = sum(1 for r in results if r[3] is None) logger.info(f"\n 批量任务结束：成功 {success_count}/{len(results)} 个文件") if success_count < len(results): failed = [r[0] for r in results if r[3] is not None] logger.warning(f"失败文件：{', '.join(failed)}") if __name__ == "__main__": main()

3.3 运行与验证（2分钟搞定）

1. 把音频放进去
   将你的音频文件（命名不含中文/空格更稳妥，如interview_01.wav）复制到/root/workspace/batch_asr/input/

2. 执行脚本

   cd /root/workspace source /opt/miniconda3/bin/activate torch25 python batch_asr/run_batch.py

3. 看结果
   几秒钟后，你会看到实时日志滚动：

   2025-04-05 14:22:18 - INFO - 开始处理：interview_01.wav 2025-04-05 14:22:25 - INFO - interview_01 处理完成 | 耗时 6.8s | 字数 1247 ... 2025-04-05 14:22:42 - INFO - 批量任务结束：成功 5/5 个文件

   同时，outputs/text/下生成interview_01.txt，outputs/srt/下生成interview_01.srt（可用VLC或PotPlayer直接播放校验时间轴）。

4. 进阶技巧：让批量处理更省心、更可控

4.1 自动清理输入文件（处理完就移走）

在main()函数末尾添加：

# 处理完成后，把已识别的音频移到done目录 done_dir = input_path / "done" done_dir.mkdir(exist_ok=True) for f in audio_files: f.rename(done_dir / f.name) logger.info(f" 已处理文件已移至 {done_dir}")

这样下次再跑，就不会重复处理同一文件。

4.2 失败重试机制（网络抖动/显存不足时自动续跑）

修改process_single_audio函数，在except块里加：

except Exception as e: error_msg = f"❌ {filename} 处理失败：{str(e)}" logger.error(error_msg) # 重试一次，加1秒延时防瞬时压力 time.sleep(1) try: res = model.generate(input=str(audio_path), batch_size_s=300) if res and len(res) > 0: logger.info(f" {filename} 重试成功") text_result = res[0]["text"] srt_content = generate_srt_from_res(res[0]) return filename, text_result, srt_content, None except Exception as e2: pass return filename, "", "", str(e)

4.3 输出结构化JSON（方便程序读取）

在保存文本/SRT后，加一段：

# 保存JSON（含原始字段：text, segments, timestamps） json_path = Path(OUTPUT_TEXT_DIR).parent / "json" / f"{filename}.json" json_path.parent.mkdir(exist_ok=True) with open(json_path, "w", encoding="utf-8") as f: json.dump(res[0], f, ensure_ascii=False, indent=2)

这样下游系统（比如用Python脚本做关键词提取、情感分析）就能直接读JSON，不用再解析TXT。

5. 常见问题与真实踩坑记录

5.1 “CUDA out of memory” 显存爆了？

这是批量处理最常遇到的问题。根本原因不是模型太大，而是batch_size_s=300在并发时叠加显存压力。

解决方案：

• 降低MAX_WORKERS（从4→2）
• 在model.generate()中显式减小batch_size_s（如设为150）
• 添加max_length=30参数限制单次处理最大时长（单位秒）

res = model.generate( input=str(audio_path), batch_size_s=150, max_length=30 # 单次最多处理30秒音频，长文件自动分段 )

5.2 MP3文件识别报错：“Unsupported format”

FunASR底层用soundfile读取，它默认不支持MP3。但镜像里已预装ffmpeg，所以只要确保ffmpeg在PATH里，FunASR会自动调用它转码。

验证方法：

which ffmpeg # 应该返回 /usr/bin/ffmpeg ffmpeg -version # 确认版本 ≥ 4.0

如果没找到，执行：

apt update && apt install -y ffmpeg

5.3 识别结果没有标点？或标点错位？

Paraformer-large的Punc模块对中文断句效果很好，但有两个前提：

• 音频必须是人声清晰、无背景音乐的干净录音
• model.generate()必须传入batch_size_s参数（哪怕设为1），否则Punc模块不启用

确认你代码里写了：

res = model.generate(input=..., batch_size_s=300) # 有这个参数 # res = model.generate(input=...) # ❌ 没这个参数，Punc不工作

6. 总结：你现在已经拥有了一个生产级ASR流水线

回顾一下，你刚刚完成的不是“又一个Python脚本”，而是一套可立即投入使用的语音处理基础设施：

• 零学习成本：不用学Gradio API、不用碰FastAPI，复用现有模型对象
• 真离线安全：所有音频、模型、结果100%留在你的服务器，不触网、不上传
• 可扩展性强：增加新功能只需改几行（比如加微信通知、存数据库、对接飞书机器人）
• 运维友好：日志自动归档、失败文件隔离、并发可控、资源用量透明

下一步你可以：
🔹 把run_batch.py做成systemd服务，开机自启监听input/目录变化
🔹 写个简单Shell脚本，每天凌晨自动拉取NAS里的新录音并触发识别
🔹 用gradio.Blocks封装一个“批量上传+状态看板”界面，给非技术人员用

语音识别的价值，从来不在单次点击的惊艳，而在千次重复的无声高效。

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。