# FSMN VAD Python调用示例：脱离WebUI的代码实例 ## 1. 引言 如果你正在寻找一个高效、精准的语音活动检测（VAD）解决方案，阿里达摩院开源的FSMN VAD模型绝对值得一试。这个模型只有1.7MB大小，却能实现工业级的语音检测精度，实时率（RTF）低至0.03，意味着处理速度是实时的33倍。 你可能已经用过它的WebUI界面，上传文件、点击按钮就能得到结果，确实很方便。但在实际项目中，我们往往需要将VAD功能集成到自己的应用里，或者批量处理成千上万的音频文件，这时候WebUI就显得不够灵活了。 今天，我就带你彻底摆脱WebUI的束缚，直接用Python代码调用FSMN VAD模型。我会分享完整的代码示例，从环境搭建到实际调用，一步步教你如何在自己的项目中集成这个强大的语音检测工具。 ## 2. 环境准备与模型下载 ### 2.1 安装必要的Python包 首先，你需要确保Python环境是3.8或更高版本。然后安装几个核心的依赖包： ```bash pip install torch torchaudio pip install funasr pip install soundfile pip install numpy ``` 这里简单说明一下每个包的作用： - `torch`和`torchaudio`：PyTorch框架，FSMN VAD基于此构建 - `funasr`：阿里达摩院的语音识别工具包，包含了VAD模型 - `soundfile`：用于读取音频文件 - `numpy`：数值计算基础库 ### 2.2 下载FSMN VAD模型 FSMN VAD模型已经集成在funasr包中，但我们需要手动下载模型文件。模型文件不大，只有1.7MB，下载很快： ```python from funasr import AutoModel # 自动下载并加载模型 model = AutoModel(model="fsmn-vad") ``` 第一次运行时会自动从阿里云下载模型文件，下载完成后会保存在本地缓存中，下次就不需要再下载了。 如果你需要指定模型保存路径，可以这样： ```python model = AutoModel( model="fsmn-vad", model_revision="v2.0.4", cache_dir="./models" # 指定模型保存目录 ) ``` ## 3. 基础调用：单个音频文件处理 ### 3.1 最简单的调用方式 让我们从一个最简单的例子开始。假设你有一个WAV格式的音频文件，想要检测其中的语音片段： ```python import soundfile as sf from funasr import AutoModel def detect_speech_segments(audio_path): """ 检测音频文件中的语音片段 参数: audio_path: 音频文件路径 返回: 语音片段列表，每个片段包含开始时间、结束时间和置信度 """ # 加载模型（第一次运行会自动下载） model = AutoModel(model="fsmn-vad") # 读取音频文件 # 注意：模型要求音频为16kHz采样率，单声道 audio, sample_rate = sf.read(audio_path) # 如果音频不是16kHz，需要重采样 if sample_rate != 16000: print(f"警告：音频采样率为{sample_rate}Hz，建议转换为16kHz") # 执行VAD检测 result = model.generate( input=audio_path, batch_size=1, vad_params={ "max_end_silence_time": 800, # 尾部静音阈值，单位毫秒 "speech_noise_thres": 0.6 # 语音-噪声阈值 } ) return result[0]["value"] # 使用示例 if __name__ == "__main__": # 替换为你的音频文件路径 audio_file = "meeting_recording.wav" segments = detect_speech_segments(audio_file) print(f"检测到 {len(segments)} 个语音片段：") for i, seg in enumerate(segments, 1): start_sec = seg["start"] / 1000 # 转换为秒 end_sec = seg["end"] / 1000 duration = (seg["end"] - seg["start"]) / 1000 print(f"片段 {i}:") print(f" 开始时间: {start_sec:.2f}秒") print(f" 结束时间: {end_sec:.2f}秒") print(f" 持续时间: {duration:.2f}秒") print(f" 置信度: {seg['confidence']:.2f}") print() ``` ### 3.2 处理不同格式的音频 实际项目中，你可能会遇到各种格式的音频文件。下面是一个更通用的函数，支持多种格式： ```python import os import subprocess import tempfile from pathlib import Path def convert_to_wav(input_path, output_dir=None): """ 将音频文件转换为WAV格式（16kHz, 单声道） 参数: input_path: 输入音频文件路径 output_dir: 输出目录，默认为临时目录 返回: 转换后的WAV文件路径 """ if output_dir is None: output_dir = tempfile.gettempdir() # 生成输出文件名 input_name = Path(input_path).stem output_path = os.path.join(output_dir, f"{input_name}_converted.wav") # 使用ffmpeg进行转换 cmd = [ "ffmpeg", "-i", input_path, "-ar", "16000", # 采样率16kHz "-ac", "1", # 单声道 "-acodec", "pcm_s16le", # 16位PCM编码 "-y", # 覆盖已存在文件 output_path ] try: subprocess.run(cmd, check=True, capture_output=True) return output_path except subprocess.CalledProcessError as e: print(f"音频转换失败: {e}") return None except FileNotFoundError: print("错误：未找到ffmpeg，请先安装ffmpeg") return None def detect_speech_from_any_format(audio_path): """ 支持多种格式的语音检测 参数: audio_path: 音频文件路径，支持wav, mp3, flac, ogg等 返回: 语音片段列表 """ # 检查文件扩展名 ext = Path(audio_path).suffix.lower() # 如果是WAV文件且符合要求，直接处理 if ext == ".wav": # 可以添加更多检查，如采样率、声道数等 wav_path = audio_path else: # 转换为WAV格式 print(f"检测到{ext}格式，正在转换为WAV...") wav_path = convert_to_wav(audio_path) if wav_path is None: print("音频转换失败，无法进行VAD检测") return [] # 进行VAD检测 model = AutoModel(model="fsmn-vad") result = model.generate( input=wav_path, batch_size=1, vad_params={ "max_end_silence_time": 800, "speech_noise_thres": 0.6 } ) # 如果是转换的文件，清理临时文件 if wav_path != audio_path and os.path.exists(wav_path): os.remove(wav_path) return result[0]["value"] ``` ## 4. 高级功能：参数调优与批量处理 ### 4.1 理解并调优VAD参数 FSMN VAD有两个核心参数，理解它们对于获得最佳检测效果至关重要： ```python def vad_with_custom_params(audio_path, max_end_silence=800, speech_threshold=0.6): """ 使用自定义参数进行VAD检测 参数: audio_path: 音频文件路径 max_end_silence: 尾部静音阈值（毫秒） 控制语音结束的判定 值越大，越不容易截断语音 值越小，语音片段切分越细 speech_threshold: 语音-噪声阈值 控制语音和噪声的判定 值越大，对语音判定越严格 值越小，越容易将噪声判定为语音 返回: 语音片段列表 """ model = AutoModel(model="fsmn-vad") result = model.generate( input=audio_path, batch_size=1, vad_params={ "max_end_silence_time": max_end_silence, "speech_noise_thres": speech_threshold } ) return result[0]["value"] # 不同场景的参数建议 def get_recommended_params(scenario): """ 根据不同场景推荐参数 参数: scenario: 场景类型 "meeting" - 会议录音 "phone" - 电话录音 "interview" - 访谈录音 "lecture" - 讲座录音 返回: (max_end_silence, speech_threshold) 参数对 """ recommendations = { "meeting": (1000, 0.6), # 会议：适当增大静音阈值，避免截断发言 "phone": (800, 0.7), # 电话：中等静音阈值，提高噪声阈值过滤电话噪声 "interview": (1200, 0.5), # 访谈：较大静音阈值，宽松语音判定 "lecture": (1500, 0.6), # 讲座：最大静音阈值，适合连续演讲 "fast_talk": (600, 0.6), # 快速对话：较小静音阈值，细分语音片段 "noisy": (800, 0.7), # 嘈杂环境：提高噪声阈值 "quiet": (800, 0.5), # 安静环境：降低噪声阈值 } return recommendations.get(scenario, (800, 0.6)) # 使用示例 if __name__ == "__main__": audio_file = "recording.wav" # 会议场景 meeting_params = get_recommended_params("meeting") meeting_segments = vad_with_custom_params(audio_file, *meeting_params) print(f"会议场景参数：静音阈值={meeting_params[0]}ms, 语音阈值={meeting_params[1]}") print(f"检测到 {len(meeting_segments)} 个语音片段") # 电话场景 phone_params = get_recommended_params("phone") phone_segments = vad_with_custom_params(audio_file, *phone_params) print(f"\n电话场景参数：静音阈值={phone_params[0]}ms, 语音阈值={phone_params[1]}") print(f"检测到 {len(phone_segments)} 个语音片段") ``` ### 4.2 批量处理音频文件 在实际应用中，我们经常需要批量处理多个音频文件。下面是一个完整的批量处理示例： ```python import json import time from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path from tqdm import tqdm # 进度条库，需要安装：pip install tqdm class BatchVADProcessor: """批量VAD处理器""" def __init__(self, model_path="fsmn-vad", max_workers=4): """ 初始化批量处理器 参数: model_path: 模型路径或名称 max_workers: 最大并发工作线程数 """ self.model = AutoModel(model=model_path) self.max_workers = max_workers def process_single_file(self, audio_path, params=None): """ 处理单个音频文件 参数: audio_path: 音频文件路径 params: VAD参数字典，默认为None使用默认参数 返回: (文件路径, 处理结果, 错误信息) """ try: if params is None: params = { "max_end_silence_time": 800, "speech_noise_thres": 0.6 } result = self.model.generate( input=str(audio_path), batch_size=1, vad_params=params ) return (audio_path, result[0]["value"], None) except Exception as e: return (audio_path, [], str(e)) def process_batch(self, audio_files, output_dir="results", params=None): """ 批量处理音频文件 参数: audio_files: 音频文件路径列表 output_dir: 结果输出目录 params: VAD参数，默认为None使用默认参数 返回: 处理统计信息 """ # 创建输出目录 Path(output_dir).mkdir(parents=True, exist_ok=True) # 准备结果存储 all_results = {} errors = [] # 使用线程池并发处理 print(f"开始批量处理 {len(audio_files)} 个文件...") start_time = time.time() with ThreadPoolExecutor(max_workers=self.max_workers) as executor: # 提交所有任务 future_to_file = { executor.submit(self.process_single_file, file, params): file for file in audio_files } # 使用tqdm显示进度 with tqdm(total=len(audio_files), desc="处理进度") as pbar: for future in as_completed(future_to_file): file_path = future_to_file[future] try: file_path, segments, error = future.result() if error: errors.append((file_path, error)) else: all_results[str(file_path)] = segments # 保存单个文件的结果 self._save_single_result( file_path, segments, output_dir ) except Exception as e: errors.append((file_path, str(e))) pbar.update(1) # 保存汇总结果 self._save_summary(all_results, errors, output_dir) # 计算统计信息 processing_time = time.time() - start_time stats = self._calculate_statistics(all_results, processing_time) return stats def _save_single_result(self, file_path, segments, output_dir): """保存单个文件的结果""" file_name = Path(file_path).stem result_file = Path(output_dir) / f"{file_name}_vad.json" result_data = { "file": str(file_path), "segments": segments, "segment_count": len(segments), "total_duration": sum( (seg["end"] - seg["start"]) / 1000 for seg in segments ) if segments else 0 } with open(result_file, 'w', encoding='utf-8') as f: json.dump(result_data, f, ensure_ascii=False, indent=2) def _save_summary(self, results, errors, output_dir): """保存汇总结果""" summary = { "total_files": len(results) + len(errors), "successful_files": len(results), "failed_files": len(errors), "results": results, "errors": errors, "timestamp": time.strftime("%Y-%m-%d %H:%M:%S") } summary_file = Path(output_dir) / "batch_summary.json" with open(summary_file, 'w', encoding='utf-8') as f: json.dump(summary, f, ensure_ascii=False, indent=2) def _calculate_statistics(self, results, processing_time): """计算统计信息""" total_segments = 0 total_duration = 0 for segments in results.values(): total_segments += len(segments) total_duration += sum( (seg["end"] - seg["start"]) / 1000 for seg in segments ) avg_segments_per_file = ( total_segments / len(results) if results else 0 ) return { "total_files_processed": len(results), "total_segments_detected": total_segments, "total_speech_duration": total_duration, "average_segments_per_file": avg_segments_per_file, "processing_time_seconds": processing_time, "files_per_second": len(results) / processing_time if processing_time > 0 else 0 } # 使用示例 if __name__ == "__main__": # 1. 初始化处理器 processor = BatchVADProcessor(max_workers=2) # 2. 准备音频文件列表 audio_dir = Path("./audio_files") audio_files = list(audio_dir.glob("*.wav")) + list(audio_dir.glob("*.mp3")) print(f"找到 {len(audio_files)} 个音频文件") # 3. 设置处理参数（可选） custom_params = { "max_end_silence_time": 1000, # 会议场景 "speech_noise_thres": 0.6 } # 4. 批量处理 stats = processor.process_batch( audio_files=audio_files, output_dir="./vad_results", params=custom_params ) # 5. 打印统计信息 print("\n批量处理完成！") print(f"处理文件数: {stats['total_files_processed']}") print(f"检测到语音片段: {stats['total_segments_detected']} 个") print(f"总语音时长: {stats['total_speech_duration']:.2f} 秒") print(f"平均每个文件片段数: {stats['average_segments_per_file']:.1f}") print(f"处理时间: {stats['processing_time_seconds']:.2f} 秒") print(f"处理速度: {stats['files_per_second']:.2f} 文件/秒") ``` ## 5. 实际应用：语音片段提取与处理 ### 5.1 提取检测到的语音片段 检测到语音片段后，我们通常需要将这些片段从原始音频中提取出来： ```python import soundfile as sf import numpy as np from pydub import AudioSegment # 需要安装：pip install pydub def extract_speech_segments(audio_path, segments, output_dir="extracted"): """ 从音频中提取语音片段 参数: audio_path: 原始音频文件路径 segments: VAD检测到的语音片段列表 output_dir: 输出目录 返回: 提取的文件路径列表 """ # 创建输出目录 Path(output_dir).mkdir(parents=True, exist_ok=True) # 加载原始音频 try: # 使用pydub加载音频（支持更多格式） audio = AudioSegment.from_file(audio_path) except: # 如果pydub失败，尝试soundfile audio_data, sample_rate = sf.read(audio_path) # 将numpy数组转换为pydub格式 audio = AudioSegment( audio_data.tobytes(), frame_rate=sample_rate, sample_width=audio_data.dtype.itemsize, channels=1 if len(audio_data.shape) == 1 else audio_data.shape[1] ) extracted_files = [] for i, seg in enumerate(segments, 1): # 转换为毫秒（pydub使用毫秒） start_ms = seg["start"] end_ms = seg["end"] # 提取片段（增加前后缓冲） buffer_ms = 100 # 前后各增加100毫秒缓冲 segment_start = max(0, start_ms - buffer_ms) segment_end = min(len(audio), end_ms + buffer_ms) speech_segment = audio[segment_start:segment_end] # 保存提取的片段 base_name = Path(audio_path).stem output_file = Path(output_dir) / f"{base_name}_segment_{i:03d}.wav" speech_segment.export( str(output_file), format="wav", parameters=["-ar", "16000", "-ac", "1"] ) extracted_files.append(str(output_file)) print(f"提取片段 {i}: {start_ms/1000:.2f}s - {end_ms/1000:.2f}s " f"(时长: {(end_ms-start_ms)/1000:.2f}s)") return extracted_files def create_audio_summary(audio_path, segments, output_file="summary.wav"): """ 创建音频摘要：将所有语音片段拼接成一个文件 参数: audio_path: 原始音频文件路径 segments: VAD检测到的语音片段列表 output_file: 输出文件路径 返回: 摘要文件路径 """ # 加载原始音频 audio = AudioSegment.from_file(audio_path) # 创建静音片段（用于片段间分隔） silence = AudioSegment.silent(duration=500) # 500毫秒静音 # 拼接所有语音片段 summary = AudioSegment.empty() for i, seg in enumerate(segments, 1): start_ms = seg["start"] end_ms = seg["end"] # 提取片段 speech_segment = audio[start_ms:end_ms] # 添加到摘要（第一个片段前不加静音） if i > 1: summary += silence summary += speech_segment print(f"添加片段 {i}: {start_ms/1000:.2f}s - {end_ms/1000:.2f}s") # 保存摘要文件 summary.export( output_file, format="wav", parameters=["-ar", "16000", "-ac", "1"] ) print(f"\n音频摘要已保存: {output_file}") print(f"原始时长: {len(audio)/1000:.2f}秒") print(f"摘要时长: {len(summary)/1000:.2f}秒") print(f"压缩比例: {len(summary)/len(audio)*100:.1f}%") return output_file # 使用示例 if __name__ == "__main__": # 1. 检测语音片段 audio_file = "long_recording.wav" model = AutoModel(model="fsmn-vad") result = model.generate( input=audio_file, batch_size=1, vad_params={"max_end_silence_time": 800, "speech_noise_thres": 0.6} ) segments = result[0]["value"] print(f"检测到 {len(segments)} 个语音片段") # 2. 提取所有语音片段 extracted = extract_speech_segments(audio_file, segments, "extracted_segments") print(f"已提取 {len(extracted)} 个片段到 extracted_segments/ 目录") # 3. 创建音频摘要 summary_file = create_audio_summary(audio_file, segments, "recording_summary.wav") ``` ### 5.2 实时音频流处理 虽然FSMN VAD主要设计用于离线处理，但我们也可以实现准实时的音频流处理： ```python import queue import threading import time from collections import deque class RealtimeVADProcessor: """实时VAD处理器""" def __init__(self, sample_rate=16000, chunk_duration=1.0): """ 初始化实时处理器 参数: sample_rate: 音频采样率（Hz） chunk_duration: 每次处理的音频块时长（秒） """ self.sample_rate = sample_rate self.chunk_size = int(sample_rate * chunk_duration) self.model = AutoModel(model="fsmn-vad") # 音频缓冲区 self.audio_buffer = deque(maxlen=10) # 保存最近10个块 self.result_queue = queue.Queue() # 处理线程 self.processing = False self.process_thread = None def start_processing(self): """开始处理""" self.processing = True self.process_thread = threading.Thread(target=self._process_loop) self.process_thread.start() print("实时VAD处理器已启动") def stop_processing(self): """停止处理""" self.processing = False if self.process_thread: self.process_thread.join() print("实时VAD处理器已停止") def add_audio_chunk(self, audio_data): """ 添加音频数据块 参数: audio_data: 音频数据（numpy数组） """ if len(audio_data) != self.chunk_size: print(f"警告：音频块大小应为{self.chunk_size}，实际为{len(audio_data)}") return self.audio_buffer.append(audio_data) def get_results(self): """获取处理结果""" results = [] while not self.result_queue.empty(): results.append(self.result_queue.get()) return results def _process_loop(self): """处理循环""" while self.processing: if len(self.audio_buffer) > 0: # 获取最新的音频块 audio_chunk = self.audio_buffer[-1] # 保存为临时文件进行处理 import tempfile import soundfile as sf with tempfile.NamedTemporaryFile(suffix='.wav', delete=False) as tmp: tmp_file = tmp.name sf.write(tmp_file, audio_chunk, self.sample_rate) try: # 进行VAD检测 result = self.model.generate( input=tmp_file, batch_size=1, vad_params={ "max_end_silence_time": 600, # 实时处理使用较小的值 "speech_noise_thres": 0.6 } ) segments = result[0]["value"] if segments: # 调整时间戳（相对于当前块） current_time = time.time() adjusted_segments = [] for seg in segments: adjusted_segments.append({ "start": seg["start"] / 1000, # 转换为秒 "end": seg["end"] / 1000, "confidence": seg["confidence"], "timestamp": current_time }) self.result_queue.put({ "chunk_time": current_time, "segments": adjusted_segments }) finally: # 清理临时文件 import os os.unlink(tmp_file) # 控制处理频率 time.sleep(0.1) # 模拟实时音频输入 def simulate_realtime_audio(audio_file, processor, duration=10): """ 模拟实时音频输入（用于测试） 参数: audio_file: 音频文件路径 processor: RealtimeVADProcessor实例 duration: 模拟时长（秒） """ import soundfile as sf # 加载音频文件 audio, sr = sf.read(audio_file) if sr != processor.sample_rate: print(f"采样率不匹配：文件{sr}Hz，处理器{processor.sample_rate}Hz") return # 分割为块 chunk_size = processor.chunk_size num_chunks = len(audio) // chunk_size print(f"开始模拟实时音频输入，共{num_chunks}个块...") for i in range(min(num_chunks, int(duration * processor.sample_rate / chunk_size))): # 获取当前块 start_idx = i * chunk_size end_idx = start_idx + chunk_size chunk = audio[start_idx:end_idx] # 添加到处理器 processor.add_audio_chunk(chunk) # 模拟实时延迟 time.sleep(1.0) # 每秒一个块 # 获取并显示结果 results = processor.get_results() if results: for result in results: print(f"时间 {result['chunk_time']:.1f}s: " f"检测到 {len(result['segments'])} 个语音片段") print("模拟结束") # 使用示例 if __name__ == "__main__": # 1. 创建实时处理器 vad_processor = RealtimeVADProcessor( sample_rate=16000, chunk_duration=1.0 # 1秒一个块 ) # 2. 启动处理器 vad_processor.start_processing() # 3. 模拟实时音频输入 try: simulate_realtime_audio("test_audio.wav", vad_processor, duration=30) except KeyboardInterrupt: print("\n用户中断") finally: # 4. 停止处理器 vad_processor.stop_processing() # 5. 显示最终结果 final_results = vad_processor.get_results() print(f"\n总共处理了 {len(final_results)} 个音频块") ``` ## 6. 总结 通过本文的示例代码，你已经掌握了如何脱离WebUI，直接使用Python调用FSMN VAD模型进行语音活动检测。让我们回顾一下关键要点： ### 6.1 核心收获 1. **环境搭建简单**：只需要安装几个Python包，模型会自动下载 2. **基础调用直接**：几行代码就能完成单个音频文件的VAD检测 3. **参数调优灵活**：通过调整`max_end_silence_time`和`speech_noise_thres`参数，可以适应不同场景 4. **批量处理高效**：利用多线程可以快速处理大量音频文件 5. **实际应用丰富**：不仅可以检测语音，还能提取片段、创建摘要，甚至实现准实时处理 ### 6.2 最佳实践建议 根据我的使用经验，给你几个实用建议： **参数选择方面**： - 会议录音：静音阈值1000-1500ms，语音阈值0.6 - 电话录音：静音阈值800ms，语音阈值0.7 - 快速对话：静音阈值500-700ms，语音阈值0.6 - 嘈杂环境：语音阈值提高到0.7-0.8 **性能优化方面**： - 批量处理时，根据CPU核心数设置合适的线程数（通常是CPU核心数的1-2倍） - 对于长时间音频，可以考虑分段处理以减少内存占用 - 如果处理速度是关键，可以尝试使用GPU加速（如果支持） **错误处理方面**： - 始终检查音频文件的采样率，确保是16kHz - 处理前验证文件格式，必要时进行转换 - 添加适当的异常处理，特别是文件IO操作 ### 6.3 扩展思路 掌握了基础用法后，你还可以进一步扩展： 1. **集成到现有系统**：将VAD功能集成到你的音频处理流水线中 2. **结合其他AI模型**：VAD检测后，可以将语音片段送入ASR（语音识别）模型进行转写 3. **开发Web服务**：基于Flask或FastAPI封装成REST API服务 4. **实时应用**：结合音频流，实现真正的实时语音活动检测 FSMN VAD虽然模型小巧，但在实际应用中表现非常出色。它的高精度和快速处理能力，使其成为各种语音处理任务的理想选择。希望本文的代码示例能帮助你在自己的项目中顺利集成这个强大的工具。 --- > **获取更多AI镜像** > > 想探索更多AI镜像和应用场景？访问 [CSDN星图镜像广场](https://ai.csdn.net/?utm_source=mirror_blog_end)，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。