controllable-tts¶
第9章 · 多模态与实时交互 · 配套项目
chapter9/controllable-tts
项目说明¶
实验 9-5：控制标记驱动的可控 TTS¶
《深入理解 AI Agent》实验 9-5 的可运行配套项目。
核心思路：让主 LLM 的输出不只是文本，还带上控制标记（情感 / 语速 / 风格 / 停顿 / 笑声等）；执行层解析这些标记，映射到一个参考语音库里对应的音色/风格档案， 再合成语音。这样「在哪里该停顿、该用什么语气」的决策交给了 LLM，同一段文本在不同 控制标记下能合成出不同风格、情感、节奏的语音。
Provider 适配（重要）¶
书中实验 9-5 使用 Fish Audio S1 的声音克隆：用 3-10 秒参考语音零样本克隆同一 音色，构建覆盖情绪 × 语速 × 风格的参考语音库，靠控制标记选择参考语音，Fish Audio 保证不同参考语音之间音色一致、只有韵律和情感变化。
本环境无 Fish Audio 可用 key，因此改用 OpenAI TTS 演示完全相同的思路：
| 书中（Fish Audio） | 本项目（OpenAI TTS） |
|---|---|
| 声音克隆保证音色一致 | 全库固定同一个 voice（alloy），音色不变 |
| 每条参考语音的韵律/情感 | 每个档案对应一段 instructions 风格提示词 |
| 控制标记选参考语音 | 控制标记解析后 -> 选 (情绪,语速,风格) 档案 |
- 首选模型
gpt-4o-mini-tts：支持instructions参数，可用一段中文提示词精确 控制情感/语速/口吻，最贴近「控制标记 → 风格化语音」的语义。 - 若首选模型不可用，代码自动兜底到
tts-1：不支持 instructions，改用多 voice +speed参数 + 文本级停顿近似。
必须用 OpenAI 直连 Key：本实验只用 TTS 语音合成端点（gpt-4o-mini-tts / tts-1），
这类音频端点只有 OpenAI 直连才有——OpenRouter 只做聊天补全、无音频合成端点，故无法回退到
OPENROUTER_API_KEY。离线查看语音库/标记映射（--list-voices / --dump-mapping）则无需任何 Key。
局限：OpenAI TTS 无法像 Fish Audio 那样原生生成笑声/叹气等非语言音。本项目对
<laugh>/[SIGH]用「匹配情绪的拟声词」（如“哈哈，”“唉——”）近似，[PAUSE]/[THINKING]等停顿则用 ffmpeg 生成真实静音插入，可被 ffprobe 验证时长。
控制标记 → TTS 参数 映射¶
状态标记（持续生效，直到被同类标记改变）¶
| 标记 | 中文写法 | 作用 |
|---|---|---|
[EMO:neutral\|happy\|frustrated\|thinking] |
[情感=中性\|高兴\|沮丧\|思考] |
切换情绪 |
[SPEED:normal\|fast\|slow] / [SPEED:0.8x] |
[语速=正常\|快\|慢] |
切换语速 |
[STYLE:formal\|casual] |
[风格=正式\|轻松] |
切换口吻 |
三个维度组合成参考语音库的一个档案（如 happy_fast_formal），
再拼成一段 instructions 提示词交给 gpt-4o-mini-tts。
内联标记（一次性事件）¶
| 标记 | 作用 |
|---|---|
[THINKING] |
切到「思考/慢速/正式」参考语音 + 插入 0.5s 停顿 |
[SEARCHING] |
同上，停顿 0.4s（搜索性犹豫） |
[PAUSE] / <pause> / [停顿] |
插入 0.5s 静音 |
[BREATH] / <breath> |
插入 0.4s 换气停顿 |
[SIGH] / <sigh> |
叹气拟声词「唉——」+ 0.3s 停顿 |
[LAUGH:small] / <laugh> |
轻笑拟声词「哈哈，」（欢快音色） |
<emphasis>…</emphasis> / [强调]…[/强调] |
对包裹文本追加「加重强调」提示词 |
参考语音库¶
voice_library.py 由 情绪(4) × 语速(3) × 风格(2) 笛卡尔积生成 24 条档案，全部固定
voice=alloy（音色一致），仅 instructions 不同。可单独运行查看：
安装与运行¶
pip install -r requirements.txt # 需系统已装 ffmpeg/ffprobe
cp env.example .env # 填入有效的 OPENAI_API_KEY
python demo.py # 生成 output/*.mp3
demo.py 做两件事：
- 三种配置对比（书中要求），同一段带标记文本：
A_no_markers.mp3无控制标记（流畅但机械）B_single_voice.mp3单一参考语音（自然但情感单调）C_voice_library.mp3多参考语音库（按标记切换情感/语速/停顿）- 同文本 / 不同控制标记 → 多个不同风格音频：
variant_*.mp3。
运行时会打印每个音频的「控制标记 → 参数」解析过程，以及 ffprobe 时长信息。
常用参数（python demo.py --help）：
| 参数 | 作用 |
|---|---|
--quick |
只跑三种配置对比（A/B/C），跳过 5 个风格变体，减少 TTS 调用与耗时 |
--text 文本 |
只合成这一段自定义文本（可内嵌控制标记，如 [情感=高兴][THINKING]…） |
--emotion / --speed / --style |
为 --text 指定情绪/语速/口吻（等价于在文本前加对应状态标记） |
-o / --output 路径 |
--text 模式的输出 mp3 路径（默认 output/custom.mp3） |
--list-voices |
离线（无需 API key）：打印完整参考语音库（24 条档案及其 instructions） |
--dump-mapping |
离线（无需 API key）：打印控制标记 → 动作映射表，并演示对示例文本的解析过程 |
预期输出示例（真实节选）¶
首选模型: gpt-4o-mini-tts（不可用时自动兜底 tts-1）
对比实验：同一段带控制标记的文本，三种配置
原始文本: [EMO:happy][SPEED:fast]太好了！您的订单已确认。[THINKING]嗯，让我查一下发货时间...[EMO:neutral][SPEED:normal]预计明天下午送达。
[C] 多参考语音库（解析控制标记 -> 逐段切换参考语音 + 停顿）
-- 控制标记解析过程 --
[EMO:happy] -> 情绪 = happy
[SPEED:fast] -> 语速 = fast
[THINKING] -> 切换到 思考/慢速/正式 参考语音
[THINKING] 停顿 -> 插入静音 500ms
-- 合成片段 --
· [happy_fast_formal ] gpt-4o-mini-tts voice=alloy text='太好了！您的订单已确认。'
· [静音 500ms]
· [thinking_slow_formal ] gpt-4o-mini-tts voice=alloy text='嗯，让我查一下发货时间...'
· [neutral_normal_formal ] gpt-4o-mini-tts voice=alloy text='预计明天下午送达。'
=> output/C_voice_library.mp3 | format_name=mp3 duration=11.324000 ...
对照三种配置的 ffprobe 时长即可看出差异：C（多参考语音库）因插入真实静音停顿，
比 A/B（8.5s 左右）更长（约 11.3s），且各片段用了不同的 (情绪,语速,风格) 档案。
（每次运行会真实调用 OpenAI TTS，时长/字节数会有小幅波动。）
文件说明¶
| 文件 | 作用 |
|---|---|
voice_library.py |
参考语音库 + 控制维度 → instructions 映射 |
markup.py |
控制标记解析器：带标记文本 → 片段列表(语音/静音) |
tts.py |
OpenAI TTS 合成 + ffmpeg 生成静音/拼接 |
demo.py |
演示入口，三种配置对比 + 风格变体 |
源代码¶
demo.py¶
"""
实验 9-5 演示：控制标记驱动的可控 TTS
======================================
演示两件事：
1) 三种配置对比（书中要求）：同一段带控制标记的文本，分别用
A. 无控制标记（流畅但机械）
B. 单一参考语音（自然但情感单调）
C. 多参考语音库（按控制标记切换情感/语速/停顿，接近真人客服）
2) 同一句文本、不同控制标记 -> 合成出多个不同风格的音频。
运行：python demo.py
输出：output/*.mp3
"""
import argparse
import os
import re
import subprocess
from dotenv import load_dotenv
from markup import parse, format_marker_reference
from tts import synthesize_segments, PREFERRED_MODEL
from voice_library import VOICE_LIBRARY, BASE_VOICE, EMOTIONS, SPEEDS, STYLES
load_dotenv()
OUT_DIR = os.path.join(os.path.dirname(__file__), "output")
TMP_DIR = os.path.join(OUT_DIR, ".tmp")
# 书中给出的 LLM 输出示例（带控制标记）
DEMO_TEXT = ("[EMO:happy][SPEED:fast]太好了！您的订单已确认。"
"[THINKING]嗯，让我查一下发货时间..."
"[EMO:neutral][SPEED:normal]预计明天下午送达。")
# 同一句文本 + 不同控制标记 -> 不同风格
STYLE_VARIANTS = {
"variant_happy_fast": "[情感=高兴][语速=快]您的订单已确认，预计明天下午送达。",
"variant_frustrated": "[情感=沮丧][语速=慢]您的订单已确认，预计明天下午送达。",
"variant_thinking": "[THINKING]您的订单已确认，[PAUSE]预计明天下午送达。",
"variant_casual_laugh": "[情感=高兴][风格=轻松]您的订单已确认<laugh>，预计明天下午送达。",
"variant_emphasis": "您的订单<emphasis>已确认</emphasis>，预计<emphasis>明天下午</emphasis>送达。",
}
def strip_markers(text: str) -> str:
"""去掉所有控制标记，得到纯文本（用于「无控制标记」基线）。"""
return re.sub(r"\[[^\]]*\]|<[^>]+>", "", text).strip()
def ffprobe(path: str) -> str:
"""打印 mp3 的时长/格式/码率，证明音频真实生成。"""
out = subprocess.run(
["ffprobe", "-v", "error", "-show_entries",
"format=duration,format_name,bit_rate", "-of",
"default=noprint_wrappers=1:nokey=0", path],
capture_output=True, text=True,
).stdout.strip().replace("\n", " ")
size = os.path.getsize(path)
return f"{out} size={size}B"
def render(name: str, segments, print_info=True):
"""合成一个音频文件并打印其合成信息 + ffprobe。"""
out_path = os.path.join(OUT_DIR, f"{name}.mp3")
info = synthesize_segments(segments, out_path, os.path.join(TMP_DIR, name))
if print_info:
for seg in info:
if seg["type"] == "silence":
print(f" · [静音 {seg['ms']}ms]")
else:
emph = " +强调" if "强调" in seg.get("instructions", "") else ""
print(f" · [{seg['profile']:26s}{emph}] {seg['model']:16s} "
f"voice={seg['voice']} text='{seg['text']}'")
print(f" => {os.path.relpath(out_path)} | {ffprobe(out_path)}")
return out_path
def print_voice_library():
"""离线打印完整参考语音库（无需 API key）。"""
print(f"参考语音库共 {len(VOICE_LIBRARY)} 条 = 情绪 {len(EMOTIONS)} × 语速 "
f"{len(SPEEDS)} × 风格 {len(STYLES)}；全库固定 base voice = {BASE_VOICE}"
f"（模拟 Fish Audio 的音色一致），仅 instructions 不同。\n")
print(f"{'档案 (情绪_语速_风格)':<28} {'base voice':<11} instructions")
print("-" * 100)
for key, v in VOICE_LIBRARY.items():
print(f"{key:<28} {v['base_voice']:<11} {v['instructions']}")
def print_marker_mapping(text: str):
"""离线打印控制标记映射表 + 对给定文本的解析过程（无需 API key）。"""
print("控制标记 -> 动作 静态映射表：\n")
print(format_marker_reference())
print("\n" + "=" * 72)
print("对示例文本的实时解析（标记 -> 参考语音 / 静音）：")
print("文本:", text)
print("=" * 72)
trace = []
segments = parse(text, trace=trace)
print("-- 解析过程 --")
for line in trace:
print(line)
print("-- 解析后的片段序列 --")
for i, seg in enumerate(segments):
if seg["type"] == "silence":
print(f" {i:02d}. [静音 {seg['ms']}ms]")
else:
profile = f"{seg['emotion']}_{seg['speed']}_{seg['style']}"
emph = " +强调" if seg.get("emphasis") else ""
print(f" {i:02d}. [语音 {profile}{emph}] '{seg['text']}'")
def build_marked_text(text: str, emotion, speed, style) -> str:
"""把 --emotion/--speed/--style 拼成状态标记前缀加到 text 前（text 自带标记时叠加）。"""
prefix = ""
if emotion:
prefix += f"[EMO:{emotion}]"
if speed:
prefix += f"[SPEED:{speed}]"
if style:
prefix += f"[STYLE:{style}]"
return prefix + text
def parse_args():
p = argparse.ArgumentParser(
description="实验 9-5：控制标记驱动的可控 TTS。默认（无参数）对比"
"「无标记 / 单一参考语音 / 多参考语音库」三种配置，并合成多个"
"风格变体（输出 output/*.mp3）。也可只合成单条自定义文本，或"
"离线查看参考语音库 / 控制标记映射（无需 API key）。",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="示例：\n"
" python demo.py # 跑完整对比 + 风格变体（需 API）\n"
" python demo.py --quick # 只跑三种配置对比\n"
" python demo.py --list-voices # 离线：打印 24 条参考语音库\n"
" python demo.py --dump-mapping # 离线：打印控制标记映射表\n"
" python demo.py --text '[情感=高兴][语速=快]您的订单已确认。' -o out.mp3\n"
" python demo.py --text '您的订单已确认。' --emotion thinking --speed slow",
)
p.add_argument(
"--text", metavar="文本",
help="只合成这一段文本（可内嵌控制标记，如 [情感=高兴][THINKING]…）。"
"不指定则跑默认的三种配置对比 + 风格变体。",
)
p.add_argument(
"--emotion", choices=list(EMOTIONS.keys()),
help="为 --text 指定情绪参考语音（等价于在文本前加 [EMO:x]）。",
)
p.add_argument(
"--speed", choices=list(SPEEDS.keys()),
help="为 --text 指定语速（等价于在文本前加 [SPEED:x]）。",
)
p.add_argument(
"--style", choices=list(STYLES.keys()),
help="为 --text 指定口吻风格（等价于在文本前加 [STYLE:x]）。",
)
p.add_argument(
"-o", "--output", metavar="路径",
help="--text 模式的输出 mp3 路径（默认 output/custom.mp3）。",
)
p.add_argument(
"--list-voices", action="store_true",
help="离线打印完整参考语音库（24 条档案及其 instructions），无需 API key。",
)
p.add_argument(
"--dump-mapping", action="store_true",
help="离线打印控制标记 -> 动作映射表，并对示例文本演示解析过程，无需 API key。",
)
p.add_argument(
"--quick", action="store_true",
help="仅跑三种配置对比（A/B/C），跳过 5 个风格变体，减少 TTS 调用与耗时。",
)
return p.parse_args()
def main():
args = parse_args()
# ---- 离线路径（无需 API key）：查看参考语音库 / 控制标记映射 ----
if args.list_voices:
print_voice_library()
return
if args.dump_mapping:
print_marker_mapping(args.text or DEMO_TEXT)
return
if not os.getenv("OPENAI_API_KEY"):
raise SystemExit("请先设置 OPENAI_API_KEY（见 env.example）；"
"或用 --list-voices / --dump-mapping 离线查看语音库与标记映射。")
os.makedirs(OUT_DIR, exist_ok=True)
# ---- 单条自定义文本合成 ----
if args.text or args.emotion or args.speed or args.style:
text = build_marked_text(args.text or "", args.emotion, args.speed, args.style)
if not text.strip():
raise SystemExit("请用 --text 提供要合成的文本。")
out_path = args.output or os.path.join(OUT_DIR, "custom.mp3")
print(f"首选模型: {PREFERRED_MODEL}（不可用时自动兜底 tts-1）\n")
print("文本:", text)
trace = []
segs = parse(text, trace=trace)
print("-- 控制标记解析过程 --")
for line in trace:
print(line)
print("-- 合成片段 --")
name = os.path.splitext(os.path.basename(out_path))[0]
render(name, segs)
if os.path.abspath(os.path.join(OUT_DIR, f"{name}.mp3")) != os.path.abspath(out_path):
os.replace(os.path.join(OUT_DIR, f"{name}.mp3"), out_path)
print(f" => 已写出 {out_path}")
return
print(f"首选模型: {PREFERRED_MODEL}（不可用时自动兜底 tts-1）"
f"{' [--quick 模式：跳过风格变体]' if args.quick else ''}\n")
# ================= 三种配置对比 =================
print("=" * 72)
print("对比实验：同一段带控制标记的文本，三种配置")
print("原始文本:", DEMO_TEXT)
print("=" * 72)
# ---- 配置 A：无控制标记（基线，流畅但机械）----
print("\n[A] 无控制标记（strip 掉所有标记，单次默认合成）")
plain = strip_markers(DEMO_TEXT)
print(" 纯文本:", plain)
seg_a = parse(plain) # 无标记 -> 单个中性片段
render("A_no_markers", seg_a)
# ---- 配置 B：单一参考语音（自然但情感单调）----
print("\n[B] 单一参考语音（去标记，全程用同一条中性/正常/正式参考语音）")
seg_b = [dict(type="speech", text=plain, emotion="neutral",
speed="normal", style="formal", emphasis=False)]
render("B_single_voice", seg_b)
# ---- 配置 C：多参考语音库（按控制标记切换）----
print("\n[C] 多参考语音库（解析控制标记 -> 逐段切换参考语音 + 停顿）")
trace = []
seg_c = parse(DEMO_TEXT, trace=trace)
print(" -- 控制标记解析过程 --")
for line in trace:
print(line)
print(" -- 合成片段 --")
render("C_voice_library", seg_c)
# ================= 同文本 / 不同控制标记 =================
if not args.quick:
print("\n" + "=" * 72)
print("同一句文本 + 不同控制标记 -> 不同风格音频")
print("=" * 72)
for name, text in STYLE_VARIANTS.items():
print(f"\n[{name}] {text}")
trace = []
segs = parse(text, trace=trace)
for line in trace:
print(line)
render(name, segs)
# ================= 汇总 =================
print("\n" + "=" * 72)
print("全部输出文件（ffprobe 时长对比）")
print("=" * 72)
for f in sorted(os.listdir(OUT_DIR)):
if f.endswith(".mp3"):
p = os.path.join(OUT_DIR, f)
print(f" {f:26s} {ffprobe(p)}")
if __name__ == "__main__":
main()
markup.py¶
"""
控制标记解析器（Control Markup Parser）
========================================
把带控制标记的文本解析成一串「片段」，每个片段要么是一段需要用某条参考语音
合成的语音（speech），要么是一段静音停顿（silence）。这一步对应书中「执行层
解析标记并映射到对应的参考语音」。
支持两类标记：
1) 状态标记（持续生效，直到被下一个同类标记改变）
[EMO:neutral|happy|frustrated|thinking] 或 [情感=中性|高兴|沮丧|思考]
[SPEED:normal|fast|slow] / [SPEED:0.8x] 或 [语速=正常|快|慢]
[STYLE:formal|casual] 或 [风格=正式|轻松]
2) 内联标记（一次性事件，插入停顿 / 填充音 / 非语言音，或临时改变状态）
[THINKING] 思考停顿 + 迟疑语气（=情绪思考/慢速/正式，并插入停顿）
[SEARCHING] 搜索性停顿（同上，停顿略短）
[PAUSE] / <pause> / [停顿] 插入停顿
[BREATH] / <breath> 换气停顿
[SIGH] / <sigh> 叹气（用叹气拟声词近似）
[LAUGH:small] / [LAUGH] / <laugh> 轻笑（用笑声拟声词近似）
<emphasis>...</emphasis> / [强调]...[/强调] 对包裹的文本加重强调
注意：OpenAI TTS 无法像 Fish Audio 那样「原生生成」笑声/叹气等非语言音，
这里用「拟声词 + 匹配情绪」的方式近似（详见 README 的 provider 适配说明）。
"""
import re
# 中文取值 -> 英文维度值的别名映射
_EMO_ALIAS = {
"中性": "neutral", "高兴": "happy", "开心": "happy", "兴奋": "happy",
"沮丧": "frustrated", "无奈": "frustrated", "思考": "thinking",
}
_SPEED_ALIAS = {"正常": "normal", "快": "fast", "快速": "fast", "慢": "slow", "慢速": "slow"}
_STYLE_ALIAS = {"正式": "formal", "轻松": "casual", "随意": "casual"}
# 各内联事件插入的停顿时长（毫秒）
PAUSE_MS = 500
BREATH_MS = 400
THINKING_MS = 500
SEARCHING_MS = 400
SIGH_TAIL_MS = 300
def _norm(value: str, alias: dict) -> str:
v = value.strip()
return alias.get(v, v.lower())
class Segment(dict):
"""一个片段：type='speech'(text, emotion, speed, style, emphasis) 或 type='silence'(ms)。"""
def parse(text: str, trace: list | None = None):
"""
解析带控制标记的文本，返回片段列表。
若传入 trace（list），会把「标记 -> 动作」的解析过程逐条记入，便于打印。
"""
def log(msg):
if trace is not None:
trace.append(msg)
# 当前状态（状态标记会持续改变它）
state = {"emotion": "neutral", "speed": "normal", "style": "formal", "emphasis": False}
segments: list[Segment] = []
buf = [] # 累积当前状态下的普通文本
def flush():
"""把缓冲区的普通文本作为一个 speech 片段输出。"""
s = "".join(buf).strip()
buf.clear()
if s:
segments.append(Segment(type="speech", text=s, **state))
def add_silence(ms, why):
flush()
segments.append(Segment(type="silence", ms=ms))
log(f" {why:22s} -> 插入静音 {ms}ms")
def add_speech_token(token, emotion, speed, style, why):
"""插入一个独立的、带指定情绪的短语音片段（用于笑声/叹气等拟声词）。"""
flush()
segments.append(Segment(type="speech", text=token, emotion=emotion,
speed=speed, style=style, emphasis=False))
log(f" {why:22s} -> 拟声语音 '{token}' (情绪={emotion},语速={speed})")
def set_state(**kw):
flush() # 状态改变前，先把旧状态的文本收尾
for k, v in kw.items():
state[k] = v
# 用一个总正则切出所有 [..] 与 <..> 标记，其余为普通文本
parts = re.split(r"(\[[^\]]*\]|<[^>]+>)", text)
for part in parts:
if not part:
continue
if not re.fullmatch(r"\[[^\]]*\]|<[^>]+>", part):
buf.append(part) # 普通文本
continue
m = part # 标记原文
inner = m[1:-1].strip()
# --- 状态标记：EMO / SPEED / STYLE（英文冒号式 或 中文等号式） ---
km = re.match(r"(?i)^(EMO|SPEED|STYLE)\s*:\s*(.+)$", inner)
cm = re.match(r"^(情感|语速|风格)\s*=\s*(.+)$", inner)
if km:
key, val = km.group(1).upper(), km.group(2)
elif cm:
key = {"情感": "EMO", "语速": "SPEED", "风格": "STYLE"}[cm.group(1)]
val = cm.group(2)
else:
key = val = None
if key == "EMO":
e = _norm(val, _EMO_ALIAS)
set_state(emotion=e)
log(f" {m:22s} -> 情绪 = {e}")
continue
if key == "SPEED":
raw = val.strip()
v = raw.lower().replace("x", "") # 兼容 0.8x
# 先认英文取值(normal/fast/slow)，再认中文别名(正常/快/慢)
if v in ("normal", "fast", "slow"):
s = v
elif raw in _SPEED_ALIAS:
s = _SPEED_ALIAS[raw]
else:
# 数字型（如 0.8）就近映射到 fast/slow/normal，仅用于展示
try:
f = float(v)
s = "fast" if f > 1.05 else ("slow" if f < 0.95 else "normal")
except ValueError:
s = "normal"
set_state(speed=s)
log(f" {m:22s} -> 语速 = {s}")
continue
if key == "STYLE":
st = _norm(val, _STYLE_ALIAS)
set_state(style=st)
log(f" {m:22s} -> 风格 = {st}")
continue
# --- 强调包裹 ---
low = inner.lower()
if low in ("emphasis", "强调"):
set_state(emphasis=True)
log(f" {m:22s} -> 开启强调")
continue
if low in ("/emphasis", "/强调"):
set_state(emphasis=False)
log(f" {m:22s} -> 关闭强调")
continue
# --- 内联事件标记 ---
tag = low.split(":")[0] # laugh:small -> laugh
if tag == "thinking":
set_state(emotion="thinking", speed="slow", style="formal")
log(f" {m:22s} -> 切换到 思考/慢速/正式 参考语音")
add_silence(THINKING_MS, "[THINKING] 停顿")
continue
if tag == "searching":
set_state(emotion="thinking", speed="slow", style="formal")
log(f" {m:22s} -> 切换到 思考/慢速/正式 参考语音")
add_silence(SEARCHING_MS, "[SEARCHING] 停顿")
continue
if tag in ("pause", "停顿"):
add_silence(PAUSE_MS, m)
continue
if tag in ("breath", "换气"):
add_silence(BREATH_MS, m)
continue
if tag == "sigh":
add_speech_token("唉——", "frustrated", "slow", "formal", m)
segments.append(Segment(type="silence", ms=SIGH_TAIL_MS))
continue
if tag == "laugh":
add_speech_token("哈哈，", "happy", "fast", "casual", m)
continue
# 未知标记：忽略但记录
log(f" {m:22s} -> [未知标记，已忽略]")
flush()
return segments
# ---------------------------------------------------------------------------
# 控制标记 -> 动作 的静态映射表（离线可查，供 demo.py --dump-mapping 打印）
# 这是「书中控制标记 -> 参考语音 / 非语言音」映射关系的单一事实来源。
# ---------------------------------------------------------------------------
# (类别, 标记写法, 中文写法, 映射到的动作)
MARKER_REFERENCE = [
("状态", "[EMO:neutral|happy|frustrated|thinking]", "[情感=中性|高兴|沮丧|思考]",
"切换情绪维度，选择参考语音"),
("状态", "[SPEED:normal|fast|slow] / [SPEED:0.8x]", "[语速=正常|快|慢]",
"切换语速维度（数字型就近映射到 fast/slow/normal）"),
("状态", "[STYLE:formal|casual]", "[风格=正式|轻松]", "切换口吻维度"),
("内联", "[THINKING]", "—", "切到「思考/慢速/正式」参考语音 + 插入 500ms 停顿"),
("内联", "[SEARCHING]", "—", "切到「思考/慢速/正式」参考语音 + 插入 400ms 停顿"),
("内联", "[PAUSE] / <pause>", "[停顿]", "插入 500ms 静音"),
("内联", "[BREATH] / <breath>", "[换气]", "插入 400ms 换气停顿"),
("内联", "[SIGH] / <sigh>", "—", "叹气拟声词「唉——」(沮丧音色) + 300ms 停顿"),
("内联", "[LAUGH:small] / [LAUGH] / <laugh>", "—", "轻笑拟声词「哈哈，」(高兴音色)"),
("内联", "<emphasis>…</emphasis>", "[强调]…[/强调]", "对包裹文本追加「加重强调」提示词"),
]
def format_marker_reference() -> str:
"""把 MARKER_REFERENCE 渲染成可打印的对齐表格字符串。"""
lines = [f"{'类别':<4} {'标记写法':<40} {'中文写法':<24} 动作", "-" * 100]
for cat, mark, zh, action in MARKER_REFERENCE:
lines.append(f"{cat:<4} {mark:<40} {zh:<24} {action}")
return "\n".join(lines)
if __name__ == "__main__":
print("控制标记 -> 动作 映射表：\n")
print(format_marker_reference())
test_empty_segments.py¶
"""回归测试：只含控制标记、没有任何语音正文的输入不应让 ffmpeg 空 concat 崩溃。
此前 `demo.py --text '[EMO:happy]'` 这类输入经 parse() 得到 0 个片段，
synthesize_segments 会拿空文件列表去跑 ffmpeg concat，
抛出晦涩的 CalledProcessError；现在应直接给出清晰的 SystemExit。
"""
import sys
from pathlib import Path
from types import ModuleType
import pytest
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 markup import parse
from tts import synthesize_segments
def test_marker_only_text_parses_to_zero_segments():
assert parse("[EMO:happy]") == []
def test_empty_segments_raise_clear_system_exit(tmp_path):
with pytest.raises(SystemExit) as excinfo:
synthesize_segments([], str(tmp_path / "out.mp3"), str(tmp_path / "work"))
assert "没有可合成的语音片段" in str(excinfo.value)
def test_normal_text_still_parses_to_speech_segment():
segs = parse("[EMO:happy]太好了！")
assert len(segs) == 1
assert segs[0]["type"] == "speech"
assert segs[0]["text"] == "太好了！"
assert segs[0]["emotion"] == "happy"
tts.py¶
"""
TTS 合成层（OpenAI TTS）
========================
Provider 适配：书中实验用 Fish Audio 的控制标记 + 声音克隆参考语音库。
Fish Audio 无可用 key，这里改用 OpenAI TTS 演示相同思路：
- 首选 gpt-4o-mini-tts：支持 `instructions` 参数，可用一段风格提示词精确
控制情感 / 语速 / 口吻，最贴近「控制标记 -> 风格化语音」的语义；
- 若该模型不可用，自动兜底到 tts-1：不支持 instructions，改用多 voice +
`speed` 参数 + 文本级停顿近似。
一段控制标记文本被解析成多个片段后：
- speech 片段：各自用对应参考语音（同一 base voice + 不同 instructions）合成；
- silence 片段：用 ffmpeg 生成真实静音；
最后用 ffmpeg 把所有片段按顺序拼成一个 mp3（音色一致、韵律/情感/停顿不同）。
"""
import os
import subprocess
import tempfile
from openai import OpenAI
from voice_library import VOICE_LIBRARY, BASE_VOICE, build_instructions, speed_factor, profile_key
# 可用 TTS_MODEL 环境变量覆盖；默认首选 gpt-4o-mini-tts
PREFERRED_MODEL = os.getenv("TTS_MODEL", "gpt-4o-mini-tts")
FALLBACK_MODEL = "tts-1"
_client = None
_active_model = None # 首次调用后确定实际使用的模型
def _get_client():
global _client
if _client is None:
# timeout + 自动重试：单次网络/SSL 抖动不至于让整段合成崩溃
_client = OpenAI(timeout=60.0, max_retries=3)
return _client
def _synth_call(model, text, voice, instructions, speed, out_path):
"""真正调用 OpenAI TTS。gpt-4o-mini-tts 用 instructions；tts-1 用 speed。"""
client = _get_client()
kwargs = dict(model=model, voice=voice, input=text, response_format="mp3")
if model == "tts-1":
# tts-1 不支持 instructions，用 speed 参数近似语速控制
kwargs["speed"] = max(0.25, min(4.0, speed))
else:
kwargs["instructions"] = instructions
with client.audio.speech.with_streaming_response.create(**kwargs) as resp:
resp.stream_to_file(out_path)
def synth_speech(text, emotion, speed, style, emphasis, out_path):
"""
合成一个 speech 片段，返回实际使用的 (model, voice, instructions/speed) 供打印。
音色固定 = BASE_VOICE，保证整段音色一致（模拟 Fish Audio 的音色一致性）。
"""
global _active_model
key = profile_key(emotion, speed, style)
profile = VOICE_LIBRARY.get(key)
voice = profile["base_voice"] if profile else BASE_VOICE
instructions = build_instructions(emotion, speed, style, emphasis)
spd = speed_factor(speed)
model = _active_model or PREFERRED_MODEL
try:
_synth_call(model, text, voice, instructions, spd, out_path)
_active_model = model
except Exception as e:
# 首选模型不可用时兜底到 tts-1
if model != FALLBACK_MODEL:
print(f" [warn] 模型 {model} 调用失败({repr(e)[:80]})，兜底到 {FALLBACK_MODEL}")
_synth_call(FALLBACK_MODEL, text, voice, instructions, spd, out_path)
_active_model = FALLBACK_MODEL
model = FALLBACK_MODEL
else:
raise
return {"model": model, "voice": voice, "profile": key,
"instructions": instructions, "speed_factor": spd}
def make_silence(ms, out_path):
"""用 ffmpeg 生成一段真实静音 mp3（会计入总时长，可被 ffprobe 验证）。"""
subprocess.run(
["ffmpeg", "-y", "-f", "lavfi", "-i", "anullsrc=r=24000:cl=mono",
"-t", f"{ms/1000:.3f}", "-q:a", "9", out_path],
check=True, capture_output=True,
)
def concat_mp3(part_paths, out_path):
"""用 ffmpeg concat demuxer 把多个 mp3 片段按顺序拼接（统一重编码，避免时基问题）。"""
with tempfile.NamedTemporaryFile("w", suffix=".txt", delete=False) as f:
for p in part_paths:
f.write(f"file '{os.path.abspath(p)}'\n")
list_path = f.name
try:
subprocess.run(
["ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", list_path,
"-ar", "24000", "-ac", "1", "-b:a", "64k", out_path],
check=True, capture_output=True,
)
finally:
os.unlink(list_path)
def synthesize_segments(segments, out_path, workdir):
"""
把 parse() 得到的片段列表合成为一个完整 mp3。
返回每个片段的合成信息列表（供打印验证）。
"""
if not segments:
# 解析结果为空（例如输入只有控制标记、没有任何语音正文）时直接给出
# 清晰报错，避免走 ffmpeg concat 空列表产生晦涩的 CalledProcessError。
raise SystemExit("错误：没有可合成的语音片段（输入可能只含控制标记，没有正文）。")
os.makedirs(workdir, exist_ok=True)
parts, info = [], []
for i, seg in enumerate(segments):
part_path = os.path.join(workdir, f"seg_{i:02d}.mp3")
if seg["type"] == "silence":
make_silence(seg["ms"], part_path)
info.append({"type": "silence", "ms": seg["ms"]})
else:
meta = synth_speech(seg["text"], seg["emotion"], seg["speed"],
seg["style"], seg.get("emphasis", False), part_path)
meta["type"] = "speech"
meta["text"] = seg["text"]
info.append(meta)
parts.append(part_path)
if len(parts) == 1:
os.replace(parts[0], out_path)
else:
concat_mp3(parts, out_path)
return info
voice_library.py¶
"""
参考语音库（Reference Voice Library）
=====================================
书中实验 9-5 使用 Fish Audio 的声音克隆：为同一个虚拟人准备 24 条不同
情绪 / 语速 / 风格的参考语音（情绪 4 × 语速 3 × 风格 2），执行层根据控制标记
选择最匹配的参考语音，Fish Audio 保证不同参考语音之间「音色一致」，
只是韵律和情感有所变化。
由于 Fish Audio 无可用 key，本项目用 OpenAI TTS 演示同一套思路：
- 「音色一致」 -> 全库固定使用同一个 base voice（如 alloy），保证音色不变；
- 「韵律/情感变化」-> 每条参考语音对应一段不同的 instructions（风格提示词），
gpt-4o-mini-tts 会据此改变语气、语速、情感。
因此这里的「一条参考语音」= 一个 (情绪 × 语速 × 风格) 组合对应的
(base_voice + instructions[+speed_factor]) 档案。整个库由维度笛卡尔积生成。
"""
# ---------------------------------------------------------------------------
# 三个控制维度的取值，以及每个取值对应的中文 instructions 片段
# ---------------------------------------------------------------------------
# 情绪维度：控制标记 [EMO:x] / [情感=x]
# 与书中实验 9-5 列出的四种情绪一致：中性 / 高兴 / 沮丧 / 思考
EMOTIONS = {
"neutral": "语气平稳、中性、不带明显情绪",
"happy": "语气欢快、上扬，带一点笑意",
"frustrated": "语气低沉、略显疲惫和无奈",
"thinking": "语气迟疑、若有所思，带轻微的犹豫和停顿感",
}
# 语速维度：控制标记 [SPEED:x] / [语速=x]
SPEEDS = {
"normal": ("语速正常", 1.0),
"fast": ("语速偏快、干脆利落", 1.25),
"slow": ("语速偏慢、从容", 0.80),
}
# 风格维度：控制标记 [STYLE:x] / [风格=x]
STYLES = {
"formal": "用正式、专业的客服口吻",
"casual": "用轻松、亲切、口语化的口吻",
}
# 全库统一的 base voice —— 用它来模拟 Fish Audio 的「音色一致」
BASE_VOICE = "alloy"
# 强调（[强调].../<emphasis>...）追加的 instructions 片段
EMPHASIS_HINT = "并对句中的关键信息加重语气、突出强调"
def build_instructions(emotion: str, speed: str, style: str, emphasis: bool = False) -> str:
"""把三个维度拼成一段 gpt-4o-mini-tts 的 instructions（风格提示词）。"""
emo = EMOTIONS.get(emotion, EMOTIONS["neutral"])
spd = SPEEDS.get(speed, SPEEDS["normal"])[0]
sty = STYLES.get(style, STYLES["formal"])
text = f"请以中文朗读。{sty}，{emo}，{spd}。"
if emphasis:
text += EMPHASIS_HINT + "。"
return text
def speed_factor(speed: str) -> float:
"""语速倍率，仅在 tts-1 兜底路径（不支持 instructions）时用作 speed 参数。"""
return SPEEDS.get(speed, SPEEDS["normal"])[1]
def profile_key(emotion: str, speed: str, style: str) -> str:
return f"{emotion}_{speed}_{style}"
def build_voice_library() -> dict:
"""生成完整的参考语音库：情绪 × 语速 × 风格 的笛卡尔积。"""
lib = {}
for emotion in EMOTIONS:
for speed in SPEEDS:
for style in STYLES:
key = profile_key(emotion, speed, style)
lib[key] = {
"emotion": emotion,
"speed": speed,
"style": style,
"base_voice": BASE_VOICE,
"instructions": build_instructions(emotion, speed, style),
"speed_factor": speed_factor(speed),
}
return lib
VOICE_LIBRARY = build_voice_library()
if __name__ == "__main__":
print(f"参考语音库共 {len(VOICE_LIBRARY)} 条（情绪 {len(EMOTIONS)} × 语速 "
f"{len(SPEEDS)} × 风格 {len(STYLES)}），全部固定 base voice = {BASE_VOICE}")
for k, v in list(VOICE_LIBRARY.items())[:6]:
print(f" {k:28s} -> {v['instructions']}")
print(" ...")