2. 基于WebRTC VAD的语音活动检测集成
任务目标
在关键词唤醒（KWS）流水线前端集成WebRTC VAD（语音活动检测），对音频帧进行实时静音/语音判别，仅将包含语音的帧送入后续唤醒引擎，从而过滤静音段、降低误唤醒率。本案例实现一个可独立运行的Python模块，读取PCM音频文件，输出每帧的VAD判定结果及过滤后的语音帧索引。
环境依赖
- Python 3.8+
webrtcvad（pip 安装：pip install webrtcvad）- 仅需标准库
wave、struct处理音频读取
核心实现
以下代码实现完整闭环：读取16kHz单声道16-bit PCM文件，按20ms帧长（320样本/帧）切分，逐帧调用VAD判定，输出语音帧索引列表，并支持将过滤后的语音段拼接保存为新PCM文件。
import wave
import struct
import webrtcvad
from typing import List, Tuple
class WebRTCVADProcessor:
"""基于WebRTC VAD的语音活动检测器，用于KWS前端静音过滤。"""
def __init__(self, aggressiveness: int = 2, sample_rate: int = 16000, frame_ms: int = 20):
"""
Args:
aggressiveness: VAD激进程度 0~3，值越大越激进（更少语音被判为语音）
sample_rate: 音频采样率，仅支持8000/16000/32000/48000
frame_ms: 帧长，仅支持10/20/30ms
"""
if sample_rate not in (8000, 16000, 32000, 48000):
raise ValueError("sample_rate must be 8000/16000/32000/48000")
if frame_ms not in (10, 20, 30):
raise ValueError("frame_ms must be 10/20/30")
self.vad = webrtcvad.Vad(aggressiveness)
self.sample_rate = sample_rate
self.frame_len = int(sample_rate * frame_ms / 1000) # 每帧样本数
self.frame_bytes = self.frame_len * 2 # 16-bit = 2字节
def read_pcm(self, filepath: str) -> Tuple[List[int], int]:
"""读取WAV文件，返回单声道PCM样本列表和采样率。"""
with wave.open(filepath, 'rb') as wf:
assert wf.getnchannels() == 1, "仅支持单声道"
assert wf.getsampwidth() == 2, "仅支持16-bit"
assert wf.getframerate() == self.sample_rate, f"采样率需为{self.sample_rate}"
raw = wf.readframes(wf.getnframes())
samples = struct.unpack(f'<{len(raw)//2}h', raw)
return list(samples), wf.getframerate()
def frame_generator(self, samples: List[int]):
"""将样本列表按帧长切分为字节块。"""
n_frames = len(samples) // self.frame_len
for i in range(n_frames):
start = i * self.frame_len
end = start + self.frame_len
frame = samples[start:end]
# 转为16-bit little-endian字节
yield struct.pack(f'<{self.frame_len}h', *frame)
def detect_speech_frames(self, samples: List[int]) -> List[int]:
"""
对每个音频帧执行VAD判定。
Returns:
语音帧索引列表（0-based）
"""
speech_frames = []
for idx, frame_bytes in enumerate(self.frame_generator(samples)):
is_speech = self.vad.is_speech(frame_bytes, self.sample_rate)
if is_speech:
speech_frames.append(idx)
return speech_frames
def filter_silence(self, input_wav: str, output_wav: str = None) -> Tuple[List[int], List[int]]:
"""
主处理入口：读取WAV，检测语音帧，可选保存过滤后的音频。
Returns:
(语音帧索引列表, 全部样本列表)
"""
samples, _ = self.read_pcm(input_wav)
speech_idx = self.detect_speech_frames(samples)
if output_wav:
# 拼接所有语音帧为连续音频
speech_samples = []
for idx in speech_idx:
start = idx * self.frame_len
end = start + self.frame_len
speech_samples.extend(samples[start:end])
with wave.open(output_wav, 'wb') as wf:
wf.setnchannels(1)
wf.setsampwidth(2)
wf.setframerate(self.sample_rate)
wf.writeframes(struct.pack(f'<{len(speech_samples)}h', *speech_samples))
return speech_idx, samples
# 最小调用入口
if __name__ == "__main__":
import sys
if len(sys.argv) < 2:
print("用法: python vad_kws.py <input.wav> [output.wav]")
print("示例: python vad_kws.py test.wav filtered.wav")
sys.exit(1)
input_path = sys.argv[1]
output_path = sys.argv[2] if len(sys.argv) > 2 else None
processor = WebRTCVADProcessor(aggressiveness=2)
speech_idx, _ = processor.filter_silence(input_path, output_path)
# 输出统计信息
total_frames = len(speech_idx) # 实际总帧数需从文件计算
# 重新计算总帧数以获得准确统计
samples, _ = processor.read_pcm(input_path)
total_frames = len(samples) // processor.frame_len
speech_ratio = len(speech_idx) / total_frames * 100 if total_frames > 0 else 0
print(f"总帧数: {total_frames}")
print(f"语音帧数: {len(speech_idx)}")
print(f"语音占比: {speech_ratio:.1f}%")
print(f"静音帧数: {total_frames - len(speech_idx)}")
if output_path:
print(f"过滤后音频已保存: {output_path}")
运行与验证
- 准备测试音频：准备一段包含语音和静音的16kHz单声道16-bit WAV文件（如
test.wav），时长建议5～10秒。 - 运行检测：
python vad_kws.py test.wav filtered.wav- 验证通过条件：
- 程序正常输出总帧数、语音帧数、语音占比，无异常报错；
- 语音占比应在合理范围（如30%～80%），若测试文件包含明显静音段，语音占比应显著小于100%；
- 生成的
filtered.wav可正常播放，听感上首尾静音被去除，语音内容连续完整； - 对比原始文件，过滤后文件时长明显缩短（若原文件含静音段）。
KWS集成说明
在实际KWS系统中，将本模块的 detect_speech_frames 逻辑嵌入实时音频流处理：每次从麦克风获取20ms音频帧，调用 vad.is_speech() 判定，仅当返回True时才将该帧送入唤醒引擎的特征提取与匹配模块。激进程度参数 aggressiveness 建议在真实唤醒场景中调参：值越大，对静音的过滤越强，但可能误切语音起始段；值越小则越保守，适合噪声环境。推荐初始值2，再根据误唤醒率与召回率平衡调整。