Agent 工程体系 · 第 57/98 篇。内容以 2026 年 9 月可验证的公开规范和稳定接口为基线；框架版本敏感能力会明确标注，不把实验行为写成通用保证。

语音 Agent：VAD、ASR、流式推理、TTS、打断和延迟预算

语音 Agent 不是“给文本 Agent 接一个麦克风和一个语音合成接口”。它是一个持续运行的实时系统：输入以音频帧到达，系统需要判断用户何时开始、何时结束、说了什么、是否仍然需要等待更多上下文；同时，模型可能已经开始生成回答，TTS 也可能已经把回答转换成音频并播放。用户在任意时刻重新开口，系统都必须决定是继续播放、暂停、取消、保留还是回滚当前回答。

因此，语音 Agent 的核心问题不是单一模型能力，而是实时事件、部分结果、并发任务和可取消副作用之间的协调。

主流实现有两类：

语音到语音（speech-to-speech）：模型直接处理实时音频并生成音频，减少显式的 ASR、文本推理和 TTS 之间的边界，适合自然对话和低首包延迟。
链式语音管线（chained voice pipeline）：显式执行 VAD、ASR、文本 Agent、TTS，每一阶段都可观测、可替换、可审计，适合客服、审批、合规和已有文本 Agent 的复用。

这一区分不是语言 SDK 的差异，而是系统控制权的差异：前者把更多实时协作交给语音模型和会话层，后者把中间文本、策略检查、工具调用和持久化交给应用。OpenAI 当前的语音 Agent 文档也将这两种架构作为主要选择，并分别对应低延迟自然交互与可控、可预测的链式流程。(developers.openai.com)

一、先定义语音 Agent 的实时闭环

一个最小的链式语音 Agent 可以表示为：

麦克风
  │
  ▼
音频采集与降噪
  │
  ▼
VAD：判断是否有人说话
  │
  ▼
ASR：音频 → 部分/最终文本
  │
  ▼
文本 Agent：推理、记忆、工具调用
  │
  ▼
TTS：文本 → 音频块
  │
  ▼
播放器
  │
  └──── 用户再次说话 ────► 打断当前输出

但是，这个图仍然过于串行。生产系统通常是并发的：

flowchart LR
    Mic[麦克风音频帧] --> VAD[VAD / Turn Detection]
    VAD --> ASR[ASR 流式转写]
    ASR --> Transcript[部分或最终转写]
    Transcript --> Agent[文本 Agent / 工具循环]
    Agent --> Delta[流式文本 Delta]
    Delta --> TTS[TTS 流式合成]
    TTS --> AudioQ[可取消音频队列]
    AudioQ --> Player[播放器]

    Mic --> Bararge[用户语音检测]
    Bararge --> Cancel[取消当前生成与播放]
    Cancel --> AudioQ
    Cancel --> Agent

这里有三个同时运行的时间轴：

输入时间轴：音频帧不断到达；
推理时间轴：模型逐步生成文本、调用工具或等待工具结果；
输出时间轴：TTS 逐块合成并播放音频。

它们不是一条队列，而是三个可能相互抢占的流水线。一个可靠的设计必须定义：

哪些事件可以启动下一阶段；
哪些事件只代表“部分结果”，不能提交为最终事实；
哪些任务可以并行；
哪些任务必须取消；
取消后哪些状态可以保留；
断线后如何恢复而不重复调用工具或重复播放语音。

二、VAD：检测声音不等于判断一句话结束

2.1 VAD 的定义

VAD（Voice Activity Detection，语音活动检测）负责判断某一段音频中是否存在人类语音，或者更准确地说，判断输入流当前是否处于：

静音；
语音开始；
持续说话；
语音结束；
噪声或非语音活动。

VAD 通常作用在短音频帧上。例如，系统可能每 10～30 毫秒接收一个音频帧，对每个帧计算语音概率：

p_t = P(\text{speech} \mid x_t)

其中：

$x_t$ 是时刻 $t$ 的音频帧；
$p_t$ 是该帧为语音的概率。

最简单的判定是：

\text{speech}_t = \begin{cases} 1, & p_t \ge \theta \\ 0, & p_t < \theta \end{cases}

$\theta$ 是语音阈值。但单帧阈值会产生抖动：一个音节可能被判为语音，下一帧被判为静音，再下一帧又被判为语音。因此工程实现通常使用连续帧、前置缓冲和静音持续时间。

2.2 三个必须分开的时间参数

一个完整的 VAD 需要至少处理三个时间概念：

前置缓冲（prefix padding）
VAD 检测到语音后，补回检测前的一小段音频，避免截掉辅音、爆破音或第一个词。
起始确认窗口
需要连续若干帧超过阈值，才确认“用户真的开始说话”，避免键盘声、碰撞声触发对话。
结束静音窗口（silence duration）
只有持续静音超过某个时间，才确认用户停止说话。

设：

$F$ 为帧长；
$N_{\text{start}}$ 为起始确认所需连续语音帧数；
$N_{\text{stop}}$ 为结束确认所需连续静音帧数。

则 VAD 的理论检测延迟近似为：

L_{\text{start}} \approx N_{\text{start}}F

L_{\text{stop}} \approx N_{\text{stop}}F

如果帧长为 20 毫秒，结束静音窗口为 500 毫秒，那么即使用户已经说完，系统也可能要等待约 500 毫秒才提交这一轮输入。这部分等待不属于 ASR 或 LLM 延迟，而是回合检测延迟。

OpenAI Realtime VAD 文档将 server_vad 定义为基于静音区间切分音频，并提供阈值、前置缓冲和静音持续时间等配置；同时还提供基于语义判断用户是否说完的 semantic_vad。(developers.openai.com)

2.3 `server_vad` 与 `semantic_vad`

基于静音的 VAD 只回答：

“用户已经静音了足够长时间吗？”

它的优点是快、简单、行为容易解释；缺点是无法区分：

我想查询一下……

和：

我想查询一下。

第一个句子的停顿可能只是用户思考，第二个句子的停顿才可能表示结束。

语义 VAD 试图回答：

“从用户已经说出的词来看，这句话是否听起来已经完整？”

它对“嗯……我再想想”和“请帮我取消订单”可以采用不同等待策略。语义 VAD 的 eagerness 本质上是在调节最大等待时间：高 eagerness 更快切分，但更容易抢断用户；低 eagerness 更愿意等待，但首响应延迟更高。(developers.openai.com)

这产生一个无法消除的取舍：

策略	反应速度	抢话风险	适合场景
短静音窗口	高	高	简短命令、免手操作
长静音窗口	低	低	客服、复杂描述
高语义 eagerness	高	中高	明确问句、事务指令
低语义 eagerness	低	低	思考型对话、长句表达

2.4 VAD 的反例：把播放回声当成用户语音

最常见的错误不是 VAD 模型精度不够，而是系统没有处理扬声器回声：

Agent 播放：“您的订单已经……”
麦克风采集到扬声器输出；
VAD 判断检测到语音；
系统认为用户打断；
播放被取消；
对话陷入自我打断循环。

因此，VAD 前面通常需要：

回声消除（AEC）；
噪声抑制；
自动增益控制；
麦克风与播放器状态关联；
必要时使用双通道参考信号。

不能把“检测到人声”直接等同于“用户有意发起新回合”。真正的打断信号应当是：

\text{interrupt} = \text{speech\_detected} \land \text{user\_channel\_active} \land \text{not\_echo}

在高风险流程中，还可以要求检测到至少一个稳定的 ASR 文本片段后再执行硬取消，从而降低误触发，但这会增加打断延迟。

三、ASR：流式转写必须区分 partial、final 和 revision

3.1 ASR 的输入输出

ASR（Automatic Speech Recognition，自动语音识别）将音频转换成文字。离线 ASR 可以等待整段音频结束后输出一个结果；流式 ASR 则在用户仍然说话时不断输出部分结果。

设完整语音为：

帮我查一下明天杭州到上海的高铁

流式 ASR 可能依次输出：

帮我
帮我查一下
帮我查一下明天杭州
帮我查一下明天杭州到上海
帮我查一下明天杭州到上海的高铁

这些不是五条用户消息，而是同一条消息的五个观察版本。

更复杂的是，ASR 可能修订前面的内容：

帮我查一下明天杭州到上……
帮我查一下明天杭州到上海……

因此，ASR 事件至少应包含：

{
  "type": "asr.delta",
  "turn_id": "turn-42",
  "text": "帮我查一下明天杭州",
  "revision": 3,
  "is_final": false,
  "audio_end_ms": 1280
}

最终提交时：

{
  "type": "asr.final",
  "turn_id": "turn-42",
  "text": "帮我查一下明天杭州到上海的高铁",
  "revision": 7,
  "is_final": true,
  "audio_end_ms": 2640
}

3.2 为什么不能把 partial 直接交给 Agent

如果每个 partial 都启动一次 Agent，会产生：

帮我
→ Agent 开始回答：“好的，我来……”

帮我查一下
→ Agent 又开始回答：“您想查询什么？”

帮我查一下明天杭州
→ Agent 再次调用交通查询工具

结果是：

重复推理；
重复工具调用；
大量无效 TTS；
上一个回答尚未取消就生成了下一个回答；
会话状态出现多个互相矛盾的用户消息。

因此，partial 文本通常只能用于：

UI 实时显示；
预测性意图识别；
提前准备候选工具；
低风险的预取；
语音端点检测辅助。

只有满足提交条件后，才应把它写入对话历史并启动正式 Agent。提交条件通常是：

\text{commit} = \text{vad\_stop} \land \text{asr\_final} \land \text{revision\_stable}

对于支持语义回合检测的系统，也可以使用：

\text{commit} = \text{semantic\_turn\_end} \land \text{asr\_final}

3.3 ASR 的准确性不只看 WER

WER（Word Error Rate）常用于衡量识别错误：

WER = \frac{S + D + I}{N}

其中：

$S$ ：替换错误；
$D$ ：删除错误；
$I$ ：插入错误；
$N$ ：参考文本词数。

但对 Agent 来说，所有错误并不等价：

查询订单 12345
查询订单 12315

只错一个数字，业务结果可能完全不同。

因此应额外评估：

专有名词错误率；
数字、日期、金额错误率；
工具参数槽位准确率；
否定词准确率；
端点截断率；
多语言或中英混说错误率；
低置信度内容是否被正确确认。

对结构化参数，建议保留 ASR 的来源和置信度：

{
  "order_id": {
    "value": "12345",
    "source": "asr",
    "confidence": 0.61,
    "requires_confirmation": true
  }
}

不要让模型把低置信度数字“猜完整”。如果 ASR 返回“订单一二三四五”和“订单一二三五”之间存在不确定性，Agent 应向用户确认，而不是直接执行取消、退款或转账。

四、流式推理：输出 Delta 不是可播放文本

4.1 流式推理的含义

流式推理是指模型在完整回答生成之前，就不断发送部分输出。文本流通常由多个 delta 组成：

好的
好的，我
好的，我先
好的，我先帮您
好的，我先帮您查询

流式的价值有两层：

降低感知延迟：用户不必等待完整回答；
支持并发流水线：TTS 可以在模型尚未结束时开始合成。

但模型的 token 边界与自然语言边界不同。不能简单地“每收到一个 token 就调用一次 TTS”。

4.2 TTS 分块需要语义边界

一个常见的文本缓冲器可以按照以下条件发出 TTS chunk：

遇到句号、问号、感叹号；
达到最小字符数；
遇到逗号且当前句子已经足够长；
检测到工具调用前的固定提示；
当前 token 流暂停超过某个时间。

例如：

好的，我先帮您查询

不宜立即合成，因为它可能继续变成：

好的，我先帮您查询明天杭州到上海的高铁余票。

更适合的策略是：

好的，我先帮您查询明天杭州到上海的高铁余票。

作为一个完整 TTS chunk。

但是，如果模型需要调用一个耗时工具，也不能一直沉默。可以使用硬编码的过渡语：

好的，我正在查询，请稍等。

然后并行调用工具。对固定确认语、拒绝语和等待提示，使用预生成音频通常比让 LLM 生成更快、更稳定。OpenAI 的延迟优化建议也明确指出，高度受限的输出可以使用硬编码，而不必默认调用语言模型。(developers.openai.com)

4.3 流式 TTS 的两种状态

TTS 输出至少需要区分：

GENERATING：模型仍可能继续追加文本
DRAINING：模型已结束，正在播放已生成音频

例如模型输出：

您的订单已经为您找到。

TTS 可能已经生成并播放了：

您的订单已经

此时用户打断，系统不能只取消 TTS 生成任务，还必须清空播放器中尚未播放的音频帧。

因此，播放队列不能是不可控的字节流，而应带有 generation_id：

{
  "generation_id": "gen-9",
  "chunk_id": 4,
  "audio": "<bytes>",
  "text_span": [12, 18],
  "cancelled": false
}

播放器只允许播放当前 generation：

def should_play(chunk, current_generation):
    return (
        chunk.generation_id == current_generation
        and not chunk.cancelled
    )

这可以防止旧回答在新回合开始后继续从队列中漏出。

五、打断：真正的打断是取消一组副作用

5.1 打断不是“停止播放”

用户说话时打断 Agent，至少涉及四个动作：

停止向扬声器发送新音频；
清空已经排队但尚未播放的音频；
取消或标记当前 TTS 任务；
取消当前模型生成、工具调用或下游工作。

但第 4 步不能机械执行。工具调用可能已经产生不可逆副作用：

Agent 调用“提交退款”
用户此时说：“等等，不是这个订单”

如果退款请求已经提交，取消 HTTP 请求并不代表业务动作被撤销。网络取消只取消了客户端等待，不一定取消了服务器端执行。

因此应区分三类任务：

任务类型	是否可取消	打断后的处理
文本生成	通常可取消	立即取消
TTS 生成	通常可取消	取消并清空音频队列
查询类工具	通常可取消或忽略结果	取消、超时或丢弃旧结果
写入、支付、退款	不应假设可取消	使用幂等键、状态查询和补偿流程

5.2 使用 generation_id 隔离旧回合

每一次用户回合或 Agent 回答都生成一个单调递增的 ID：

turn-1 → generation-1
turn-2 → generation-2
turn-3 → generation-3

任何异步结果返回时，都必须检查它是否仍属于当前 generation：

async def deliver_tool_result(result, generation_id):
    if generation_id != session.current_generation:
        return  # 旧回合结果，不能再驱动当前回答

    await agent.resume(result)

打断流程：

async def interrupt(session):
    session.current_generation += 1
    new_generation = session.current_generation

    await session.audio_player.stop()
    session.audio_player.clear()

    session.cancel_task("tts")
    session.cancel_task("llm")

    # 对已提交的工具调用不假设可撤销
    # 通过幂等状态查询或补偿逻辑处理
    return new_generation

这里的关键不是 Python 代码本身，而是旧任务即使无法物理取消，也不能继续影响新回合。

5.3 打断的状态机

stateDiagram-v2
    [*] --> Listening
    Listening --> UserSpeaking: speech_started
    UserSpeaking --> Transcribing: speech_stopped
    Transcribing --> Thinking: asr_final
    Thinking --> Speaking: first_tts_chunk
    Speaking --> Speaking: tts_chunk
    Speaking --> Interrupted: speech_started
    Thinking --> Interrupted: speech_started
    Interrupted --> Listening: cancel_and_flush
    Thinking --> WaitingTool: tool_call
    WaitingTool --> Thinking: tool_result
    WaitingTool --> Interrupted: speech_started
    Speaking --> Listening: finish_and_drain
    Transcribing --> Listening: asr_error
    WaitingTool --> Recovery: tool_timeout
    Recovery --> Listening: fallback_or_retry

几个状态容易被混淆：

UserSpeaking：检测到输入语音，但还没有最终文本；
Transcribing：等待 ASR 最终结果；
Thinking：模型正在生成或决定是否调用工具；
WaitingTool：模型暂停，等待工具结果；
Speaking：至少有一部分音频已经交给播放器；
Interrupted：当前输出被新用户回合抢占；
Recovery：发生超时、断线或工具状态不确定。

一个错误设计是把“模型返回了文本”直接当成“回答完成”。对于流式 TTS，真正的完成至少需要：

模型完成
+ TTS 完成
+ 播放队列排空

六、延迟预算：从用户说完到听到首个音频

6.1 端到端延迟公式

语音 Agent 的首音频延迟可以拆成：

L_{\text{first-audio}} = L_{\text{endpoint}} + L_{\text{ASR}} + L_{\text{queue}} + L_{\text{agent}} + L_{\text{tool}} + L_{\text{TTS-first}} + L_{\text{network}} + L_{\text{playback}}

其中：

$L_{\text{endpoint}}$ ：VAD 判断用户结束的等待；
$L_{\text{ASR}}$ ：从音频结束到最终文本可用；
$L_{\text{queue}}$ ：排队、调度和限流等待；
$L_{\text{agent}}$ ：模型开始输出可播放文本前的时间；
$L_{\text{tool}}$ ：工具调用时间；
$L_{\text{TTS-first}}$ ：TTS 生成首个音频块的时间；
$L_{\text{network}}$ ：网络传输时间；
$L_{\text{playback}}$ ：播放器缓冲和设备输出时间。

如果模型直接处理音频，则显式的 $L_{\text{ASR}}$ 和一部分协议转换延迟可能被内部吸收，但并不意味着它们消失了。它们仍然会体现在模型的首事件延迟和回合判断延迟中。

6.2 完整算例

假设目标是：

用户说完后，300 毫秒内开始听到语音。

测得各阶段预算如下：

阶段	预算
VAD 结束检测	80 ms
ASR 最终结果	60 ms
Agent 首个可播放片段	70 ms
TTS 首包	50 ms
网络与播放器	40 ms
总计	300 ms

总和为：

80 + 60 + 70 + 50 + 40 = 300\text{ ms}

如果引入一个必须先完成的工具调用，工具耗时为 800 毫秒，则：

L = 80 + 60 + 800 + 70 + 50 + 40 = 1100\text{ ms}

此时不应继续假设“优化模型”能解决问题，因为主要瓶颈是工具。可以改成两阶段输出：

0～300 ms：
好的，我现在帮您查询。

300～1100 ms：
并行执行工具调用。

工具返回后：
为您查到明天 09:15 有余票……

固定的过渡语使首音频仍然满足 300 毫秒预算，而完整答案延迟变为约 1100 毫秒。对于用户体验而言，“已经开始响应”与“完整事实已经返回”是两个不同指标。

6.3 首音频延迟与完整回答延迟

必须分别记录：

TTFA（Time to First Audio）：用户说完到第一个音频块开始播放；
TTFT（Time to First Token）：请求提交到模型首个文本 token；
TTFC（Time to First Content）：首个具有实质内容的输出；
Completion Latency：完整回答完成；
Barge-in Latency：用户开口到旧音频停止；
Tool Latency：工具调用耗时；
Audio Drain Latency：回答结束到播放器完全排空。

只看平均值是不够的。语音交互对尾延迟非常敏感，应至少观察 P50、P90、P95 和 P99：

TTFA P50 = 280 ms
TTFA P95 = 920 ms
Barge-in P95 = 110 ms

这两个系统的平均 TTFA 可能接近，但第二个系统的 P95 会让用户频繁感到“有时很聪明，有时完全没反应”。

七、如何压缩延迟：先识别串行依赖

设一个请求包含：

ASR → Agent → Tool → Agent → TTS

如果每一步都必须等待前一步完全结束，总延迟为：

L_{\text{serial}} = L_{\text{ASR}} + L_{\text{Agent1}} + L_{\text{Tool}} + L_{\text{Agent2}} + L_{\text{TTS}}

如果其中两个步骤没有数据依赖，可以并行：

           ┌→ 输入安全检查 ─┐
ASR ───────┤                ├→ Agent
           └→ 意图预分类 ───┘

则这部分延迟变成：

L_{\text{parallel}} = L_{\text{ASR}} + \max(L_{\text{safety}}, L_{\text{classification}}) + L_{\text{Agent}}

OpenAI 的延迟优化文档也把并行执行、流式传输和分块处理列为降低等待的重要手段；同时提醒，展示进度主要改善感知体验，而流式和分块才会真正减少应用完成可用内容的等待。(developers.openai.com)

但是，不能为了并行而并行。以下操作不应在用户意图尚未稳定时执行：

提交支付；
发送消息；
修改订单；
删除数据；
产生不可逆外部副作用。

可以提前做的是：

读取候选数据；
查询只读接口；
预加载常用知识；
准备候选 TTS；
对可能的工具进行 schema 校验。

这也是 Agent 与普通工作流的边界之一：固定、可预测的流程适合由代码编排；需要动态选择工具和步骤时，才让模型控制流程。Anthropic 将 workflow 定义为由预定义代码路径编排 LLM 和工具，而 Agent 则由 LLM 动态决定过程和工具使用，并特别强调 Agent 会以更高延迟、成本和错误累积为代价换取灵活性。(anthropic.com)

八、工具调用：语音只是输入输出形式，不改变业务一致性

语音 Agent 经常被误认为是“对话系统”，但只要它能调用工具，就必须遵守普通 Agent 的一致性原则。

一次工具调用应至少包含：

{
  "type": "tool.call",
  "call_id": "call-17",
  "generation_id": "gen-9",
  "tool": "query_train_ticket",
  "arguments": {
    "from": "杭州",
    "to": "上海",
    "date": "2026-09-02"
  },
  "idempotency_key": "session-8-turn-4-call-17"
}

工具结果：

{
  "type": "tool.result",
  "call_id": "call-17",
  "generation_id": "gen-9",
  "status": "success",
  "data": {
    "trains": [
      {
        "number": "G123",
        "departure": "09:15",
        "available": true
      }
    ]
  }
}

8.1 旧结果不能污染新回合

用户可能在工具执行期间说：

不用查了，改查后天。

此时 call-17 的结果即使稍后返回，也不能直接进入当前回答。系统应：

创建新的 turn；
递增 generation；
取消可取消查询；
对不可取消查询丢弃旧结果；
对新请求使用新的 call_id 和幂等键。

8.2 工具结果和语音输出的证据对齐

当 Agent 说：

明天 09:15 的 G123 还有票。

这句话的事实来源应能追溯到：

tool.call call-17
tool.result call-17
text span [0, 22]
audio span [0 ms, 1560 ms]

可以把输出建模为：

\text{AudioSegment} \rightarrow \text{TextSpan} \rightarrow \text{EvidenceSet}

例如：

{
  "audio_segment_id": "audio-3",
  "text": "明天 09:15 的 G123 还有票。",
  "evidence": [
    {
      "source_type": "tool.result",
      "call_id": "call-17",
      "json_path": "$.data.trains[0]"
    }
  ]
}

这对于多模态 Agent 尤其重要。图像、音频、视频、文档和工具结果都可能成为证据，但模型的口头回答不能只保留最终音频。系统至少要保存：

原始音频或可审计引用；
ASR 的 revision 链；
模型生成的文本 delta；
工具调用和结果；
最终 TTS 文本；
输出音频与文本的时间映射。

否则，用户说“我明明说的是 12345”，系统却无法判断错误来自 VAD 截断、ASR 识别、Agent 改写还是工具参数解析。

九、事件协议：把语音 Agent 当成事件流系统

语音 Agent 不应只返回一个最终字符串。客户端需要接收一系列事件：

{"type":"turn.started","turn_id":"turn-4"}
{"type":"asr.delta","turn_id":"turn-4","text":"帮我查一下"}
{"type":"asr.delta","turn_id":"turn-4","text":"帮我查一下明天杭州到上海"}
{"type":"asr.final","turn_id":"turn-4","text":"帮我查一下明天杭州到上海的高铁"}
{"type":"agent.started","generation_id":"gen-4"}
{"type":"text.delta","generation_id":"gen-4","text":"好的，我先帮您查询。"}
{"type":"tool.call","generation_id":"gen-4","call_id":"call-9"}
{"type":"tool.result","generation_id":"gen-4","call_id":"call-9"}
{"type":"text.delta","generation_id":"gen-4","text":"明天 09:15 的 G123 还有票。"}
{"type":"tts.audio","generation_id":"gen-4","chunk_id":1}
{"type":"finish","generation_id":"gen-4","reason":"completed"}

事件需要具备以下属性：

event_id：用于去重；
session_id：标识会话；
turn_id：标识用户回合；
generation_id：标识回答版本；
sequence：同一流内的顺序；
created_at：服务端时间；
usage：token、音频时长和工具消耗；
finish.reason：完成、打断、错误、超时或断线。

9.1 Delta 不是状态快照

如果事件是：

{"type":"text.delta","text":"上海"}

客户端应追加它；如果事件是：

{"type":"text.snapshot","text":"帮我查一下明天杭州到上海"}

客户端应替换当前快照。

把 delta 当 snapshot 会造成重复文本：

帮我查一下明天杭州帮我查一下明天杭州到上海

同理，音频块不是完整音频文件。客户端必须根据 chunk_id、顺序和 generation 进行拼接或播放。

9.2 断线和恢复

断线恢复不能简单地“重新发一遍用户文本”，因为可能造成：

工具重复执行；
退款、下单等副作用重复提交；
TTS 重复播放；
客户端和服务端状态不一致。

一种安全恢复流程是：

客户端携带最后确认的 event_id 或 sequence 重连；
服务端返回缺失事件；
如果事件日志不可恢复，则返回当前会话快照；
对正在运行的 generation 查询状态；
对工具调用按 idempotency_key 查询执行状态；
只有在状态明确为未执行时才重试；
对已完成但未播放的音频重新发送，或从文本重新合成。

状态快照示例：

{
  "session_id": "session-8",
  "last_event_sequence": 42,
  "current_turn_id": "turn-4",
  "current_generation_id": "gen-4",
  "agent_status": "waiting_tool",
  "active_tools": [
    {
      "call_id": "call-9",
      "idempotency_key": "session-8-turn-4-call-9",
      "status": "succeeded"
    }
  ],
  "audio_played_until_chunk": 3
}

对于实时语音，事件协议和音频协议最好分离：控制事件使用 WebSocket 或数据通道，音频使用适合实时媒体的传输方式。OpenAI 的当前语音文档将 WebRTC、WebSocket 和 SIP 列为不同连接方式，并在浏览器场景中推荐通过服务端创建临时凭证后建立实时会话。(developers.openai.com)

十、一个最小可运行的链式管线模型

下面的代码不是绑定某个厂商 API 的完整客户端，而是一个可以运行和测试的本地管线骨架。它用于说明状态、取消和 generation 隔离。

import asyncio
from dataclasses import dataclass


@dataclass
class AudioFrame:
    pcm: bytes
    timestamp_ms: int
    speech_probability: float


class VoiceSession:
    def __init__(self):
        self.generation = 0
        self.state = "listening"
        self.tasks: dict[str, asyncio.Task] = {}
        self.audio_queue: asyncio.Queue[tuple[int, str]] = asyncio.Queue()

    def new_generation(self) -> int:
        self.generation += 1
        return self.generation

    def register(self, name: str, task: asyncio.Task):
        self.tasks[name] = task

    async def cancel_generation(self):
        for task in self.tasks.values():
            if not task.done():
                task.cancel()

        self.tasks.clear()

        while not self.audio_queue.empty():
            self.audio_queue.get_nowait()

    async def interrupt(self):
        # 先切换 generation，使旧结果即使返回也无法提交
        self.new_generation()
        await self.cancel_generation()
        self.state = "listening"

    async def emit_audio(self, generation: int, text: str):
        if generation != self.generation:
            return

        await self.audio_queue.put((generation, text))

    async def play(self):
        while True:
            generation, text = await self.audio_queue.get()

            if generation != self.generation:
                continue

            print(f"[PLAY] {text}")


async def fake_agent(session: VoiceSession, user_text: str, generation: int):
    await asyncio.sleep(0.05)

    if generation != session.generation:
        return

    await session.emit_audio(generation, "好的，我先帮您查询。")

    # 模拟一个较慢的工具调用
    await asyncio.sleep(0.8)

    if generation != session.generation:
        return

    await session.emit_audio(
        generation,
        f"查询完成：{user_text}"
    )


async def demo():
    session = VoiceSession()
    asyncio.create_task(session.play())

    generation = session.new_generation()
    task = asyncio.create_task(
        fake_agent(session, "明天杭州到上海的高铁", generation)
    )
    session.register("agent", task)

    # 用户在工具调用期间打断
    await asyncio.sleep(0.2)
    print("[INTERRUPT] 用户开始说话")
    await session.interrupt()

    # 新回合开始
    generation = session.new_generation()
    task = asyncio.create_task(
        fake_agent(session, "改查后天的航班", generation)
    )
    session.register("agent", task)

    await asyncio.sleep(1)


if __name__ == "__main__":
    asyncio.run(demo())

预期输出类似：

[PLAY] 好的，我先帮您查询。
[INTERRUPT] 用户开始说话
[PLAY] 好的，我先帮您查询。
[PLAY] 查询完成：改查后天的航班

第一代 Agent 的慢工具结果不会被播放，因为：

if generation != session.generation:
    return

这条判断解决的是结果隔离，不是物理取消。即使底层 HTTP 请求仍在执行，只要结果返回后不再驱动当前输出，旧回合就不会污染新回合。

生产环境还需要增加：

工具幂等键；
任务超时；
断线重连；
事件持久化；
播放器真实停止；
ASR revision 合并；
TTS 音频块序列；
错误事件和降级语音。

十一、失败路径必须按组件定位

11.1 用户说了话但没有响应

可能原因：

麦克风没有采集到音频；
音频采样率、声道或编码格式不匹配；
VAD 阈值过高；
前置缓冲太短，首音节被切掉；
结束静音窗口过长；
ASR 没有产生 final；
Agent 仍在等待一个没有返回的工具。

诊断顺序应从事件日志开始：

audio.frame.received
speech.started
speech.stopped
asr.final
agent.started
text.delta
tts.audio
player.started

缺失在哪个事件之后，就优先检查该组件，而不是直接换模型。

11.2 Agent 经常抢话

可能原因：

静音窗口过短；
semantic VAD eagerness 过高；
播放回声被当成用户语音；
用户停顿时系统过早提交 partial；
播放器没有正确区分远端音频和本地麦克风输入。

解决方法不是简单地把所有阈值调大。阈值变大可能使真正的打断也变慢。应分别测量：

误打断率；
真实打断检测延迟；
用户回合平均停顿；
回声触发比例。

11.3 打断后仍听到旧回答

通常是以下原因之一：

只取消了模型，没有清空 TTS 队列；
只清空队列，没有停止播放器；
旧 TTS 任务仍在向队列写入；
音频包没有 generation_id；
客户端收到旧事件后未做版本检查；
服务端取消成功，但网络缓冲中仍有旧音频。

完整打断必须形成闭环：

检测打断
→ 递增 generation
→ 标记旧事件无效
→ 取消模型
→ 取消 TTS
→ 清空音频队列
→ 停止播放器
→ 丢弃旧网络包
→ 接收新回合

11.4 工具结果正确，但回答错误

如果工具结果正确而语音回答错误，问题可能发生在：

工具结果被压缩或摘要时丢失字段；
模型引用了旧 generation 的结果；
TTS 文本和屏幕文本不一致；
ASR 识别的日期或数字已经错误；
模型把候选项说成确定事实。

此时应做跨模态追踪：

用户音频时间段
→ ASR 文本 revision
→ Agent 输入消息
→ 工具参数
→ 工具结果
→ Agent 文本输出
→ TTS 文本
→ 播放音频时间段

只记录最终文本，无法定位这条链中的错误来源。

十二、架构选择：什么时候用语音到语音，什么时候用链式管线

12.1 语音到语音适合什么

语音到语音适合：

开放式闲聊；
低延迟问答；
需要自然语气和连续回合；
频繁打断；
不要求每个中间文本都被业务系统审核；
工具调用主要是只读或可安全重试。

它的主要优点是减少显式管线边界，模型可以直接利用音频中的韵律、停顿和语气。但系统调试会更难：错误不一定能简单归因到某个 ASR 或 TTS 阶段。

12.2 链式管线适合什么

链式管线适合：

客服和质检；
审批、支付、退款；
需要保存完整转写；
需要在回复前执行政策检查；
需要替换 ASR、LLM 或 TTS；
需要在中间步骤插入确定性代码；
已经有成熟文本 Agent。

其代价是：

L_{\text{chain}} > L_{\text{direct}}

在多数实现中，显式 ASR、文本 Agent 和 TTS 会增加边界、序列化和传输延迟。但它同时带来更清晰的可观测性和更强的业务控制。OpenAI 当前文档将链式管线描述为显式管理 speech-to-text、Agent workflow 和 text-to-speech，尤其适用于需要持久化转写、确定性逻辑和审批的流程。(developers.openai.com)

12.3 混合架构

实际系统常采用混合模式：

实时语音模型
  ├── 负责自然回合、闲聊和低延迟反馈
  └── 遇到高风险操作时
        ↓
      转入链式流程
        ├── ASR final
        ├── 参数校验
        ├── 人工确认
        ├── 工具调用
        └── 审计后的 TTS

例如：

用户：帮我把这笔订单退款。
Agent：我可以处理。订单号是多少？
用户：12345。
Agent：订单 12345 将退款 299 元，确认退款吗？
用户：确认。

在“确认退款”之前，可以使用低延迟语音交互；在真正提交退款时，必须进入带参数校验、幂等键和状态确认的确定性流程。

十三、生产基线：从“能说话”升级为“可诊断的实时 Agent”

一个可上线的语音 Agent 至少应具备以下可验证属性：

事件完整性

每个用户回合都能追踪：

audio → vad → asr → agent → tool → tts → player

版本隔离

每个异步结果都携带：

session_id
turn_id
generation_id
sequence

可取消性

系统明确知道：

哪些任务可以取消；
哪些任务只能忽略结果；
哪些任务需要补偿；
取消后如何清理音频队列。

延迟可分解

不能只记录“整体响应时间”，而应记录：

endpoint_ms
asr_final_ms
agent_first_delta_ms
tool_ms
tts_first_audio_ms
barge_in_ms
audio_drain_ms

事实可追溯

最终语音中的关键事实能够关联到：

ASR 片段；
工具调用；
工具结果；
文本生成区间；
TTS 音频区间。

降级可用

当某组件失败时，系统应有明确路径：

ASR 失败：请求用户重说；
TTS 失败：显示文本或切换备用语音；
工具超时：说明正在处理，而不是编造结果；
模型失败：播放固定错误提示；
连接中断：恢复会话或明确结束。

语音 Agent 的质量最终不是由“声音像不像真人”决定的，而是由系统能否在不完整输入、部分输出、用户打断、工具延迟和网络故障同时发生时，仍然保持正确的状态转换。

结语

VAD 决定系统何时把音频视为一个回合，ASR 决定系统如何把连续声音变成可处理的文本，流式推理决定系统何时开始产生回答，TTS 决定回答何时变成可播放的音频，而打断机制决定新回合能否真正夺回控制权。

延迟预算则把这些组件放在同一条可计算的因果链上：

\text{用户说完} \rightarrow \text{回合结束} \rightarrow \text{文本稳定} \rightarrow \text{模型首输出} \rightarrow \text{TTS 首音频} \rightarrow \text{播放器开始播放}

只要其中任意一个阶段的状态、版本或取消语义不清晰，系统就会出现重复回答、抢话、旧音频泄漏、工具重复执行或无法恢复等问题。

可靠的语音 Agent 不是把多个模型串起来，而是建立一套能处理部分结果、并发任务、可取消输出、不可逆工具和断线恢复的实时事件系统。

系列导航与关联阅读

官方资料

本文依据 Agent、模型、协议与框架官方资料重新梳理；正文、示例与生产清单由 WR BLOG 编写。