SenseVoice-Small ONNX实战教程批量音频识别脚本扩展开发指南1. 引言如果你已经体验过SenseVoice-Small ONNX语音识别工具的便捷可能会发现一个问题每次只能处理一个音频文件对于需要处理大量录音、会议纪要或播客内容的场景来说效率实在太低了。想象一下你手头有几十个会议录音需要整理成文字或者需要批量处理客户访谈音频。一个个上传、点击识别、复制结果这个过程不仅耗时耗力还容易出错。有没有一种方法能让电脑自动帮你完成这些重复性工作这正是本教程要解决的问题。我们将基于现有的SenseVoice-Small ONNX工具开发一个批量音频识别脚本让你能够一键处理整个文件夹的音频文件自动保存识别结果到文本文件支持中断续传不怕程序意外退出生成处理报告清晰了解处理情况无论你是内容创作者、研究人员还是需要处理大量语音数据的企业用户这个脚本都能让你的工作效率提升数倍。接下来我将手把手带你从零开始构建这个实用的批量处理工具。2. 环境准备与项目结构2.1 环境检查在开始之前确保你已经成功运行过SenseVoice-Small ONNX工具。如果还没有请先按照项目README完成基础部署。我们需要确认几个关键组件# 检查Python环境 python --version # 应该显示Python 3.8或更高版本 # 检查关键依赖 pip list | grep -E onnxruntime|funasr|streamlit # 应该能看到相关包已安装2.2 创建项目结构我们将创建一个独立的批量处理项目不干扰原有的Web界面工具。建议按照以下结构组织文件sensevoice-batch-processor/ ├── batch_processor.py # 主脚本文件 ├── config.yaml # 配置文件 ├── requirements.txt # 依赖文件 ├── input_audio/ # 输入音频文件夹 ├── output_text/ # 输出文本文件夹 └── logs/ # 日志文件夹先创建基础目录结构# 创建项目目录 mkdir sensevoice-batch-processor cd sensevoice-batch-processor # 创建必要的子目录 mkdir -p input_audio output_text logs # 创建主脚本文件 touch batch_processor.py touch config.yaml touch requirements.txt2.3 安装额外依赖批量处理脚本需要一些额外的库来支持文件操作和配置管理。编辑requirements.txt文件# requirements.txt pyyaml6.0 # 用于读取配置文件 tqdm4.66.0 # 用于显示进度条 watchdog3.0.0 # 用于监控文件夹变化可选安装这些依赖pip install -r requirements.txt3. 核心脚本开发3.1 配置文件设计我们先设计一个灵活的配置文件让用户可以根据自己的需求调整参数。编辑config.yaml文件# config.yaml - 批量处理配置文件 # 路径配置 paths: input_dir: ./input_audio # 输入音频文件夹 output_dir: ./output_text # 输出文本文件夹 model_dir: ../models # 模型文件夹路径相对于脚本位置 log_dir: ./logs # 日志文件夹 # 模型配置 model: model_name: SenseVoiceSmall-ONNX-int8 # 模型名称 vad_model: fsmn-vad # 语音活动检测模型 punc_model: ct-punc # 标点模型 batch_size: 1 # 批处理大小 use_itn: true # 是否使用逆文本正则化 language: auto # 语种识别模式 # 处理配置 processing: supported_formats: [.wav, .mp3, .m4a, .ogg, .flac] # 支持的音频格式 max_duration: 600 # 最大音频时长秒超过会跳过 skip_existing: true # 是否跳过已处理文件 save_intermediate: false # 是否保存中间结果 # 性能配置 performance: num_workers: 1 # 并行处理数CPU核心数 chunk_size: 2000 # 音频分块大小毫秒 device: cpu # 运行设备cpu 或 cuda这个配置文件的好处是用户不需要修改代码就能调整各种参数非常灵活。3.2 批量处理器类设计现在我们来编写核心的批量处理器类。编辑batch_processor.py文件# batch_processor.py import os import sys import yaml import time import logging from pathlib import Path from datetime import datetime from typing import List, Dict, Optional, Tuple from tqdm import tqdm # 添加项目根目录到Python路径 project_root Path(__file__).parent sys.path.append(str(project_root.parent)) # 指向原始工具目录 class BatchAudioProcessor: 批量音频识别处理器 def __init__(self, config_path: str config.yaml): 初始化批量处理器 Args: config_path: 配置文件路径 # 加载配置 self.config self._load_config(config_path) # 设置日志 self._setup_logging() # 初始化路径 self._init_paths() # 延迟加载模型首次使用时加载 self.model None self.logger.info(批量处理器初始化完成) def _load_config(self, config_path: str) - Dict: 加载配置文件 try: with open(config_path, r, encodingutf-8) as f: config yaml.safe_load(f) # 设置默认值 defaults { paths: { input_dir: ./input_audio, output_dir: ./output_text, model_dir: ../models, log_dir: ./logs }, processing: { skip_existing: True, max_duration: 600 } } # 合并配置 for section, values in defaults.items(): if section not in config: config[section] values else: for key, value in values.items(): if key not in config[section]: config[section][key] value return config except FileNotFoundError: print(f配置文件 {config_path} 不存在使用默认配置) return defaults except Exception as e: print(f加载配置文件失败: {e}) return defaults def _setup_logging(self): 设置日志系统 log_dir Path(self.config[paths][log_dir]) log_dir.mkdir(exist_okTrue) # 创建日志文件名带时间戳 timestamp datetime.now().strftime(%Y%m%d_%H%M%S) log_file log_dir / fbatch_process_{timestamp}.log # 配置日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(log_file, encodingutf-8), logging.StreamHandler(sys.stdout) ] ) self.logger logging.getLogger(BatchProcessor) def _init_paths(self): 初始化路径并创建必要目录 paths self.config[paths] # 创建输入输出目录 Path(paths[input_dir]).mkdir(exist_okTrue) Path(paths[output_dir]).mkdir(exist_okTrue) self.logger.info(f输入目录: {paths[input_dir]}) self.logger.info(f输出目录: {paths[output_dir]}) def _load_model(self): 加载语音识别模型 if self.model is not None: return self.model try: # 导入原始工具的模型加载代码 # 这里需要根据实际的项目结构进行调整 from sensevoice_onnx_tool.model_loader import load_model model_config self.config[model] model load_model( model_dirself.config[paths][model_dir], model_namemodel_config[model_name], vad_modelmodel_config[vad_model], punc_modelmodel_config[punc_model], batch_sizemodel_config[batch_size], use_itnmodel_config[use_itn], languagemodel_config[language], deviceself.config[performance][device] ) self.model model self.logger.info(模型加载成功) return model except ImportError as e: self.logger.error(f导入模型失败: {e}) self.logger.info(请确保sensevoice_onnx_tool在Python路径中) raise except Exception as e: self.logger.error(f加载模型时出错: {e}) raise def get_audio_files(self) - List[Path]: 获取待处理的音频文件列表 input_dir Path(self.config[paths][input_dir]) supported_formats self.config[processing][supported_formats] audio_files [] for format_ext in supported_formats: audio_files.extend(input_dir.glob(f*{format_ext})) # 按文件名排序确保处理顺序一致 audio_files.sort() self.logger.info(f找到 {len(audio_files)} 个音频文件) return audio_files def process_single_audio(self, audio_path: Path) - Tuple[bool, str, Optional[str]]: 处理单个音频文件 Args: audio_path: 音频文件路径 Returns: (成功标志, 消息, 识别结果) try: # 检查文件是否存在 if not audio_path.exists(): return False, f文件不存在: {audio_path}, None # 检查文件大小 file_size audio_path.stat().st_size if file_size 0: return False, f文件为空: {audio_path}, None # 加载模型首次使用时加载 if self.model is None: self._load_model() # 执行识别 start_time time.time() # 这里调用原始工具的识别函数 # 需要根据实际项目结构调整 result self.model.transcribe(str(audio_path)) processing_time time.time() - start_time # 提取识别文本 if hasattr(result, text): text result.text elif isinstance(result, dict) and text in result: text result[text] else: text str(result) self.logger.info(f处理完成: {audio_path.name} ({processing_time:.2f}秒)) return True, f处理成功 ({processing_time:.2f}秒), text except Exception as e: error_msg f处理失败 {audio_path.name}: {str(e)} self.logger.error(error_msg) return False, error_msg, None def save_result(self, audio_path: Path, text: str): 保存识别结果到文件 try: # 创建对应的输出文件名 output_dir Path(self.config[paths][output_dir]) output_file output_dir / f{audio_path.stem}.txt # 写入文本 with open(output_file, w, encodingutf-8) as f: f.write(text) self.logger.debug(f结果已保存: {output_file}) except Exception as e: self.logger.error(f保存结果失败 {audio_path.name}: {e}) def process_batch(self, specific_files: Optional[List[str]] None): 批量处理音频文件 Args: specific_files: 指定要处理的文件列表可选 # 获取要处理的文件 if specific_files: audio_files [Path(self.config[paths][input_dir]) / f for f in specific_files] else: audio_files self.get_audio_files() if not audio_files: self.logger.warning(没有找到可处理的音频文件) return self.logger.info(f开始批量处理 {len(audio_files)} 个文件) # 统计信息 stats { total: len(audio_files), success: 0, failed: 0, skipped: 0, processing_time: 0 } # 处理报告 report [] # 使用进度条显示处理进度 with tqdm(totallen(audio_files), desc处理进度) as pbar: for audio_path in audio_files: # 检查是否跳过已处理文件 if self.config[processing][skip_existing]: output_file Path(self.config[paths][output_dir]) / f{audio_path.stem}.txt if output_file.exists(): stats[skipped] 1 report.append({ file: audio_path.name, status: skipped, message: 已存在输出文件 }) pbar.update(1) continue # 处理单个文件 start_time time.time() success, message, text self.process_single_audio(audio_path) processing_time time.time() - start_time stats[processing_time] processing_time if success: stats[success] 1 # 保存结果 self.save_result(audio_path, text) report.append({ file: audio_path.name, status: success, message: message, processing_time: processing_time }) else: stats[failed] 1 report.append({ file: audio_path.name, status: failed, message: message }) pbar.update(1) pbar.set_postfix({ 成功: stats[success], 失败: stats[failed], 跳过: stats[skipped] }) # 生成处理报告 self._generate_report(stats, report) self.logger.info(批量处理完成) self.logger.info(f统计: 总计{stats[total]}个, 成功{stats[success]}个, f失败{stats[failed]}个, 跳过{stats[skipped]}个) self.logger.info(f总处理时间: {stats[processing_time]:.2f}秒) def _generate_report(self, stats: Dict, report: List[Dict]): 生成处理报告 try: timestamp datetime.now().strftime(%Y%m%d_%H%M%S) report_file Path(self.config[paths][log_dir]) / freport_{timestamp}.txt with open(report_file, w, encodingutf-8) as f: f.write( * 50 \n) f.write(批量音频识别处理报告\n) f.write(f生成时间: {datetime.now().strftime(%Y-%m-%d %H:%M:%S)}\n) f.write( * 50 \n\n) # 统计信息 f.write(【处理统计】\n) f.write(f总文件数: {stats[total]}\n) f.write(f成功处理: {stats[success]}\n) f.write(f处理失败: {stats[failed]}\n) f.write(f跳过文件: {stats[skipped]}\n) f.write(f总耗时: {stats[processing_time]:.2f}秒\n) if stats[total] 0: avg_time stats[processing_time] / (stats[success] stats[failed]) f.write(f平均每个文件: {avg_time:.2f}秒\n) f.write(\n * 50 \n\n) # 详细报告 f.write(【详细处理记录】\n) for i, item in enumerate(report, 1): f.write(f{i}. {item[file]}\n) f.write(f 状态: {item[status]}\n) f.write(f 信息: {item[message]}\n) if processing_time in item: f.write(f 耗时: {item[processing_time]:.2f}秒\n) f.write(\n) self.logger.info(f处理报告已生成: {report_file}) except Exception as e: self.logger.error(f生成报告失败: {e}) def monitor_folder(self, interval: int 10): 监控文件夹自动处理新文件 Args: interval: 检查间隔秒 try: from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler class AudioFileHandler(FileSystemEventHandler): def __init__(self, processor): self.processor processor self.processed_files set() def on_created(self, event): if not event.is_directory: file_path Path(event.src_path) if file_path.suffix.lower() in self.processor.config[processing][supported_formats]: if str(file_path) not in self.processed_files: self.processor.logger.info(f检测到新文件: {file_path.name}) self.processor.process_batch([file_path.name]) self.processed_files.add(str(file_path)) input_dir Path(self.config[paths][input_dir]) event_handler AudioFileHandler(self) observer Observer() observer.schedule(event_handler, str(input_dir), recursiveFalse) observer.start() self.logger.info(f开始监控文件夹: {input_dir}) self.logger.info(f按 CtrlC 停止监控) try: while True: time.sleep(interval) except KeyboardInterrupt: observer.stop() observer.join() except ImportError: self.logger.warning(未安装watchdog无法使用文件夹监控功能) self.logger.info(请运行: pip install watchdog) except Exception as e: self.logger.error(f文件夹监控失败: {e}) def main(): 主函数 import argparse parser argparse.ArgumentParser(description批量音频识别处理器) parser.add_argument(--config, defaultconfig.yaml, help配置文件路径) parser.add_argument(--files, nargs, help指定要处理的文件列表) parser.add_argument(--monitor, actionstore_true, help启用文件夹监控模式) parser.add_argument(--interval, typeint, default10, help监控间隔秒) args parser.parse_args() # 创建处理器实例 processor BatchAudioProcessor(args.config) if args.monitor: # 监控模式 processor.monitor_folder(args.interval) elif args.files: # 处理指定文件 processor.process_batch(args.files) else: # 处理所有文件 processor.process_batch() if __name__ __main__: main()这个批量处理器类包含了所有核心功能从文件扫描、单个文件处理、批量处理到结果保存和报告生成一应俱全。3.3 与原始工具的集成由于原始SenseVoice-Small ONNX工具可能采用不同的代码结构我们需要创建一个适配层。在项目根目录创建model_adapter.py文件# model_adapter.py import sys from pathlib import Path # 添加原始工具路径 original_tool_path Path(__file__).parent.parent / sensevoice_onnx_tool sys.path.append(str(original_tool_path)) def load_model(model_dir, model_name, vad_model, punc_model, batch_size, use_itn, language, device): 适配函数根据实际项目结构调整 这里需要根据原始工具的实际代码结构进行调整 try: # 方法1如果原始工具提供了直接的导入方式 from pipeline import ASRService # 初始化ASR服务 asr_service ASRService( model_dirmodel_dir, model_namemodel_name, vad_modelvad_model, punc_modelpunc_model, batch_sizebatch_size, use_itnuse_itn, languagelanguage, devicedevice ) return asr_service except ImportError: try: # 方法2如果原始工具使用不同的结构 from funasr import AutoModel # 使用FunASR的AutoModel model AutoModel( modelmodel_name, model_revisionv2.0.4, vad_modelvad_model, vad_model_revisionv2.0.4, punc_modelpunc_model, punc_model_revisionv2.0.4, devicedevice ) # 包装成统一的接口 class ModelWrapper: def __init__(self, model, use_itn, language): self.model model self.use_itn use_itn self.language language def transcribe(self, audio_path): # 调用FunASR的生成接口 result self.model.generate( inputaudio_path, cache{}, languageself.language, use_itnself.use_itn ) return result return ModelWrapper(model, use_itn, language) except Exception as e: print(f加载模型失败请检查原始工具结构: {e}) print(可能需要根据实际项目结构调整 model_adapter.py) raise # 测试函数 def test_connection(): 测试与原始工具的连接 try: # 尝试导入原始工具 import sensevoice_onnx_tool print(✓ 成功导入原始工具) return True except ImportError: print(✗ 无法导入原始工具) print(请确保原始工具在Python路径中) return False if __name__ __main__: test_connection()这个适配层的作用是桥接批量处理器和原始工具确保无论原始工具的具体实现如何批量处理器都能正常调用。4. 使用指南与实战演示4.1 基础使用方法现在让我们看看如何使用这个批量处理脚本。首先确保你的目录结构是这样的your_project/ ├── sensevoice_onnx_tool/ # 原始Web工具 │ ├── app.py │ ├── model_loader.py │ └── ... └── sensevoice-batch-processor/ # 批量处理工具 ├── batch_processor.py ├── model_adapter.py ├── config.yaml ├── input_audio/ ├── output_text/ └── logs/4.1.1 准备音频文件将要处理的音频文件放入input_audio文件夹# 复制或移动音频文件到输入目录 cp /path/to/your/audio/*.mp3 ./input_audio/ cp /path/to/your/audio/*.wav ./input_audio/ # 查看文件列表 ls -la ./input_audio/4.1.2 运行批量处理最简单的使用方式直接运行脚本处理所有音频文件python batch_processor.py你会看到类似这样的输出2024-01-15 10:30:25 - BatchProcessor - INFO - 批量处理器初始化完成 2024-01-15 10:30:25 - BatchProcessor - INFO - 输入目录: ./input_audio 2024-01-15 10:30:25 - BatchProcessor - INFO - 输出目录: ./output_text 2024-01-15 10:30:25 - BatchProcessor - INFO - 找到 8 个音频文件 2024-01-15 10:30:25 - BatchProcessor - INFO - 开始批量处理 8 个文件 处理进度: 100%|██████████| 8/8 [02:1500:00, 16.92s/it, 成功8, 失败0, 跳过0] 2024-01-15 10:32:40 - BatchProcessor - INFO - 处理报告已生成: ./logs/report_20240115_103240.txt 2024-01-15 10:32:40 - BatchProcessor - INFO - 批量处理完成 2024-01-15 10:32:40 - BatchProcessor - INFO - 统计: 总计8个, 成功8个, 失败0个, 跳过0个 2024-01-15 10:32:40 - BatchProcessor - INFO - 总处理时间: 135.42秒4.1.3 查看处理结果处理完成后所有识别结果会保存在output_text文件夹中# 查看输出文件 ls -la ./output_text/ # 查看某个文件的识别结果 cat ./output_text/meeting_20240115.txt同时详细的处理报告会保存在logs文件夹中# 查看处理报告 cat ./logs/report_20240115_103240.txt报告内容示例 批量音频识别处理报告 生成时间: 2024-01-15 10:32:40 【处理统计】 总文件数: 8 成功处理: 8 处理失败: 0 跳过文件: 0 总耗时: 135.42秒 平均每个文件: 16.93秒 【详细处理记录】 1. meeting_1.mp3 状态: success 信息: 处理成功 (15.23秒) 耗时: 15.23秒 2. interview_2.wav 状态: success 信息: 处理成功 (18.45秒) 耗时: 18.45秒 ... 更多文件记录 ...4.2 高级使用技巧4.2.1 处理指定文件如果你只想处理特定的几个文件可以使用--files参数python batch_processor.py --files meeting_1.mp3 interview_2.wav4.2.2 文件夹监控模式对于需要持续处理新文件的场景可以使用监控模式。脚本会自动检测input_audio文件夹中的新文件并立即处理python batch_processor.py --monitor --interval 30这会在后台运行每30秒检查一次文件夹。当有新音频文件放入时会自动处理并保存结果。4.2.3 自定义配置文件如果你想调整处理参数可以修改config.yaml文件或者使用不同的配置文件# 使用自定义配置文件 cp config.yaml my_config.yaml # 编辑my_config.yaml调整参数 python batch_processor.py --config my_config.yaml常用的配置调整包括修改batch_size提高处理速度需要更多内存调整num_workers使用多核CPU并行处理修改supported_formats支持更多音频格式调整max_duration限制处理时长4.2.4 跳过已处理文件默认情况下脚本会跳过已经存在输出文件的音频避免重复处理。如果你需要重新处理所有文件可以在配置文件中设置# config.yaml processing: skip_existing: false # 改为false会重新处理所有文件4.3 实战案例会议录音批量整理让我们通过一个实际案例来看看这个脚本如何提升工作效率。场景你每周有5个团队会议每个会议约60分钟需要整理会议纪要。传统方式找到会议录音文件5个打开Web工具上传第一个文件等待识别完成约15分钟复制识别结果到文档重复步骤2-4处理剩余4个文件整理格式添加标题和时间戳总耗时约75-90分钟使用批量脚本将所有会议录音放入input_audio文件夹运行命令python batch_processor.py去喝杯咖啡脚本自动处理所有文件处理完成后所有文本已在output_text文件夹简单整理格式即可总耗时约15分钟处理时间 5分钟整理时间 20分钟效率提升3-4倍5. 常见问题与解决方案5.1 模型加载失败问题运行脚本时提示模型加载失败。解决方案检查原始工具路径是否正确确保模型文件已下载到指定目录检查config.yaml中的model_dir配置# 可以添加调试代码检查路径 import os print(当前工作目录:, os.getcwd()) print(模型目录:, os.path.abspath(../models))5.2 内存不足问题问题处理大文件或多个文件时内存不足。解决方案在配置中减小batch_size默认为1限制单个文件时长在配置中设置max_duration分批处理文件# config.yaml model: batch_size: 1 # 减小批处理大小 processing: max_duration: 300 # 限制为5分钟5.3 音频格式不支持问题某些音频文件无法识别。解决方案检查文件格式是否在支持列表中添加新的格式支持使用工具预先转换格式# config.yaml processing: supported_formats: [.wav, .mp3, .m4a, .ogg, .flac, .aac] # 添加.aac支持5.4 处理速度慢问题处理速度不如预期。优化建议使用GPU加速如果有调整num_workers使用多核优化音频文件降低采样率、单声道# config.yaml performance: device: cuda # 使用GPU num_workers: 4 # 使用4个CPU核心6. 功能扩展建议6.1 添加格式转换功能如果遇到不支持的音频格式可以集成格式转换功能# 在batch_processor.py中添加 import subprocess def convert_audio_format(self, input_path: Path, output_format: str wav): 转换音频格式 output_path input_path.with_suffix(f.{output_format}) try: # 使用ffmpeg转换 cmd [ ffmpeg, -i, str(input_path), -acodec, pcm_s16le, -ar, 16000, -ac, 1, str(output_path) ] subprocess.run(cmd, checkTrue, capture_outputTrue) return output_path except subprocess.CalledProcessError as e: self.logger.error(f格式转换失败 {input_path.name}: {e}) return None6.2 添加结果后处理可以在保存结果前进行一些后处理def post_process_text(self, text: str, audio_name: str) - str: 后处理识别文本 # 添加文件名作为标题 processed f【{audio_name}】\n\n # 清理多余的空行 lines [line.strip() for line in text.split(\n) if line.strip()] processed \n.join(lines) # 添加时间戳 timestamp datetime.now().strftime(%Y-%m-%d %H:%M:%S) processed f\n\n---\n识别时间: {timestamp} return processed6.3 添加邮件通知功能对于长时间运行的批量任务可以添加完成通知import smtplib from email.mime.text import MIMEText def send_notification(self, stats: Dict, report_file: Path): 发送处理完成通知 try: # 配置邮件信息 msg MIMEText(f批量处理完成\n\n成功: {stats[success]}\n失败: {stats[failed]}) msg[Subject] 音频批量识别处理完成 msg[From] your_emailexample.com msg[To] recipientexample.com # 发送邮件 with smtplib.SMTP(smtp.example.com, 587) as server: server.starttls() server.login(username, password) server.send_message(msg) self.logger.info(通知邮件已发送) except Exception as e: self.logger.error(f发送通知失败: {e})7. 总结通过本教程我们成功开发了一个功能完整的SenseVoice-Small ONNX批量音频识别脚本。这个工具不仅解决了单个文件处理的效率问题还提供了丰富的功能和灵活的配置选项。主要收获效率大幅提升从手动一个个处理到一键批量处理工作效率提升数倍灵活配置通过配置文件可以轻松调整各种参数适应不同需求完善的功能支持进度显示、结果保存、处理报告、错误处理等易于扩展模块化设计方便添加新功能如格式转换、后处理等生产就绪包含日志系统、错误处理、中断恢复等生产环境需要的功能使用建议对于常规使用直接运行python batch_processor.py即可对于持续收集音频的场景使用--monitor监控模式定期检查logs文件夹中的报告了解处理情况根据硬件性能调整配置参数获得最佳效果这个批量处理脚本可以很好地与原有的Web界面工具互补。Web界面适合交互式使用和实时预览而批量脚本适合处理大量文件的后台任务。两者结合构成了一个完整的语音识别解决方案。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。