语音打断、流式播报、前置指令:打造工业级AI语音交互体验
2026/5/7 9:33:27 网站建设 项目流程

发布时间: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-whisper244MB较快

选择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秒。

七、整体交互体验总结

用户使用这台语音助手的典型流程:

  1. 说"你好" → 0.1秒收到回应

  2. 问"ROS中的TF2是什么" → 听到检索中轻微等待(约5秒上下文处理),然后开始流式播报

  3. 听到一半发现问题 → 说"停" → 播报立即中断

  4. 换个方式追问 → 继续对话

  5. 说"再见" → 程序退出

整个过程不碰键盘,不联网,所有数据留在本地。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询