AI 原生开放教材

外观

AI 工程从零到一（中文）源码研读·中文原文 ↗原件 ↓仓库 ↗

语音 Agent：Pipecat 与 LiveKit ​

语音 agent 在 2026 年是一等的生产类别。Pipecat 给你一条 Python 的、基于帧的流水线（VAD → STT → LLM → TTS → 传输）。LiveKit Agents 通过 WebRTC 把 AI 模型桥接给用户。高端技术栈的生产延迟目标落在端到端 450–600ms。

类型： Learn 语言： Python（标准库） 前置要求： 阶段 14 · 01（Agent 循环）、阶段 14 · 12（工作流模式） 预计时间： ~60 分钟

学习目标 ​

  • 描述 Pipecat 基于帧的流水线：DOWNSTREAM（源→汇）和 UPSTREAM（控制）。
  • 说出标准语音流水线的各个阶段，以及 Pipecat 支持哪些传输。
  • 解释 LiveKit Agents 的两种语音 agent 类（MultimodalAgent、VoicePipelineAgent）以及各自何时合适。
  • 总结 2026 年的生产延迟预期，以及它们如何驱动架构选择。

问题背景 ​

语音 agent 不是一个文本循环硬加上 TTS。延迟预算很残酷（~600ms），部分音频是默认情况，回合检测是个模型，传输从电话 SIP 到 WebRTC 都有。要么你构建一条基于帧的流水线（Pipecat），要么你倚靠一个平台（LiveKit）。

核心概念 ​

Pipecat（pipecat-ai/pipecat） ​

  • Python 的基于帧的流水线框架。
  • Frame → FrameProcessor 链。
  • 两个流动方向：
    • DOWNSTREAM —— 源 → 汇（音频进，TTS 出）。
    • UPSTREAM —— 反馈与控制（取消、指标、插话）。
  • PipelineTask 用事件（on_pipeline_started、on_pipeline_finished、on_idle_timeout）和用于指标/tracing/RTVI 的观察者来管理生命周期。

典型流水线：

VAD (Silero) → STT → LLM (context alternates user/assistant) → TTS → transport

传输：Daily、LiveKit、SmallWebRTCTransport、FastAPI WebSocket、WhatsApp。

Pipecat Flows 加了结构化对话（状态机）。Pipecat Cloud 是托管运行时。

LiveKit Agents（livekit/agents） ​

  • 通过 WebRTC 把 AI 模型桥接给用户。
  • 关键概念：Agent、AgentSession、entrypoint、AgentServer。
  • 两种语音 agent 类：
    • MultimodalAgent —— 通过 OpenAI Realtime 或等价物的直接音频。
    • VoicePipelineAgent —— STT → LLM → TTS 级联；给你文本级控制。
  • 通过一个 transformer 模型做语义回合检测。
  • 原生 MCP 集成。
  • 通过 SIP 做电话。
  • 通过 LiveKit Inference 无需 API key 的 50+ 模型；通过插件再加 200+。

商业平台 ​

Vapi（在一个优化过的高端技术栈上约 450–600ms）和 Retell（180 通测试呼叫中端到端约 600ms）建在这些之上。当你想要一个托管语音栈、又不想要一支 WebRTC 团队时，挑一个平台。

这个模式在哪里会出错 ​

  • 没处理插话。 用户打断；agent 还在说。在 Pipecat 里需要 UPSTREAM 取消帧，LiveKit 里有等价物。
  • 忽略 STT 置信度。 把低置信度的转录当圣旨喂给 LLM。按置信度设关卡或请求确认。
  • TTS 句中截断。 当流水线在话语中途取消时，TTS 需要知道或者切掉音频。
  • 忽略延迟预算。 每个组件加 50–200ms。上线前把你的链路加总。

2026 年典型延迟 ​

  • VAD：20–60ms
  • STT 部分结果：100–250ms
  • LLM 首 token：150–400ms
  • TTS 首段音频：100–200ms
  • 传输 RTT：30–80ms

端到端 450–600ms 算高端。800–1200ms 很常见。任何 > 1500ms 都感觉坏了。

figure
voice-pipeline

动手构建 ​

code/main.py 是一条基于帧的玩具流水线，带：

  • Frame 类型（audio、transcript、text、tts_audio、control）。
  • 带 process(frame) 的 Processor 接口。
  • 一条五阶段流水线（VAD → STT → LLM → TTS → 传输），用脚本化的 processor。
  • 一个 UPSTREAM 取消帧来演示插话。

运行它：

python3 code/main.py

轨迹展示正常流程，以及一次在话语中途停掉 TTS 的插话取消。

实际使用 ​

  • Pipecat 用于完全控制 —— 自定义 processor、Python 优先、可插拔提供方。
  • LiveKit Agents 用于 WebRTC 优先的部署和电话。
  • Vapi / Retell 用于不要 WebRTC 团队的托管语音 agent。
  • OpenAI Realtime / Gemini Live 用于直接音频进/音频出（MultimodalAgent）。

拿去用 ​

outputs/skill-voice-pipeline.md 脚手架出一条 Pipecat 形态的语音流水线，带 VAD + STT + LLM + TTS + 传输外加插话处理。

练习 ​

  1. 给你的玩具流水线加一个指标观察者：统计每阶段每秒的帧数。延迟在哪里累积？
  2. 实现置信度设关卡的 STT：低于阈值时请求「能再说一遍吗？」
  3. 加语义回合检测：简单规则 —— 如果转录以「?」结尾，就是回合结束。
  4. 读 Pipecat 的传输文档。把标准库传输换成 SmallWebRTCTransport 配置（桩）。
  5. 在同一个查询上度量 OpenAI Realtime vs STT+LLM+TTS 级联。文本级控制带来多少延迟成本？

关键术语 ​

术语大家怎么说它实际是什么
Frame「事件」流水线里带类型的数据单元（音频、转录、文本、控制）
Processor「流水线阶段」带 process(frame) 的处理器
DOWNSTREAM「前向流」源到汇：音频进，语音出
UPSTREAM「反馈流」控制：取消、指标、插话
VAD「语音活动检测」检测用户何时在说话
Semantic turn detection「智能回合结束」基于模型判断用户说完了
MultimodalAgent「直接音频 agent」音频进，音频出；中间没有文本
VoicePipelineAgent「级联 agent」STT + LLM + TTS；文本级控制

延伸阅读 ​