AI Engineering · 第 6 章 语音助手流水毕业项目

语音助手流水线:第 6 章毕业项目

本节摘要:把第 01~11 节的全部内容缝起来——搭一个能听、能推理、能回话的语音助手。2026 年这已经是一个被解决的工程问题,不再是研究问题,但集成细节决定它能不能上线。本节毕业项目要求你搭一条端到端流水线:采集麦克风(16 kHz 单声道)、检测用户语音的起止、流式转录、把转录喂给能调工具(定时器、天气、日历)的 LLM、把 LLM 文本流式喂给 TTS、把音频回放给用户,并在用户中途打断时停止。延迟目标:笔记本 CPU 上,用户说完话后 800 毫秒内出第一个 TTS 字节。质量目标:不漏字、静音上不幻觉字幕、无声音克隆泄漏、提示注入不成功。本节会逐一拆解七个组件、三个你必将撞上的失败模式(首字被切、中途打断混乱、静音幻觉),以及 2026 年的四种生产参考栈(LiveKit 全商业、Pipecat 半开源、Moshi 全双工、Whisper.cpp+llama.cpp 全离线)。读完本节,你能为给定预算、规模、语种、合规约束产出一份完整的语音助手技术栈规格。

学习目标

阅读完本节,你应当能够:

1. 把 VAD、流式 ASR、带工具调用的 LLM、流式 TTS、抢话中断集成成一条端到端语音助手流水线。
2. 实现端点检测(end-pointing)与预滚动(pre-roll)缓冲,避免首字被切。
3. 设计工具调用循环,让 LLM 在生成过程中调用天气、定时器等外部函数并把结果回灌。
4. 实现抢话中断(VAD 触发 → 停 TTS → 取消 LLM → 重启 STT)的异步取消链路。
5. 在延迟、质量、隐私、合规四维度权衡下,从 LiveKit / Pipecat / Moshi / 全离线 四类栈中选对方案。

一、问题与直觉

搭一个端到端助手:

1. 采集麦克风输入(16 kHz 单声道)。
2. 检测用户语音的起止。
3. 流式转录。
4. 把转录传给能调工具(定时器、天气、日历)的 LLM。
5. 把 LLM 文本流式传给 TTS。
6. 把音频回放给用户。
7. 用户中途打断时停止。

延迟目标:笔记本 CPU 上,用户说完话后 800 毫秒内出第一个 TTS 字节。质量目标:不漏字、静音上不幻觉字幕、无声音克隆泄漏、提示注入不成功。

七个组件

1. 音频采集:麦克风 → 16 kHz 单声道 → 20 毫秒块。Python 里通常用 sounddevice,生产用原生 AudioUnit/ALSA/WASAPI。
2. VAD(第 11 节):Silero VAD 阈值 0.5,最短语音 250 毫秒,静音悬挂 500 毫秒。发出「开始」「结束」信号。
3. 流式 STT(第 4-5 节):Whisper-streaming、Parakeet-TDT 或 Deepgram Nova-3(API)。部分 + 最终转录。
4. 带工具调用的 LLM:GPT-4o / Claude 3.5 / Gemini 2.5 Flash。工具用 JSON schema 定义。流式 token。
5. 流式 TTS(第 7 节):Kokoro-82M(最快开源)或 Cartesia Sonic(商业)。LLM 出 20 个 token 后启动 TTS。
6. 回放:喇叭输出;低带宽网络用 opus 编码。
7. 中断处理器:TTS 回放期间 VAD 触发,则停回放、取消 LLM、重启 STT。

三个你必将撞上的失败模式

1. 首字被切:VAD 慢一拍启动,用户的「hey」丢了。阈值从 0.5 调到 0.3。
2. 中途打断混乱:用户打断后 LLM 还在生成,助手和用户抢话。要把 VAD 接到「取消 LLM」。
3. 静音幻觉:Whisper 在静音预热帧上输出「Thanks for watching」。始终 VAD 门控。

2026 年生产参考栈

💡 注意「延迟数字」与「延迟体感」的区别。一个标称 350 ms 的栈,如果首字被切、抢话不灵敏,体感会比标称 800 ms 但抢话灵敏的栈更差。所以衡量语音助手要把「首响应延迟」与「抢话响应延迟」分开测,后者往往决定用户是否觉得它「活着」。

二、从零实现

第 1 步:带分块的麦克风采集(伪代码)

import sounddevice as sd

def mic_stream(chunk_ms=20, sr=16000):
    q = queue.Queue()
    def cb(indata, frames, time, status):
        q.put(indata.copy().flatten())
    with sd.InputStream(channels=1, samplerate=sr, blocksize=int(sr * chunk_ms/1000), callback=cb):
        while True:
            yield q.get()
​

第 2 步:VAD 门控的轮次捕获

def capture_turn(stream, vad, pre_roll_ms=300, silence_ms=500):
    buf, pre, triggered = [], collections.deque(maxlen=pre_roll_ms // 20), False
    silent = 0
    for chunk in stream:
        pre.append(chunk)
        if vad(chunk):
            if not triggered:
                buf = list(pre)          # 预滚动:把 VAD 触发前的音频也带上
                triggered = True
            buf.append(chunk)
            silent = 0
        elif triggered:
            silent += 20
            buf.append(chunk)
            if silent >= silence_ms:
                return b"".join(buf)
​

设计要点:pre_roll_ms=300 的预滚动缓冲是避免首字被切的关键——VAD 检测到语音时,用户的第一个字往往已经过去几十到几百毫秒。用一个固定长度的环形缓冲把 VAD 触发前的音频也保留下来,拼到正式缓冲前面,就能完整捕获「hey」「ok」这类短促的开头。

第 3 步:流式 STT → LLM → TTS

async def turn(audio_bytes):
    transcript = await stt.transcribe(audio_bytes)
    async for token in llm.stream(transcript):
        async for audio in tts.stream(token):
            await speaker.play(audio)
​

第 4 步:LLM 循环内的工具调用

tools = [
    {"name": "get_weather", "parameters": {"location": "string"}},
    {"name": "set_timer", "parameters": {"seconds": "int"}},
]

async for chunk in llm.stream(user_text, tools=tools):
    if chunk.type == "tool_call":
        result = dispatch(chunk.name, chunk.args)
        continue_streaming(result)
    if chunk.type == "text":
        await tts.stream(chunk.text)
​

第 5 步:中断处理

tts_task = asyncio.create_task(tts_loop())
while True:
    chunk = await mic.get()
    if vad(chunk):
        tts_task.cancel()
        await speaker.stop()
        await new_turn()
        break
​

三、框架对比

code/main.py 提供一个可运行的模拟,用桩模型把七个组件接起来,即使没有硬件也能看到流水线形状。真实实现把桩换成:

• silero-vad(pip install silero-vad)
• deepgram-sdk 或 openai-whisper
• openai(gpt-4o)或 anthropic
• kokoro 或 cartesia
• sounddevice 做输入输出

四类栈的取舍:

• LiveKit 全商业:延迟最低(350~500 ms),上线快,但全 API 依赖,成本与合规要算清。
• Pipecat 半开源:延迟略高(500~800 ms),可换组件,定制空间大,适合有工程团队的项目。
• Moshi 全双工:延迟最低(200~300 ms),单模型架构(第 15 节详讲),但对话风格与可控性不如级联栈。
• 全离线(Whisper.cpp + llama.cpp + Kokoro-ONNX):隐私与边缘场景必备,延迟与质量取决于本地硬件。

⚠️ 六类坑:① 永远记录 PII——完整轮次音频在多数司法管辖区是 PII,30 天留存、静态加密;② 无抢话——用户一定会打断,助手必须停;③ TTS 阻塞——同步 TTS 阻塞事件循环,用 async 或独立线程;④ 工具调用无错误处理——工具会失败,LLM 必须收到错误 + 重试一次,然后优雅降级;⑤ 幻觉过滤过激——过滤过头助手只会复读「我帮不了这个」,过滤不够又会乱说,要在留出集上调;⑥ 无唤醒词选项——常听是隐私负债,加唤醒词门(Porcupine 或 openWakeWord)。

四、可复用产物

本节产出技能文档(原课程 outputs/skill-voice-assistant-architect.md),给定预算 + 规模 + 语种 + 合规约束,产出完整技术栈规格。模板:

• 延迟目标:首响应 < 800 ms,抢话响应 < 100 ms。
• 栈选择:LiveKit(商业快)/ Pipecat(开源定制)/ Moshi(全双工)/ 全离线(隐私)。
• 组件:VAD(Silero 0.5 + 预滚动 300 ms)、STT(Parakeet-TDT 或 Deepgram)、LLM(GPT-4o/Claude,带工具 schema)、TTS(Kokoro 或 Cartesia,20 token 后启动)。
• 合规:轮次音频 30 天加密留存,唤醒词门控,提示注入护栏,静音 VAD 把关。
• 评估闭环:WER(STT)、首响应延迟、抢话响应延迟、提示注入成功率、主观 MOS。

五、练习

1. 基础:运行 code/main.py。它用桩模块端到端模拟一个完整轮次,打印逐级延迟。

2. 进阶:把 STT 桩换成真实 Whisper 模型,对一段预录 .wav 跑,测 WER 与端到端延迟。

3. 挑战:加工具调用:实现 get_weather(任意 API)和 set_timer。让 LLM 经工具路由,验证用户说「设个 5 分钟定时器」时正确的函数被触发,且回话确认。

本节要点回顾

1. 语音助手 = VAD + 流式 ASR + 带工具的 LLM + 流式 TTS + 抢话中断,2026 年是工程问题而非研究问题。
2. 七个组件:音频采集、VAD、流式 STT、带工具的 LLM、流式 TTS、回放、中断处理器。
3. 延迟目标:首响应 < 800 ms,质量目标不漏字/不幻觉/不泄漏/防注入。
4. 三个必撞失败模式:首字被切(阈值调 0.3 + 预滚动)、中途打断混乱(VAD 接取消 LLM)、静音幻觉(始终 VAD 门控)。
5. 预滚动缓冲 300 ms 是避免首字被切的关键——把 VAD 触发前的音频拼到正式缓冲前。
6. 工具调用循环:LLM 流式生成中遇到 tool_call 就 dispatch,结果回灌继续生成。
7. 抢话中断= 停 TTS + 取消 LLM + 重启 STT,asyncio 取消链路贯穿全程。
8. 四类生产栈:LiveKit(350500ms 商业)、Pipecat(500800ms 开源)、Moshi(200~300ms 全双工)、全离线(隐私)。
9. 六类坑:PII 永存、无抢话、TTS 阻塞、工具无错误处理、幻觉过滤过激、无唤醒词。

下一节,我们下沉到流水线之下,讲神经音频编解码——EnCodec、SoundStream、DAC 如何把波形压成离散 token,这是 MusicGen、VALL-E、AudioLM 等所有「音频 token」模型的共同基座。