多模态开发（09）：全双工对话 Agent

第 7 篇讲 TTS 解决了"说"，第 8 篇讲 ASR 解决了"听"，把两者与 LLM 串成流水线，看起来就能对话了——识别成文字、交给模型、合成语音返回。真跑起来你会发现每轮问答要卡一两秒，而人对话的换句间隔只有两三百毫秒：延迟在语音交互里是第一等公民，慢半拍用户觉得"它在想"，慢一秒就是"卡死了"。本篇讲"边听边想边说"的全双工实时语音，先算清链式拼接为什么到不了实时，再对比三条实时路线，重点拆 OpenAI Realtime 与 Gemini Live 两套接口，落到打断、端点检测、首包优化三个深坑和一个语音客服实战。全文按"毫秒账 → 选型 → 接口 → 深坑 → 实战 → 成本"组织。

1、链式拼接的毫秒账

ASR → LLM → TTS 三段级联是多数团队的第一版语音 Agent，因为每段都是现成 API。问题出在账要一起算：三段串行，还叠加两笔容易被忽略的隐藏开销。

合计 1～2.5 s，这是每轮回答都逃不掉的固定开销。第一笔隐藏账是 VAD（Voice Activity Detection，语音活动检测）尾巴：你得等用户"说完"才能拿到完整语音，静音阈值压得太低又会把"嗯——"后面的半句截掉，这是结构性等待，优化空间有限。第二笔是段间完整性依赖：ASR 要完整音频才出定稿文本，LLM 要完整文本才组提示词，TTS 要完整首句才好切分合成——每一段都在等上一段的"完成信号"。

级联还有个只在实战里才疼的问题：打断。用户插话时，你要同时停掉 LLM 生成、清空 TTS 缓冲、掐断播放器，还要让三段状态机归位——每段单独都行，合在一起就是分布式取消问题。结论：级联适合录音转写、播客入库这类异步场景；实时对话，换架构。

2、实时方案三条路

2026 年做实时语音有三条路：模型厂商的原生 Realtime（端到端语音模型，一条会话同时听和说）、编排框架（把三段流水线全部流式化拼起来）、自建全链路（VAD + 流式三段自己拼）。

原生方案的体验天花板最高：端到端模型直接吃 PCM 吐 PCM，没有"转文字再转回来"的信息损耗，还能听出语气与犹豫。编排框架的代表是 pipecat——Daily.co 开源的 Python 框架，BSD-2 协议，既支持级联三段，也支持接入 OpenAI Realtime 与 Gemini Live 的语音到语音模式，传输层聚合 Daily、WebRTC、电话等渠道。要接电话线、要多厂商容灾的团队，用它省下的状态机代码是实打实的。自建三段自由度最高也最贵：把每段流式化、把打断做成一等公民，工程量约等于自己写一个 pipecat。

级联与端到端的链路差异如下：

图上最直观的是中间环节的数量：级联每段都要等上一段的完整产物才能开工，端到端把中间三跳折叠成一跳——首响延迟的差距本质是跳数差距。

选型一句话：先上原生 Realtime 验证产品假设，需要多渠道或容灾时引入 pipecat，自建只留给有特殊合规与成本约束的场景。

3、OpenAI Realtime 接入

OpenAI 的 Realtime API 用一条长连接承载整个会话，通道两种：WebSocket 适合服务端中继，WebRTC 适合浏览器与移动端直连（配临时 token，媒体流不绕服务器）。当前模型家族是 gpt-realtime（含 2、2.1 版本与 mini 档），整体事件驱动，细节见官方 Realtime 指南。WebSocket 会话骨架如下：

# openai_realtime.py —— WebSocket 会话骨架（websockets 库）
import asyncio, base64, json
from websockets import connect

URL = "wss://api.openai.com/v1/realtime?model=gpt-realtime"
HEADERS = {"Authorization": "Bearer $OPENAI_API_KEY",
           "OpenAI-Beta": "realtime=v1"}

async def main():
    async with connect(URL, additional_headers=HEADERS) as ws:
        await ws.send(json.dumps({"type": "session.update", "session": {
            "voice": "sage",                          # 音色
            "input_audio_format": "pcm16",            # 16 kHz 单声道
            "output_audio_format": "pcm16",
            "turn_detection": {"type": "server_vad"}, # 服务端断句
            "input_audio_transcription": {"model": "whisper-1"},
        }}))
        await asyncio.gather(push_mic(ws), handle_events(ws))

async def push_mic(ws):
    async for chunk in mic_stream():                  # 100 ms 一片
        await ws.send(json.dumps({
            "type": "input_audio_buffer.append",
            "audio": base64.b64encode(chunk).decode()}))

asyncio.run(main())

连接后先发 session.update 配置音色、音频格式与 VAD；push_mic 以 100 ms 一片推麦克风 PCM，两个协程并发才叫全双工。真正让 Realtime 好用的是事件循环里顺手就能做打断与函数调用：

# handle_events —— 主循环：音频、打断、函数调用
async def handle_events(ws):
    buffer = []                                       # 播放缓冲
    async for raw in ws:
        ev = json.loads(raw)
        t = ev["type"]
        if t == "response.audio.delta":               # 模型在说话
            buffer.append(base64.b64decode(ev["delta"]))
        elif t == "input_audio_buffer.speech_started":  # 用户开口
            buffer.clear()                            # 打断：清播放
            await ws.send(json.dumps({"type": "response.cancel"}))
        elif (t == "response.output_item.done"
                and ev["item"]["type"] == "function_call"):
            out = await call_tool(ev["item"])         # 查订单等本地函数
            await ws.send(json.dumps({
                "type": "conversation.item.create",
                "item": {"type": "function_call_output",
                         "call_id": ev["item"]["call_id"], "output": out}}))
            await ws.send(json.dumps({"type": "response.create"}))

打断的关键两行在 speech_started 分支：清播放缓冲、发 response.cancel 掐掉在途响应，这 20 行就是全双工的核心。函数调用走"工具定义进会话、模型产出调用、客户端回填结果、再触发响应"的回路——tools 在 session.update 里注册，参数格式与 Chat Completions 同构，第 6 节实战会接上真实的工具函数。当前计费音频按 token 计（见 OpenAI 价格页）：

两个省钱点：缓存命中的音频输入降到 $0.40/1M（mini 档 $0.30），重复的问候语与系统提示都会命中；同家族另有按分钟计费的分支——gpt-realtime-whisper 实时转写 $0.017/分钟、gpt-realtime-translate 同传 $0.034/分钟，价格页还给出按秒结算的 gpt-live 会话 $0.05/分钟。对话场景社区实测单分钟成本约 $0.04～0.20，语速与缓存命中影响极大，上量前务必用自己的语料压测。

4、Gemini Live 接入

Gemini Live API 是 Google 的对位产品：一条双向流连接，官方 SDK 封装为 client.aio.live.connect，也提供 WebSocket 接入，细节见 Live API 文档。模型从 gemini-2.5-flash-native-audio-preview 起步，2026 年已迭代到 gemini-3.8-live 家族（含 extended-thinking 变体），面向实时语音 Agent 优化。SDK 版接入骨架：

# gemini_live.py —— google-genai SDK 双向流式会话
import asyncio
from google import genai
from google.genai import types

client = genai.Client()                               # 读 GEMINI_API_KEY
CONFIG = types.LiveConnectConfig(
    response_modalities=["AUDIO"],
    system_instruction="你是电商语音客服，答不上来转人工。")

async def main():
    async with client.aio.live.connect(
            model="gemini-2.5-flash-native-audio-preview-12-2025",
            config=CONFIG) as session:
        await asyncio.gather(send_mic(session), recv(session))

async def send_mic(session):                          # 麦克风 → 模型
    async for chunk in mic_stream():                  # PCM 16 kHz 分片
        await session.send_realtime(types.Blob(
            data=chunk, mime_type="audio/pcm;rate=16000"))

async def recv(session):                              # 模型 → 扬声器
    async for msg in session.receive():
        sc = msg.server_content
        if sc and sc.output_audio:
            play(sc.output_audio.data)                # PCM 24 kHz
        if sc and sc.interrupted:                     # 用户打断
            clear_playback()

与 OpenAI 事件流一一对应：send_realtime 推麦克风分片，session.receive() 收增量音频与状态。打断是内置行为——服务端检测到插话会下发 interrupted 标记，客户端要做的是清空播放队列，别让已被否决的半句话继续播。函数调用同样支持：msg.tool_call 拿到调用、以 FunctionResponse 回填，思路与 OpenAI 一致。计费口径见 Gemini 价格页：gemini-3.8-live 音频输入 $3/1M token（官方折算 $0.005/分钟）、音频输出 $12/1M（折算 $0.018/分钟），文本输入 $0.75/1M、输出 $4.50/1M。

对开发者最有吸引力的是标准层免费额度：Live 系列模型在免费层标注免费，做原型与内测基本不花钱。两个注意：免费层数据可能被用于改进产品，生产环境别默认走它；限流口径价格页未明示，社区实测配额比名义请求数更早耗尽，压测前先读官方 rate limits 文档。

5、打断与首包延迟

三条路线都绕不开三个坑：打断、端点检测、首包延迟，逐个说。

5.1、打断处理

**打断（barge-in）**指用户在 Agent 播报时插话，正确处理要做三件事：停生成，用 response.cancel 之类指令掐掉在途响应；清播放，播放缓冲里未播的音频全部丢弃，否则用户已经听到新回应、扬声器还在念旧稿；记上下文，被打断的那句不要捡起来复读——OpenAI 在会话里标记未播完的音频，Gemini 用 interrupted 事件区分。两处细节容易翻车：播放端骤停会出爆音，留 20 ms 淡出；打断风暴（用户连说带咳）要加 200 ms 去抖，别让响应反复起停。级联方案这三件事要在三个服务间自己协调，这正是原生方案最大的体验红利。

5.2、VAD 端点检测

端点检测回答"用户说完了吗"。OpenAI 的 server_vad 默认静音约 500 ms 判定断句，silence_duration_ms 等参数可调：压到 300 ms 换句更快，代价是拖音被拦腰截断、模型答非所问。背景噪音与电视声还会持续触发假断句，生产上要么在采集端开噪声抑制，要么在业务关键节点（用户报手机号、订单号时）改用显式确认——"好的，订单号 8866，对吗"。自建 VAD 选型上，Silero VAD 一类轻量模型单核 CPU 实时跑几十路，是流式管线的常客。

5.3、首包延迟优化

首包延迟是从用户说完到你出声的间隔。原生方案厂商已做完大部分优化：端到端模型省掉段间往返，你要做的是会话预热——提前建连、提前发 session.update，别让用户第一句话等握手。级联方案的优化清单长得多：LLM 流式输出按句切给 TTS、首句用预合成的短话术垫场（一句"正在为您查询"能盖住 500 ms）、接入点选离模型最近的区域、播放缓冲压到 100 ms 以内。每项都是工程活，这也是第 2 节把级联首响标到 800～1500 ms 的原因——那是把优化全做完的下限，不是默认值。

6、语音客服 Agent 实战

串一个完整场景：电商语音客服，用户开口查物流，能被打断，工具超时有兜底。架构是浏览器麦克风 ⇄ FastAPI WebSocket 中继 ⇄ Realtime 会话，中继层拦两类消息：浏览器的音频分片直通，模型的函数调用就地处理。会话与工具注册：

# agent_relay.py —— FastAPI 中继：浏览器 ⇄ Realtime
import asyncio, json
from fastapi import FastAPI, WebSocket

app = FastAPI()
TOOLS = [{"type": "function", "name": "get_order_status",
          "description": "按订单号查询物流状态",
          "parameters": {"type": "object", "properties": {
              "order_id": {"type": "string"}},
              "required": ["order_id"]}}]

@app.websocket("/voice")
async def voice(ws: WebSocket):
    await ws.accept()
    async with realtime_connect("gpt-realtime") as rt:  # 第 3 节骨架
        await rt.send(json.dumps({"type": "session.update", "session": {
            "instructions": "电商客服，查物流先调 get_order_status。",
            "tools": TOOLS,
            "turn_detection": {"type": "server_vad"}}}))
        await asyncio.gather(pump(ws, rt), pump(rt, ws))

realtime_connect 复用第 3 节的会话建立；两个 pump 把浏览器与 Realtime 连成双向管道，两端都按"文本消息的异步迭代器"看待。转发器的核心是拦截函数调用并回填：

async def pump(src, dst):                          # 双向转发消息
    async for text in src:
        ev = json.loads(text)
        if ev.get("type") == "response.output_item.done" \
                and ev["item"]["type"] == "function_call":
            text = json.dumps({                    # 拦截函数调用
                "type": "conversation.item.create",
                "item": {"type": "function_call_output",
                         "call_id": ev["item"]["call_id"],
                         "output": await call_tool(ev["item"])}})
        await dst.send(text)

async def call_tool(item):                         # 2 秒超时兜底
    try:
        args = json.loads(item["arguments"])
        return await asyncio.wait_for(
            get_order_status(**args), timeout=2)
    except asyncio.TimeoutError:
        return json.dumps({"error": "timeout",
                           "say": "查询超时，为您转人工。"})

普通消息原样透传（音频分片、转写增量都不动），函数调用被拦截后就地执行并改写成 function_call_output 回填，服务端自动续上响应。兜底逻辑在 call_tool：内部订单接口 2 s 不回，就回填一条带兜底话术的超时结果——Realtime 会话里 function_call_output 缺席会让对话挂起，这是新手最容易撞上的卡死点。打断沿用第 3 节 speech_started → response.cancel 的拦截逻辑，浏览器端收到清空播放队列即可。

7、实时会话成本账

实时是按"会话时长 × 语速"烧钱，与级联各段各计各的有本质区别。按 10 分钟客服对话折算：

级联三段沿用系列第 7、8 篇口径：ASR $0.006/分钟、TTS 约 $0.015/分钟、对话轮次的 LLM 文本 token 几乎可忽略。两个结论：其一，Gemini Live 付费层已经与级联打平，"实时必然贵一个数量级"在 2026 年不再成立，选型该回到体验与生态；其二，OpenAI 标准档仍是级联的 3～5 倍，mini 档与音频缓存是省钱的常规操作。

成本警告：实时会话挂起不清是事故级浪费——用户已挂机、连接还活着、VAD 还在喂环境音。生产必配最大会话时长与闲置超时断连，监控盯 token 消耗而不是请求数。

8、总结

链式拼接输在串行等待与段间完整性依赖，1～2.5 s 的首响撑不起对话。三条路里，原生 Realtime 体验最好、开发最省；pipecat 换来多厂商与电话渠道；自建只留给硬约束场景。OpenAI Realtime 用 WebSocket/WebRTC 加一套事件流把全双工、打断、函数调用都收进会话，gpt-realtime 按 token 计费，mini 档与缓存是省钱位；Gemini Live 用双向流接入，官方折算约 $0.023/分钟且标准层免费，开发内测成本近零。工程上盯死三件事：打断三连（停生成、清播放、记上下文）、端点阈值与截断兜底、会话预热与首句垫场。

练习：给现有的 Agent 加一条语音通道，分三步走。

1. 用第 3 节骨架跑通浏览器到 Realtime 的回声对话；
2. 注册一个函数调用，让它真的查到一条数据；
3. 对着它插话，验证打断三连全部生效。

对于想把文本 Agent 升级为语音交互的团队，建议先用 Gemini 免费层验证交互设计，再按并发与成本预算落到 OpenAI 或混合部署。