语音 Agent 架构：Realtime API + 端侧 ASR/TTS 实战

语音 Agent（Voice Agent）是 2024-2025 年 AI 应用最热的赛道之一。OpenAI Realtime API、LiveKit Agents、Pipecat 等框架的成熟，让"会说话的 AI 助手"从 demo 走向产品。但语音 Agent 远比文本 Agent 复杂——延迟、回声打断、流式合成、网络抖动、端侧部署——每一个环节都可能让用户体验崩溃。本文从工程实战出发，系统讲解语音 Agent 的分层架构、延迟优化、回声打断处理和端侧部署策略。

语音 Agent 的能力分层

一个完整的语音 Agent 由四层组成：

第 1 层：音频 I/O 层

• 麦克风采集（VAD 检测说话开始/结束）
• 扬声器播放（TTS 流式合成）
• 音频格式转换（PCM/Opus/AAC）
• 噪声抑制（AEC、NS、AGC）

第 2 层：ASR 层（语音识别）

• 实时流式识别
• 多语言切换
• 说话人识别（Speaker Diarization）
• 关键词唤醒（Wake Word）

第 3 层：Agent 推理层

• LLM 理解 + 决策
• 工具调用
• 记忆与上下文

第 4 层：TTS 层（语音合成）

• 流式合成
• 情感与语气控制
• 多声音切换
• 实时语速调整

这四层都必须以"低延迟 + 流式"为核心设计，任何一层卡顿都会让整个交互显得不自然。

延迟：语音 Agent 的生命线

延迟标准（人耳对延迟的敏感度）：

• < 200ms：无缝对话，用户感觉像跟人说话
• 200-500ms：可接受对话，但能感受到"机器感"
• 500-1000ms：明显卡顿，对话节奏破坏

• 1000ms：不可用

典型延迟分布：

环节	延迟
麦克风 → ASR	100-300ms
ASR → LLM	50-200ms
LLM 推理（流式首 token）	200-500ms
TTS 流式合成	100-300ms
TTS → 扬声器	50-100ms
总延迟	500-1400ms

优化目标：把 P50 延迟压到 600ms 以下，P95 压到 1200ms 以下。

OpenAI Realtime API 架构

OpenAI 在 2024 年底发布的 Realtime API 是当前最易用的语音 Agent 接口：

from openai import AsyncOpenAI
import asyncio

client = AsyncOpenAI()

async def voice_agent():
    async with client.beta.realtime.connect(model="gpt-4o-realtime-preview") as conn:
        # 配置会话
        await conn.session.update(session={
            "modalities": ["text", "audio"],
            "voice": "alloy",
            "instructions": "你是一个客服助手，回答用户关于订单的问题。",
            "tools": [query_order_tool, send_email_tool],
            "turn_detection": {
                "type": "server_vad",  # 服务端 VAD 自动检测说话结束
                "threshold": 0.5,
            },
        })
        
        # 启动音频流
        async def send_mic():
            # 从麦克风读取音频，发送给 OpenAI
            async for audio_chunk in microphone_stream():
                await conn.input_audio_buffer.append(audio=audio_chunk)
        
        async def receive_response():
            # 接收 OpenAI 的回复（可能是音频或工具调用）
            async for event in conn:
                if event.type == "response.audio.delta":
                    # 增量音频数据，直接播放
                    play_audio(event.delta)
                elif event.type == "response.function_call_arguments.done":
                    # 工具调用完成
                    result = await execute_tool(event)
                    await conn.conversation.item.create(
                        item={
                            "type": "function_call_output",
                            "call_id": event.call_id,
                            "output": json.dumps(result),
                        }
                    )
        
        await asyncio.gather(send_mic(), receive_response())

Realtime API 的优势：

• 集成度高：ASR + LLM + TTS 在一个 API 里
• 服务器端 VAD：自动检测说话结束，无需客户端逻辑
• 流式响应：音频是 token 级增量返回，延迟低
• 内置工具调用：支持 Function Calling

Realtime API 的局限：

• 依赖网络：必须连接到 OpenAI，不支持离线
• 成本高：按音频时长计费，比文本贵 10-100 倍
• 延迟受限：受网络往返影响，最优 500-800ms
• 私有化难：数据必须传到 OpenAI

LiveKit Agents 框架

LiveKit 是开源的实时通信框架，LiveKit Agents 是其 Agent 框架：

from livekit import agents
from livekit.agents import Agent, AgentSession, AutoSubscribe
from livekit.agents.stt import openai_stt
from livekit.agents.tts import openai_tts
from livekit.agents.llm import openai_llm
from livekit.plugins import silero  # 客户端 VAD

class MyAgent(Agent):
    def __init__(self):
        super().__init__(instructions="你是一个智能助手")
    
    async def on_enter(self):
        # Agent 进入会话
        self.session.say("你好，有什么可以帮你的？")

async def entrypoint(ctx: agents.JobContext):
    # 初始化各组件
    session = AgentSession(
        stt=openai_stt.STT(),
        llm=openai_llm.LLM(model="gpt-4o"),
        tts=openai_tts.TTS(voice="alloy"),
        vad=silero.VAD.load(),  # 客户端 VAD
    )
    
    await session.start(
        room=ctx.room,
        agent=MyAgent(),
    )

if __name__ == "__main__":
    agents.cli.run_app(agents.WorkerOptions(entrypoint_fnc=entrypoint))

LiveKit 的优势：

• 完全开源：可以私有化部署
• WebRTC 传输：低延迟、抗网络抖动
• 组件可插拔：STT/LLM/TTS/VAD 可独立替换
• 多 Agent 房间：支持多人多 Agent 协作

Pipecat 框架

Pipecat 是 Daily 公司的语音 Agent 框架，强调"pipeline"模式：

from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.frames.frames import TextFrame, AudioFrame, EndFrame
from pipecat.services.openai_stt import OpenAISTTService
from pipecat.services.openai_tts import OpenAITTSService
from pipecat.services.openai_llm import OpenAILLMService
from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketTransport

transport = FastAPIWebsocketTransport(websocket=websocket)

stt = OpenAISTTService(api_key=os.environ["OPENAI_API_KEY"])
llm = OpenAILLMService(model="gpt-4o")
tts = OpenAITTSService(voice="alloy")

pipeline = Pipeline([
    transport.input(),
    stt,
    llm,
    tts,
    transport.output(),
])

runner = PipelineRunner(pipeline)
runner.run()

Pipecat 的优势：

• Pipeline 模式：数据流清晰，易于调试
• Daily.co 集成：内置 WebRTC 传输
• 丰富的 Services：50+ 预集成 STT/TTS/LLM 服务
• 实时可视化：pipeline debugger

端侧 ASR/TTS 部署

对于隐私敏感场景（医疗、法律、企业内部），必须端侧部署：

# 端侧 ASR：whisper.cpp
from pywhispercpp.model import Model

model = Model("base.en", n_threads=4)
segments = model.transcribe("audio.wav")
for segment in segments:
    print(segment.text)

# 端侧 TTS：piper
from piper import PiperVoice
voice = PiperVoice.load("en_US-lessac-medium.onnx")
voice.synthesize("Hello, world!", "output.wav")

# 端侧 LLM：llama.cpp / Ollama
from llama_cpp import Llama
llm = Llama(model_path="llama-3.1-8b-instruct.gguf")
response = llm.create_chat_completion(
    messages=[{"role": "user", "content": "Hi"}],
    max_tokens=100,
)

端侧部署的工程权衡：

维度	云端	端侧
延迟	500-1000ms	200-500ms
成本	按量计费	一次性硬件投入
隐私	数据上传	数据本地
语音质量	高（商用模型）	中（开源模型）
维护	零	高（模型更新）

端侧典型配置：

• 硬件：Apple Silicon M2+ / 高通 8 Gen 2+ / Intel 12 代+
• ASR：whisper.cpp base 模型，~1GB 内存
• TTS：piper，~100MB 内存
• LLM：Qwen 2.5 3B / Llama 3.1 8B，~5GB 内存
• 总内存占用：~6-8GB

回声与打断：语音 Agent 的 UX 难点

问题 1：TTS 播放时收到自己的声音（Echo） 当 Agent 用 TTS 播放回复时，麦克风会再次拾取到这段声音，导致 ASR 把它当成"用户继续说话"。

解决方案：

• AEC（Acoustic Echo Cancellation）：硬件或软件回声消除
• 播放时静音麦克风：TTS 播放期间关闭录音
• VAD 阈值提高：播放时提高 VAD 阈值

# Pipecat 中的回声处理
class EchoCancellation:
    def __init__(self):
        self.is_playing = False
    
    async def on_tts_start(self):
        self.is_playing = True
        # 暂停 ASR
        await self.stt.pause()
    
    async def on_tts_end(self):
        self.is_playing = False
        # 恢复 ASR
        await self.stt.resume()

问题 2：用户打断 Agent

用户体验：Agent 正在说话时，用户想插话。

解决方案：

• Barging-in 检测：检测到新声音时立即停止 TTS
• VAD 持续时间：检测到 300ms 持续声音就视为打断
• 优先级切换：用户声音优先于 Agent 声音

class BargeInHandler:
    def __init__(self, vad, tts):
        self.vad = vad
        self.tts = tts
    
    async def on_voice_detected(self, confidence: float):
        if confidence > 0.7:  # 阈值
            await self.tts.interrupt()
            # 立即开始处理新输入

性能优化

1. 流式 ASR

不要等用户说完才开始识别——流式 ASR 在用户说话过程中就开始理解：

from livekit.agents.stt import openai_stt

stt = openai_stt.STT(
    model="whisper-1",
    interim_results=True,  # 流式返回中间结果
)

流式 ASR 可以把识别延迟从 1-2s 压到 200-500ms。

2. LLM 首 token 延迟

LLM 的 TTFT（Time To First Token）直接决定 Agent 的"反应速度"：

• 用更小的模型：3B / 7B 模型比 70B 快 5-10 倍
• Prompt 缓存：相同 system prompt 缓存，减少重复计算
• Speculative decoding：用小模型预测大模型输出

3. TTS 流式合成

TTS 不要等整句生成完再播放，而是边生成边播放：

from livekit.agents.tts import openai_tts

tts = openai_tts.TTS(
    model="tts-1",  # 流式模型
    voice="alloy",
    streaming=True,
)

流式 TTS 可以让首字节延迟从 500ms 压到 100ms。

失败模式与处理

失败模式	表现	处理
网络抖动	音频卡顿	WebRTC + 自适应码率
ASR 误识别	文字错误	关键词置信度过滤
LLM 超时	长时间无响应	缩短 prompt + 重试
TTS 合成失败	静音	降级到预录音频
用户打断失败	用户体验差	缩短 VAD 时间 + 优化 TTS 取消
多说话人混淆	识别错误	说话人识别 + 关键词锁定

实施路径

第 1 周：选框架（OpenAI Realtime API / LiveKit / Pipecat），跑通 demo。第 2 周：实现基本语音 Agent（STT → LLM → TTS），测量延迟基线。第 3 周：优化延迟（流式 ASR、流式 TTS、Prompt 缓存）。第 4 周：实现回声消除和打断处理。第 5 周：建立端侧 ASR/TTS 方案（如需隐私）。第 6 周：性能监控（延迟分布、错误率、用户满意度）。

总结

语音 Agent 的核心挑战是延迟和 UX。从 500ms 到 1500ms，体验差距是"流畅对话"与"明显卡顿"的区别。OpenAI Realtime API 适合快速验证；LiveKit Agents 适合私有化部署；Pipecat 适合复杂 pipeline 场景。

回声和打断是 UX 难点：TTS 播放时关闭麦克风、检测到新声音立即停止 TTS、缩短 VAD 时间——这些细节决定了用户是否愿意继续使用。

端侧 ASR/TTS 在隐私敏感场景是必须的，但需要权衡语音质量和硬件成本。

参考工具：LiveKit Agents（开源实时 Agent 框架）、Pipecat（Daily.co 的 pipeline 框架）、OpenAI Realtime Agents（OpenAI 官方 Realtime 示例）、ElevenLabs Python（高质量 TTS）和 Microsoft VibeVoice（微软语音工具）覆盖了语音 Agent 工具链的核心节点。