VAD 语音活动检测在 LiveKit Agents
范围:VAD 模块的架构设计、Silero 实现、CPU 部署及与其他模块的协作
综合自:livekit-agents
优先级:P1
概述
VAD(Voice Activity Detection)是 LiveKit Agents 中用于检测用户语音活动的核心组件。在实时语音对话中,VAD 负责:
- 语音边界检测 - 检测用户何时开始说话、何时结束
- 语音活动监控 - 实时跟踪音频中的语音活动
- 打断检测 - 结合
AdaptiveInterruptionDetector实现用户打断
VAD 是实现自然语音对话的基础,它让 Agent 能够知道用户的会话轮次。
核心抽象层
类型体系
vad.py (核心抽象)
├── VADEventType (枚举)
│ ├── START_OF_SPEECH - 语音开始
│ ├── INFERENCE_DONE - 推理完成(每帧)
│ └── END_OF_SPEECH - 语音结束
├── VADEvent (数据类)
│ ├── type, timestamp, speech_duration, silence_duration
│ ├── frames, probability, inference_duration
│ └── speaking, raw_accumulated_silence/speech
├── VADCapabilities
│ └── update_interval: float (默认 0.032s)
└── VAD (ABC)
└── stream() -> VADStream
VADStream (ABC)
├── push_frame(frame) - 推送音频帧
├── flush() - 标记当前段结束
├── end_input() - 标记输入结束
├── aclose() - 异步关闭
└── __anext__() - 迭代获取 VADEvent
核心设计
异步流式处理:VADStream 继承自 AsyncIterator[VADEvent],通过 aio.Chan 实现异步事件通道。
指标收集:_metrics_monitor_task 收集推理延迟、推理次数等指标,通过 metrics_collected 事件发布。
# livekit/agents/vad.py:110-112
self._metrics_task = asyncio.create_task(
self._metrics_monitor_task(monitor_aiter), name="TTS._metrics_task"
)
Silero VAD 实现
目录结构
livekit-plugins-silero/
├── vad.py # VAD 和 VADStream 实现
├── onnx_model.py # ONNX 模型加载和推理
└── resources/
└── silero_vad.onnx # 模型文件
配置参数
| 参数 | 默认值 | 说明 |
|---|---|---|
min_speech_duration | 0.05s | 最小语音持续时间(防误检) |
min_silence_duration | 0.55s | 结束语音前的静音时长 |
prefix_padding_duration | 0.5s | 前缀填充(包含上下文) |
activation_threshold | 0.5 | 语音激活阈值 |
deactivation_threshold | 0.35 | 语音停阈值 |
max_buffered_speech | 60.0s | 最大缓冲语音时长 |
sample_rate | 16000 | 仅支持 8k/16k |
force_cpu | True | 默认 CPU 推理 |
推理流程
# vad.py:291-553 (_main_task)
async for input_frame in self._input_ch:
# 1. 重采样(如需要)
if self._input_sample_rate != self._opts.sample_rate:
resampler = rtc.AudioResampler(...)
# 2. 累积样本直到满足窗口大小
while available_inference_samples >= self._model.window_size_samples:
# 3. 运行 ONNX 推理
p = await self._loop.run_in_executor(None, self._model, inference_f32_data)
p = self._exp_filter.apply(exp=1.0, sample=p) # 指数平滑
# 4. 状态机判断
if p >= activation_threshold:
speech_threshold_duration += window_duration
if speech_threshold_duration >= min_speech_duration:
# 触发 START_OF_SPEECH
else:
silence_threshold_duration += window_duration
if silence_threshold_duration >= min_silence_duration:
# 触发 END_OF_SPEECH
性能指标
| 采样率 | 窗口大小 | 窗口时长 | 帧处理间隔 |
|---|---|---|---|
| 16kHz | 512 samples | 32ms | 32ms |
| 8kHz | 256 samples | 32ms | 32ms |
update_interval = 0.032sSLOW_INFERENCE_THRESHOLD = 0.2s(200ms 延迟告警)
ONNX Runtime 配置
# onnx_model.py:36-41
opts.inter_op_num_threads = 1 # 线程池 1 线程
opts.intra_op_num_threads = 1 # 单线程执行算子
opts.execution_mode = ORT_SEQUENTIAL # 顺序执行
opts.add_session_config_entry("session.intra_op.allow_spinning", "0")
设计理由:VAD 模型轻量,单线程串行减少线程竞争开销,CPU 推理已远超实时需求。
CPU 部署
默认行为
# silero/vad.py:69
force_cpu: bool = True # 默认强制 CPU
部署选项
# 1. 默认 CPU 推理(推荐)
vad = silero.VAD.load()
# 2. 强制 CPU
vad = silero.VAD.load(force_cpu=True)
# 3. 允许 GPU(如有可用)
vad = silero.VAD.load(force_cpu=False)
生产环境建议
| 场景 | 推荐配置 |
|---|---|
| 通用部署 | CPU (force_cpu=True) |
| 高并发 | CPU + 水平扩展 |
| 嵌入式 | CPU(GPU 功耗高) |
| 低延迟优先 | CPU(避免 GPU 拷贝开销) |
结论:VAD 模型极轻(ONNX 文件),CPU 推理毫无压力,GPU 对此类轻量模型反而增加开销。
模块协作关系
1. VAD + STT → 流式语音识别
AudioInput → VADStream → START/END_OF_SPEECH → STT.recognize() → Transcript
关键文件:stt/stream_adapter.py
当 STT 不支持流式时(如 Deepgram、OpenAI STT),VAD 作为桥梁:
- VAD 检测语音边界(START_OF_SPEECH, END_OF_SPEECH)
- END_OF_SPEECH 时,将缓存的音频 frames 合并
- 调用 STT.recognize() 进行识别
# stt/stream_adapter.py:112-127
async for event in vad_stream:
if event.type == VADEventType.START_OF_SPEECH:
self._event_ch.send_nowait(SpeechEvent(STT.START_OF_SPEECH))
elif event.type == VADEventType.END_OF_SPEECH:
merged_frames = utils.merge_frames(event.frames)
t_event = await self._wrapped_stt.recognize(buffer=merged_frames, ...)