语音 Agent:Pipecat 与 LiveKit

语音 Agent:Pipecat 与 LiveKit

本节摘要:语音 Agent 不是「文本循环外挂个 TTS」那么简单。它的延迟预算极其残酷——生产目标端到端 450~600ms,超过 1500ms 用户就觉得「卡坏了」;部分音频(partial audio)是默认状态而非异常(用户边说边出结果);轮次检测本身是个模型(怎么判断用户说完了?);传输从电话 SIP 到 WebRTC 五花八门。两条路:Pipecat(pipecat-ai/pipecat)给你一个 Python 的、基于帧(frame)的流水线框架——Frame 流经 FrameProcessor 链,分两个方向:DOWNSTREAM(源→汇,音频进、TTS 出)与 UPSTREAM(反馈与控制:取消、指标、打断 barge-in),典型五段是 VAD(Silero 语音活动检测)→ STT(语音转文字)→ LLM(上下文在用户/助手间交替)→ TTS(文字转语音)→ transport;LiveKit Agents(livekit/agents)把 AI 模型经 WebRTC 桥接给用户,两个语音 Agent 类——MultimodalAgent(OpenAI Realtime 式直连音频)、VoicePipelineAgent(STT→LLM→TTS 级联、给你文本级控制),带语义轮次检测与原生 MCP 集成。商业平台 Vapi 与 Retell 在其上构建,优化后的高端栈能压到 450600ms。本节吃透两套架构、2026 年各环节典型延迟,并用标准库实现一个带 VAD→STT→LLM→TTS→传输五段、含 UPSTREAM 取消帧演示打断的帧式流水线。

对应原课程:Phase 14 · Lesson 22 · voice-agents-pipecat-livekit(原英文 phases/14-agent-engineering/22-voice-agents-pipecat-livekit/docs/en.md)。前置:第 01 节(Agent 循环)、第 12 节(工作流模式)。

学习目标

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

1. 描述 Pipecat 的帧式流水线:DOWNSTREAM(源→汇)与 UPSTREAM(控制)。
2. 说出标准语音流水线的各段,以及 Pipecat 支持哪些传输。
3. 解释 LiveKit Agents 的两个语音 Agent 类(MultimodalAgent、VoicePipelineAgent)及各自适合的场景。
4. 总结 2026 年生产的延迟预期,以及它如何驱动架构选择。
5. 识别四种语音 Agent 失败:无打断处理、忽略 STT 置信度、TTS 句中截断、忽视延迟预算。

一、问题与直觉

文本 Agent 的延迟预算是「秒级」——用户能等。语音 Agent 的延迟预算是「亚秒级」——超过 600ms 用户就觉得迟钝,超过 1.5 秒对话就崩了。这彻底改变了架构:

• 部分音频是默认:用户还在说话时,STT 已经在吐部分转录,LLM 已经在生成首 token,TTS 已经在合成首段音频。每一级都要流式。
• 轮次检测是个模型:文本对话里「用户发完消息」是显式的回车;语音里,沉默 200ms 是「说完了」还是「喘口气」?需要一个模型来判断(语义轮次检测)。
• 打断(barge-in)是核心交互:用户中途插话,Agent 必须立即停止 TTS——否则就是各说各的。
• 传输五花八门:电话 SIP、WebRTC、WebSocket、WhatsApp,各有延迟与编码差异。

Pipecat(pipecat-ai/pipecat)

• Python 的帧式流水线框架。
• Frame(帧)→ FrameProcessor(帧处理器)链。
• 两个流向:
  • DOWNSTREAM —— 源→汇(音频进、TTS 出)。
  • UPSTREAM —— 反馈与控制(取消、指标、打断 barge-in)。
• PipelineTask 管理生命周期,带事件(on_pipeline_started、on_pipeline_finished、on_idle_timeout)与指标/追踪/RTVI 观察者。

典型流水线:

VAD(Silero) → STT → LLM(上下文在用户/助手间交替)→ TTS → transport
​

传输:Daily、LiveKit、SmallWebRTCTransport、FastAPI WebSocket、WhatsApp。Pipecat Flows 加结构化对话(状态机);Pipecat Cloud 是托管运行时。

LiveKit Agents(livekit/agents)

• 把 AI 模型经 WebRTC 桥接给用户。
• 关键概念:Agent、AgentSession、entrypoint、AgentServer。
• 两个语音 Agent 类:
  • MultimodalAgent —— 经 OpenAI Realtime 或等价物直连音频(音频进、音频出,中间无文本)。
  • VoicePipelineAgent —— STT → LLM → TTS 级联;给你文本级控制(可以在中间改文本)。
• 经 transformer 模型做语义轮次检测。
• 原生 MCP 集成。
• 经 SIP 支持电话。
• LiveKit Inference 提供 50+ 模型(无需 API key),另有 200+ 经插件。

商业平台

Vapi(优化后的高端栈约 450~600ms)与 Retell(180 次测试电话端到端约 600ms)在其上构建。当你想要托管语音栈、又不想养一个 WebRTC 团队时,选平台。

典型 2026 延迟

• VAD:20~60ms
• STT 部分:100~250ms
• LLM 首 token:150~400ms
• TTS 首音频:100~200ms
• 传输 RTT:30~80ms

端到端 450600ms 是高端;8001200ms 常见;超过 1500ms 感觉坏了。每个组件加 50~200ms,上线前先把链路加一遍。

⚠️ 四种失败模式:① 无打断处理——用户插话,Agent 还在说;Pipecat 需 UPSTREAM 取消帧,LiveKit 需等价物。② 忽略 STT 置信度——低置信转录当真理喂给 LLM;应按置信度门控或请求确认。③ TTS 句中截断——流水线在话说到一半取消时,TTS 要知道、或切断音频,否则出杂音。④ 忽视延迟预算——每个组件加 50~200ms,不先加一遍链路就上线,常超 1.5 秒。

二、从零实现

原课程 code/main.py 是一个帧式玩具流水线:

• Frame 类型(音频、转录、文本、tts_audio、控制)。
• Processor 接口,带 process(frame)。
• 一个五段流水线(VAD → STT → LLM → TTS → transport),作为脚本化处理器。
• 一个 UPSTREAM 取消帧,演示打断。

Step 1:帧与处理器接口

@dataclass
class Frame:
    kind: str        # audio / transcript / text / tts_audio / control
    payload: Any
    confidence: float = 1.0

class Processor:
    def process(self, frame: Frame): ...
    downstream: "Processor" = None
    upstream:   "Processor" = None
​

Step 2:五段流水线(DOWNSTREAM)

class VAD(Processor):       # 语音活动检测
    def process(self, f):
        if f.kind == "audio" and self.is_speech(f.payload):
            self.downstream.process(Frame("audio", f.payload))

class STT(Processor):       # 语音转文字(流式部分转录)
    def process(self, f):
        text, conf = self.transcribe(f.payload)
        self.downstream.process(Frame("transcript", text, conf))

class LLM(Processor):       # 上下文交替,首 token 流式
    def process(self, f):
        if f.kind == "transcript":
            if f.confidence < 0.6:              # 置信度门控
                self.downstream.process(Frame("text", "能再说一遍吗?")); return
            for tok in self.generate(f.payload):
                self.downstream.process(Frame("text", tok))

class TTS(Processor):       # 文字转语音,首音频流式
    playing = True
    def process(self, f):
        if not self.playing: return             # 被 UPSTREAM 取消
        audio = self.synthesize(f.payload)
        self.downstream.process(Frame("tts_audio", audio))
​

Step 3:UPSTREAM 取消帧(打断)

def barge_in(pipeline):
    cancel = Frame("control", {"cmd": "cancel"})   # 用户插话
    # 反向传播:TTS 停止、LLM 中断生成
    pipeline.tts.playing = False
    pipeline.llm.stop()
​

运行 python3 code/main.py 会展示正常流,以及一次打断取消在 TTS 句中停止的轨迹。

💡 设计要点:语音 Agent 的两个流向同等重要——DOWNSTREAM 产出语音,UPSTREAM 让用户能打断。没有 UPSTREAM 的取消路径,Agent 就是个「自说自话不听人」的设备,体验直接崩塌。这与第 01 节的 Agent 循环同源,只是循环的「观察」变成了流式音频帧。

三、框架对比

四、可复用产物

原课程 outputs/skill-voice-pipeline.md:脚手架一个 Pipecat 形状的语音流水线——VAD + STT + LLM + TTS + transport,带打断处理、置信度门控、指标观察者。

五、练习

1. (Easy) 给玩具流水线加一个指标观察者:每段每秒帧数。延迟累积在哪?
2. (Medium) 实现置信度门控 STT:低于阈值就请求「能再说一遍吗?」
3. (Medium) 加语义轮次检测:简单规则——转录以「?」结尾即视为轮次结束。
4. (Hard) 读 Pipecat 传输文档,把标准库传输换成 SmallWebRTCTransport 配置(stub)。
5. (Hard) 在同一查询上测 OpenAI Realtime vs STT+LLM+TTS 级联。文本级控制带来多少延迟成本?

本节要点回顾

1. 语音 Agent ≠ 文本 + TTS:延迟预算亚秒级(450~600ms 高端,>1500ms 崩),部分音频是默认,轮次检测是模型。
2. Pipecat 帧式流水线:Frame→FrameProcessor 链,DOWNSTREAM(源→汇,音频进 TTS 出)+ UPSTREAM(控制:取消/指标/打断)。
3. 典型五段:VAD(Silero)→ STT → LLM(上下文交替)→ TTS → transport;传输含 Daily/LiveKit/WebSocket/WhatsApp/SIP。
4. LiveKit 两个语音类:MultimodalAgent(直连音频)、VoicePipelineAgent(STT→LLM→TTS 级联,文本级控制)。
5. 语义轮次检测:transformer 模型判断用户是否说完;不是简单静默计时。
6. 商业平台:Vapi/Retell 在 Pipecat/LiveKit 上构建,托管 450~600ms,适合不要 WebRTC 团队的团队。
7. 2026 各段延迟:VAD 2060ms、STT 部分 100250ms、LLM 首 token 150400ms、TTS 首音频 100200ms、传输 RTT 30~80ms。
8. 四种失败:无打断处理、忽略 STT 置信度、TTS 句中截断、忽视延迟预算(上线前先加链路)。
9. 两个流向同等重要:DOWNSTREAM 产出、UPSTREAM 让用户打断;没 UPSTREAM 取消路径体验崩。
10. 选型:要全控制用 Pipecat,WebRTC/电话用 LiveKit,托管用 Vapi/Retell,直连音频用 OpenAI Realtime/Gemini Live。

下一节,我们进入 Agent 工程的可观测性地基——OpenTelemetry GenAI 语义约定,看如何用统一的标准 span 形状,让 LLM 调用、工具调用、Agent 跨进程的追踪能汇成一条可查询的轨迹。