如何批量处理音频?编写脚本调用SenseVoiceSmall模型教程
1. 为什么需要批量语音处理能力?
你有没有遇到过这样的场景:手头有几十段客服录音、上百条会议片段,或者一整个课程的音频资料,却只能靠人工一条条点开、上传、等待识别、再复制结果?效率低、易出错、还特别耗时间。
SenseVoiceSmall 不是普通的语音转文字工具。它像一个懂情绪、会听环境的“语音助理”——不仅能准确听清中、英、日、韩、粤五种语言,还能告诉你说话人是开心还是烦躁,背景里有没有突然响起的掌声或BGM音乐。但它的真正价值,往往被藏在那个漂亮的Gradio界面背后:批量处理能力才是落地到真实工作流的关键。
这篇教程不讲怎么点几下就能识别单条音频,而是带你从零写一个可复用、可调度、能一次跑完几十个文件的Python脚本。你会学到:如何绕过WebUI直接调用模型核心逻辑、怎样自动适配不同采样率和格式的音频、如何结构化保存带情感标签的识别结果,以及最关键的——怎么让这个脚本稳定运行在服务器上,成为你日常工作的“语音处理小助手”。
不需要你提前掌握语音模型原理,只要你会写基础Python、能看懂命令行,就能跟着一步步完成。
2. 理解SenseVoiceSmall的核心能力与限制
2.1 它不是“另一个ASR”,而是一个富文本语音理解器
传统语音识别(ASR)的目标只有一个:把声音变成文字。SenseVoiceSmall 的目标更进一步——它输出的是一段带语义标记的富文本。比如:
<|HAPPY|>今天这个方案太棒了!<|APPLAUSE|><|BGM|>
这串文本里藏着三层信息:
- 文字内容:“今天这个方案太棒了!”
- 情感状态:
<|HAPPY|>表示说话人语气积极、情绪高涨 - 环境事件:
<|APPLAUSE|>是现场掌声,<|BGM|>是背景音乐
这种结构化输出,让后续分析变得简单:你可以用一行代码统计所有“ANGRY”出现次数,筛选出含“LAUGHTER”的片段做趣味剪辑,或者把“SAD”和“CRY”同时出现的录音优先转给心理支持团队。
2.2 模型对输入音频的实际要求很友好
很多语音模型对音频格式吹毛求疵:必须WAV、必须16kHz、必须单声道……SenseVoiceSmall 的设计更贴近真实场景:
- 支持常见格式:MP3、WAV、M4A、FLAC,甚至部分MOV视频里的音频流
- 自动重采样:输入44.1kHz或8kHz音频,模型内部会通过
av库自动转成16kHz - 自动分段:内置VAD(语音活动检测),能智能切分长音频中的静音段,避免整段识别导致错误累积
- ❌ 唯一硬性要求:音频时长建议控制在30分钟以内。超过后内存占用明显上升,推荐按5–10分钟切分后再批量处理
这意味着你不用花时间预处理音频——扔进去,它自己搞定。
2.3 WebUI只是入口,真正的能力在AutoModel.generate()里
Gradio界面很直观,但它本质是个“演示包装层”。所有识别逻辑都封装在model.generate()这个方法里。只要拿到这个对象,你就拥有了全部能力:
- 可以传入本地文件路径(
input="/path/to/audio.mp3") - 也可以传入numpy数组(适合从麦克风实时捕获)
- 还能控制关键参数:
language="auto"(自动检测)、merge_vad=True(合并相邻语音段)、batch_size_s=60(每批最多处理60秒音频)
重点记住这句话:WebUI是给你看的,generate()是给你用的。
3. 编写第一个批量处理脚本
3.1 环境准备:三步确认,避免后续报错
在开始写代码前,请先确认你的运行环境已满足以下三点。这不是可选项,而是避免卡在第一步的关键检查:
Python版本是否为3.11?
运行python --version,如果不是3.11,请切换虚拟环境或重装。FunASR对Python版本敏感,3.10或3.12都可能触发兼容性问题。CUDA是否可用?
运行以下命令,确认返回True:import torch print(torch.cuda.is_available()) # 应输出 True print(torch.cuda.device_count()) # 至少为 1必要库是否已安装?
执行这条命令一次性装齐(注意顺序,av必须在funasr前安装):pip install av funasr modelscope gradio
如果某一步失败,请暂停阅读,先解决环境问题。后面所有代码都依赖这三步成功。
3.2 核心脚本:batch_sensevoice.py
下面这段代码,就是你批量处理能力的起点。它不依赖WebUI,不打开浏览器,只做一件事:读取指定文件夹下的所有音频,逐个识别,把结果存成结构化JSON文件。
# batch_sensevoice.py import os import json import time from pathlib import Path from datetime import datetime from funasr import AutoModel from funasr.utils.postprocess_utils import rich_transcription_postprocess # ========== 1. 初始化模型(只需执行一次) ========== print("⏳ 正在加载 SenseVoiceSmall 模型...") model = AutoModel( model="iic/SenseVoiceSmall", trust_remote_code=True, vad_model="fsmn-vad", vad_kwargs={"max_single_segment_time": 30000}, device="cuda:0", # 使用GPU加速,如无GPU可改为 "cpu" ) # ========== 2. 定义批量处理函数 ========== def process_audio_file(audio_path: str, language: str = "auto") -> dict: """ 处理单个音频文件,返回结构化结果 """ start_time = time.time() try: # 调用模型识别 res = model.generate( input=audio_path, cache={}, language=language, use_itn=True, batch_size_s=60, merge_vad=True, merge_length_s=15, ) # 后处理:将富文本标签转为易读格式 raw_text = res[0]["text"] if res else "" clean_text = rich_transcription_postprocess(raw_text) # 构建结构化结果 result = { "audio_path": str(Path(audio_path).resolve()), "duration_sec": round(res[0]["duration"] if res else 0, 2), "language_detected": res[0].get("language", "unknown"), "raw_output": raw_text, "clean_output": clean_text, "processing_time_sec": round(time.time() - start_time, 2), "timestamp": datetime.now().isoformat(), } print(f" {Path(audio_path).name} 处理完成 | {result['duration_sec']}s | {result['processing_time_sec']}s") return result except Exception as e: error_msg = f"❌ {Path(audio_path).name} 处理失败:{str(e)}" print(error_msg) return { "audio_path": str(Path(audio_path).resolve()), "error": str(e), "timestamp": datetime.now().isoformat(), } # ========== 3. 主流程:遍历文件夹,批量处理 ========== if __name__ == "__main__": # 配置项(你只需要改这里!) INPUT_FOLDER = "./audio_samples" # 替换为你自己的音频文件夹路径 OUTPUT_JSON = "./batch_results.json" LANGUAGE = "auto" # 可选:"zh", "en", "yue", "ja", "ko", "auto" # 创建输入文件夹(如果不存在) os.makedirs(INPUT_FOLDER, exist_ok=True) # 获取所有支持的音频文件 supported_exts = {".mp3", ".wav", ".m4a", ".flac", ".ogg"} audio_files = [ f for f in Path(INPUT_FOLDER).rglob("*") if f.is_file() and f.suffix.lower() in supported_exts ] if not audio_files: print(f" 在 {INPUT_FOLDER} 中未找到音频文件,请检查路径和格式") exit(1) print(f" 发现 {len(audio_files)} 个音频文件,开始批量处理...") # 逐个处理 all_results = [] for idx, audio_path in enumerate(audio_files, 1): print(f"\n[{idx}/{len(audio_files)}] 正在处理:{audio_path.name}") result = process_audio_file(str(audio_path), LANGUAGE) all_results.append(result) # 保存结果到JSON with open(OUTPUT_JSON, "w", encoding="utf-8") as f: json.dump(all_results, f, ensure_ascii=False, indent=2) print(f"\n 批量处理完成!结果已保存至:{OUTPUT_JSON}") print(f" 总计处理 {len(all_results)} 个文件,平均耗时 {sum(r.get('processing_time_sec', 0) for r in all_results)/len(all_results):.2f} 秒/文件")3.3 脚本使用说明:三步上手
准备音频文件
把你要处理的所有音频文件(MP3/WAV/M4A等)放进一个文件夹,比如./audio_samples。脚本会自动扫描子目录。修改配置项
打开batch_sensevoice.py,找到INPUT_FOLDER和LANGUAGE这两行,按需修改:INPUT_FOLDER = "./my_meetings"→ 指向你的实际文件夹LANGUAGE = "zh"→ 如果所有音频都是中文,固定语言可提升准确率;设为"auto"则自动检测
运行脚本
在终端执行:python batch_sensevoice.py第一次运行会自动下载模型(约1.2GB),之后再运行就快了。处理完成后,你会得到一个
batch_results.json文件,内容类似这样:[ { "audio_path": "/home/user/audio_samples/call_001.mp3", "duration_sec": 128.4, "language_detected": "zh", "clean_output": "客户说:<|HAPPY|>这个价格我很满意!<|APPLAUSE|>", "processing_time_sec": 4.21, "timestamp": "2025-04-05T10:22:33.123456" } ]
4. 进阶技巧:让脚本更实用、更稳定
4.1 自动过滤无效音频,避免中断
真实场景中,常有损坏文件、静音片段或极短录音(<1秒)。它们会导致model.generate()报错并中断整个流程。加一段预检逻辑即可解决:
# 在 process_audio_file 函数开头添加 def is_valid_audio(audio_path: str) -> bool: """检查音频是否可读且时长合理""" try: import av container = av.open(audio_path) audio_stream = next((s for s in container.streams if s.type == 'audio'), None) if not audio_stream: return False # 获取时长(秒) duration = float(container.duration) / av.time_base if container.duration else 0 return 1.0 <= duration <= 1800.0 # 1秒到30分钟 except Exception: return False # 在主循环中调用 for idx, audio_path in enumerate(audio_files, 1): if not is_valid_audio(str(audio_path)): print(f" {audio_path.name} 无效或损坏,跳过") continue # ... 后续处理4.2 导出为CSV,方便Excel分析
JSON对程序员友好,但业务同事更习惯Excel。加几行代码就能导出带表头的CSV:
# 在保存JSON后,追加CSV导出 import csv def save_as_csv(results: list, csv_path: str): if not results: return # 提取所有键作为表头(取第一个非空结果) sample = next((r for r in results if "clean_output" in r), {}) fieldnames = ["audio_path", "duration_sec", "language_detected", "clean_output", "processing_time_sec"] with open(csv_path, "w", newline="", encoding="utf-8-sig") as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() for r in results: # 只写入关键字段,避免JSON嵌套导致CSV混乱 row = {k: r.get(k, "") for k in fieldnames} writer.writerow(row) # 在主流程末尾调用 save_as_csv(all_results, "./batch_results.csv") print(" 同时导出为CSV,可直接用Excel打开分析")4.3 添加进度条,告别“黑屏等待”
用tqdm库让进度可视化(先安装:pip install tqdm):
from tqdm import tqdm # 替换原来的 for 循环 for audio_path in tqdm(audio_files, desc="🔊 处理中", unit="file"): if not is_valid_audio(str(audio_path)): continue result = process_audio_file(str(audio_path), LANGUAGE) all_results.append(result)运行时你会看到清晰的进度条和预估剩余时间,体验大幅提升。
5. 实战案例:10分钟搞定一场3小时会议的结构化整理
我们用一个真实工作流来验证这套方案的价值。
5.1 场景还原
假设你刚参加完一场3小时的产品评审会,录了4段各45分钟的音频(meeting_part1.mp3到meeting_part4.mp3),需要:
- 提取所有发言内容
- 标记产品经理提到“性能优化”时的情绪倾向
- 找出所有客户反馈环节(含“建议”、“希望”、“问题”等关键词)
- 统计技术负责人发言总时长
5.2 操作步骤
- 将4个MP3文件放入
./meeting_audio文件夹 - 修改脚本配置:
INPUT_FOLDER = "./meeting_audio" LANGUAGE = "zh" - 运行
python batch_sensevoice.py→ 生成batch_results.json - 用Python快速分析(新建
analyze_meeting.py):
import json with open("./batch_results.json", "r", encoding="utf-8") as f: results = json.load(f) # 统计技术负责人发言(假设他说话常带“我们后端”、“API响应”等词) tech_keywords = ["后端", "API", "响应", "延迟", "QPS", "压测"] tech_total_sec = 0 for r in results: if "clean_output" not in r: continue text = r["clean_output"] if any(kw in text for kw in tech_keywords): tech_total_sec += r["duration_sec"] print(f"🔧 技术负责人总发言时长:{tech_total_sec:.1f} 秒(约 {tech_total_sec/60:.1f} 分钟)") # 情绪分析:找出所有“HAPPY”和“ANGRY”出现的片段 happy_count = sum(1 for r in results if "HAPPY" in r.get("raw_output", "")) angry_count = sum(1 for r in results if "ANGRY" in r.get("raw_output", "")) print(f"😊 开心片段数:{happy_count} | 😠 愤怒片段数:{angry_count}")运行后,你立刻得到量化结论,而不是面对一堆原始文字发愁。
6. 常见问题与避坑指南
6.1 “CUDA out of memory” 错误怎么解决?
这是GPU显存不足的典型提示。别急着换显卡,先尝试三个低成本方案:
- 降低 batch_size_s:把
batch_size_s=60改成30或15,减少单次处理时长 - 关闭VAD合并:将
merge_vad=True改为False,让模型分更小段处理,显存压力下降约40% - 强制CPU推理:把
device="cuda:0"改成device="cpu",速度慢3–5倍,但100%可用
6.2 为什么有些音频识别结果全是<|NOISE|>?
这通常表示音频质量差:信噪比低、有持续电流声、或录音设备离人太远。SenseVoiceSmall 对噪声较敏感。解决方案:
- 用Audacity等工具先做基础降噪(推荐“效果→降噪”)
- 或在脚本中加入简单能量阈值过滤:
import numpy as np from scipy.io import wavfile def get_rms_energy(wav_path): try: sr, data = wavfile.read(wav_path) if data.ndim > 1: data = data.mean(axis=1) # 转单声道 return np.sqrt(np.mean(data.astype(float)**2)) except: return 0 # 在预检中加入 if get_rms_energy(str(audio_path)) < 100: print(f" {audio_path.name} 能量过低,疑似静音或损坏,跳过") continue
6.3 如何部署为定时任务,每天自动处理新录音?
Linux服务器上,用crontab即可。例如每天早上9点处理/data/new_audios下的新文件:
# 编辑定时任务 crontab -e # 添加这一行(假设脚本在 /opt/sensevoice/) 0 9 * * * cd /opt/sensevoice && python batch_sensevoice.py >> /var/log/sensevoice.log 2>&1记得在脚本开头加上#!/usr/bin/env python3,并给执行权限:chmod +x batch_sensevoice.py。
7. 总结:批量处理不是功能,而是工作流的起点
你现在已经掌握了 SenseVoiceSmall 的批量处理能力,但这只是开始。真正的价值在于,它让你能把语音数据当作一种可编程、可分析、可集成的一等公民:
- 你可以把它接入企业微信机器人,收到新录音自动推送摘要
- 可以和Notion API联动,把每次会议结果直接生成结构化页面
- 甚至训练自己的情绪分类模型,用 SenseVoiceSmall 的输出作为高质量标注数据
没有复杂的模型微调,没有烧脑的工程架构,只有一份干净的脚本、几个明确的配置项,和一个随时待命的语音理解引擎。
下一步,不妨从你手边最头疼的那堆音频开始——把它拖进./audio_samples,运行一次python batch_sensevoice.py。10分钟后,你会收到第一份结构化结果。那一刻,你就已经跨过了“想用AI”和“正在用AI”的分界线。
--- > **获取更多AI镜像** > > 想探索更多AI镜像和应用场景?访问 [CSDN星图镜像广场](https://ai.csdn.net/?utm_source=mirror_blog_end),提供丰富的预置镜像,覆盖大模型推理、图像生成、视频生成、模型微调等多个领域,支持一键部署。