如何为whisperX创建实时语音活动检测（VAD）循环？

一、实时音频流处理示例代码 #

以下是结合PyAudio获取麦克风输入、WhisperX VAD检测语音活动，并完成实时转录的完整代码：

import pyaudio
import numpy as np
import torch
from whisperx import load_vad_model, transcribe

# 配置参数（匹配WhisperX要求）
SAMPLE_RATE = 16000
CHUNK_SIZE = 1024
CHANNELS = 1
FORMAT = pyaudio.paInt16

# 初始化VAD模型
vad_model, vad_utils = load_vad_model("cpu")  # 按需替换为"cuda"
get_speech_timestamps = vad_utils[0]

# 初始化PyAudio音频流
p = pyaudio.PyAudio()
stream = p.open(
    format=FORMAT,
    channels=CHANNELS,
    rate=SAMPLE_RATE,
    input=True,
    frames_per_buffer=CHUNK_SIZE
)

print("开始监听，按Ctrl+C停止...")

audio_buffer = []
speech_active = False

try:
    while True:
        # 读取麦克风音频块并转换为兼容格式
        data = stream.read(CHUNK_SIZE)
        audio_chunk = np.frombuffer(data, dtype=np.int16).astype(np.float32) / 32768.0
        audio_buffer.append(audio_chunk)

        # 每积累2秒音频执行一次VAD检测（平衡延迟与精度）
        if len(audio_buffer) * CHUNK_SIZE / SAMPLE_RATE >= 2:
            full_audio = np.concatenate(audio_buffer)
            speech_ts = get_speech_timestamps(full_audio, vad_model)

            if speech_ts:
                speech_active = True
                # 检测到语音结束（当前块无后续语音），触发转录
                last_speech_end = speech_ts[-1]["end"]
                if last_speech_end < len(full_audio) - (CHUNK_SIZE/SAMPLE_RATE)*0.5:
                    speech_segment = full_audio[:int(last_speech_end * SAMPLE_RATE)]
                    # 执行转录（按需调整模型、语言、设备）
                    result = transcribe("large-v2", speech_segment, language="zh", device="cpu")
                    print("\n转录结果：", result["text"])
                    # 保留未处理的后续音频，避免截断
                    remaining_audio = full_audio[int(last_speech_end * SAMPLE_RATE):]
                    audio_buffer = [remaining_audio]
                    speech_active = False
            else:
                if speech_active:
                    audio_buffer = []
                    speech_active = False
                else:
                    audio_buffer = []

except KeyboardInterrupt:
    print("\n停止监听")
finally:
    stream.stop_stream()
    stream.close()
    p.terminate()

关键说明： #

• 强制对齐WhisperX要求的16kHz单声道音频格式，避免格式不兼容问题
• 采用“积累音频块+批量VAD检测”的方式，平衡实时性与检测准确性
• 语音结束后自动触发转录，并保留后续未处理音频，防止截断说话内容
• 可根据硬件性能调整模型大小（如base/medium）和计算设备（cpu/cuda）

二、开源项目推荐 #

• whisperX-realtime：专为实时场景优化的WhisperX封装，内置麦克风流处理、VAD分段和实时转录输出，支持多语言切换与自定义阈值
• live-whisperx：轻量级实时语音转文本工具，集成麦克风捕获、VAD语音分割和WhisperX转录，支持输出格式自定义
• whisper-streaming：支持WhisperX作为后端的实时流转录项目，除麦克风输入外，还支持网络音频流接入，适配多场景需求

内容的提问来源于stack exchange，提问作者Jay Ferments