实时语音转文字系统的第一步，是拿到"对方正在说话"的音频。麦克风录到的是你自己的声音和环境噪音，而你需要的是系统正在播放的那路音频。本篇来解决这个问题，并在此基础上搭建实时语音检测（VAD），为后续的语音识别铺好路。

---

纯语音转文字系列导航

• 第 1 篇：系统设计与四层流水线
• 第 2 篇：音频采集 + 语音活动检测（本篇）
• 第 3 篇：流式 ASR —— 从 3 秒延迟重构到亚秒级增量解码（核心篇）
• 第 4 篇：PySide6 悬浮字幕窗 —— 透明置顶 + 双层字幕
• 第 5 篇：工程化与打包发布

---

一、本篇要解决的两个问题

1. 怎么捕获"正在播放的声音"？（会议里对方的声音、网课老师的声音、视频里的对话）——这需要 WASAPI Loopback。
2. 怎么实时判断"现在有没有人说话"？——这需要 VAD（语音活动检测），它是整个 ASR 流水线的触发器：有人说话才启动识别，静音时保持静默，避免把键盘声、空调噪音送去识别。

---

二、音频采集：WASAPI Loopback

2.1 原理：在功放输出口接一根监听线

Windows 的音频架构（WASAPI）中，所有应用播放的声音最终都流经系统的音频引擎（Audio Engine），再送往声卡/扬声器输出。WASAPI Loopback 模式允许我们以只读方式旁路监听这个输出——就像在功放的输出口上接了一根监听线，你能听到功放正在播放的一切，但你并不干扰正常的播放流程。

对比一下两种采集方式：

方式	比喻	录到什么	信噪比
麦克风采集	站在房间中央用耳朵听	你的声音 + 环境噪音 + 扬声器漏音	低——空调、键盘声全混进来
WASAPI Loopback	在功放输出口接监听线	系统正在播放的所有声音	高——纯净的数字信号，零环境噪音

所以 Loopback 天然适合"捕获对方说话"这个场景：信号干净、不受环境干扰、不需要额外硬件。

2.2 用 pyaudiowpatch 实现

原生的 PyAudio 不支持 Loopback 模式，需要用它的 Windows 补丁版 pyaudiowpatch：

import pyaudiowpatch as pyaudio
import numpy as np

# 1. 初始化 PyAudio（补丁版，支持 WASAPI Loopback）
pa = pyaudio.PyAudio()

# 2. 获取默认 Loopback 设备
#    这个设备对应系统"正在往哪个扬声器/耳机输出声音"
default_speaker = pa.get_default_wasapi_loopback()
print(f"默认 Loopback 设备: {default_speaker['name']}")
# 输出示例: 默认 Loopback 设备: 扬声器 (Realtek High Definition Audio)

# 3. 打开捕获流——以"录音"的方式读取扬声器的输出
stream = pa.open(
    format=pyaudio.paFloat32,     # Loopback 输出固定为 Float32 格式
    channels=2,                    # 双声道（立体声），后续需要混成单声道
    rate=44100,                    # 设备原生采样率——注意！不一定是 16kHz
    input=True,
    input_device_index=default_speaker["index"],
    frames_per_buffer=512,         # 每次读 512 个样本（约 11.6ms @44.1kHz）
)

# 4. 循环读取音频数据
#    阻塞等待 512 个样本就绪后返回
#    在 44.1kHz 下，512 样本 ≈ 11.6ms，满足实时性要求
raw_data = stream.read(512, exception_on_overflow=False)
audio = np.frombuffer(raw_data, dtype=np.float32)
# exception_on_overflow=False：忽略溢出警告，生产环境中用队列缓冲更稳妥

关键坑：Loopback 设备的采样率是设备原生值（常见 44100Hz 或 48000Hz），而且通常是双声道立体声。但 ASR 模型需要的是 16kHz 单声道。所以必须做重采样和声道混合——见 2.4 节。

2.3 双通道设计：系统音频 + 麦克风

实际场景中，我们可能既想捕获对方的声音（系统音频），也想捕获自己的声音（麦克风）。因此设计了双通道并行采集架构：

class AudioCapture:
    def __init__(self, system_queue, mic_queue, config):
        # 两路音频各用独立的队列传递数据，互不干扰
        self._system_queue = system_queue  # 系统音频 → 独立队列
        self._mic_queue = mic_queue        # 麦克风   → 独立队列

    def start(self):
        # 每路音频一个独立线程，线程名方便调试时定位
        threading.Thread(
            target=self._capture_system_audio,
            name="Audio-System",  # Loopback 采集线程
            daemon=True
        ).start()
        threading.Thread(
            target=self._capture_mic_audio,
            name="Audio-Mic",     # 麦克风采集线程
            daemon=True
        ).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 成 (帧数, 声道数) 后对每帧取左右声道均值
    if orig_channels > 1 and target_channels == 1:
        num_frames = len(data) // orig_channels
        data = data[:num_frames * orig_channels] \
            .reshape(num_frames, orig_channels) \
            .mean(axis=1)  # 左右声道均值 → 单声道

    # 第二步：采样率转换
    # 使用 scipy 的 resample，内置抗混叠滤波器，音质优于简单的线性插值
    if orig_rate != target_rate:
        from scipy.signal import resample
        num_samples = int(len(data) * target_rate / orig_rate)
        data = resample(data, num_samples).astype(np.float32)
        # 例：44100Hz → 16000Hz，512 样本 → 约 186 样本

    return data.astype(np.float32)

注意：重采样后每块的样本数可能不再是 512（比如 44100→16000 时 512→186），而 VAD 模型要求输入恰好 512 样本。所以 VAD 内部需要一个输入缓冲来累积对齐——见 3.2 节。

2.5 工程细节：断线重连与热切换

真实使用中，用户可能拔掉耳机、切换输出设备、或者系统音频引擎重启。采集模块必须能优雅地处理这些异常。

断线自动重连：当 stream.read() 抛出 OSError 时，说明音频流已断开。此时关闭旧流、等待 1 秒后重试，最多重连 10 次：

except OSError as e:
    # 流已断开，清理旧资源
    self._close_system_stream()
    reconnect_count += 1
    if reconnect_count > self._max_reconnect_attempts:  # 最多 10 次
        logger.error("超过最大重连次数，放弃重连")
        break
    logger.warning(f"音频流断开，{self._reconnect_interval}s 后第 {reconnect_count} 次重连...")
    time.sleep(self._reconnect_interval)  # 间隔 1 秒

运行时热切换设备：用户在系统设置里切换了输出设备（比如从扬声器切到蓝牙耳机），我们需要跟着切换 Loopback 源。流程是：暂停采集 → 关闭旧流 → 获取新的默认 Loopback 设备 → 用新配置重开流。捕获线程检测到流为 None 时，自动用新配置重连，无需重启整个采集模块。

---

三、VAD：Silero VAD 语音检测

3.1 为什么需要 VAD

拿到系统音频后，不能无脑全部送去 ASR——原因有三：

1. 检测语音边界：需要知道"什么时候开始说话"和"什么时候说完了"，才能把一句话完整地切出来送去识别。
2. 过滤噪音：系统里不只有人声——通知提示音、键盘声、鼠标点击、视频背景乐……VAD 负责把这些非语音信号过滤掉。
3. 触发下游 ASR：VAD 是整条流水线的"扳机"。检测到语音 → 开始累积音频 → 一句话结束 → 整段送去 ASR。没有 VAD，ASR 会被大量静音和噪音片段淹没，既浪费算力又产生垃圾输出。

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 噪音过滤

• 如果静音累计达到 300ms（silence_threshold_ms）→ 判定"说完了"。此时检查整段语音时长：如果 >= 250ms（min_speech_duration_ms），就把缓冲区里的完整音频段打包输出，送去 ASR。日志打印：VAD: 语音结束, 时长=736ms。
• 如果语音时长 < 250ms → 太短了，大概率是咳嗽、按键声等噪音，直接丢弃。日志打印：VAD: 噪声过滤, 时长=96ms < 250ms。

然后状态机回到"静默"状态，等待下一句话的开始。

class VADDetector:
    def __init__(self, config):
        # ---- 状态机参数 ----
        self._silence_threshold_ms = 300      # 静音持续 300ms → 判定一句话结束
        self._min_speech_duration_ms = 250    # 语音时长 < 250ms → 视为噪音，丢弃
        self._max_speech_duration_ms = 15000  # 语音时长 > 15s → 强制分段，防止延迟过高
        self._threshold = 0.1                 # 语音概率阈值：>= 此值判定为"语音"

        # ---- 运行时状态 ----
        self._is_speaking = False             # 当前是否处于"语音中"状态
        self._speech_buffer = []              # 当前语音段的音频样本缓冲
        self._input_buffer = np.array([], dtype=np.float32)  # 输入对齐缓冲（凑满 512 样本）
        self._silence_duration_ms = 0.0       # 当前静音累计时长
        self._speech_duration_ms = 0.0        # 当前语音段累计时长

    def process_chunk(self, chunk):
        """
        处理一块音频，返回 (is_speech, segment)。
        - is_speech: 当前块是否被判定为语音（中间态，可用于 UI 指示）
        - segment: 当一句话结束时返回完整音频段（numpy array），否则返回 None
        """
        # ---- 第 1 步：输入对齐，凑满 512 样本 ----
        self._input_buffer = np.concatenate([self._input_buffer, chunk])
        if len(self._input_buffer) < 512:
            return (False, None)  # 样本不够，等下一块

        segment_result = None

        # 可能累积了多块，逐块处理
        while len(self._input_buffer) >= 512:
            vad_chunk = self._input_buffer[:512]
            self._input_buffer = self._input_buffer[512:]

            # ---- 第 2 步：VAD 推理 ----
            # 模型输入：512 个 Float32 样本（32ms @16kHz）
            # 模型输出：0~1 的语音概率
            speech_prob = self._model(torch.from_numpy(vad_chunk), 16000).item()
            is_speech = speech_prob >= self._threshold

            if is_speech:
                # ---- 第 3 步：语音开始 / 持续 ----
                if not self._is_speaking:
                    # 状态跳转：静默 → 语音中
                    self._is_speaking = True
                    self._speech_buffer = []
                    self._speech_duration_ms = 0.0
                    logger.debug("VAD: 语音开始")

                self._speech_buffer.extend(vad_chunk.tolist())
                self._speech_duration_ms += 32.0  # 每块 512 样本 = 32ms @16kHz
                self._silence_duration_ms = 0.0   # 重置静音计数

                # 超过 15s 强制分段——防止演讲式长句导致延迟爆炸
                if self._speech_duration_ms >= self._max_speech_duration_ms:
                    segment_result = np.array(self._speech_buffer, dtype=np.float32)
                    self._is_speaking = False
                    self._speech_buffer = []
                    logger.debug(f"VAD: 强制分段, 时长={self._speech_duration_ms:.0f}ms")

            else:
                # ---- 第 4 步：静音处理 ----
                if self._is_speaking:
                    # 还在"语音中"状态——可能是说话间的短暂停顿
                    self._speech_buffer.extend(vad_chunk.tolist())
                    self._silence_duration_ms += 32.0  # 累加静音时长

                    # ---- 第 5 步：静音够长 → 一句话结束 ----
                    if self._silence_duration_ms >= self._silence_threshold_ms:
                        self._is_speaking = False
                        if self._speech_duration_ms >= self._min_speech_duration_ms:
                            # 时长足够 → 有效语音，打包输出
                            segment_result = np.array(self._speech_buffer, dtype=np.float32)
                            logger.debug(
                                f"VAD: 语音结束, 时长={self._speech_duration_ms:.0f}ms, "
                                f"样本数={len(self._speech_buffer)}"
                            )
                        else:
                            # 时长不足 → 判定为噪音（咳嗽、按键声等），丢弃
                            logger.debug(
                                f"VAD: 噪声过滤, 时长={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_ms	300	静音持续多久后判定"一句话说完"	太小会把长句切碎（“我今天…去了…超市”），太大会增加端到端延迟	300~400ms 适合大多数会议/网课场景；如果对方语速快、停顿短，可以降到 200ms
min_speech_duration_ms	250	短于此值的语音段视为噪音丢弃	太大会吞掉短促但有效的语音（如"嗯"“对”），太小会频繁触发噪音识别	250ms 是比较好的平衡点；如果环境噪音多（如机械键盘），可以提到 300ms
max_speech_duration_ms	15000	超过此时长强制切出一段送 ASR	防止演讲式长句积压导致延迟爆炸；太小会切碎正常长句	15s 是延迟与语义完整性的折中；如果 ASR 支持流式（第 3 篇），可以放宽到 20~30s
threshold	0.1	语音概率超过此值判定为"有人说话"	有背景噪音（空调、风扇）时调高，减少误触发；安静环境可以保持默认	安静环境用 0.1；有背景噪音时调到 0.3~0.5；如果对方声音很小（如远程网课音量低），降到 0.05

真实运行日志，可以看到状态机在实际场景中的表现：

VAD: 语音开始
VAD: 语音结束, 时长=736ms, 样本数=20480   ← 一句话完成，整段送 ASR
VAD: 语音开始
VAD: 噪声过滤, 时长=96ms < 250ms          ← 太短，判定为噪音（可能是按键声），丢弃
VAD: 语音开始
VAD: 强制分段, 时长=15032ms               ← 对方连续说了 15 秒，强制切出送 ASR

3.4 双通道 VAD + 说话人判定

系统音频和麦克风两路都经过 VAD，但优先级不同——我们更关心对方在说什么（系统音频），而不是自己的声音：

# 主循环：优先处理系统音频（对方/网课老师的声音）
try:
    # 从系统音频队列取数据，超时 200ms
    audio_chunk = self._system_audio_queue.get(timeout=0.2)
    is_speech, segment = self._vad_detector.process_chunk(audio_chunk)
    if segment is not None:
        # 一句话结束 → 送 ASR 识别
        self._process_speech_segment(segment, source="system")
except queue.Empty:
    # 系统音频队列空了（对方没在说话 / 没在开会）
    # 此时才去处理麦克风（自己的声音）
    # 额外条件：系统音频静默超过 2 秒才处理麦克风
    #   → 避免对方短暂停顿时误把自己的声音送去识别
    silence_duration = time.time() - self._system_silence_since
    if silence_duration > 2.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 ms（CPU）	2MB 轻量模型，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 重构到亚秒级？"边说边出字"的实时字幕到底怎么实现？