实时语音转文字系统的第一步是拿到对方正在说话的音频。麦克风录到的是你自己的声音和环境噪音而你需要的是系统正在播放的那路音频。本篇来解决这个问题并在此基础上搭建实时语音检测VAD为后续的语音识别铺好路。纯语音转文字系列导航第 1 篇系统设计与四层流水线第 2 篇音频采集 语音活动检测本篇第 3 篇流式 ASR —— 从 3 秒延迟重构到亚秒级增量解码核心篇第 4 篇PySide6 悬浮字幕窗 —— 透明置顶 双层字幕第 5 篇工程化与打包发布一、本篇要解决的两个问题怎么捕获正在播放的声音会议里对方的声音、网课老师的声音、视频里的对话——这需要WASAPI Loopback。怎么实时判断现在有没有人说话——这需要VAD语音活动检测它是整个 ASR 流水线的触发器有人说话才启动识别静音时保持静默避免把键盘声、空调噪音送去识别。二、音频采集WASAPI Loopback2.1 原理在功放输出口接一根监听线Windows 的音频架构WASAPI中所有应用播放的声音最终都流经系统的音频引擎Audio Engine再送往声卡/扬声器输出。WASAPI Loopback 模式允许我们以只读方式旁路监听这个输出——就像在功放的输出口上接了一根监听线你能听到功放正在播放的一切但你并不干扰正常的播放流程。对比一下两种采集方式方式比喻录到什么信噪比麦克风采集站在房间中央用耳朵听你的声音 环境噪音 扬声器漏音低——空调、键盘声全混进来WASAPI Loopback在功放输出口接监听线系统正在播放的所有声音高——纯净的数字信号零环境噪音所以 Loopback 天然适合捕获对方说话这个场景信号干净、不受环境干扰、不需要额外硬件。2.2 用 pyaudiowpatch 实现原生的PyAudio不支持 Loopback 模式需要用它的 Windows 补丁版pyaudiowpatchimportpyaudiowpatchaspyaudioimportnumpyasnp# 1. 初始化 PyAudio补丁版支持 WASAPI Loopbackpapyaudio.PyAudio()# 2. 获取默认 Loopback 设备# 这个设备对应系统正在往哪个扬声器/耳机输出声音default_speakerpa.get_default_wasapi_loopback()print(f默认 Loopback 设备:{default_speaker[name]})# 输出示例: 默认 Loopback 设备: 扬声器 (Realtek High Definition Audio)# 3. 打开捕获流——以录音的方式读取扬声器的输出streampa.open(formatpyaudio.paFloat32,# Loopback 输出固定为 Float32 格式channels2,# 双声道立体声后续需要混成单声道rate44100,# 设备原生采样率——注意不一定是 16kHzinputTrue,input_device_indexdefault_speaker[index],frames_per_buffer512,# 每次读 512 个样本约 11.6ms 44.1kHz)# 4. 循环读取音频数据# 阻塞等待 512 个样本就绪后返回# 在 44.1kHz 下512 样本 ≈ 11.6ms满足实时性要求raw_datastream.read(512,exception_on_overflowFalse)audionp.frombuffer(raw_data,dtypenp.float32)# exception_on_overflowFalse忽略溢出警告生产环境中用队列缓冲更稳妥关键坑Loopback 设备的采样率是设备原生值常见 44100Hz 或 48000Hz而且通常是双声道立体声。但 ASR 模型需要的是16kHz 单声道。所以必须做重采样和声道混合——见 2.4 节。2.3 双通道设计系统音频 麦克风实际场景中我们可能既想捕获对方的声音系统音频也想捕获自己的声音麦克风。因此设计了双通道并行采集架构classAudioCapture:def__init__(self,system_queue,mic_queue,config):# 两路音频各用独立的队列传递数据互不干扰self._system_queuesystem_queue# 系统音频 → 独立队列self._mic_queuemic_queue# 麦克风 → 独立队列defstart(self):# 每路音频一个独立线程线程名方便调试时定位threading.Thread(targetself._capture_system_audio,nameAudio-System,# Loopback 采集线程daemonTrue).start()threading.Thread(targetself._capture_mic_audio,nameAudio-Mic,# 麦克风采集线程daemonTrue).start()为什么用独立线程 队列音频采集是严格的实时任务——每 11.6ms 必须读一次数据否则就会丢样本。如果直接在采集线程里做 VAD 或 ASR下游处理一旦卡顿比如模型推理慢了采集线程就会被阻塞导致音频丢失。所以采用经典的生产者-消费者模式采集线程只管读数据、往队列里塞生产者VAD/ASR 线程从队列里取数据处理消费者。队列满时丢弃最旧的数据put_nowait 溢出时get_nowait保证采集线程永不阻塞。2.4 重采样与多声道混单声道Loopback 拿到的原始音频是设备原生采样率 多声道而 ASR 模型要求 16kHz 单声道。这一步做两件事def_resample(self,data,orig_rate,target_rate,orig_channels,target_channels): 将原始音频转换为 ASR 所需的格式。 data: 原始音频数据numpy array orig_rate: 原始采样率如 44100 target_rate: 目标采样率16000 orig_channels: 原始声道数如 2 立体声 target_channels: 目标声道数1 单声道 # 第一步多声道 → 单声道# 立体声数据是交织存储的 [L0, R0, L1, R1, ...]# reshape 成 (帧数, 声道数) 后对每帧取左右声道均值iforig_channels1andtarget_channels1:num_frameslen(data)//orig_channels datadata[:num_frames*orig_channels]\.reshape(num_frames,orig_channels)\.mean(axis1)# 左右声道均值 → 单声道# 第二步采样率转换# 使用 scipy 的 resample内置抗混叠滤波器音质优于简单的线性插值iforig_rate!target_rate:fromscipy.signalimportresample num_samplesint(len(data)*target_rate/orig_rate)dataresample(data,num_samples).astype(np.float32)# 例44100Hz → 16000Hz512 样本 → 约 186 样本returndata.astype(np.float32)注意重采样后每块的样本数可能不再是 512比如 44100→16000 时 512→186而 VAD 模型要求输入恰好 512 样本。所以 VAD 内部需要一个输入缓冲来累积对齐——见 3.2 节。2.5 工程细节断线重连与热切换真实使用中用户可能拔掉耳机、切换输出设备、或者系统音频引擎重启。采集模块必须能优雅地处理这些异常。断线自动重连当stream.read()抛出OSError时说明音频流已断开。此时关闭旧流、等待 1 秒后重试最多重连 10 次exceptOSErrorase:# 流已断开清理旧资源self._close_system_stream()reconnect_count1ifreconnect_countself._max_reconnect_attempts:# 最多 10 次logger.error(超过最大重连次数放弃重连)breaklogger.warning(f音频流断开{self._reconnect_interval}s 后第{reconnect_count}次重连...)time.sleep(self._reconnect_interval)# 间隔 1 秒运行时热切换设备用户在系统设置里切换了输出设备比如从扬声器切到蓝牙耳机我们需要跟着切换 Loopback 源。流程是暂停采集 → 关闭旧流 → 获取新的默认 Loopback 设备 → 用新配置重开流。捕获线程检测到流为 None 时自动用新配置重连无需重启整个采集模块。三、VADSilero VAD 语音检测3.1 为什么需要 VAD拿到系统音频后不能无脑全部送去 ASR——原因有三检测语音边界需要知道什么时候开始说话和什么时候说完了才能把一句话完整地切出来送去识别。过滤噪音系统里不只有人声——通知提示音、键盘声、鼠标点击、视频背景乐……VAD 负责把这些非语音信号过滤掉。触发下游 ASRVAD 是整条流水线的扳机。检测到语音 → 开始累积音频 → 一句话结束 → 整段送去 ASR。没有 VADASR 会被大量静音和噪音片段淹没既浪费算力又产生垃圾输出。3.2 集成与状态机一句话的生命周期Silero VAD 是一个仅 2MB 的轻量神经网络模型在 CPU 上处理 512 样本32ms的推理耗时不到 1ms。它的核心逻辑是一个有限状态机下面用一个句子的一生来讲述这个故事场景对方开始说话——第 1 步输入对齐重采样后的音频块可能不足 512 样本比如只有 186 个所以先放入输入缓冲区累积。每凑满 512 样本就送入 VAD 模型做一次推理。第 2 步语音开始“有人说话了”模型输出一个 0~1 的语音概率值。当概率超过阈值默认 0.1时状态机从静默跳转到语音中。此时清空语音缓冲区开始把当前及后续的语音样本存入缓冲区。日志打印VAD: 语音开始。第 3 步语音持续中“还在说……”只要模型持续判定为语音概率 阈值就不断往缓冲区里追加样本。但如果语音持续超过 15 秒max_speech_duration_ms状态机会强制切出一段送去 ASR——防止演讲式的长段落积压太多延迟。第 4 步静音降临“说完了还是停顿”当模型判定某一块不是语音时状态机进入等待状态继续把静音样本追加到缓冲区因为说话中间可能有短暂停顿同时开始累计静音时长。第 5 步一句话结束 or 噪音过滤如果静音累计达到300mssilence_threshold_ms→ 判定说完了。此时检查整段语音时长如果 250msmin_speech_duration_ms就把缓冲区里的完整音频段打包输出送去 ASR。日志打印VAD: 语音结束, 时长736ms。如果语音时长 250ms → 太短了大概率是咳嗽、按键声等噪音直接丢弃。日志打印VAD: 噪声过滤, 时长96ms 250ms。然后状态机回到静默状态等待下一句话的开始。classVADDetector:def__init__(self,config):# ---- 状态机参数 ----self._silence_threshold_ms300# 静音持续 300ms → 判定一句话结束self._min_speech_duration_ms250# 语音时长 250ms → 视为噪音丢弃self._max_speech_duration_ms15000# 语音时长 15s → 强制分段防止延迟过高self._threshold0.1# 语音概率阈值 此值判定为语音# ---- 运行时状态 ----self._is_speakingFalse# 当前是否处于语音中状态self._speech_buffer[]# 当前语音段的音频样本缓冲self._input_buffernp.array([],dtypenp.float32)# 输入对齐缓冲凑满 512 样本self._silence_duration_ms0.0# 当前静音累计时长self._speech_duration_ms0.0# 当前语音段累计时长defprocess_chunk(self,chunk): 处理一块音频返回 (is_speech, segment)。 - is_speech: 当前块是否被判定为语音中间态可用于 UI 指示 - segment: 当一句话结束时返回完整音频段numpy array否则返回 None # ---- 第 1 步输入对齐凑满 512 样本 ----self._input_buffernp.concatenate([self._input_buffer,chunk])iflen(self._input_buffer)512:return(False,None)# 样本不够等下一块segment_resultNone# 可能累积了多块逐块处理whilelen(self._input_buffer)512:vad_chunkself._input_buffer[:512]self._input_bufferself._input_buffer[512:]# ---- 第 2 步VAD 推理 ----# 模型输入512 个 Float32 样本32ms 16kHz# 模型输出0~1 的语音概率speech_probself._model(torch.from_numpy(vad_chunk),16000).item()is_speechspeech_probself._thresholdifis_speech:# ---- 第 3 步语音开始 / 持续 ----ifnotself._is_speaking:# 状态跳转静默 → 语音中self._is_speakingTrueself._speech_buffer[]self._speech_duration_ms0.0logger.debug(VAD: 语音开始)self._speech_buffer.extend(vad_chunk.tolist())self._speech_duration_ms32.0# 每块 512 样本 32ms 16kHzself._silence_duration_ms0.0# 重置静音计数# 超过 15s 强制分段——防止演讲式长句导致延迟爆炸ifself._speech_duration_msself._max_speech_duration_ms:segment_resultnp.array(self._speech_buffer,dtypenp.float32)self._is_speakingFalseself._speech_buffer[]logger.debug(fVAD: 强制分段, 时长{self._speech_duration_ms:.0f}ms)else:# ---- 第 4 步静音处理 ----ifself._is_speaking:# 还在语音中状态——可能是说话间的短暂停顿self._speech_buffer.extend(vad_chunk.tolist())self._silence_duration_ms32.0# 累加静音时长# ---- 第 5 步静音够长 → 一句话结束 ----ifself._silence_duration_msself._silence_threshold_ms:self._is_speakingFalseifself._speech_duration_msself._min_speech_duration_ms:# 时长足够 → 有效语音打包输出segment_resultnp.array(self._speech_buffer,dtypenp.float32)logger.debug(fVAD: 语音结束, 时长{self._speech_duration_ms:.0f}ms, f样本数{len(self._speech_buffer)})else:# 时长不足 → 判定为噪音咳嗽、按键声等丢弃logger.debug(fVAD: 噪声过滤, 时长{self._speech_duration_ms:.0f}ms f{self._min_speech_duration_ms}ms)self._speech_buffer[]return(is_speech,segment_result)返回值设计(is_speech, segment)——is_speech标记当前块是否语音可用于 UI 上的实时波形/指示灯segment只在一句话结束或强制分段时返回完整音频段其余时候为 None表示还没说完继续等。3.3 参数调优实战下面这张表汇总了四个核心参数的含义、调优方向和实测建议参数默认值作用调优建议实测建议silence_threshold_ms300静音持续多久后判定一句话说完太小会把长句切碎“我今天…去了…超市”太大会增加端到端延迟300~400ms适合大多数会议/网课场景如果对方语速快、停顿短可以降到 200msmin_speech_duration_ms250短于此值的语音段视为噪音丢弃太大会吞掉短促但有效的语音如嗯“对”太小会频繁触发噪音识别250ms是比较好的平衡点如果环境噪音多如机械键盘可以提到 300msmax_speech_duration_ms15000超过此时长强制切出一段送 ASR防止演讲式长句积压导致延迟爆炸太小会切碎正常长句15s是延迟与语义完整性的折中如果 ASR 支持流式第 3 篇可以放宽到 20~30sthreshold0.1语音概率超过此值判定为有人说话有背景噪音空调、风扇时调高减少误触发安静环境可以保持默认安静环境用0.1有背景噪音时调到0.3~0.5如果对方声音很小如远程网课音量低降到0.05真实运行日志可以看到状态机在实际场景中的表现VAD: 语音开始 VAD: 语音结束, 时长736ms, 样本数20480 ← 一句话完成整段送 ASR VAD: 语音开始 VAD: 噪声过滤, 时长96ms 250ms ← 太短判定为噪音可能是按键声丢弃 VAD: 语音开始 VAD: 强制分段, 时长15032ms ← 对方连续说了 15 秒强制切出送 ASR3.4 双通道 VAD 说话人判定系统音频和麦克风两路都经过 VAD但优先级不同——我们更关心对方在说什么系统音频而不是自己的声音# 主循环优先处理系统音频对方/网课老师的声音try:# 从系统音频队列取数据超时 200msaudio_chunkself._system_audio_queue.get(timeout0.2)is_speech,segmentself._vad_detector.process_chunk(audio_chunk)ifsegmentisnotNone:# 一句话结束 → 送 ASR 识别self._process_speech_segment(segment,sourcesystem)exceptqueue.Empty:# 系统音频队列空了对方没在说话 / 没在开会# 此时才去处理麦克风自己的声音# 额外条件系统音频静默超过 2 秒才处理麦克风# → 避免对方短暂停顿时误把自己的声音送去识别silence_durationtime.time()-self._system_silence_sinceifsilence_duration2.0:self._process_mic_audio()局限VAD 只能判断有没有人说话判断不了谁在说话。当对方停顿 2s 时你插话麦克风那路会被丢弃因为系统音频还没安静够2 秒。更优的方案是双路能量门控或说话人分离模型如 pyannote-audio第 5 篇会给出改进方向。四、性能实测以下数据在 Intel i5-12400 / 16GB RAM / Windows 11 环境下测得指标数值说明采集粒度512 样本 / 32ms 16kHz每 32ms 处理一次远快于实时Silero VAD 单块推理 1 msCPU2MB 轻量模型CPU 推理几乎无感重采样44.1k→16k~2 ms / 块scipy 抗混叠重采样音质好双通道总 CPU 占用 5%采集 VAD 合计非常轻量结论采集和 VAD 层的处理速度远快于实时要求32ms 的音频只用不到 3ms 处理瓶颈在下游的 ASR 层——这正是第 3 篇要解决的核心问题。五、小结与预告本篇完成了流水线的前两层关键模块WASAPI Loopback 音频采集通过在功放输出口接监听线的方式干净地捕获系统播放的任何声音。配套实现了双通道并行采集、重采样/混声道、断线自动重连、运行时设备热切换等工程能力。Silero VAD 语音检测用 2MB 的轻量模型搭建了一个有限状态机实现语音边界检测、噪音过滤、强制分段。详解了四个核心参数的调优策略和实测建议。下一篇系列核心篇进入 ASR 层为什么 Qwen3-ASR 全量识别延迟高达 3 秒如何用 sherpa-onnx 流式 Zipformer 重构到亚秒级边说边出字的实时字幕到底怎么实现