YYDS Space

Appearance

实时语音视频流测试 ​

2026 年的"对话式 AI"已经不是文本聊天的代名词。GPT-5 Realtime、Gemini 3 Live、豆包实时语音 v2 把端到端的音视频流模型推到了商用门槛——延迟降到 300ms、可以听懂打断、可以看着摄像头实时回答你。但相应地，测试方法也彻底变了：以前测 ASR+LLM+TTS 是三段拆开测，今天必须用全双工的视角去测一个会"听-想-说-看"的连续过程。这一篇把实时多模态的指标体系、噪声测试、Barge-in 自动化、视频流压测完整讲清楚。

教学导读

**定位：**这一章解决"实时音视频对话场景下，传统功能/性能测试方法完全失效"的问题。它是把第 22 篇《多模态测试》和第 31 篇《延迟与吞吐性能测试》缝合到全双工流式场景下的工程实践。 **前置依赖：**建议已读 第 22 篇 多模态测试、第 31 篇 延迟与吞吐性能测试，对 WebSocket、PCM 音频、采样率、帧率有基本概念。 **适用场景：**智能客服语音 Bot、车载语音助手、手机系统级 AI 助手（小爱同学/Siri/Bixby 类）、AR/VR 实时交互、远程教育、远程医疗咨询、视频监控异常检测。 **学完产出：**你能搭建 WebSocket 自动化测试客户端，跑通 TTSF/E2E/Barge-in 三大延迟指标采集，能用 librosa + sounddevice 注入工业级噪声，能给端到端模型和传统 pipeline 出量化对比报告。

第1章：实时多模态——从异步 ASR/TTS 到端到端流式 ​

1.1 一段时间线 ​

要理解 2026 年实时语音模型的位置，必须把它放在过去十年语音对话系统的演进里看：

时期典型架构典型延迟典型产品
2014~2018串行 ASR → 意图分类 → 槽位填充 → TTS2~5 秒Siri 1 / 小爱同学早期
2019~2022流式 ASR + 模板回答 + 流式 TTS1~2 秒Google Assistant / Alexa
2023~2024流式 ASR + LLM（非流式）+ 流式 TTS1.2~3 秒ChatGPT 语音 v1 / 文心一言语音
2024.10OpenAI 发布首个端到端语音模型 gpt-4o-realtime0.3~0.6 秒ChatGPT Advanced Voice
2025Gemini 2 Live、豆包实时语音 v1、GLM-Realtime 跟进0.3~0.5 秒Gemini App / 豆包 App
2026gpt-5-realtime / gemini-3-live / Doubao-Realtime-1.5 / DeepSeek-Voice-1.00.2~0.4 秒系统级语音助手 / 车机 / IoT

2024 年下半年是分水岭：从这一刻开始，"语音对话 = ASR + LLM + TTS"的串联式认知被打破。模型直接接收原始音频字节，直接吐出原始音频字节，中间不再有可见的文本中转。这种端到端的设计带来三个根本性变化：

  1. 延迟从"和好"降到"和体感无感"。人类对话的自然停顿大约 200ms，端到端模型把首字延迟压到这个量级以内，于是产生了"和真人讲话差不多"的感觉。
  2. 语气、情绪、停顿成为模型的"输入特征"。串联式架构里，ASR 把"啊？真的吗？"和"啊真的吗"识别成同一段文本，模型完全感受不到惊讶或质疑的语气。端到端模型直接吃声学波形，能听到这种细节。
  3. Barge-in（打断）成为内建能力。串联式架构里，TTS 在播音频，模型在等下一个用户输入；用户一打断，整个流水线都要重置。端到端模型设计上就是全双工的——边说边听，能在自己说话时识别到用户打断意图，立即停说改听。

1.2 端到端流式的工程定义 ​

"端到端流式"在 2026 年的工程语境里有四个最低要求，少一个就不算：

  • 输入流式：客户端可以一边录一边发，模型不等说完就开始处理。
  • 输出流式：模型一边想一边吐音频帧，客户端可以一边收一边播。
  • 全双工：输入和输出可以同时发生，互不阻塞。
  • 状态保持：服务器端维持会话上下文（语音、文本、工具调用历史），客户端只发增量。

测试视角的根本转变。

对测试人员来说，最重要的认知是："实时流式" = "时间维度上不可重入的状态机"。你不能像测 REST API 那样"发请求-收响应-断言"，因为输入和输出在时间上是交织的。每一次测试都要在 时间轴 上记录事件序列，再针对事件序列做断言。这一点在第 6 章会展开。

1.3 真实 2026 模型清单 ​

这一章会反复出现的实时多模态模型，你需要先认识它们：

模型厂商主要能力主要 SDK / 协议
gpt-5-realtimeOpenAI语音输入 + 语音输出 + 工具调用 + 视觉（vision-realtime 子型号）WebSocket / WebRTC， openai-realtime-api
gpt-5-realtime-visionOpenAI音频 + 视频帧（base64 JPEG 或 RTP）+ 文本WebSocket， openai v2 SDK
gemini-3-liveGoogle语音 + 文本，多语言原生WebSocket， google-genai
gemini-3-live-visionGoogle音频 + 视频流 + 屏幕共享WebSocket， google-genai 1.10+
豆包实时语音 v2火山引擎中文优化 + 方言 + 音色克隆WebSocket（火山自有协议），volcengine-sdk-python
Doubao-Realtime-1.5火山引擎音视频联合，工具调用WebSocket + WebRTC（v2.1+）
GLM-Realtime智谱中英双语，企业私有化部署WebSocket， zhipuai SDK
DeepSeek-Voice-1.0DeepSeek开源权重，本地推理友好HTTP/2 + Server Sent Events

第2章：实时模型测试和传统多模态测试的本质差异 ​

2.1 测试对象的差异 ​

传统多模态测试里，测试对象是一个"输入图片/音频，输出文本"的函数。延迟、准确率、鲁棒性都可以围绕"单次请求"展开。实时模型完全不同：

传统多模态测试 ​

**输入：**一段完整的音频文件 / 一张完整的图片 **输出：**一段完整的文本结果 **断言对象：**结果文本与 ground truth 比对 **延迟指标：**请求-响应总时长（一个数） **失败模式：**识别错误、答案错误

实时多模态测试 ​

**输入：**持续 N 分钟的音频/视频流，过程中可能有用户行为变化 **输出：**持续的事件流（音频帧、文本片段、工具调用、状态变化） **断言对象：**时间轴上事件序列的属性（顺序、间隔、内容） **延迟指标：**多个分阶段指标（TTSF、TTLF、Barge-in latency 等） **失败模式：**抢话、卡顿、冷场、打断失败、上下文丢失、回声循环

2.2 失败模式的差异 ​

实时模型有一些"传统测试根本看不到"的失败模式，下面这五个在 2025-2026 真实事故复盘中最常见：

  1. 抢话（Premature Speaking）：用户还没说完，模型已经开始回答。常见原因：VAD（语音活动检测）阈值过低，把句子停顿误判为说完。
  2. 冷场（Dead Air）：用户说完了，模型超过 1.5 秒没反应，用户以为没听到，重复一遍——结果触发两轮回答叠加。
  3. 打断失败（Barge-in Failure）：模型在长篇大论，用户说"等等"想打断，模型没识别到，继续讲完。这是 2026 年最被用户诟病的问题。
  4. 上下文丢失（Context Drop）：会话进行 5 分钟后，模型突然"忘记"前面提到的人名、订单号。本质是上下文窗口被音频 token 占满。
  5. 回声自激（Echo Loop）：扬声器播放的模型语音被麦克风采集回去，模型听到自己的声音又开始回答自己——形成无限递归。

测试挑战。

这五个失败模式没有任何一个能用传统 unit test 复现。它们都需要在 真实的全双工时间轴上 才能暴露。这就解释了为什么实时模型的测试栈天生就是 "脚本驱动 + 时间戳采集 + 事件序列断言"，而不是简单的 "input-output 比对"。

2.3 工具链的差异 ​

传统多模态测试用 pytest + transformers + datasets 就够了。实时多模态测试需要更厚的工具链：

能力传统多模态实时多模态
API 调用HTTP 同步请求WebSocket 长连接 / WebRTC
音频处理soundfile 读写sounddevice 实时采集 + numpy 流式拼接
视频处理PIL.Image 静态加载opencv-python 帧迭代 + base64 编码
噪声注入预先合成 fixturelibrosa 在线生成 + 实时叠加
事件采集response.json()异步 callback + asyncio.Queue + 时间戳
断言方式assert text == expected对事件序列做模式匹配 + 时间窗口检查

第3章：OpenAI Realtime / Gemini Live / 豆包实时语音架构 ​

3.1 OpenAI Realtime API 架构 ​

OpenAI Realtime API 是 2024-10 首发、2026-01 升级到 gpt-5-realtime 的产品形态。它的协议核心是 WebSocket 上的 JSON 事件流，事件分两类：

  • 客户端事件（Client → Server）：session.update、input_audio_buffer.append、input_audio_buffer.commit、response.create、response.cancel。
  • 服务端事件（Server → Client）：session.created、response.audio.delta、response.audio.done、response.text.delta、input_audio_buffer.speech_started、input_audio_buffer.speech_stopped、response.function_call_arguments.done。

客户端采集 PCM → base64 编码 → input_audio_buffer.append (循环) → 服务端 VAD 自动检测说完 → response.audio.delta (流式) → 客户端实时播放

关键设计：服务端内置了 Semantic VAD（gpt-5-realtime 引入的语义级语音活动检测）。它不仅看声音强度是否下降，还会判断"用户讲到一个完整的语义边界没有"。这一改动把抢话率从 gpt-4o 时代的 8.4% 降到了 1.7%。

3.2 Gemini Live API 架构 ​

Google 的 Gemini Live API 在 2025 年 5 月发布 v1 (gemini-2-live)，2026 年 2 月升级到 gemini-3-live。架构上和 OpenAI 类似（WebSocket + JSON 事件），但有几个差异：

  • 事件命名风格不同：用 BidiGenerateContentSetup、BidiGenerateContentClientContent、BidiGenerateContentServerContent 这种"双向生成内容"命名。
  • 音频采样率默认 16kHz（OpenAI 默认 24kHz），需要重采样。
  • 原生支持视频流：客户端可以同时发送音频和视频帧（JPEG base64），不需要切到 vision 子型号。
  • Activity Detection 由客户端控制：不像 OpenAI 默认 server-side VAD，Gemini Live 默认让客户端决定何时认为"用户说完了"——这给测试带来更多控制权。

3.3 豆包实时语音 v2 / Doubao-Realtime-1.5 架构 ​

火山引擎的实时语音协议是自有协议，不兼容 OpenAI/Gemini 风格。其特点：

  • 使用 WebSocket 上的二进制帧（不是 JSON），自定义了 4 字节头 + payload 的帧格式。
  • 支持音色克隆和方言（粤语、四川话、东北话），是中国市场客服业务的事实标准。
  • 对音频的编解码默认用 OPUS（不是 PCM），客户端需要做编码。
  • Doubao-Realtime-1.5 在 2026 年 3 月增加了视觉通道，用 WebRTC 而不是 WebSocket。

3.4 三家协议对比 ​

能力OpenAI RealtimeGemini Live豆包实时语音 v2
协议WebSocket + JSONWebSocket + JSONWebSocket + 二进制帧
默认采样率24 kHz PCM1616 kHz PCM1616 kHz OPUS
VADServer-side Semantic VAD客户端控制 / Server VAD 可选Server-side 能量 VAD
视觉通道子型号 vision-realtime（同 WS）原生支持（同 WS）WebRTC 通道（v1.5+）
工具调用function callingtool usefunction call（中文 schema）
音色定制8 个预设30+ 多语言预设支持音色克隆（5 秒样本）
典型 TTSF320~520 ms280~480 ms300~600 ms

选型建议（仅技术视角）。

英文场景或全球化产品 → OpenAI 或 Gemini，二者半斤八两。Gemini 在视频流多模态上稍领先。中文客服 → 豆包是事实选择，方言和音色克隆是其他家短期补不上的。私有化部署 → DeepSeek-Voice-1.0 是唯一可商用的开源选项，但工具链不如商业 API 完善。

第4章：视频流模型原理 ​

4.1 视频流模型解决什么问题 ​

静态图像理解（看一张图回答问题）在 2024 年就已经成熟。视频流模型解决的是 "持续看一段动态画面" 的问题。它在以下场景有不可替代的价值：

  • 实时辅助：用户用手机摄像头对着冰箱说"帮我看看哪个调料快用完了"。
  • 视障辅助：摄像头持续描述周围环境（OpenAI 在 2025 年和 Be My Eyes 合作的产品）。
  • 远程巡检：维修工拿摄像头对着设备，AI 实时指导拆装步骤。
  • 系统级 Agent：手机 AI 看屏幕实时帮你操作 App。

4.2 视频流模型的两种工程实现 ​

方案 A · 离散帧采样（GPT-5 Vision Realtime / Gemini 3 Live Vision） ​

客户端按一定间隔（典型 1~4 帧/秒）采样视频帧，每帧编码为 base64 JPEG 后通过 WebSocket 发送。服务端把每帧当作一张图片处理，并用时间戳维护时序。这是 2026 年商业 API 的主流做法，因为它对带宽友好（每秒 200KB 量级）。

方案 B · 连续视频 token（Gemini 3 Live Vision Premium） ​

客户端推送 H.264 编码的视频流，服务端先做帧抽取再统一 token 化。延迟更低（端到端 200ms 内），但带宽要求高（每秒 1MB 量级）。仅用于专业场景如远程医疗。

4.3 视频流模型的关键参数 ​

参数典型默认值对延迟影响对效果影响
采样帧率（fps）1 ~ 4 fps越高带宽越大、处理越慢越高对快速动作越准
分辨率640×480 ~ 1280×720越高编码越慢越高细节识别越准
JPEG 质量70 ~ 85越高数据越大过低会丢细节
滑动窗口长度近 30 秒帧窗口大上下文 token 多关键信息能跨帧

4.4 一个关键概念：时间锚点（Temporal Anchoring） ​

视频流模型的 ground truth 不是"图里有什么"，而是"在第 X 秒发生了什么"。GPT-5 Vision Realtime 在 2026 年增加了 temporal_anchor 字段，让模型回答里可以引用具体时间戳。这给测试带来一个新的断言维度：不仅要测它说对没，还要测它定位到的时间戳是否准确（典型容忍度 ±0.5 秒）。

第5章：端到端语音模型 vs 传统 ASR + LLM + TTS pipeline ​

5.1 两种架构的对比 ​

对比维度传统 pipeline端到端模型
架构ASR → LLM → TTS 三段串行音频 in → 音频 out 单模型
语气感知丢失（ASR 只输出文本）保留（声学特征直接进模型）
典型 TTSF800ms ~ 1500ms200ms ~ 500ms
调试可见性高（中间文本可看可改）低（黑盒）
可定制性高（每段都可换模型）低（只能换整模型）
成本三段 API 各计费，复杂按音频时长统一计费
典型故障断句错、口音识别错抢话、回声、语气怪

5.2 一张延迟分解对比表（基于真实压测数据） ​

这是一份实测数据，场景为 5 秒中文用户输入："你帮我查一下明天去上海的高铁票，最好是上午 10 点之前的"。运行环境为同机房直连，10 次平均：

阶段传统 pipeline (Whisper-large + GPT-5 + ElevenLabs)端到端 (gpt-5-realtime)差距
用户说完 → ASR 出 final 文本420 ms——
LLM 首字（含网络）720 ms——
TTS 合成首段音频340 ms——
客户端首音频帧到达—340 ms—
TTSF（首音）1480 ms340 ms4.4×
用户说完 → 一句话回答完整播放5200 ms2400 ms2.2×
Barge-in 中断响应（用户打断到模型停说）~ 850 ms（需要清空 TTS buffer）~ 180 ms4.7×

关键观察。

从数据可见，端到端模型在首音延迟（TTSF）和打断响应上的优势是数量级的，而不是百分比的。这就是为什么 2026 年大部分新建的语音对话产品都直接押注端到端方案。但传统 pipeline 也没死——在需要严格审计、需要可定制 ASR 词典、需要本地化 TTS 的场景（医疗、政务、法务）依然是首选。

5.3 测试视角的差异化对待 ​

因为两种架构差异巨大，测试方法也要分开：

  • 传统 pipeline 测试：单测每段（ASR 准确率、LLM 准确率、TTS MOS），加端到端延迟链路追踪。
  • 端到端模型测试：必须做全双工时间轴测试，单测每段没有意义（没法拆）。

第6章：实时语音的关键指标 ​

6.1 四大延迟指标 ​

这四个指标是 2026 年评价实时语音模型的核心，缺一不可：

TTSF（Time To Speak First） ​

**定义：**从"用户说完最后一个字"到"客户端收到第一帧模型音频"的耗时。 商用基线：≤ 800ms 为优秀，≤ 1200ms 为可用，> 2000ms 用户会明显感知冷场。 **测量方法：**客户端打两个时间戳——VAD 判定用户说完的时刻 t1，第一个 audio.delta 到达的时刻 t2，TTSF = t2 - t1。

E2E Latency（端到端延迟） ​

**定义：**从用户说完到模型完整答完一句话的总耗时。 商用基线：≤ 500ms（短回答）或 ≤ 3000ms（长回答）为商用门槛。 **测量方法：**VAD 用户说完时刻 → response.audio.done 事件时刻。

Barge-in Latency（打断响应延迟） ​

**定义：**模型正在说话时用户打断，从"用户开口"到"模型停说"的耗时。 商用基线：≤ 200ms 用户体感无感；> 500ms 用户会重复说一遍。 **测量方法：**客户端注入用户音频时刻 → 服务端发出 response.cancelled 事件时刻。

Turn-taking Accuracy（轮转准确率） ​

**定义：**模型正确识别"用户说完了"和"用户没说完只是停顿"的准确率。 **商用基线：**抢话率 ≤ 3%、冷场率 ≤ 2%。 **测量方法：**用包含 1.0/1.5/2.0 秒不同停顿长度的对话样本测试，统计抢话和冷场次数。

6.2 质量指标 ​

除了延迟，质量维度也要看：

指标定义测量工具商用基线
WER（Word Error Rate）识别错误率（用作 ground truth 比对）jiwer 库中文 ≤ 5%、英文 ≤ 3%
MOS（Mean Opinion Score）合成语音自然度主观评分众包评测 / NISQA 工具≥ 4.0/5.0
语义准确率回答是否回答了问题LLM-as-Judge≥ 90%
音色一致性同一会话内音色是否漂移speaker embedding 余弦相似度≥ 0.92
情绪适配对急切/沮丧用户是否调整语气人工评测 + 情绪分类器≥ 80%

6.3 鲁棒性指标 ​

  • 噪声容忍度：在 SNR=10dB 的白噪声环境下 WER 上升 ≤ 30%。
  • 回声抑制：扬声器回采耦合下，模型能正确识别用户音 vs 自身回采。
  • 多说话人鲁棒性：背景人声叠加时，能锁定主说话人。
  • 方言鲁棒性：粤语、四川话、东北话覆盖下的 WER。

测试团队经常踩的坑。

很多团队第一次跑实时语音性能测试，只看一个"端到端延迟"，结果隐藏了 80% 的问题。同样是 1500ms 端到端延迟，TTSF=400ms / 完整答完=1500ms 的体验和 TTSF=1200ms / 完整答完=1500ms 的体验差异巨大。必须把四大延迟指标分开看。

6.4 P50 与 P95 同样重要 ​

在实时场景里只看 P50（中位数）会严重低估用户痛感。原因是用户对"偶尔卡顿"的记忆比"平均流畅"强烈得多——一次 3 秒的冷场会让用户对整段对话留下"很卡"的印象，即使前 9 句平均只有 400ms。 建议同时报告 P50/P95/P99 三档，并把 P95 作为门控阈值；P99 只用于事后归因（多由网络抖动或 token 缓存 miss 引起）。

指标P50 阈值P95 阈值P99 阈值
TTSF≤ 500 ms≤ 800 ms≤ 1500 ms
E2E（短答）≤ 1500 ms≤ 2500 ms≤ 4000 ms
Barge-in≤ 150 ms≤ 250 ms≤ 500 ms

第7章：噪声鲁棒性测试 ​

7.1 噪声分类 ​

实战中常见的噪声类型可以分四类：

噪声类型声学特征典型来源对模型影响
白噪声 / 粉噪声宽频均匀空调、风扇、电流轻度提升 WER
多说话人 / 鸡尾酒会语谱重叠办公室、咖啡馆严重提升抢话和误识别
回声 / 混响多径延时叠加会议室、车内触发回声自激循环
突发噪声瞬态高能关门、咳嗽、拍桌触发误打断

7.2 信噪比（SNR）的工程定义 ​

SNR（Signal-to-Noise Ratio）= 10 × log10(信号功率 / 噪声功率)。SNR 越高代表语音越清晰。工程经验值：

  • SNR > 30 dB：录音棚级别，几乎听不到背景音。
  • SNR 20~30 dB：安静办公室。
  • SNR 10~20 dB：街头、地铁、商场。
  • SNR 0~10 dB：嘈杂酒吧、施工现场。
  • SNR < 0 dB：噪声比语音还响，人类也听不清。

7.3 噪声测试矩阵 ​

建议建立"噪声类型 × SNR 等级"的二维矩阵作为标准测试集。每格放至少 50 条样本：

SNR 30dBSNR 20dBSNR 10dBSNR 5dB
白噪声50505050
办公室杂音50505050
街道交通50505050
多说话人50505050
车内行驶50505050

7.4 librosa 噪声生成的核心 API ​

python
import librosa
import numpy as np

def add_white_noise(signal: np.ndarray, snr_db: float) -> np.ndarray:
    """按指定 SNR 注入白噪声"""
    sig_power = np.mean(signal ** 2)
    noise_power = sig_power / (10 ** (snr_db / 10))
    noise = np.random.normal(0, np.sqrt(noise_power), signal.shape)
    return signal + noise

def add_pink_noise(signal: np.ndarray, snr_db: float) -> np.ndarray:
    """粉噪声 = 1/f 频谱，比白噪声更接近真实环境噪声"""
    n = len(signal)
    white = np.random.randn(n)
    fft = np.fft.rfft(white)
    freqs = np.fft.rfftfreq(n)
    freqs[0] = freqs[1]
    fft = fft / np.sqrt(freqs)
    pink = np.fft.irfft(fft, n)
    pink = pink / np.std(pink)
    sig_power = np.mean(signal ** 2)
    noise_power = sig_power / (10 ** (snr_db / 10))
    return signal + pink * np.sqrt(noise_power)

def add_reverb(signal: np.ndarray, sr: int = 16000,
               decay: float = 0.4, n_echos: int = 6) -> np.ndarray:
    """简单冲激响应卷积模拟混响"""
    impulse = np.zeros(int(sr * 0.6))
    impulse[0] = 1.0
    for i in range(1, n_echos):
        delay = int(sr * 0.05 * i)
        if delay < len(impulse):
            impulse[delay] = decay ** i
    out = np.convolve(signal, impulse, mode="same")
    return out / (np.max(np.abs(out)) + 1e-8) * np.max(np.abs(signal))

7.5 多说话人模拟（鸡尾酒会场景） ​

python
def mix_speakers(main_path: str, bg_paths: list, sr: int = 16000,
                 main_gain: float = 1.0, bg_gain: float = 0.3) -> np.ndarray:
    """把主说话人音频与多个背景说话人音频按比例叠加"""
    main, _ = librosa.load(main_path, sr=sr)
    main = main * main_gain
    bg_mix = np.zeros_like(main)
    for p in bg_paths:
        bg, _ = librosa.load(p, sr=sr)
        if len(bg) < len(main):
            bg = np.tile(bg, (len(main) // len(bg) + 1))[:len(main)]
        else:
            bg = bg[:len(main)]
        bg_mix += bg
    bg_mix = bg_mix / len(bg_paths) * bg_gain
    mix = main + bg_mix
    return mix / (np.max(np.abs(mix)) + 1e-8) * np.max(np.abs(main))

实战经验。

多说话人测试最容易暴露端到端模型的弱点。我们曾经在测试某主流厂商的实时语音 API 时发现：在主说话人音量是背景人声 3 倍的情况下，模型仍有 12% 的概率把背景人声当作用户输入响应。这种 bug 用纯白噪声测试根本测不出来——必须用真实人声叠加。

第8章：视频流测试方法 ​

8.1 视频流的扰动维度 ​

音频可以加噪声，视频可以加什么？工业上有四类常用扰动：

扰动实现方式模拟真实场景对模型影响
运动模糊（Motion Blur）cv2.filter2D + 运动核手抖、走路拍摄识别准确率下降 10~30%
光照变化HSV V 通道增减逆光、夜间、隧道暗部目标漏检
帧丢失随机丢帧 / 重复帧弱网、设备过热时序事件错位
压缩伪影低 JPEG 质量循环编解码低带宽传输细节文字识别失败

8.2 关键视频流指标 ​

  • 识别延迟（Detection Latency）：目标出现在画面到模型描述出来的耗时，目标 ≤ 1500ms。
  • 识别一致性（Consistency）：连续帧中对同一物体的描述是否前后一致。
  • 跨帧时序准确率：模型回答"这个动作发生在第 X 秒"的精度，目标 ±0.5s。
  • 抗扰动鲁棒性：在不同扰动等级下识别准确率的衰减曲线。

8.3 视频流扰动注入示例 ​

python
import cv2
import numpy as np

def motion_blur(frame: np.ndarray, kernel_size: int = 15) -> np.ndarray:
    """水平运动模糊"""
    kernel = np.zeros((kernel_size, kernel_size))
    kernel[kernel_size // 2, :] = 1.0 / kernel_size
    return cv2.filter2D(frame, -1, kernel)

def adjust_lighting(frame: np.ndarray, gain: float = 0.4) -> np.ndarray:
    """模拟暗光/逆光：gain < 1 变暗、> 1 过曝"""
    hsv = cv2.cvtColor(frame, cv2.COLOR_BGR2HSV).astype(np.float32)
    hsv[..., 2] = np.clip(hsv[..., 2] * gain, 0, 255)
    return cv2.cvtColor(hsv.astype(np.uint8), cv2.COLOR_HSV2BGR)

def frame_drop(frames: list, drop_rate: float = 0.2) -> list:
    """按概率丢帧（保留前一帧）"""
    out = []
    last = frames[0]
    for f in frames:
        if np.random.random() > drop_rate:
            last = f
        out.append(last)
    return out

def jpeg_recompress(frame: np.ndarray, quality: int = 30) -> np.ndarray:
    """低质量 JPEG 编解码循环模拟压缩伪影"""
    _, enc = cv2.imencode(".jpg", frame, [cv2.IMWRITE_JPEG_QUALITY, quality])
    return cv2.imdecode(enc, cv2.IMREAD_COLOR)

第9章：实时性能 vs 质量的权衡测试 ​

9.1 三角权衡 ​

实时多模态模型存在一个不可绕开的三角：延迟 ↔ 质量 ↔ 成本。压低任何一个维度都会让另外两个变差。下面是 2026 年实测的几个权衡点：

调节延迟变化质量变化成本变化
把视频流帧率从 4fps 降到 1fps↓ 30%快速动作识别 ↓ 25%↓ 75%
开启 server-side VAD↓ 100ms（少一次往返）抢话率 ↑ 2%不变
把音频采样率从 24kHz 降到 16kHz↓ 5~10%口音识别 ↓ 4%↓ 33%
把 system prompt 从 800 token 缩到 200↓ 80~150ms多任务依从性 ↓ 8%↓ 75%
关闭 RAG 检索（仅用模型记忆）↓ 200~400ms事实准确性 ↓ 15%↓ 30%

9.2 帕累托前沿测试方法 ​

不要把"延迟"和"质量"分别报告。建议绘制帕累托前沿图：横轴 P95 端到端延迟，纵轴语义准确率，每个配置一个点。这样产品/业务方能直观看到取舍空间，做出 trade-off 决策。

9.3 配置矩阵实验法 ​

常见做法是固定 5 维配置：模型版本 × VAD 模式 × 采样率 × system prompt 长度 × 视频帧率，做笛卡尔积测试。每个组合跑 200 条样本，最终输出一个 CSV，业务方在 Excel 里就能拖拽筛选。

9.4 决策表模板 ​

真实项目里"是否启用 server-side semantic VAD"的决策表，可以作为模板复用。把所有候选项的延迟、质量、风险都列出来，业务方一眼能看清取舍：

候选配置TTSF P95抢话率实施成本结论
客户端能量 VAD（旧方案）540 ms4.8%已实现抢话率超基线
客户端能量 VAD + 200ms 缓冲720 ms2.1%极低TTSF 超标
Server-side semantic VAD490 ms1.4%需迁移协议推荐
Server-side + 客户端双重 VAD510 ms0.8%较高合规场景采用

这种"配置候选 × 关键指标"的决策表本身就是一份合规归档材料——审计时只要能拿出"我们评估过哪些方案、为什么选了这个"的记录，监管基本不会再追问技术细节。

一个反直觉的发现。

2026 年我们在多个项目中重复观察到一个反直觉现象：把 system prompt 从 800 token 缩到 200 token，语义准确率不一定下降——有时反而上升。原因是过长的 prompt 经常包含"如果用户问 X 你就答 Y"这种规则，模型在实时对话场景下来不及精细执行这些规则，反而被它干扰。实时模型的 prompt 工程要倾向"少而锐"，而不是"多而全"——这和文本对话场景的经验相反。

9.5 在 CI 里固化配置基线 ​

权衡决策一旦做出，就要把它"冻结"到 CI 脚本里。具体做法：把决策选定的配置参数写到 config_baseline.yaml，每次 PR 跑 smoke 测试时同时跑"基线配置"和"PR 改动后配置"两份，把延迟和质量的差异自动评论到 PR——任何人想偷偷换 prompt 或换 VAD 模式都会被回归测试拦下来。

第10章：OpenAI Realtime API 测试脚本（WebSocket） ​

10.1 安装依赖 ​

bash
pip install openai==2.4.0 websockets==13.1 sounddevice soundfile librosa numpy

10.2 完整可运行：连接 + 推音频 + 收音频 + 计时 ​

这是一个 200+ 行的最小可运行测试客户端，覆盖了：建立 WebSocket、推送 PCM 音频、接收音频回流、记录关键时间戳、计算 TTSF/E2E。

python
import asyncio
import base64
import json
import os
import time
import wave
from dataclasses import dataclass, field
from typing import List, Optional

import numpy as np
import soundfile as sf
import websockets

OPENAI_REALTIME_URL = (
    "wss://api.openai.com/v1/realtime?model=gpt-5-realtime-2026-01-15"
)

@dataclass
class TimingMarks:
    """记录全双工时间轴上的关键事件"""
    started_at: float = field(default_factory=time.perf_counter)
    user_speech_started: Optional[float] = None
    user_speech_stopped: Optional[float] = None
    first_audio_delta: Optional[float] = None
    response_text_done: Optional[float] = None
    response_audio_done: Optional[float] = None
    cancelled_at: Optional[float] = None
    audio_chunks: List[float] = field(default_factory=list)

    def ttsf_ms(self) -> Optional[float]:
        if self.user_speech_stopped and self.first_audio_delta:
            return (self.first_audio_delta - self.user_speech_stopped) * 1000
        return None

    def e2e_ms(self) -> Optional[float]:
        if self.user_speech_stopped and self.response_audio_done:
            return (self.response_audio_done - self.user_speech_stopped) * 1000
        return None

    def barge_in_ms(self) -> Optional[float]:
        if self.cancelled_at and self.user_speech_started:
            return (self.cancelled_at - self.user_speech_started) * 1000
        return None

async def send_audio_file(ws, path: str, chunk_ms: int = 100):
    """把 wav 按 100ms 块流式推送"""
    audio, sr = sf.read(path, dtype="int16")
    if sr != 24000:
        raise ValueError(f"需要 24kHz PCM16, 实际 {sr}Hz, 请预处理")
    samples_per_chunk = int(sr * chunk_ms / 1000)
    for i in range(0, len(audio), samples_per_chunk):
        chunk = audio[i:i + samples_per_chunk].tobytes()
        await ws.send(json.dumps({
            "type": "input_audio_buffer.append",
            "audio": base64.b64encode(chunk).decode(),
        }))
        await asyncio.sleep(chunk_ms / 1000)
    await ws.send(json.dumps({"type": "input_audio_buffer.commit"}))
    await ws.send(json.dumps({"type": "response.create"}))

async def collect_events(ws, marks: TimingMarks, audio_out: list):
    """异步收事件，给关键事件打时间戳"""
    async for raw in ws:
        evt = json.loads(raw)
        t = time.perf_counter()
        etype = evt.get("type")
        if etype == "input_audio_buffer.speech_started":
            marks.user_speech_started = t
        elif etype == "input_audio_buffer.speech_stopped":
            marks.user_speech_stopped = t
        elif etype == "response.audio.delta":
            if marks.first_audio_delta is None:
                marks.first_audio_delta = t
            marks.audio_chunks.append(t)
            audio_out.append(base64.b64decode(evt["delta"]))
        elif etype == "response.audio.done":
            marks.response_audio_done = t
        elif etype == "response.text.done":
            marks.response_text_done = t
        elif etype == "response.cancelled":
            marks.cancelled_at = t
            return
        elif etype == "error":
            raise RuntimeError(f"server error: {evt}")
        if marks.response_audio_done:
            return

async def run_test(audio_path: str, instructions: str = "You are a concise assistant."):
    headers = [
        ("Authorization", f"Bearer {os.environ['OPENAI_API_KEY']}"),
        ("OpenAI-Beta", "realtime=v1"),
    ]
    marks = TimingMarks()
    audio_out: list = []
    async with websockets.connect(OPENAI_REALTIME_URL, extra_headers=headers,
                                  max_size=None) as ws:
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "modalities": ["audio", "text"],
                "instructions": instructions,
                "voice": "marin",
                "input_audio_format": "pcm16",
                "output_audio_format": "pcm16",
                "turn_detection": {"type": "semantic_vad", "eagerness": "auto"},
            },
        }))
        send_task = asyncio.create_task(send_audio_file(ws, audio_path))
        recv_task = asyncio.create_task(collect_events(ws, marks, audio_out))
        await asyncio.gather(send_task, recv_task)

    if audio_out:
        pcm = b"".join(audio_out)
        with wave.open("response.wav", "wb") as f:
            f.setnchannels(1); f.setsampwidth(2); f.setframerate(24000)
            f.writeframes(pcm)
    return marks

if __name__ == "__main__":
    m = asyncio.run(run_test("user_query_24k.wav"))
    print(f"TTSF: {m.ttsf_ms():.1f} ms")
    print(f"E2E:  {m.e2e_ms():.1f} ms")
    print(f"音频块数: {len(m.audio_chunks)}")

10.3 把单次测试变批量回归 ​

python
import asyncio
import statistics
from pathlib import Path

async def batch_run(samples_dir: str, n_per_sample: int = 5):
    rows = []
    for wav in Path(samples_dir).glob("*.wav"):
        ttsfs = []
        e2es = []
        for _ in range(n_per_sample):
            m = await run_test(str(wav))
            if m.ttsf_ms(): ttsfs.append(m.ttsf_ms())
            if m.e2e_ms():  e2es.append(m.e2e_ms())
        rows.append({
            "sample": wav.name,
            "ttsf_p50": statistics.median(ttsfs) if ttsfs else None,
            "ttsf_p95": np.percentile(ttsfs, 95) if ttsfs else None,
            "e2e_p50": statistics.median(e2es) if e2es else None,
            "e2e_p95": np.percentile(e2es, 95) if e2es else None,
        })
    return rows

if __name__ == "__main__":
    import json as _j
    rows = asyncio.run(batch_run("./test_samples", n_per_sample=10))
    print(_j.dumps(rows, indent=2, ensure_ascii=False))

测试小贴士。

跑 P95 至少要有 30 个样本，否则置信区间太宽。每个样本至少跑 5 次以平均网络抖动。如果是 CI 场景，建议把测试拆成两层：smoke（3 个样本 × 1 次，跑通即过）和 full（30 个样本 × 5 次，作为发版准入）。

第11章：Gemini Live API 测试脚本 ​

11.1 安装与认证 ​

bash
pip install google-genai==1.10.0 numpy soundfile

export GOOGLE_API_KEY=sk-...

11.2 用官方 SDK 跑 Gemini 3 Live ​

Gemini 的 SDK 把 WebSocket 细节封装在了 client.aio.live.connect() 上下文管理器中，对测试友好——但要拿到底层时间戳，仍需要手工打点。

python
import asyncio
import os
import time
from google import genai
from google.genai import types

MODEL = "gemini-3-live-2026-02"

async def gemini_run(audio_bytes: bytes, sr: int = 16000):
    client = genai.Client(api_key=os.environ["GOOGLE_API_KEY"])
    config = types.LiveConnectConfig(
        response_modalities=[types.Modality.AUDIO],
        speech_config=types.SpeechConfig(
            voice_config=types.VoiceConfig(
                prebuilt_voice_config=types.PrebuiltVoiceConfig(voice_name="Zephyr")
            )
        ),
    )
    timing = {"send_done": None, "first_recv": None, "all_done": None}
    audio_chunks = []
    async with client.aio.live.connect(model=MODEL, config=config) as session:
        async def push():
            chunk_size = sr * 2 // 10  # 100ms PCM16
            for i in range(0, len(audio_bytes), chunk_size):
                await session.send_realtime_input(
                    audio=types.Blob(data=audio_bytes[i:i+chunk_size],
                                     mime_type=f"audio/pcm;rate={sr}")
                )
                await asyncio.sleep(0.1)
            timing["send_done"] = time.perf_counter()

        async def pull():
            async for resp in session.receive():
                t = time.perf_counter()
                if resp.data:
                    if timing["first_recv"] is None:
                        timing["first_recv"] = t
                    audio_chunks.append(resp.data)
                if resp.server_content and resp.server_content.turn_complete:
                    timing["all_done"] = t
                    return

        await asyncio.gather(push(), pull())

    if timing["send_done"] and timing["first_recv"]:
        ttsf = (timing["first_recv"] - timing["send_done"]) * 1000
    else:
        ttsf = None
    if timing["send_done"] and timing["all_done"]:
        e2e = (timing["all_done"] - timing["send_done"]) * 1000
    else:
        e2e = None
    return {"ttsf_ms": ttsf, "e2e_ms": e2e, "n_chunks": len(audio_chunks)}

if __name__ == "__main__":
    with open("user_query_16k.pcm", "rb") as f:
        data = f.read()
    print(asyncio.run(gemini_run(data)))

11.3 OpenAI 与 Gemini 的延迟对照基准 ​

把上面的脚本和第 10 章的 OpenAI 脚本接到同一个测试驱动器里，跑同一份样本集，能拿到客观对比。我们在 2026-03 在阿里云东京 region + 北京到东京专线上的实测数据：

样本类型OpenAI gpt-5-realtime TTSF P50/P95Gemini 3 Live TTSF P50/P95OpenAI E2E P50/P95Gemini 3 Live E2E P50/P95
3 秒短问 (50 条)340 / 520 ms290 / 480 ms1850 / 2400 ms1620 / 2150 ms
10 秒长问 (50 条)410 / 680 ms380 / 610 ms3920 / 5100 ms3540 / 4720 ms
含 function call (30 条)720 / 1080 ms—（v1.10 SDK 工具调用未稳定）2890 / 3700 ms—

第12章：噪声注入测试框架 ​

12.1 框架设计目标 ​

把第 7 章的噪声生成函数接进第 10 章的测试客户端，形成一个"在线注入 + 延迟和准确率双维度记录"的可复用框架。设计目标：

  • 噪声类型和 SNR 等级在 YAML 中配置，不改代码就能调矩阵。
  • 每条样本在不同扰动下的结果落到结构化日志，方便事后画曲线。
  • 可插拔 Judge：可换成人工评测、本地 ASR 算 WER、或 LLM-as-Judge 算语义准确率。

12.2 完整可运行框架 ​

python
import asyncio
import csv
import json
import time
from dataclasses import dataclass, asdict
from pathlib import Path
from typing import Callable, Optional

import librosa
import numpy as np
import soundfile as sf
import yaml

@dataclass
class NoiseRunResult:
    sample: str
    noise_type: str
    snr_db: float
    ttsf_ms: Optional[float]
    e2e_ms: Optional[float]
    transcript: str
    wer: Optional[float]
    semantic_score: Optional[float]

NOISE_FUNCS: dict = {}

def register_noise(name: str):
    def deco(fn: Callable):
        NOISE_FUNCS[name] = fn
        return fn
    return deco

@register_noise("white")
def _white(signal, snr_db):
    sig_p = np.mean(signal ** 2)
    n_p = sig_p / (10 ** (snr_db / 10))
    return signal + np.random.normal(0, np.sqrt(n_p), signal.shape)

@register_noise("pink")
def _pink(signal, snr_db):
    n = len(signal)
    white = np.random.randn(n)
    fft = np.fft.rfft(white)
    freqs = np.fft.rfftfreq(n)
    freqs[0] = freqs[1]
    pink = np.fft.irfft(fft / np.sqrt(freqs), n)
    pink = pink / np.std(pink)
    sig_p = np.mean(signal ** 2)
    n_p = sig_p / (10 ** (snr_db / 10))
    return signal + pink * np.sqrt(n_p)

@register_noise("multi_speaker")
def _multi(signal, snr_db, bg_paths=("bg1.wav", "bg2.wav")):
    bg_mix = np.zeros_like(signal)
    for p in bg_paths:
        bg, _ = librosa.load(p, sr=24000)
        if len(bg) < len(signal):
            bg = np.tile(bg, (len(signal) // len(bg) + 1))[:len(signal)]
        else:
            bg = bg[:len(signal)]
        bg_mix += bg
    bg_mix = bg_mix / len(bg_paths)
    sig_p = np.mean(signal ** 2)
    bg_p = np.mean(bg_mix ** 2) + 1e-12
    target_bg_p = sig_p / (10 ** (snr_db / 10))
    return signal + bg_mix * np.sqrt(target_bg_p / bg_p)

@register_noise("reverb")
def _reverb(signal, snr_db, sr=24000, decay=0.4, n_echos=6):
    impulse = np.zeros(int(sr * 0.6))
    impulse[0] = 1.0
    for i in range(1, n_echos):
        d = int(sr * 0.05 * i)
        if d < len(impulse):
            impulse[d] = decay ** i
    out = np.convolve(signal, impulse, mode="same")
    return out / (np.max(np.abs(out)) + 1e-8) * np.max(np.abs(signal))

def inject_and_save(src_path: str, noise_type: str, snr_db: float,
                    out_path: str) -> str:
    sig, sr = librosa.load(src_path, sr=24000)
    noisy = NOISE_FUNCS[noise_type](sig.copy(), snr_db)
    noisy = np.clip(noisy, -1.0, 1.0)
    sf.write(out_path, (noisy * 32767).astype(np.int16), sr,
             subtype="PCM_16")
    return out_path

# 复用第10章的 run_test
from realtime_client import run_test  # type: ignore

def compute_wer(ref: str, hyp: str) -> float:
    """简化版 WER，工程上用 jiwer 库"""
    ref_w = ref.split()
    hyp_w = hyp.split()
    n, m = len(ref_w), len(hyp_w)
    if n == 0:
        return 0.0
    dp = [[0]*(m+1) for _ in range(n+1)]
    for i in range(n+1): dp[i][0] = i
    for j in range(m+1): dp[0][j] = j
    for i in range(1, n+1):
        for j in range(1, m+1):
            cost = 0 if ref_w[i-1] == hyp_w[j-1] else 1
            dp[i][j] = min(dp[i-1][j]+1, dp[i][j-1]+1, dp[i-1][j-1]+cost)
    return dp[n][m] / n

async def run_matrix(config_path: str, out_csv: str):
    cfg = yaml.safe_load(open(config_path))
    rows = []
    for sample in cfg["samples"]:
        for noise_type in cfg["noise_types"]:
            for snr in cfg["snr_levels"]:
                tmp_path = f"/tmp/noisy_{Path(sample['audio']).stem}_{noise_type}_{snr}.wav"
                inject_and_save(sample["audio"], noise_type, snr, tmp_path)
                marks = await run_test(tmp_path)
                wer = compute_wer(sample["reference"],
                                  sample.get("transcript_hyp", ""))  # 实际需 ASR
                rows.append(NoiseRunResult(
                    sample=Path(sample["audio"]).name,
                    noise_type=noise_type, snr_db=snr,
                    ttsf_ms=marks.ttsf_ms(), e2e_ms=marks.e2e_ms(),
                    transcript="", wer=wer, semantic_score=None,
                ))
                print(f"{sample['audio']:30s} {noise_type:14s} SNR={snr:5.1f} "
                      f"TTSF={marks.ttsf_ms():.0f}ms WER={wer:.3f}")
    with open(out_csv, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=list(asdict(rows[0]).keys()))
        writer.writeheader()
        for r in rows:
            writer.writerow(asdict(r))

if __name__ == "__main__":
    asyncio.run(run_matrix("noise_matrix.yaml", "noise_results.csv"))

12.3 矩阵配置示例 ​

# noise_matrix.yaml
samples:
  - audio: ./samples/q1_book_train.wav
    reference: "帮我查明天去上海的高铁票"
  - audio: ./samples/q2_check_weather.wav
    reference: "上海明天天气怎么样"
  - audio: ./samples/q3_translate.wav
    reference: "把这段话翻译成英文"

noise_types:
  - white
  - pink
  - multi_speaker
  - reverb

snr_levels: [30, 20, 10, 5, 0]

第13章：Barge-in 自动化测试方案 ​

13.1 测试动作分解 ​

Barge-in 测试的难点不是技术，而是"在模型说话的过程中精确地插入用户音频"。要做对，需要拆三个时刻：

  1. t0：用户问完触发模型回答（通过 input_audio_buffer.commit + response.create 触发）。
  2. t1：等模型说出第 N 个 audio chunk（确保模型确实在说话状态）。
  3. t2：注入打断音频（再次发 input_audio_buffer.append）。
  4. t3：监听 response.cancelled 或 response.audio.done 截断事件，记录服务端响应时刻。

Barge-in latency = t3 - t2。商用基线 ≤ 200ms。

13.2 完整可运行 Barge-in 测试 ​

python
import asyncio
import base64
import json
import os
import time
import numpy as np
import soundfile as sf
import websockets

OPENAI_URL = "wss://api.openai.com/v1/realtime?model=gpt-5-realtime-2026-01-15"

async def barge_in_test(initial_wav: str, interrupt_wav: str,
                        interrupt_after_chunks: int = 4,
                        interrupt_after_ms: float = 600):
    """
    1. 推入 initial_wav 让模型开始一段较长回答
    2. 等模型吐出 interrupt_after_chunks 个 audio.delta（或经过 interrupt_after_ms）
    3. 推入 interrupt_wav 模拟用户打断
    4. 记录从打断音频起始到 response.cancelled 的时间
    """
    headers = [
        ("Authorization", f"Bearer {os.environ['OPENAI_API_KEY']}"),
        ("OpenAI-Beta", "realtime=v1"),
    ]
    audio1, sr1 = sf.read(initial_wav, dtype="int16")
    audio2, sr2 = sf.read(interrupt_wav, dtype="int16")
    assert sr1 == sr2 == 24000

    marks = {"interrupt_sent": None, "cancelled_received": None,
             "first_audio_delta": None, "audio_chunks": 0}
    interrupt_event = asyncio.Event()

    async def writer(ws):
        chunk = int(24000 * 0.1)
        for i in range(0, len(audio1), chunk):
            await ws.send(json.dumps({
                "type": "input_audio_buffer.append",
                "audio": base64.b64encode(audio1[i:i+chunk].tobytes()).decode(),
            }))
            await asyncio.sleep(0.1)
        await ws.send(json.dumps({"type": "input_audio_buffer.commit"}))
        await ws.send(json.dumps({
            "type": "response.create",
            "response": {
                "instructions": "请详细回答，至少 30 秒长度。",
                "modalities": ["audio", "text"],
            }
        }))
        await interrupt_event.wait()
        marks["interrupt_sent"] = time.perf_counter()
        for i in range(0, len(audio2), chunk):
            await ws.send(json.dumps({
                "type": "input_audio_buffer.append",
                "audio": base64.b64encode(audio2[i:i+chunk].tobytes()).decode(),
            }))
            await asyncio.sleep(0.1)
        await ws.send(json.dumps({"type": "input_audio_buffer.commit"}))

    async def reader(ws):
        start = time.perf_counter()
        async for raw in ws:
            evt = json.loads(raw)
            now = time.perf_counter()
            etype = evt.get("type")
            if etype == "response.audio.delta":
                if marks["first_audio_delta"] is None:
                    marks["first_audio_delta"] = now
                marks["audio_chunks"] += 1
                if (marks["audio_chunks"] >= interrupt_after_chunks
                        and not interrupt_event.is_set()):
                    interrupt_event.set()
                if (now - start > interrupt_after_ms / 1000
                        and not interrupt_event.is_set()):
                    interrupt_event.set()
            elif etype in ("response.cancelled", "response.done"):
                marks["cancelled_received"] = now
                return

    async with websockets.connect(OPENAI_URL, extra_headers=headers,
                                  max_size=None) as ws:
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "modalities": ["audio", "text"],
                "voice": "marin",
                "input_audio_format": "pcm16",
                "output_audio_format": "pcm16",
                "turn_detection": {"type": "server_vad",
                                   "threshold": 0.5,
                                   "prefix_padding_ms": 200,
                                   "silence_duration_ms": 400},
            },
        }))
        await asyncio.gather(writer(ws), reader(ws))

    if marks["interrupt_sent"] and marks["cancelled_received"]:
        latency = (marks["cancelled_received"] - marks["interrupt_sent"]) * 1000
    else:
        latency = None
    return {"barge_in_ms": latency, **marks}

if __name__ == "__main__":
    out = asyncio.run(barge_in_test(
        "long_question.wav", "interrupt_stop.wav",
        interrupt_after_chunks=6,
    ))
    print(out)

13.3 三种 Barge-in 测试场景 ​

场景打断音频内容期望模型行为检查点
明确停止"停一下"、"等等"200ms 内停说response.cancelled 出现
追问"那这个能不能改成蓝色"停止当前并响应新问题新 response.id 出现
误打断（咳嗽）咳嗽声、关门声不应被打断无 response.cancelled

容易踩的坑。

很多团队第一次跑 Barge-in 测试，结果显示 latency 经常负数或 0。原因是 "interrupt_sent" 打的时间戳是 客户端发送时刻，但服务端 cancel 是基于 服务端 VAD 触发时刻——两者时序复杂。建议在断言时同时记录服务端的 input_audio_buffer.speech_started 事件时刻作为参考。

第14章：视频流模型测试脚本（GPT-5 Vision Realtime） ​

14.1 用 OpenCV 模拟摄像头喂帧 ​

python
import asyncio
import base64
import cv2
import json
import os
import time
import websockets

OPENAI_URL = "wss://api.openai.com/v1/realtime?model=gpt-5-realtime-vision-2026-02"

def encode_frame(frame, quality=80) -> str:
    """把 BGR ndarray 编码成 base64 JPEG"""
    _, buf = cv2.imencode(".jpg", frame, [cv2.IMWRITE_JPEG_QUALITY, quality])
    return base64.b64encode(buf.tobytes()).decode()

async def video_stream_test(video_path: str, fps: int = 2,
                            question: str = "请描述这段视频中发生的事件，"
                                            "并标注出现的时间戳。"):
    cap = cv2.VideoCapture(video_path)
    src_fps = cap.get(cv2.CAP_PROP_FPS) or 30
    skip = max(1, int(src_fps / fps))
    headers = [
        ("Authorization", f"Bearer {os.environ['OPENAI_API_KEY']}"),
        ("OpenAI-Beta", "realtime=v1"),
    ]
    sent_frames = []
    first_text_at = None
    text_buf = []

    async with websockets.connect(OPENAI_URL, extra_headers=headers,
                                  max_size=None) as ws:
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "modalities": ["text"],
                "instructions": "你是视频分析助手，回答需带时间戳，"
                               "格式 [hh:mm:ss] 描述",
            },
        }))

        async def push_frames():
            idx = 0
            while True:
                ret, frame = cap.read()
                if not ret:
                    break
                if idx % skip == 0:
                    sent_at = time.perf_counter()
                    sent_frames.append(sent_at)
                    await ws.send(json.dumps({
                        "type": "input_video_frame.append",
                        "image": encode_frame(frame),
                        "timestamp_ms": int(idx / src_fps * 1000),
                    }))
                idx += 1
                await asyncio.sleep(1.0 / fps)
            await ws.send(json.dumps({
                "type": "conversation.item.create",
                "item": {"type": "message", "role": "user",
                         "content": [{"type": "input_text", "text": question}]},
            }))
            await ws.send(json.dumps({"type": "response.create"}))

        async def collect():
            nonlocal first_text_at
            async for raw in ws:
                evt = json.loads(raw)
                now = time.perf_counter()
                if evt.get("type") == "response.text.delta":
                    if first_text_at is None:
                        first_text_at = now
                    text_buf.append(evt["delta"])
                elif evt.get("type") == "response.done":
                    return

        await asyncio.gather(push_frames(), collect())
    cap.release()
    return {
        "n_frames_sent": len(sent_frames),
        "first_text_latency_ms": (first_text_at - sent_frames[-1]) * 1000
                                  if first_text_at else None,
        "transcript": "".join(text_buf),
    }

if __name__ == "__main__":
    print(asyncio.run(video_stream_test("kitchen_cooking.mp4", fps=2)))

14.2 视频识别延迟基准 ​

测量"目标进入画面 → 模型给出对应描述"的耗时。下面这套测试集是 2026 年我们在内部做手机助手压测时用的：

测试集典型 fpsP50 识别延迟P95 识别延迟识别准确率
厨房场景（拿调料瓶）2 fps1320 ms2410 ms91%
办公场景（指着屏幕文字）2 fps1840 ms3120 ms87%
户外步行（识别建筑）1 fps2480 ms4350 ms78%
儿童危险动作（攀爬）4 fps980 ms1820 ms94%

14.3 时序锚点准确率检查 ​

视频流模型回答里如果包含"在第 X 秒发生了 Y"，要单独验证 X 的准确性。建议做法：

  • 测试视频每段标注 ground truth：事件 → 出现时刻（毫秒级）。
  • 用正则提取模型回答里的时间戳。
  • 计算"预测时间 - 真实时间"的绝对误差（MAE），目标 ≤ 500ms。

第15章：案例：客服实时语音 Bot 上线前压测 ​

15.1 背景 ​

2026 年 Q1，某头部银行决定把客服热线 30% 的话务量切给基于 Doubao-Realtime-1.5 的语音 Bot。上线前 QA 团队需要给出"在真实业务负载下，语音 Bot 的延迟、抢话率、转人工率是否达标"的测试报告。

15.2 测试目标与基线 ​

指标基线（业务方设定）测试方法
TTSF P95≤ 800 ms50 个高频问句 × 30 次重复
E2E P95≤ 3500 ms同上
抢话率≤ 2%200 条带停顿样本
Barge-in P95≤ 250 ms50 条打断样本 × 5 次
SNR=15dB 下 WER 上升≤ 25%30 条嘈杂背景样本
5 分钟会话上下文丢失率≤ 1%20 个长对话剧本

15.3 测试方案 ​

  1. 样本采集：从真实呼叫中心 6 个月的录音中抽样 1000 条，按业务类型（账单查询/转账纠纷/卡片挂失/理财咨询）分层采样，去敏后作为测试集。
  2. 单条延迟基线：把第 10 章脚本改造接 Doubao 协议，在凌晨低峰跑 10 次平均，得到无负载下基线。
  3. 并发压测：模拟 50 / 100 / 200 / 500 路并发，看延迟随并发数的退化曲线。
  4. 噪声环境矩阵：用第 12 章框架在 5 个 SNR × 4 类噪声下跑全部样本。
  5. Barge-in 抽样：从样本里挑 50 条预期会被打断的（如长篇风控提示），用第 13 章方案测响应延迟。
  6. 5 分钟长对话：构造 20 个真实业务对话剧本，每条 8~12 轮，看模型是否记住开头提到的卡号、姓名。

15.4 关键结果 ​

压测在 2026-02 跑完，报告核心数据：

无负载基线 (50 路并发)：
  TTSF  P50/P95: 380 / 720 ms        ✓ 达标
  E2E   P50/P95: 2120 / 3340 ms      ✓ 达标
  抢话率: 1.3%                        ✓ 达标
  Barge-in P95: 220 ms               ✓ 达标

100 路并发：
  TTSF  P50/P95: 460 / 890 ms        ✗ P95 超标
  E2E   P50/P95: 2480 / 3920 ms      ✗ P95 超标

200 路并发：
  TTSF  P50/P95: 640 / 1480 ms       ✗ 严重超标
  E2E   P50/P95: 3210 / 6280 ms      ✗ 严重超标

噪声矩阵 SNR=15dB 多说话人：
  WER 上升 38%                       ✗ 超标
  抢话率 5.7%                         ✗ 超标

5 分钟长对话：
  上下文丢失率 4.2%                  ✗ 超标
  典型失败：第 8 轮后忘记开头提到的银行卡号

15.5 整改与上线 ​

根据测试结果，团队和供应商联合做了三项整改：

  1. 并发能力：从单 region 部署改为三 region 多活，把单 region 压到 80 路上限以下，靠负载均衡分流。
  2. 噪声处理：客户端接入 RNNoise 做前置降噪，在 SNR=15dB 测试中 WER 上升降到 18%。
  3. 上下文丢失：把对话历史用业务字段提取出关键实体（卡号、订单号、姓名），在每轮 system prompt 里以结构化形式注入，作为模型上下文窗口之外的"侧通道记忆"。

整改后第二轮压测全部指标达标，分阶段灰度上线 5%、20%、30%。上线 4 周后客户满意度从 78% 升到 84%，转人工率从 31% 降到 19%。

启示。

这个案例的关键不是"语音 Bot 不行"——基线下指标都达标。关键是"在真实业务负载和真实噪声环境下，性能会显著退化"。如果只跑无负载基线就上线，三个月后用户投诉爆掉。所以实时语音 Bot 上线前必须做三件事：并发压测、噪声压测、长对话压测。少做一项就是埋雷。

第16章：案例：手机系统在嘈杂环境下的视频识别误判 ​

16.1 背景 ​

2026 年 3 月，某手机厂商基于 GPT-5 Vision Realtime 推出"视频问答"功能：用户用摄像头对着任何物体提问，AI 实时回答。功能上线后 2 周，社区出现批量投诉："在地铁/商场等嘈杂环境下，AI 经常答非所问，甚至说出根本没看到的东西"。QA 团队需要复现并定位。

16.2 复现挑战 ​

这个 bug 难复现，因为它是"音视频双模态在嘈杂环境下的耦合失效"，普通的视频识别测试和普通的语音识别测试都看不出来。团队最终设计了"音视频双扰动"测试矩阵：

条件视频内容背景音用户问题正确回答
A1桌上一杯咖啡静音"这是什么"咖啡 / 饮料
A2桌上一杯咖啡地铁广播：本次列车开往……"这是什么"咖啡 / 饮料（不应受广播影响）
A3桌上一杯咖啡另一人对话："今天吃面条""这是什么"咖啡 / 饮料（不应说"面条"）

16.3 测试结果 ​

条件 A1 (静音):     正确识别率 96%
条件 A2 (地铁广播): 正确识别率 88%（轻度退化）
条件 A3 (旁人对话): 正确识别率 71%（严重退化）

A3 错误分析：
- 25% 把"咖啡"答成"面条"等旁人对话提到的食物
- 12% 直接拒答 "我没看清楚"
- 8%  开始回应旁人对话内容

16.4 根因分析 ​

把日志拉出来看，根因清晰：模型把"旁人对话内容"当作了"用户问题的上下文"。GPT-5 Vision Realtime 在多模态融合时，音频是"用户问什么"+"现场环境音"的混合，模型没有能力区分"哪段是用户在问、哪段是环境噪音"。当环境音也是清晰的人声时，特别容易混淆。

16.5 整改方案 ​

  1. 方案 A · 客户端音频路由：让用户问话用近距离麦克风（手机底部），通过 beam-forming 优先采集用户声，背景音不进 API。代价是需要硬件配合。
  2. 方案 B · 双段提示词：在 system prompt 中明确："只回应用户对着摄像头说的问题，忽略环境对话"。代价是依赖模型自身分辨能力，不稳定。
  3. 方案 C · 用户问句单独标记：让用户按住按钮说话（push-to-talk），按住期间的音频用专门的事件类型推送，环境音用另一种事件类型。这是 OpenAI 在 2026-04 新增的 input_audio.role 字段支持的。

厂商选择了 A + C 的组合：硬件层面默认开启 beam-forming，软件层面允许用户切到 push-to-talk 模式。整改后 A3 条件下识别率从 71% 升到 92%。

启示。

这个案例的核心教训是：多模态融合的失败模式不是单模态失败模式的简单累加。视频识别 OK + 语音识别 OK 不代表音视频联合识别 OK。测试团队必须设计专门的"双模态扰动矩阵"，把音视频的交叉影响显式覆盖到。

16.6 复盘的副产品：一份多模态扰动测试集 ​

这个案例之后，团队把"音视频双扰动测试"沉淀成了一份内部基础测试集，结构如下：

  • 视频维度：8 个常见物体类（食物、文字、人脸、屏幕、车辆、植物、宠物、工具）。
  • 音频维度：5 类背景音（静音、白噪声、地铁广播、旁人对话、儿童哭闹）。
  • 问句维度：3 类（明确指代"这是什么"、模糊指代"它是怎么用的"、追问"换一个角度看呢"）。
  • 笛卡尔积：8 × 5 × 3 = 120 个组合，每组 5 条样本，共 600 条。

这份测试集每月跑一次回归，任何模型升级、prompt 变更、客户端音频路由调整都需要先过这套测试再上线。

第17章：课堂练习 ​

  1. 延迟分解：你团队的语音 Bot 端到端延迟 P95 是 4200ms，业务方要求降到 3000ms 以内。请设计一个测试方案，把这 4200ms 拆解成 ASR/LLM/TTS（如果是传统 pipeline）或 网络/服务端处理/客户端播放（如果是端到端模型）的耗时分布，并给出至少 3 个具体的优化候选项。
  2. 抢话误判：观察一个真实场景：用户说"我想……（停 1.2 秒）……查一下账单"，模型在停顿处抢话答了"好的，请问您想查什么"。请设计一份不少于 30 条样本的测试集，把这种"语义未完成的停顿"覆盖到，并写出统计抢话率的脚本。
  3. Barge-in 自动化：在第 13 章脚本基础上，增加"误打断防御"测试——注入一段 500ms 的咳嗽声（用 librosa 合成或现成音频），断言模型不应该停说。给出预期通过条件和失败时的归因方法。
  4. 视频流帧率取舍：你的视频问答产品需要在 1fps、2fps、4fps 中选一个默认帧率。请设计一个二维实验矩阵（帧率 × 场景类型），跑完后绘制"识别准确率 vs 单次问答成本"的帕累托前沿，给出推荐配置和理由。
  5. 合规对照：你的客服语音 Bot 即将在中国市场上线。对照《生成式 AI 服务管理暂行办法》《个人信息保护法》以及《电信服务规范》，列出至少 6 项实时语音特有的合规检查项（提示：声纹、留存、AI 主动声明、转人工触达等）。
  6. 测试报告写作：基于第 15 章案例的数据，模拟你是 QA 负责人，给业务方写一份不超过一页的"是否同意上线"建议书。要求结构清晰，数据先行，结论明确（建议上线 / 建议有条件上线 / 建议暂缓上线，三选一并说明理由）。

本章小结。

实时语音视频流测试是 2026 年 LLM 测试体系里最复杂、最对工具链有要求的一块。它要求测试团队同时掌握：异步编程（asyncio + WebSocket）、音频 DSP（librosa + sounddevice）、视频处理（OpenCV）、统计分析（P50/P95、显著性检验）、用户体验直觉（什么样的延迟和抢话用户能容忍）。如果你的团队还停留在"录一段音、调一次 API、听一下结果"的手工测试阶段，那么实时语音产品上线后必然踩雷。这一篇给的脚本框架可以直接复用，但更重要的是把 "全双工时间轴 + 事件序列断言" 的思维方式建立起来——它不是这一章的方法，而是整个实时多模态测试的基础。

课程外延阅读建议。

这一章把"实时多模态"作为一个独立的测试维度讲完了。如果你想进一步深入，建议接着读：第 31 篇《延迟与吞吐性能测试》补足并发与服务端调优视角，第 22 篇《多模态测试》补足图像/音频/视频的离线评测方法，第 55 篇《AI 应用合规与备案测试》看实时语音对话的备案与留存要求。把这四篇串起来，你就拥有完整的实时多模态质量保证体系。

实时语音视频流测试 大模型测试体系教程 · 第 54 篇 · 内部培训资料