end-to-end-speech¶
第9章 · 多模态与实时交互 · 配套项目
chapter9/end-to-end-speech
项目说明¶
实验 9-4 配套：端到端语音思考 vs 级联流水线¶
对应《深入理解 AI Agent》第 9 章 实验 9-4 ★★★：使用 Step-Audio R1 实现端到端语音思考。
目的¶
书中实验 9-4 的核心是端到端语音思考模型 Step-Audio R1：单一模型直接「听 → 想 → 说」，把 ASR、LLM、TTS 三段合而为一，在隐空间中直接传递副语言信息（情绪、语气、语速、环境声），延迟更低、韵律更自然。
Step-Audio R1 无公开 endpoint、需多卡 GPU 部署，读者难以直接跑通。本 demo 因此用 OpenAI 的 speech-to-speech 模型 gpt-audio 作为真实可跑的端到端代表：它同样是「音频进 → 单模型 → 音频出」，一次调用就返回语音答案与其转写，中间没有独立的 ASR/LLM/TTS 三段。demo 让你同题对照端到端与级联两条真实管道，直观看到二者在延迟与信息损失（语气、副语言）上的差异——两条路的延迟都是本次运行真实测得，两段输出音频都用 ffprobe 校验。
demo 提供两类任务（--task），对应书中两条对照轴：
math（默认）：口述数学题（Spoken-MQA 风格）。答案只取决于「说了什么」，级联与端到端准确率相当（对应书 9.3「自级联」一节），因此这一任务主要对照延迟维度。paralinguistic：一句「答案取决于怎么说」的话。级联在 ASR 处把语音压成纯文本，LLM 拿到的只有字面，任何情绪判断都是文本代理思考（Textual Surrogate Reasoning，书中 9.3「文本代理思考问题」）——凭词汇猜情绪；端到端则直接听声学特征（语速、音调），据此回应。这一任务对照信息损失维度，正是书中 MGRD（模态锚定思考蒸馏） 要解决的核心问题。
原理：三种语音范式¶
书中把语音架构分为三种范式（chapter9.md「语音架构的三种范式」）：
| 范式 | 结构 | 特点 |
|---|---|---|
| 级联（Cascaded） | ASR → LLM → TTS 三个独立模型串联 | 模块清晰、可独立调优、可解释性好；但延迟串行累加，模型间以纯文本接口相连，情绪/语速/语调/环境声在交接时几乎全部丢失 |
| 端到端全模态（Omni / End-to-End） | 单一模型「听→想→说」（Step-Audio R1、gpt-audio 等） | 延迟更低、保留副语言信息、韵律自然，可「边想边说」；代价是训练数据需求大、可解释性差。仍假设「轮流说话」，靠 VAD 划分轮次 |
| 全双工（Full-Duplex） | 单模型边听边说，取消「轮流」假设（Moshi、GPT-Live） | 持续同时处理输入与输出，每秒多次决策该说/该听/该停/该打断；本 demo 不涉及 |
本 demo 聚焦前两种范式的对照：端到端（范式二）vs 级联（范式一）。
Step-Audio R1 是范式二里把「思考」也内化进单模型的代表：通过 MGRD（模态锚定思考蒸馏）真正基于声学特征思考，通过 MPS 双脑架构实现「边想边说」。gpt-audio 是同一范式（音频进单模型出音频）里读者可直接调用的代表。
运行¶
# 1. 安装依赖
pip install -r requirements.txt
# 2. 配置 Key
cp env.example .env
# 编辑 .env，填入有效的 OPENAI_API_KEY（需有权访问 gpt-audio / whisper-1 / tts-1）。
# 音频端点（端到端 gpt-audio、级联 ASR/TTS）只有 OpenAI 直连才有——OpenRouter 无音频端点，
# 故本实验必须用 OpenAI 直连 Key。可选：额外配 OPENROUTER_API_KEY，则级联中间的纯文本
# LLM 思考自动改走 OpenRouter 的 gpt-5.6-luna（绕开 gpt-5.6* 直连的组织实名认证）。
# 3. 运行（默认：math 任务，同题跑通端到端 + 级联，并打印真实延迟对照）
python demo.py
# 副语言任务：凸显端到端相对级联的优势（端到端听得到语气，级联只看到字面）
python demo.py --task paralinguistic
# 用真实带情绪的录音作输入（比 TTS 合成更能体现语气差异；与 --question 互斥）
python demo.py --task paralinguistic --audio-input my_voice.wav
# 只跑端到端 / 自定义提问 / 换音色 / 换输出目录
python demo.py --skip-cascade
python demo.py --question "从北京到上海高铁 4 小时，我 9 点出发，几点到？"
python demo.py --voice nova --output-dir out
python demo.py --help
# 切到自部署的真实 Step-Audio R1（覆盖环境变量）
python demo.py --step-audio-endpoint http://localhost:8000/v1/audio/chat
# 4.（可选）试听生成的语音答案
ffplay audio/answer_end_to_end.wav # 端到端
ffplay audio/answer_cascade.mp3 # 级联
CLI 全部参数见 python demo.py --help：--task、--question、--audio-input、--e2e-model、--step-audio-endpoint、--voice、--output-dir、--skip-cascade。默认行为（python demo.py 跑 math 任务、端到端 + 级联对照）与改造前保持一致。
依赖 ffprobe/ffplay（用于校验、试听音频）：brew install ffmpeg（macOS）。
预期输出示例（真实节选）¶
范式一：端到端语音思考（后端=gpt-audio，模型=gpt-audio）
形态：音频进 → 单模型「听→想→说」→ 音频出（一次调用，无独立 ASR/LLM/TTS 段）
[单阶段] 端到端（听→想→说，单模型一次调用） | 延迟=7.21s
语音答案转写（模型顺带产出，非中间文本阶段）：咱们先算一下：小明有12块钱，买了3支铅笔……最后不剩钱了。
ffprobe 校验：{'格式': 'wav', '时长(秒)': 20.75, ...}
端到端总延迟（单模型一次前向）：7.21s
范式二（对照基线）：级联流水线 ASR → LLM → TTS
[阶段 1] ASR 语音识别 | 模型=whisper-1 | 延迟=1.35s
[阶段 2] LLM 思考 | 模型=gpt-5.6-luna | 延迟=1.92s
[阶段 3] TTS 语音合成 | 模型=tts-1 | 延迟=3.66s
级联总延迟（各阶段串行相加）：6.94s
端到端 vs 级联：真实延迟对照
【1】实测总延迟 端到端 7.21s；级联 ASR(1.35)+LLM(1.92)+TTS(3.66)=6.94s
【3】书中表 9-1（引自 Step-Audio R1 论文，非本 demo 产出）
MPS Speak-First 92.8% / 完整 TBS 93.0% ……
副语言任务的真实节选（--task paralinguistic，一次真实运行）¶
同一句「行吧，那就这样吧，我没什么意见。」（TTS 合成，情感偏平），两条路的回答：
[阶段 1] ASR 语音识别 | whisper-1 → 「行吧,那就这样吧,我没什么意见」 ← 级联 LLM 只拿得到这行纯文本
两条路对同一段音频的回答（可直观对比是否「听到了怎么说」）：
级联（只凭 ASR 文本，文本代理思考）→ 「听起来你有些无奈，可能对这个决定不太满意。……」
端到端（直接听声学特征） → 「听起来你有点无奈，语速不快，音调挺平稳的。……」
差异一目了然：级联只能从字面「没什么意见」猜出无奈，说不出任何声学依据（书中「文本代理思考」）；端到端则直接引用了「语速不快、音调平稳」这类声学特征——这正是书中 MGRD 想让模型学会的「用耳朵思考」。注意默认输入是 TTS 合成、情感偏平，差异已被削弱；用 --audio-input 喂一段真实带情绪的录音，对照会更明显。
关于表 9-1：demo 里打印的 Spoken-MQA / URO-Bench 分数是引自 Step-Audio R1 论文（书中表 9-1 转引），并非本 demo 跑出来的——本 demo 真实产出的只有实测延迟与两段真实音频。延迟数字随网络与 OpenAI 负载波动；gpt-audio 是「一次前向出整段音频」，真流式的端到端（如 Step-Audio R1 的 MPS）还能「边想边说」把首字延迟进一步压低，这一点非本 demo 的整段延迟所能体现。
如何适配真实 Step-Audio R1¶
端到端一路的后端可切换：
- 默认：不配置
STEP_AUDIO_ENDPOINT→ 用gpt-audio（模型名可用E2E_MODEL覆盖）。 - 真实 Step-Audio R1：自行多卡 GPU 部署后，把服务地址写入
STEP_AUDIO_ENDPOINT，demo 的端到端一路即改走它。speech_model.py的EndToEndSpeechModel._run_step_audio给出了「上传音频、取回音频」的调用骨架；不同部署方案（vLLM / 自定义 HTTP 服务）的请求体各异，接入时按需改写。
局限¶
gpt-audio是端到端范式的代表，不是 Step-Audio R1 本身：它没有暴露 MPS「边想边说」的流式首字延迟，也没有 Step-Audio R1 论文报告的副语言基准分数。表 9-1 因此是引用而非复现。- demo 用「文本 TTS → 输入语音」构造输入，仅为演示闭环；真实场景中输入来自用户麦克风，副语言信息更丰富，级联的损失也更明显。
- 端到端与级联的延迟高低会随网络、模型负载、答案长度波动，单次运行不代表稳定结论；范式差异（副语言信息是否保留）才是更稳健的对照维度。
文件¶
demo.py：可运行主程序（python demo.py），同题跑通端到端 + 级联并打印真实对照。speech_model.py：EndToEndSpeechModel（gpt-audio / 可切 Step-Audio R1）与CascadedSpeechModel（whisper-1 → gpt-5.6-luna → tts-1）。requirements.txt/env.example：依赖与环境变量样例。audio/：运行时生成的输入/输出音频（已在.gitignore中忽略）。
源代码¶
demo.py¶
"""
实验 9-4 配套 demo：端到端语音思考 vs 级联流水线（真实跑通两条路并对照）。
书中实验 9-4 的核心是端到端语音思考模型 Step-Audio R1（直接「听→想→说」）。
Step-Audio R1 需多卡 GPU 部署、无公开 endpoint，本 demo 因此用 OpenAI 的
speech-to-speech 模型 `gpt-audio` 作为**真实可跑的端到端代表**（音频进、单模型、
音频出，无独立 ASR/LLM/TTS 三段），与**级联基线**（whisper-1 → gpt-5.6-luna → tts-1）
同题对照，两条路的延迟都是真实测得、两段输出音频都用 ffprobe 校验。
流程：
1. 用 TTS 合成一段「用户提问」的语音（一道需要多步推理的数学题，Spoken-MQA 风格），
作为两条管道共同的输入；
2. 端到端：一次 gpt-audio 调用，音频进 → 语音答案 + 转写出（单模型融合）；
3. 级联：whisper-1 转写 → gpt-5.6-luna 思考 → tts-1 合成回答语音（三段串行）；
4. 打印两条路的真实延迟对照，并用 ffprobe 确认两段输出音频真实生成；
5. 给出端到端 vs 级联的范式对照（延迟、副语言信息损失），并引用书中表 9-1。
"""
from __future__ import annotations
import argparse
import json
import os
import shutil
import subprocess
import sys
from pathlib import Path
from dotenv import load_dotenv
from openai import OpenAI
from speech_model import (
CascadedSpeechModel,
EndToEndSpeechModel,
PipelineResult,
synthesize_question_audio,
)
HERE = Path(__file__).parent
AUDIO_DIR = HERE / "audio"
# 一道需要多步推理的口述数学题（Spoken-MQA 风格：先听懂，再多步计算）
USER_QUESTION = (
"小明有 12 块钱，他买了 3 支铅笔，每支 2 块钱，"
"然后又用剩下的钱买了尽可能多的、每块 1 块 5 的橡皮，"
"请问他最后还剩多少钱？"
)
# 一句「答案取决于怎么说」的话：字面是顺从/无所谓，真实情绪要靠语气、语速、语调
# 才能判断（是真的没意见，还是无奈、赌气）。这类副语言（Paralinguistic）任务正是
# 端到端相对级联的分水岭——见 chapter9.md「文本代理思考」与 MGRD。
PARALINGUISTIC_QUESTION = "行吧，那就这样吧，我没什么意见。"
# 两类任务，对应书中两条对照轴：
# math —— 语义任务：级联与端到端准确率相当（书 9.3「自级联」一节），主看延迟。
# paralinguistic—— 副语言任务：答案取决于「怎么说」。级联在 ASR 处把语音压成纯文本，
# LLM 只能凭字面猜情绪（书中「文本代理思考」）；端到端直接听声学特征。
TASKS: dict[str, dict] = {
"math": {
"question": USER_QUESTION,
"e2e_prompt": (
"你是一个中文语音助手。请先在内部完成必要的推理，再用简洁、口语化、"
"适合朗读的中文说出结论，控制在三句话以内。"
),
"cascade_prompt": (
"你是一个语音助手。请先进行必要的推理，再给出简洁、口语化、"
"适合朗读的中文回答。回答控制在三句话以内。"
),
"axis": (
"语义（Spoken-MQA 风格）：答案只取决于「说了什么」。级联与端到端准确率相当，"
"本任务主要对照【延迟】维度。"
),
},
"paralinguistic": {
"question": PARALINGUISTIC_QUESTION,
"e2e_prompt": (
"你是一个中文语音助手。请先判断说话人的情绪、语速和语调（尽量依据声学特征"
"本身，而非仅凭字面意思），说出你听到的语气，再据此给出体贴、口语化、"
"适合朗读的中文回应，三句以内。"
),
"cascade_prompt": (
"你是一个中文语音助手。请先判断说话人的情绪、语速和语调，说出你听到的语气，"
"再据此给出体贴、口语化、适合朗读的中文回应，三句以内。"
),
"axis": (
"副语言（Paralinguistic）：答案取决于「怎么说」。级联的 LLM 只拿得到 ASR 纯"
"文本，任何情绪判断都是「文本代理思考」（凭字面猜）；端到端直接听声学特征。"
"本任务对照【信息损失】维度。注意：默认输入由 TTS 合成、情感偏平，差异被削弱；"
"用 --audio-input 喂一段真实带情绪的录音，对照会明显得多。"
),
},
}
def hr(char: str = "-", n: int = 68) -> str:
return char * n
def ffprobe_info(audio_path: str) -> dict:
"""用 ffprobe 读取音频真实信息（时长、格式、码率），确认文件已生成。"""
if not shutil.which("ffprobe"):
return {"error": "未安装 ffprobe，跳过音频校验（brew install ffmpeg）"}
try:
out = subprocess.run(
[
"ffprobe", "-v", "error",
"-show_entries", "format=duration,format_name,bit_rate,size",
"-of", "json", audio_path,
],
capture_output=True, text=True, check=True,
)
fmt = json.loads(out.stdout).get("format", {})
return {
"格式": fmt.get("format_name"),
"时长(秒)": round(float(fmt.get("duration", 0)), 2),
"码率(bps)": fmt.get("bit_rate"),
"文件大小(字节)": fmt.get("size"),
}
except Exception as e: # noqa: BLE001
return {"error": f"ffprobe 失败：{e}"}
def print_e2e_result(result: PipelineResult, backend: str) -> None:
s = result.stages[0]
print(hr("="))
print(f"范式一：端到端语音思考（后端={backend}，模型={s.model}）")
print(hr("="))
print("形态：音频进 → 单模型「听→想→说」→ 音频出（一次调用，无独立 ASR/LLM/TTS 段）")
print(f"\n[单阶段] {s.name} | 延迟={s.latency_s:.2f}s")
if s.text is not None:
print(f" 语音答案转写（模型顺带产出，非中间文本阶段）：{s.text}")
print(f" 语音答案：{s.audio_path}")
print(f" ffprobe 校验：{ffprobe_info(s.audio_path)}")
print(f"\n端到端总延迟（单模型一次前向）：{result.total_latency_s:.2f}s")
def print_cascade_result(result: PipelineResult) -> None:
print("\n" + hr("="))
print(f"范式二（对照基线）：级联流水线 ASR → LLM → TTS")
print(hr("="))
for i, s in enumerate(result.stages, 1):
print(f"\n[阶段 {i}] {s.name} | 模型={s.model} | 延迟={s.latency_s:.2f}s")
if s.text is not None:
print(f" 文本：{s.text}")
if s.audio_path is not None:
print(f" 音频：{s.audio_path}")
print(f" ffprobe 校验：{ffprobe_info(s.audio_path)}")
print(f"\n级联总延迟（各阶段串行相加）：{result.total_latency_s:.2f}s")
def print_comparison(e2e: PipelineResult, e2e_backend: str, cas: PipelineResult,
task: str) -> None:
"""打印端到端 vs 级联的真实对照（实测延迟）+ 范式概念差异 + 书中表 9-1（引用）。"""
stages = {s.name: s for s in cas.stages}
asr = stages["ASR 语音识别"]
llm = stages["LLM 思考"]
tts = stages["TTS 语音合成"]
print("\n" + hr("="))
print("端到端 vs 级联：真实对照")
print(hr("="))
print(f"\n【任务】{task}：{TASKS[task]['axis']}")
print("\n【1】实测总延迟（本次运行，随网络与负载波动）")
print(f" 端到端（{e2e_backend}，单模型一次调用）：{e2e.total_latency_s:.2f}s")
print(f" 级联 ASR({asr.latency_s:.2f}s) + LLM({llm.latency_s:.2f}s) "
f"+ TTS({tts.latency_s:.2f}s) = {cas.total_latency_s:.2f}s（三段串行累加）")
delta = cas.total_latency_s - e2e.total_latency_s
faster = "端到端更快" if delta > 0 else "级联更快"
print(f" 差值：{abs(delta):.2f}s（{faster}）。注意端到端是「一次前向出整段音频」，"
"真流式端到端还能「边想边说」把首字延迟进一步压低；级联的三段则天然串行累加。")
print("\n【2】信息损失（副语言 / 语气）——范式差异，非本次延迟数字")
print(" 级联在 ASR 处把语音压成纯文本，说话人的情绪、语速、语调、重音、停顿，")
print(" 以及背景环境声/音乐在交接时几乎全部丢失——LLM 只看到「说了什么」，")
print(" 看不到「怎么说的」。本次输入语音在 ASR 处被抹平为纯文本：")
print(f" ASR 文本 → 「{asr.text}」")
print(" 端到端模型在隐空间（Latent Space）中直接传递这些副语言信息，能感知")
print(" 情绪/语速/语调，并据此生成有表现力、韵律匹配的回复。")
print(" 两条路对同一段音频的回答（可直观对比是否「听到了怎么说」）：")
print(f" 级联（只凭 ASR 文本，文本代理思考）→ 「{llm.text}」")
print(f" 端到端（直接听声学特征） → 「{e2e.stages[0].text}」")
print("\n【3】书中表 9-1：Step-Audio R1 不同语音思考配置（引自 Step-Audio R1 论文，"
"非本 demo 产出）")
print(" 配置 Spoken-MQA URO-Bench")
print(" 不思考直接回答（基线） 70.6% 77.4")
print(" MPS Speak-First（零延迟） 92.8% 82.5")
print(" MPS Think-First（~80 tok 延迟） 93.9% 84.8")
print(" 完整 TBS（无延迟约束） 93.0% —")
print(" 出处：Step-Audio R1 论文（书中表 9-1 转引）。这些是论文报告的评测分数，")
print(" 不是本 demo 跑出来的——本 demo 只产出上面【1】的真实延迟与两段真实音频。")
print(" 要点：Speak-First 几乎不损推理精度（92.8% ≈ TBS 93.0%），因为 CoT")
print(" 开头往往只是复述问题；这正是端到端「边想边说」能低延迟又不失准的原因。")
print("\n【4】取舍小结")
print(" 级联：模块清晰、每段可独立调优、可解释性好；但延迟串行累加、副语言损失大。")
print(" 端到端：延迟更低、保留非文字信息、韵律自然；代价是训练数据需求大、")
print(" 可解释性差。二者在 2026 年的生产系统中长期并存。")
def parse_args() -> argparse.Namespace:
p = argparse.ArgumentParser(
prog="demo.py",
description="实验 9-4：端到端语音思考 vs 级联流水线。合成（或读取）一段「用户"
"提问」语音，同题跑通端到端（gpt-audio，可切 Step-Audio R1）与级联"
"（whisper-1 → gpt-5.6-luna → tts-1），打印真实延迟与信息损失对照。",
epilog="示例：\n"
" python demo.py # 默认：数学题，端到端+级联对照\n"
" python demo.py --task paralinguistic # 副语言任务：凸显端到端优势\n"
" python demo.py --audio-input my_voice.wav # 用真实录音作输入（更能体现语气差异）\n"
" python demo.py --e2e-model gpt-audio --voice nova --output-dir out\n"
" python demo.py --step-audio-endpoint http://localhost:8000/v1/audio/chat",
formatter_class=argparse.RawDescriptionHelpFormatter,
)
p.add_argument("--task", choices=sorted(TASKS), default="math",
help="任务类型：math=口述数学题（语义任务，主看延迟）；"
"paralinguistic=副语言任务（答案取决于「怎么说」，凸显端到端"
"对级联的优势）。默认 math。")
p.add_argument("--question", default=None,
help="自定义口述提问文本，覆盖 --task 的默认题目。与 --audio-input 互斥。")
p.add_argument("--audio-input", metavar="FILE", default=None,
help="直接用一段已有的音频文件（.wav/.mp3）作为输入，跳过 TTS 合成"
"提问；适合喂真实带情绪的录音。与 --question 互斥。")
p.add_argument("--e2e-model", default=None,
help="端到端 speech-to-speech 模型名（默认取环境变量 E2E_MODEL，"
"再默认 gpt-audio）。")
p.add_argument("--step-audio-endpoint", default=None,
help="自部署的 Step-Audio R1 服务地址；给定后端到端一路改走它"
"（覆盖环境变量 STEP_AUDIO_ENDPOINT）。")
p.add_argument("--voice", default="alloy",
help="回答语音音色（端到端与级联 TTS 共用，默认 alloy）。")
p.add_argument("--output-dir", metavar="DIR", default=None,
help="音频输出目录（默认为脚本同级的 audio/）。")
p.add_argument("--skip-cascade", action="store_true",
help="只跑端到端，不跑级联对照基线。")
args = p.parse_args()
if args.question and args.audio_input:
p.error("--question 与 --audio-input 互斥，请二选一。")
return args
def _map_openrouter_model(model: str) -> str:
"""把常见模型名映射为 OpenRouter 的 provider/model 形式。"""
if "/" in model:
return model
if model.startswith("gpt-"):
return "openai/" + model
if model.startswith("claude-"):
return "anthropic/claude-opus-4.8"
return "openai/gpt-5.6-luna"
def _resolve_llm_client(openai_client: OpenAI):
"""为级联的纯文本 LLM 思考选路：优先 OpenRouter，否则回落到 OpenAI 直连。
返回 (client, model, route_label)。ASR/TTS/端到端音频仍固定用 openai_client。
"""
model = os.getenv("LLM_MODEL") or os.getenv("OPENAI_MODEL") or "gpt-5.6-luna"
if os.getenv("OPENROUTER_API_KEY"):
client = OpenAI(
base_url="https://openrouter.ai/api/v1",
api_key=os.getenv("OPENROUTER_API_KEY"),
timeout=120.0,
max_retries=3,
)
return client, _map_openrouter_model(model), "OpenRouter"
return openai_client, model, "OpenAI 直连"
def main() -> int:
args = parse_args()
task = args.task
task_cfg = TASKS[task]
question = args.question or task_cfg["question"]
load_dotenv(HERE / ".env")
api_key = os.getenv("OPENAI_API_KEY")
if not api_key:
print("错误：未配置 OPENAI_API_KEY。端到端 gpt-audio 与级联的 ASR(whisper)/TTS(tts-1) "
"都是音频端点，只有 OpenAI 直连才有——OpenRouter 无音频端点，故本实验必须提供"
" OpenAI 直连 Key。请复制 env.example 为 .env 并填入。", file=sys.stderr)
return 1
# 音频客户端：端到端 gpt-audio、级联的 whisper/tts 都必须走 OpenAI 直连。
# timeout + 自动重试：单次网络/SSL 抖动不至于让整条流水线崩溃
client = OpenAI(api_key=api_key, timeout=120.0, max_retries=3)
# 级联中间的纯文本 LLM 思考：优先走 OpenRouter（绕开 gpt-5.6* 直连的组织实名认证），
# 没有 OPENROUTER_API_KEY 时回落到上面的 OpenAI 直连客户端。
llm_client, llm_model, llm_route = _resolve_llm_client(client)
out_dir = Path(args.output_dir) if args.output_dir else AUDIO_DIR
out_dir.mkdir(parents=True, exist_ok=True)
e2e_audio = str(out_dir / "answer_end_to_end.wav")
cascade_audio = str(out_dir / "answer_cascade.mp3")
# -- 步骤 1：准备「用户提问」语音（两条管道共同输入） ----------------------
print(hr("="))
print(f"步骤 1：准备一段「用户提问」语音（任务={task}，作为两条管道共同的输入）")
print(hr("="))
if args.audio_input:
question_audio = args.audio_input
if not Path(question_audio).is_file():
print(f"错误：--audio-input 指定的音频文件不存在：{question_audio}",
file=sys.stderr)
return 1
print(f"使用已有输入音频（跳过 TTS 合成）：{question_audio}")
print("提问文本：（未知，来自音频文件，将由级联的 ASR 转写还原）")
else:
question_audio = str(out_dir / "user_question.mp3")
print(f"提问文本：{question}")
synthesize_question_audio(client, question, question_audio)
print(f"已用 TTS 合成输入音频：{question_audio}")
print(f"ffprobe 校验：{ffprobe_info(question_audio)}")
# -- 步骤 2：端到端语音思考（真实跑通） -----------------------------------
print("\n" + hr("="))
print("步骤 2：端到端语音思考（音频进 → 单模型 → 音频出）")
print(hr("="))
e2e_model = EndToEndSpeechModel(
client,
model=args.e2e_model,
voice=args.voice,
system_prompt=task_cfg["e2e_prompt"],
endpoint=args.step_audio_endpoint,
)
backend = e2e_model.backend
if backend == "step-audio-r1":
print(f"检测到 STEP_AUDIO_ENDPOINT={e2e_model.endpoint}，走真实 Step-Audio R1。")
else:
print(f"未配置 STEP_AUDIO_ENDPOINT，用 OpenAI speech-to-speech 模型 "
f"{e2e_model.model} 作为端到端代表（真正的音频→单模型→音频）。")
print("如需换成书中的 Step-Audio R1，自部署后把地址写入 STEP_AUDIO_ENDPOINT 即可。\n")
try:
e2e_result = e2e_model.run(question_audio, e2e_audio)
except Exception as e: # noqa: BLE001
print(f"错误：端到端调用失败：{e}", file=sys.stderr)
return 2
print_e2e_result(e2e_result, backend)
# -- 步骤 3：级联流水线（对照基线） ---------------------------------------
cascade_result = None
if not args.skip_cascade:
print(f"\n级联 LLM 思考路由：{llm_route}（模型 {llm_model}）；"
f"ASR/TTS 仍走 OpenAI 直连。")
cascaded = CascadedSpeechModel(
client,
llm_model=llm_model,
tts_voice=args.voice,
system_prompt=task_cfg["cascade_prompt"],
llm_client=llm_client,
)
cascade_result = cascaded.run(question_audio, cascade_audio)
print_cascade_result(cascade_result)
# -- 步骤 4：真实延迟对照 + 范式差异 + 表 9-1（引用） ---------------------
if cascade_result is not None:
print_comparison(e2e_result, backend, cascade_result, task)
print("\n完成。可试听：")
print(f" ffplay {e2e_audio} # 端到端语音答案")
if cascade_result is not None:
print(f" ffplay {cascade_audio} # 级联语音答案")
return 0
if __name__ == "__main__":
raise SystemExit(main())
speech_model.py¶
"""
可插拔的语音模型接口：端到端（end-to-end）与级联（cascaded）两种范式。
对应《深入理解 AI Agent》实验 9-4「使用 Step-Audio R1 实现端到端语音思考」。
- 端到端（EndToEndSpeechModel）：单一模型直接「听 → 想 → 说」，音频进、音频出，
中间没有暴露给外部的纯文本推理阶段。Step-Audio R1 是书中的代表模型（音频编码器
+ 音频适配器 + Qwen2.5 32B 解码器，需多卡 GPU，无公开 endpoint）。为了让读者不
依赖自建 GPU 集群也能真实跑通「端到端语音思考」这一范式，本 demo 默认改用 OpenAI
的 speech-to-speech 模型 `gpt-audio`：它同样是「音频 → 单模型 → 音频」，一次调用
就同时返回语音答案与其文字转写，不经过独立的 ASR/LLM/TTS 三段。若你已经自行部署
了真正的 Step-Audio R1，把服务地址写入 STEP_AUDIO_ENDPOINT 即可切换到它。
- 级联（CascadedSpeechModel）：把 ASR → LLM → TTS 三个独立模型串成流水线，
一棒接一棒。可用 OpenAI 的 whisper-1 / gpt-5.6-luna / tts-1 真实跑通完整闭环。
代价：模型间以离散文本接口相连，说话人的情绪、语气、语调等副语言信息在交接时
几乎损失殆尽（见 chapter9.md 范式二 · 端到端全模态模型）。这一路在 demo 中作为
与端到端对照的**基线**。
"""
from __future__ import annotations
import base64
import os
import time
from dataclasses import dataclass, field
from typing import Optional
from openai import OpenAI
# ---------------------------------------------------------------------------
# 数据结构
# ---------------------------------------------------------------------------
@dataclass
class StageResult:
"""流水线中单个阶段的结果与延迟。"""
name: str # 阶段名（如 "ASR 语音识别"）
model: str # 使用的模型
latency_s: float # 该阶段耗时（秒）
text: Optional[str] = None # 文本产物（ASR 转写 / LLM 回答 / 端到端转写）
audio_path: Optional[str] = None # 音频产物（TTS 合成 / 端到端语音答案）
@dataclass
class PipelineResult:
"""一次完整「语音输入 → 思考 → 语音输出」的结果。"""
paradigm: str # "cascaded" 或 "end_to_end"
input_audio: str # 输入音频路径
output_audio: Optional[str] # 输出音频路径
stages: list[StageResult] = field(default_factory=list)
@property
def total_latency_s(self) -> float:
return sum(s.latency_s for s in self.stages)
# ---------------------------------------------------------------------------
# 端到端范式（可运行：默认 gpt-audio；可切换到自部署的 Step-Audio R1）
# ---------------------------------------------------------------------------
class EndToEndSpeechModel:
"""端到端语音思考模型：音频进 → 单模型「听→想→说」→ 音频出。
两种后端，二选一：
1. **gpt-audio（默认，OpenAI）**：真正的 speech-to-speech 模型，通过 Chat
Completions 调用（modalities=["text","audio"]，audio={voice,format}，
user 消息里放一个 input_audio 内容块）。一次调用即返回语音答案 + 其转写，
中间没有独立的 ASR/LLM/TTS 阶段——这正是端到端范式的形态。模型名可用
环境变量 E2E_MODEL 覆盖（默认 gpt-audio）。
2. **Step-Audio R1（可选，自部署）**：书中的端到端语音思考模型，由音频编码器 +
音频适配器 + Qwen2.5 32B 解码器组成，需多卡 GPU，无公开 endpoint。它通过
MGRD（模态锚定思考蒸馏）真正基于声学特征思考，并通过 MPS 双脑架构实现
「边想边说」。若配置了 STEP_AUDIO_ENDPOINT，本类改为向该地址上传音频、取回
音频（请求体因部署方案而异，见 run() 中的骨架，接入时按需改写）。
"""
def __init__(
self,
client: OpenAI,
model: Optional[str] = None,
voice: str = "alloy",
system_prompt: Optional[str] = None,
endpoint: Optional[str] = None,
) -> None:
self.client = client
self.model = model or os.getenv("E2E_MODEL", "gpt-audio")
self.voice = voice
# 优先用显式传入的 endpoint（CLI --step-audio-endpoint），否则回落到环境变量
self.endpoint = (endpoint if endpoint is not None
else os.getenv("STEP_AUDIO_ENDPOINT", "")).strip()
self.system_prompt = system_prompt or (
"你是一个中文语音助手。请先在内部完成必要的推理，再用简洁、口语化、"
"适合朗读的中文说出结论，控制在三句话以内。"
)
@property
def backend(self) -> str:
"""当前生效的端到端后端标识。"""
return "step-audio-r1" if self.endpoint else "gpt-audio"
# -- 后端一：自部署 Step-Audio R1（可选） --------------------------------
def _run_step_audio(self, input_audio: str, output_audio: str) -> PipelineResult:
# 不同部署方案（vLLM / 自定义 HTTP 服务）的请求体各异，此处给出最常见的
# 「上传音频、取回音频」形态，供接入真实 Step-Audio R1 时改写。
import requests # 延迟导入：仅在真正配置 endpoint 时才需要该依赖
t0 = time.perf_counter()
with open(input_audio, "rb") as f:
resp = requests.post(self.endpoint, files={"audio": f}, timeout=120)
resp.raise_for_status()
with open(output_audio, "wb") as out:
out.write(resp.content)
latency = time.perf_counter() - t0
stage = StageResult(
name="端到端（听→想→说，单模型融合）",
model="Step-Audio R1",
latency_s=latency,
audio_path=output_audio,
)
return PipelineResult(
paradigm="end_to_end",
input_audio=input_audio,
output_audio=output_audio,
stages=[stage],
)
# -- 后端二：gpt-audio speech-to-speech（默认） --------------------------
def _run_gpt_audio(self, input_audio: str, output_audio: str) -> PipelineResult:
in_fmt = "mp3" if input_audio.lower().endswith(".mp3") else "wav"
out_fmt = "wav" if output_audio.lower().endswith(".wav") else "mp3"
with open(input_audio, "rb") as f:
in_b64 = base64.b64encode(f.read()).decode()
t0 = time.perf_counter()
resp = self.client.chat.completions.create(
model=self.model,
modalities=["text", "audio"],
audio={"voice": self.voice, "format": out_fmt},
messages=[
{"role": "system", "content": self.system_prompt},
{
"role": "user",
"content": [
{
"type": "input_audio",
"input_audio": {"data": in_b64, "format": in_fmt},
}
],
},
],
)
latency = time.perf_counter() - t0
audio = resp.choices[0].message.audio
if audio is None or not audio.data:
raise RuntimeError(
f"端到端模型 {self.model} 未返回音频。请确认该 Key 有权访问 gpt-audio，"
"或用 E2E_MODEL 指定可用的 speech-to-speech 模型。"
)
with open(output_audio, "wb") as out:
out.write(base64.b64decode(audio.data))
# 端到端只有「一段」：听→想→说融合在单次前向中。transcript 是模型顺带
# 吐出的语音文字稿，并非一个独立的、可供 TTS 消费的中间文本阶段。
stage = StageResult(
name="端到端（听→想→说，单模型一次调用）",
model=self.model,
latency_s=latency,
text=audio.transcript,
audio_path=output_audio,
)
return PipelineResult(
paradigm="end_to_end",
input_audio=input_audio,
output_audio=output_audio,
stages=[stage],
)
def run(self, input_audio: str, output_audio: str) -> PipelineResult:
if self.endpoint:
return self._run_step_audio(input_audio, output_audio)
return self._run_gpt_audio(input_audio, output_audio)
# ---------------------------------------------------------------------------
# 级联范式（对照基线）
# ---------------------------------------------------------------------------
class CascadedSpeechModel:
"""级联语音流水线：ASR → LLM → TTS，三个独立模型串联。
默认使用 OpenAI：
- ASR：whisper-1 （语音 → 文本）
- LLM：gpt-5.6-luna （文本思考 → 文本回答）
- TTS：tts-1 （文本 → 语音）
ASR/TTS 只有 OpenAI 直连才有对应端点（OpenRouter 无音频端点），因此固定用 `client`
（必须是直连 OpenAI）；中间的纯文本 LLM 思考可用单独的 `llm_client` 走 OpenRouter，
从而绕开 gpt-5.6* 直连所需的组织实名认证。`llm_client` 缺省时回落到 `client`。
"""
def __init__(
self,
client: OpenAI,
asr_model: str = "whisper-1",
llm_model: str = "gpt-5.6-luna",
tts_model: str = "tts-1",
tts_voice: str = "alloy",
system_prompt: Optional[str] = None,
llm_client: Optional[OpenAI] = None,
) -> None:
self.client = client
self.llm_client = llm_client or client
self.asr_model = asr_model
self.llm_model = llm_model
self.tts_model = tts_model
self.tts_voice = tts_voice
self.system_prompt = system_prompt or (
"你是一个语音助手。请先进行必要的推理，再给出简洁、口语化、"
"适合朗读的中文回答。回答控制在三句话以内。"
)
# -- 阶段 1：ASR 语音识别 ------------------------------------------------
def transcribe(self, audio_path: str) -> StageResult:
t0 = time.perf_counter()
with open(audio_path, "rb") as f:
resp = self.client.audio.transcriptions.create(
model=self.asr_model,
file=f,
)
latency = time.perf_counter() - t0
return StageResult(
name="ASR 语音识别",
model=self.asr_model,
latency_s=latency,
text=resp.text.strip(),
)
# -- 阶段 2：LLM 思考 ----------------------------------------------------
def think(self, question_text: str) -> StageResult:
t0 = time.perf_counter()
resp = self.llm_client.chat.completions.create(
model=self.llm_model,
messages=[
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": question_text},
],
temperature=0.3,
)
latency = time.perf_counter() - t0
return StageResult(
name="LLM 思考",
model=self.llm_model,
latency_s=latency,
# content 可能为 None（如被截断/拒答）；用空串兜底，避免 .strip() 崩溃。
text=(resp.choices[0].message.content or "").strip(),
)
# -- 阶段 3：TTS 语音合成 ------------------------------------------------
def synthesize(self, text: str, output_audio: str) -> StageResult:
t0 = time.perf_counter()
resp = self.client.audio.speech.create(
model=self.tts_model,
voice=self.tts_voice,
input=text,
)
resp.stream_to_file(output_audio)
latency = time.perf_counter() - t0
return StageResult(
name="TTS 语音合成",
model=self.tts_model,
latency_s=latency,
audio_path=output_audio,
)
# -- 完整流水线 ----------------------------------------------------------
def run(self, input_audio: str, output_audio: str) -> PipelineResult:
asr = self.transcribe(input_audio)
llm = self.think(asr.text)
tts = self.synthesize(llm.text, output_audio)
return PipelineResult(
paradigm="cascaded",
input_audio=input_audio,
output_audio=output_audio,
stages=[asr, llm, tts],
)
def synthesize_question_audio(
client: OpenAI,
question_text: str,
output_audio: str,
tts_model: str = "tts-1",
voice: str = "shimmer",
) -> None:
"""用 TTS 先合成一段「用户提问」的语音，作为两条管道共同的输入。"""
resp = client.audio.speech.create(model=tts_model, voice=voice, input=question_text)
resp.stream_to_file(output_audio)
test_none_content.py¶
"""回归测试：LLM 思考阶段返回 content=None 时不应在 .strip() 处崩溃。
某些模型在截断/拒答时 message.content 为 None；此前
CascadedSpeechModel.think 会以 AttributeError 中断整条级联流水线，
现在与项目里其它调用点一样用空串兜底。不依赖真实 API：客户端为假对象。
"""
import sys
from pathlib import Path
from types import ModuleType, SimpleNamespace
sys.path.insert(0, str(Path(__file__).parent))
try:
import openai # noqa: F401
except ImportError:
sys.modules["openai"] = ModuleType("openai")
sys.modules["openai"].OpenAI = object
from speech_model import CascadedSpeechModel
def _client_returning(content):
msg = SimpleNamespace(content=content)
completions = SimpleNamespace(
create=lambda **kw: SimpleNamespace(choices=[SimpleNamespace(message=msg)]))
return SimpleNamespace(chat=SimpleNamespace(completions=completions))
def test_think_tolerates_none_content():
model = CascadedSpeechModel(client=None, llm_client=_client_returning(None))
stage = model.think("还剩多少钱？")
assert stage.text == ""
def test_think_normal_content_unchanged():
model = CascadedSpeechModel(client=None, llm_client=_client_returning(" 还剩 3 元。 "))
stage = model.think("还剩多少钱？")
assert stage.text == "还剩 3 元。"