跳转至

实验 9-3：模拟流式语音感知（Streaming Speech Perception）

配套《深入理解 AI Agent》第 9 章「实验 9-3：使用 Qwen2-Audio 模拟流式语音感知」。

目的

演示流式语音感知的核心权衡：把连续音频按递增长度分块喂给 ASR，每收到一小段 就产出「当前部分识别结果」，从而以极低的首包延迟尽早拿到文本；代价是——早期分块 因为缺少后半句上下文、语音被拦腰截断，识别可能不完整或出错，随着音频累积才逐步 收敛到正确文本。作为对照，「等完整音频到齐再识别一次」最准，但必须等整句说完 + 推理， 首字延迟最高。

书中原实验的现象是：带停顿的句子里「大概两点左右」在过早分块中被误识别为「大概零点 左右」。本 demo 用同类现象复现这一「过早决策的代价」。

模型适配说明（重要）

  • 书中原实验用 Qwen2-Audio（音频原生模型，可输出 <|noise|> 等声学事件 token）。 但其当前没有可直接调用的 key/endpoint，故本 demo 改用可用的 ASR 替代： OpenAI Whisper（whisper-1），读取 OPENAI_API_KEY。
  • 选择 Whisper 恰好合适：它和 Qwen2-Audio 一样是整段输入的非流式模型——编码器需要 一整段音频才能开始工作、且非增量（每处理一个更长的前缀都要从头重新编码识别）。 因此「按递增前缀切块、逐块识别」正好复现书中所述「模拟流式」的机制与代价。
  • 测试音频用 OpenAI TTS（tts-1） 现场合成，句子含时间信息「两点半」，前半句被截断 时容易识别不全 / 出错。
  • 必须用 OpenAI 直连 Key：本实验只用音频端点（ASR whisper-1 / TTS tts-1）， 这类端点只有 OpenAI 直连才有——OpenRouter 只做聊天补全、无音频端点，故无法回退到 OPENROUTER_API_KEY。若只想验证分块/计时逻辑，用 python demo.py --offline 即可，无需任何 Key。

与真正的流式模型（如采用分块/因果编码器的 Qwen3-Omni）相比，本 demo 的延迟数字只反映 「分块粒度 + 每块从头识别」的开销，并不等于真流式的首包延迟；这一点书中也已说明。

流式分块机制

  1. 用 TTS 合成整段中文测试音频（约 7~8 秒），保存为 audio/sentence.wav。
  2. 用 ffmpeg 按递增长度切出「到目前为止收到的全部音频」：t = 0.5s, 1.0s, 1.5s ...， 模拟音频流不断到达。
  3. 每个前缀块调用 Whisper 得到「当前部分识别结果」，记录单块识别延迟与累计到达延迟。
  4. 对照：仅在结尾对整段音频识别一次，记录其结果与延迟。
  5. 打印逐块识别表 + 整段对照，量化「首个可用识别」与「整段识别」的延迟/准确率权衡。

运行

cd chapter9/streaming-speech
pip install -r requirements.txt          # 另需本机 ffmpeg：brew install ffmpeg
cp env.example .env                       # 填入 OPENAI_API_KEY（或直接 export）
python demo.py                             # 默认：TTS 合成 + 0.5s 粒度真实 Whisper 流式识别
python demo.py --quick                     # 分块粒度放大到 1.5s，Whisper 调用减到约 1/3
python demo.py --sentence "..." --chunk-step 0.5   # 自定义测试句与分块粒度
python demo.py --audio my.wav              # 用现成音频作输入，跳过 TTS 合成
python demo.py --compare-chunks            # 跨 0.5/1.0/2.0s 的分块粒度延迟对照表
python demo.py --offline                   # 离线自检：不联网、不需 ffmpeg，用合成识别器
python demo.py --offline --compare-chunks  # 离线合成的跨粒度延迟对照表
python demo.py --output result.json        # 结果（逐块表/对照表）另存为 JSON
python demo.py --help                      # 查看全部参数

常用参数（python demo.py --help）：

  • --sentence：测试句（默认为书中同类的带时间信息的句子）。
  • --chunk-step：分块粒度（秒），默认 0.5，越小分块越多越慢。
  • --quick：把粒度放大到 1.5s 快速演示（Whisper 调用约 1/3）。
  • --audio PATH：用现成音频文件作输入，跳过 TTS 合成（离线模式忽略）。
  • --compare-chunks [S1,S2,...]：在多个分块粒度上各跑一遍，输出跨粒度延迟对照表；不带值时用默认 0.5,1.0,2.0（秒）。
  • --offline：离线自检，不联网、不需 ffmpeg、不需 Key，用合成识别器（SYNTHETIC）驱动同一套分块/计时逻辑——文本按前缀比例揭示、延迟为合成值，仅验证流程、不代表任何真实模型性能。
  • --duration SEC：离线模式下的整段时长（缺省按句子长度估算）。
  • --tts-model / --voice / --asr-model / --language：覆盖 TTS/ASR 模型与语言（默认 tts-1 / alloy / whisper-1 / zh）。
  • --output PATH：把结果另存为 JSON。

真实运行输出（节选，供参考）

一次真实运行（句子：「麻烦你帮我把明天下午的会议改到两点半，地点还是在三号会议室， 别忘了通知大家。」，整段 7.75s）的逐块识别：

块#01  音频前缀  0.5s | 识别：麻烦你们                         ← 过早截断：误识别 + 不完整
块#03  音频前缀  1.5s | 识别：麻烦你帮我把明天下午的
块#05  音频前缀  2.5s | 识别：...改到两点                       ← 时间只听到一半
块#06  音频前缀  3.0s | 识别：...改到两点半                     ← 补足上下文后收敛
块#10  音频前缀  5.0s | 识别：...地点还是在四川                 ← 截断误识别（应为「三号会议室」）
块#12  音频前缀  6.0s | 识别：...地点还是在三号会议室           ← 随音频增长收敛
块#14  音频前缀  7.0s | 识别：...别忘了通知我                   ← 又一处过早误判（应为「通知大家」）
块#16  音频前缀  7.8s | 识别：...别忘了通知大家（完整正确）

整段识别（等完整音频）：麻烦你帮我把明天下午的会议改到两点半 地点还是在三号会议室 别忘了通知大家
  需等待：7.75s（录完）+ 2.25s（推理）
流式首个可用识别：仅需约 0.5s 音频即产出部分结果，比整段提前 7.2s 拿到第一版

可见：流式分块把「首个部分结果」的延迟从「录完整句（7.75s）之后」提前到「收到 0.5s 音频」， 但代价是早期块出现「你帮我→你们」「三号会议室→四川」「通知大家→通知我」等过早决策误识别， 随音频累积逐步收敛。整段识别最准，延迟最高。这正是流式语音感知的延迟 vs 准确率权衡。

注：每次运行会真实调用 TTS + Whisper，具体识别文本与延迟数字会略有波动； 上表为某一次真实运行的结果，不同运行早期块的误识别位置可能不同，但「早期不准、 随音频收敛」的规律稳定复现。

文件说明

  • demo.py：主程序（合成音频 → 递增分块流式识别 → 整段对照）。
  • requirements.txt：Python 依赖（另需本机 ffmpeg / ffprobe）。
  • env.example：环境变量示例，复制为 .env 填入 OPENAI_API_KEY。
  • audio/：运行时生成的音频（已 gitignore）。