语音打断、流式播报、前置指令：打造工业级AI语音交互体验
发布时间：5月5日
标签：#语音交互 #VAD #TTS #ASR #人机交互
字数：约2000字
一、为什么语音交互这么难做
做语音助手的都知道一个痛点：AI还没说完，你就知道它错了。但你必须忍着听完。
手机上用Siri或小爱同学时，这种体验特别常见。AI进入幻觉模式，开始编造答案，而你能做的只有拿起手机点"停止"。
好的语音交互，打断体验和回答质量同等重要。
二、语音活动检测（VAD）：让程序知道你在说话
为什么不用简单的音量阈值？
一开始我尝试用音量的RMS值判断是否有人说话：
import pyaudio import numpy as np def simple_vad(audio_chunk, threshold=500): rms = np.sqrt(np.mean(np.square(audio_chunk))) return rms > threshold问题：环境噪声变化时阈值失效。风扇声、键盘声、翻书声都会误触发。而且无法区分人声和非人声。
最终方案：TEN VAD
TEN VAD是一个轻量级深度学习模型，专门训练用于区分人声和噪声：
class VADDetector: def __init__(self, model_path, sample_rate=16000): self.model = TenVAD(model_path) self.sample_rate = sample_rate self.speech_started = False self.silence_duration = 0 self.silence_threshold = 0.8 # 0.8秒静音判定说话结束 def is_speaking(self, audio_chunk): """返回是否检测到人声""" result = self.model.detect(audio_chunk, self.sample_rate) return result["speech_probability"] > 0.7 def should_end_utterance(self, audio_chunk, dt): """判断是否应该结束录音""" if self.is_speaking(audio_chunk): self.silence_duration = 0 return False else: self.silence_duration += dt return self.silence_duration > self.silence_threshold三、打断机制的双模设计
语音打断
用户开始说话 → VAD检测到人声 → 触发打断 → 进入2秒冷却期
关键设计：冷却期
不设冷却期的话，TTS播报的声音会被VAD检测为"用户在说话"，导致刚打断恢复就立即再次触发打断，形成振荡。
按键打断
import keyboard class KeyInterrupt: def __init__(self, callback): self.callback = callback def start(self): keyboard.on_press_key("space", lambda _: self.callback()) def stop(self): keyboard.unhook_all()空格键作为万能打断键，不依赖VAD状态，不受冷却期限制。
四、语音识别（ASR）：faster-whisper选型
对比了几个方案：
| 方案 | 模型大小 | 中文准确率 | 推理速度 |
|---|---|---|---|
| Vosk | ~50MB | 一般 | 快 |
| SpeechRecognition | 在线 | 好 | 依赖网络 |
| whisper.cpp | ~200MB | 好 | 中 |
| faster-whisper | 244MB | 好 | 较快 |
选择faster-whisper的原因：
CTranslate2加速，比原版whisper快4-6倍
INT8量化，内存占用减半
16GB笔记本上只占不到1GB内存
from faster_whisper import WhisperModel class ASREngine: def __init__(self, model_size="small"): # 第一次运行会自动下载模型 self.model = WhisperModel( model_size, device="cpu", compute_type="int8" # INT8量化节省内存 ) def transcribe(self, audio_data): segments, info = self.model.transcribe( audio_data, language="zh", beam_size=5 ) return "".join([seg.text for seg in segments])五、语音合成（TTS）：流式播报
为什么不用Edge TTS或讯飞？
离线优先。pyttsx3完全本地运行，不需要网络。
流式播报实现
import pyttsx3 import re class StreamSpeaker: def __init__(self): self.engine = pyttsx3.init() self.engine.setProperty("rate", 180) # 语速 def speak_stream(self, text_generator, interrupt_handler): """流式播报，支持打断""" buffer = "" for token in text_generator: if interrupt_handler.is_interrupted(): self.engine.stop() break buffer += token # 遇到标点就播报当前句子 if re.search(r'[。！？\n]', buffer): self.engine.say(buffer) self.engine.runAndWait() buffer = "" # 播报剩余内容 if buffer and not interrupt_handler.is_interrupted(): self.engine.say(buffer) self.engine.runAndWait()效果：模型每生成完一句话，立刻开始朗读，用户不需要等全部生成完。停顿自然的句号和问号位置刚好成为语音播报的断点。
六、前置指令：让常用操作秒回
PRESET_COMMANDS = { "你好": "你好，我是你的本地AI助手。你可以问我知识库中的任何问题。", "在吗": "我在。请随时提问。", "再见": "再见！", "退出": "退出系统。", "谢谢": "不客气！" } def handle_query(query): # 先检查是否前置指令 if query.strip() in PRESET_COMMANDS: return PRESET_COMMANDS[query.strip()] # 前置指令"再见""退出"需要退出程序 if query.strip() in ["再见", "退出"]: sys.exit(0) # 否则走正常RAG流程 return rag_pipeline(query)这类问候和告别只做字符串匹配，跳过整个RAG流水线，响应时间从10秒降到0.1秒。
七、整体交互体验总结
用户使用这台语音助手的典型流程：
说"你好" → 0.1秒收到回应
问"ROS中的TF2是什么" → 听到检索中轻微等待（约5秒上下文处理），然后开始流式播报
听到一半发现问题 → 说"停" → 播报立即中断
换个方式追问 → 继续对话
说"再见" → 程序退出
整个过程不碰键盘，不联网，所有数据留在本地。