教程系列:AI 创作入门到精通 · L1B 技术入门轨道 适用读者:有内容创作经验、想用代码解放双手的自媒体人;不需要你是程序员,但愿意复制粘贴并改两个参数。 前置基础:会用命令行运行
python xxx.py,知道 JSON 大概长什么样。 教程版本:v1.0 · 截至 2026 年 8 月 · 以市面通用 TTS API 为例 预计阅读:25-35 分钟 · 预计动手:2-3 小时
读完本篇并跟着做完,你将能够:
python --version 确认)API_KEY(或叫 Subscription Key、Bearer Token,各平台叫法不同)BASE_URL(例如 https://api.example-tts.com/v1)ffmpeg(brew install ffmpeg / apt install ffmpeg)用于音频格式转换和质检my_tts_project/
├── config.py # API 地址、密钥、默认音色
├── tts_client.py # 封装好的调用 + 重试逻辑
├── batch_generate.py # 批量生成主脚本
├── quality_check.py # 质检脚本
├── scripts/
│ ├── episode_01.json # 一期节目的文本清单
│ └── episode_02.json
├── audio/
│ └── (生成的音频文件)
├── cache/
│ └── (缓存索引)
└── voice_library/
└── (音色库目录)
照着建文件夹就行,后面每个文件我都会给你完整内容。
市面上的 TTS API(无论 Azure、阿里、讯飞、ElevenLabs、MiniMax 还是 IndexTTS)在调用模式上几乎只有三种。看懂这三种,你就能举一反三。
你发一个请求,服务器阻塞式地把音频生成完,直接在 HTTP 响应体里返回二进制音频。最简单,适合短文本(几百字以内)。
POST /v1/tts
{ "text": "你好世界", "voice": "narrator_v2" }
→ 200 OK, body = audio/wav 二进制
特点:实时,延迟低(几百毫秒到几秒),但单次文本长度有限(通常 1000 字以内),超长会被截断或报错。
你提交一个"任务",服务器立刻返回一个 job_id,然后你去干别的活;隔几秒轮询一次状态,成功了再下载音频。适合长文本和批量任务(几分钟到几十分钟的音频)。
POST /v1/tts/jobs → 返回 { "job_id": "abc123", "status": "running" }
GET /v1/tts/jobs/abc123 → 返回 { "status": "succeeded", "download_url": "..." }
GET download_url → 下载 zip/wav
这是 2026 年主流长音频方案。Azure 的 Batch Synthesis、阿里云的长文本合成、各家"异步 TTS"都属于这一类。延迟通常在 10-120 秒(Azure 官方数据:P50 约 10-20 秒,P95 约 120 秒)。
服务器边生成边推音频片段(SSE / WebSocket / chunked HTTP),你边收边播放。适合实时对话、直播字幕配音。本篇面向"批量生成节目"的场景,流式暂不展开,你只需要知道它的存在。
| 维度 | 同步 | 异步 | 流式 |
|---|---|---|---|
| 单次文本上限 | ~1000 字 | ~1万-10万字 | ~数百字/片 |
| 延迟 | 秒级 | 10-120 秒 | 毫秒级首帧 |
| 适合场景 | 短句、UI 提示音 | 整期节目、有声书 | 实时交互 |
| 实现复杂度 | 低 | 中(需轮询) | 高(需处理流) |
本教程重点:同步用于单段生成 + 质检重试,异步用于整批跑流水线。
把 API 调用逻辑封装到一个文件里,后续所有脚本都复用它。这是工程化的第一步:单一职责、统一入口。
tts_client.py —— 同步调用 + 自动重试 + 超时控制:
"""
通用 TTS 客户端封装(同步模式)
适用大多数市面 TTS API:输入文本,输出音频文件。
"""
import os
import time
import hashlib
import requests
from pathlib import Path
class TTSClient:
def __init__(self, base_url: str, api_key: str, voice: str = "default",
timeout: int = 60, max_retries: int = 3):
self.base_url = base_url.rstrip("/")
self.api_key = api_key
self.voice = voice
self.timeout = timeout
self.max_retries = max_retries
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
})
def synthesize(self, text: str, voice: str = None,
output_path: str = None) -> str:
"""
同步合成一段语音,返回保存的文件路径。
内置指数退避重试。
"""
voice = voice or self.voice
payload = {"text": text, "voice": voice, "format": "wav"}
# --- 指数退避重试 ---
for attempt in range(self.max_retries):
try:
resp = self.session.post(
f"{self.base_url}/tts",
json=payload,
timeout=self.timeout,
)
# 429 限流 / 5xx 服务端错误 → 重试
if resp.status_code == 429 or resp.status_code >= 500:
wait = 2 ** attempt # 1s, 2s, 4s
print(f" [重试 {attempt+1}/{self.max_retries}] "
f"HTTP {resp.status_code}, 等待 {wait}s")
time.sleep(wait)
continue
resp.raise_for_status()
# 音频在响应体(同步模式典型行为)
audio_bytes = resp.content
if not audio_bytes or len(audio_bytes) < 1000:
raise ValueError(f"音频内容异常:仅 {len(audio_bytes)} 字节")
# 决定输出路径
if not output_path:
# 用文本哈希命名,天然去重
h = hashlib.md5(text.encode()).hexdigest()[:12]
output_path = f"cache/{voice}_{h}.wav"
Path(output_path).parent.mkdir(parents=True, exist_ok=True)
with open(output_path, "wb") as f:
f.write(audio_bytes)
print(f" [OK] {output_path} ({len(audio_bytes)} bytes)")
return output_path
except (requests.ConnectionError, requests.Timeout) as e:
wait = 2 ** attempt
print(f" [网络错误 {attempt+1}] {e}, 等待 {wait}s")
time.sleep(wait)
raise RuntimeError(f"重试 {self.max_retries} 次后仍失败: text={text[:30]}...")
def synthesize_async(self, text: str, voice: str = None,
poll_interval: int = 5) -> str:
"""
异步合成:提交任务 → 轮询状态 → 下载音频。
适合长文本(>1000 字)。
"""
voice = voice or self.voice
payload = {"text": text, "voice": voice, "format": "wav"}
# 1. 提交
resp = self.session.post(f"{self.base_url}/tts/jobs",
json=payload, timeout=30)
resp.raise_for_status()
job = resp.json()
job_id = job["job_id"]
# 2. 轮询
while True:
time.sleep(poll_interval)
r = self.session.get(f"{self.base_url}/tts/jobs/{job_id}",
timeout=30)
r.raise_for_status()
status_data = r.json()
status = status_data["status"]
if status == "succeeded":
download_url = status_data["download_url"]
audio = self.session.get(download_url, timeout=120)
audio.raise_for_status()
h = hashlib.md5(text.encode()).hexdigest()[:12]
out = f"cache/{voice}_{h}.wav"
Path(out).parent.mkdir(parents=True, exist_ok=True)
with open(out, "wb") as f:
f.write(audio.content)
return out
elif status == "failed":
raise RuntimeError(f"异步任务失败: {status_data.get('error')}")
# else: running / pending, 继续轮询
这个封装做了三件关键的事:
config.py:
"""配置文件:改这里就行,别动其他文件。"""
import os
# === API 配置(改成你的) ===
API_KEY = os.environ.get("TTS_API_KEY", "你的密钥写这里")
BASE_URL = os.environ.get("TTS_BASE_URL", "https://api.example-tts.com/v1")
# === 默认音色 ===
DEFAULT_VOICE = "narrator_warm_v2" # 你的旁白音色 ID
# === 路径 ===
AUDIO_DIR = "audio"
CACHE_DIR = "cache"
VOICE_LIBRARY_DIR = "voice_library"
SCRIPTS_DIR = "scripts"
# === 生成参数 ===
SAMPLE_RATE = 24000 # 采样率,24kHz 是 2026 年主流
FORMAT = "wav" # 生成用 wav(无损),发布前再转 mp3/aac
MAX_RETRIES = 3
安全提示:别把密钥写死在
config.py里然后提交到 Git。用环境变量export TTS_API_KEY=xxx,或用.env文件 +python-dotenv。
这是核心生产力工具。你的节目文本放在 JSON 里,脚本一次性跑完。
scripts/episode_01.json —— 节目文本清单:
{
"episode_id": "ep01",
"title": "AI 时代的创作焦虑",
"segments": [
{"id": "intro", "text": "欢迎收听本期节目,今天我们聊一个绕不开的话题:AI 时代的创作焦虑。"},
{"id": "point_01", "text": "第一个观点是,焦虑的本质不是 AI 太强,而是你的不可替代性不够清晰。"},
{"id": "point_02", "text": "第二个观点是,工具会变,但选题直觉和对人性的理解不会过时。"},
{"id": "outro", "text": "以上就是本期全部内容,如果你有想法,欢迎在评论区留言。我们下期见。"}
]
}
batch_generate.py —— 批量生成主脚本:
"""
批量 TTS 生成:读取 JSON 脚本 → 逐段合成 → 质检 → 拼接(可选)
用法:python batch_generate.py scripts/episode_01.json
"""
import json
import sys
import time
from pathlib import Path
from config import *
from tts_client import TTSClient
def load_script(path: str) -> dict:
with open(path, "r", encoding="utf-8") as f:
return json.load(f)
def batch_generate(script_path: str, voice: str = None):
script = load_script(script_path)
ep_id = script["episode_id"]
voice = voice or DEFAULT_VOICE
out_dir = Path(AUDIO_DIR) / ep_id
out_dir.mkdir(parents=True, exist_ok=True)
client = TTSClient(BASE_URL, API_KEY, voice=voice, max_retries=MAX_RETRIES)
results = []
total = len(script["segments"])
print(f"=== 开始生成: {script['title']} ({total} 段) ===")
for i, seg in enumerate(script["segments"], 1):
print(f"\n[{i}/{total}] {seg['id']}: {seg['text'][:40]}...")
try:
audio_path = client.synthesize(
text=seg["text"],
voice=voice,
output_path=str(out_dir / f"{seg['id']}.wav")
)
results.append({
"id": seg["id"],
"text": seg["text"],
"audio": audio_path,
"status": "ok"
})
except Exception as e:
print(f" [失败] {e}")
results.append({
"id": seg["id"],
"text": seg["text"],
"audio": None,
"status": f"failed: {e}"
})
# 写入生成报告
report_path = out_dir / "generation_report.json"
with open(report_path, "w", encoding="utf-8") as f:
json.dump({
"episode": ep_id,
"voice": voice,
"total": total,
"succeeded": sum(1 for r in results if r["status"] == "ok"),
"failed": sum(1 for r in results if r["status"] != "ok"),
"segments": results,
}, f, ensure_ascii=False, indent=2)
print(f"\n=== 完成: {sum(1 for r in results if r['status']=='ok')}/{total} 成功 ===")
print(f"报告: {report_path}")
return results
if __name__ == "__main__":
if len(sys.argv) < 2:
print("用法: python batch_generate.py <script.json> [voice_id]")
sys.exit(1)
voice = sys.argv[2] if len(sys.argv) > 2 else None
batch_generate(sys.argv[1], voice)
运行:
export TTS_API_KEY="你的密钥"
python batch_generate.py scripts/episode_01.json
你会看到每段文本的生成进度、重试日志、最终报告。所有音频落在 audio/ep01/ 下,每段一个文件,方便后续拼接和替换。
生成"成功"不代表音频"能用"。常见问题:静音、截断(文本 200 字但音频只有 3 秒)、爆音(电平爆表)、采样率不对。
quality_check.py:
"""
质检脚本:扫描音频目录,检测:
1. 文件大小为 0 或过小
2. 时长异常(与文本字数估算不匹配)
3. 静音段占比过高
4. 爆音(峰值电平超过 -0.5dBFS)
依赖:ffmpeg(ffprobe)
用法:python quality_check.py audio/ep01/
"""
import json
import subprocess
import sys
from pathlib import Path
def ffprobe_duration(path: str) -> float:
"""用 ffprobe 获取时长(秒)"""
cmd = [
"ffprobe", "-v", "quiet", "-print_format", "json",
"-show_format", path
]
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
return -1
info = json.loads(result.stdout)
return float(info["format"]["duration"])
def ffprobe_volume_stats(path: str) -> dict:
"""检测音量统计:均值、峰值、静音比例"""
cmd = [
"ffmpeg", "-i", path, "-af",
"volumedetect", "-f", "null", "-"
]
result = subprocess.run(cmd, capture_output=True, text=True)
stderr = result.stderr
stats = {}
for line in stderr.split("\n"):
if "mean_volume" in line:
stats["mean_db"] = float(line.split(":")[1].strip().replace(" dB", ""))
if "max_volume" in line:
stats["max_db"] = float(line.split(":")[1].strip().replace(" dB", ""))
return stats
def check_segment(seg: dict, audio_dir: Path) -> dict:
issues = []
audio_path = audio_dir / f"{seg['id']}.wav"
if not audio_path.exists():
return {"id": seg["id"], "status": "missing", "issues": ["文件不存在"]}
size = audio_path.stat().st_size
if size < 1000:
issues.append(f"文件过小: {size} bytes")
duration = ffprobe_duration(str(audio_path))
if duration <= 0:
issues.append("无法读取时长")
else:
# 粗略估算:中文约 4-5 字/秒
expected_min = len(seg["text"]) / 6.0 # 偏快
expected_max = len(seg["text"]) / 3.0 # 偏慢
if duration < expected_min * 0.5:
issues.append(f"疑似截断: 时长 {duration:.1f}s, "
f"预期 {expected_min:.1f}-{expected_max:.1f}s")
if duration > expected_max * 2:
issues.append(f"时长过长: {duration:.1f}s, 可能有大量静音")
vol = ffprobe_volume_stats(str(audio_path))
if vol.get("max_db", -99) > -0.5:
issues.append(f"爆音风险: 峰值 {vol.get('max_db')} dB")
if vol.get("mean_db", -99) < -40:
issues.append(f"音量过低: 均值 {vol.get('mean_db')} dB")
return {
"id": seg["id"],
"status": "pass" if not issues else "warn",
"issues": issues,
"duration": duration,
"size": size,
"volume": vol,
}
def run_qc(report_path: str):
report = json.loads(Path(report_path).read_text(encoding="utf-8"))
ep_id = report["episode"]
audio_dir = Path("audio") / ep_id
segments = report["segments"]
print(f"=== 质检: {ep_id} ({len(segments)} 段) ===\n")
passed, warned, failed = 0, 0, 0
for seg in segments:
if seg["status"] != "ok":
print(f" [SKIP] {seg['id']}: 生成阶段已失败")
failed += 1
continue
result = check_segment(seg, audio_dir)
icon = {"pass": "✓", "warn": "!", "missing": "X"}.get(result["status"], "?")
print(f" [{icon}] {result['id']}: {result['status']}")
if result["issues"]:
for issue in result["issues"]:
print(f" → {issue}")
warned += 1
else:
passed += 1
print(f"\n=== 质检结果: {passed} 通过, {warned} 警告, {failed} 失败 ===")
return warned + failed # 返回需要处理的数量
if __name__ == "__main__":
if len(sys.argv) < 2:
print("用法: python quality_check.py audio/ep01/generation_report.json")
sys.exit(1)
problem_count = run_qc(sys.argv[1])
if problem_count > 0:
print(f"\n有 {problem_count} 个片段需要检查/重生成。")
sys.exit(1)
质检发现问题后,不要手动一条条重跑。写一个"修复"脚本,只针对有问题的片段重新生成:
repair_failed.py:
"""
根据质检报告,自动重新生成有问题的片段。
用法:python repair_failed.py audio/ep01/generation_report.json
"""
import json
import sys
from pathlib import Path
from config import *
from tts_client import TTSClient
def repair(report_path: str, voice: str = None):
report = json.loads(Path(report_path).read_text(encoding="utf-8"))
ep_id = report["episode"]
voice = voice or report.get("voice", DEFAULT_VOICE)
audio_dir = Path("audio") / ep_id
client = TTSClient(BASE_URL, API_KEY, voice=voice, max_retries=MAX_RETRIES)
# 找出所有失败/缺失的片段
to_repair = [s for s in report["segments"] if s["status"] != "ok"]
print(f"需要修复 {len(to_repair)} 段")
for seg in to_repair:
print(f"\n重试: {seg['id']}")
try:
audio_path = client.synthesize(
text=seg["text"],
voice=voice,
output_path=str(audio_dir / f"{seg['id']}.wav")
)
seg["status"] = "ok"
seg["audio"] = audio_path
except Exception as e:
seg["status"] = f"still_failed: {e}"
print(f" 仍然失败: {e}")
# 更新报告
with open(report_path, "w", encoding="utf-8") as f:
json.dump(report, f, ensure_ascii=False, indent=2)
if __name__ == "__main__":
repair(sys.argv[1])
这是从"能用"到"工程化"的分水岭。你的参考音频(voice sample)、音色参数、提示词,都应该有版本号,像管理代码一样管理。
voice_library/ 目录结构:
voice_library/
├── narrator_warm/
│ ├── v1/
│ │ ├── sample.wav # 参考音频
│ │ ├── params.json # 生成参数
│ │ └── README.md # 这个版本的说明
│ ├── v2/
│ │ ├── sample.wav
│ │ ├── params.json
│ │ └── README.md
│ └── current -> v2 # 软链接指向当前版本
├── character_lao_wang/
│ ├── v1/
│ └── current -> v1
└── index.json # 音色索引
voice_library/narrator_warm/v2/params.json:
{
"voice_id": "narrator_warm_v2",
"source_sample": "sample.wav",
"sample_duration_sec": 15.2,
"language": "zh-CN",
"tts_params": {
"speed": 1.0,
"pitch": 0,
"energy": 1.0,
"style": "warm"
},
"prompt": "用温暖沉稳的语调朗读,语速适中,适合知识类节目旁白",
"created": "2026-07-15",
"notes": "相比 v1:更换了更清晰的参考音频,去掉了背景噪声",
"tested_on": ["ep01", "ep02"],
"parent_version": "v1"
}
voice_library/index.json —— 全局索引:
{
"voices": {
"narrator_warm": {
"current": "v2",
"versions": {
"v1": {"created": "2026-06-01", "deprecated": true},
"v2": {"created": "2026-07-15", "deprecated": false}
}
},
"character_lao_wang": {
"current": "v1",
"versions": {
"v1": {"created": "2026-07-20", "deprecated": false}
}
}
}
}
版本管理的核心原则:
v3 目录,current 软链接更新指向。这样万一新版翻车,一秒切回旧版。params.json 里的 notes 字段写清楚"这次改了什么、为什么改",未来你不会记得。index.json 和 params.json。参考音频(wav)太大不放 Git,但用 Git LFS 或单独的对象存储管理。关键是元数据有版本历史。一个辅助脚本 switch_voice.py,快速切换当前版本:
"""切换音色当前版本:python switch_voice.py narrator_warm v1"""
import json
import os
import sys
from pathlib import Path
def switch(voice_name: str, version: str):
lib = Path("voice_library")
idx_path = lib / "index.json"
idx = json.loads(idx_path.read_text(encoding="utf-8"))
if voice_name not in idx["voices"]:
print(f"音色不存在: {voice_name}")
sys.exit(1)
versions = idx["voices"][voice_name]["versions"]
if version not in versions:
print(f"版本不存在: {version}")
print(f"可用版本: {list(versions.keys())}")
sys.exit(1)
idx["voices"][voice_name]["current"] = version
idx_path.write_text(json.dumps(idx, ensure_ascii=False, indent=2),
encoding="utf-8")
# 更新软链接(macOS/Linux)
link_path = lib / voice_name / "current"
if link_path.is_symlink() or link_path.exists():
link_path.unlink()
link_path.symlink_to(version)
print(f"已切换 {voice_name} → {version}")
if __name__ == "__main__":
if len(sys.argv) != 3:
print("用法: python switch_voice.py <voice_name> <version>")
sys.exit(1)
switch(sys.argv[1], sys.argv[2])
TTS API 按字符数或时长计费,不经控制的话账单会吓到你。两个核心策略:
策略一:文本哈希缓存(已在 tts_client.py 中实现)
同一段文本 + 同一个音色,生成的哈希值相同,文件名相同。第二次调用时直接命中缓存,不发 API 请求,不花钱。
# tts_client.py 的缓存逻辑(回顾)
h = hashlib.md5(text.encode()).hexdigest()[:12]
output_path = f"cache/{voice}_{h}.wav"
if Path(output_path).exists():
print(f" [缓存命中] {output_path}")
return output_path
在 synthesize() 方法开头加一段缓存检查就能生效:
def synthesize(self, text: str, voice: str = None, output_path: str = None):
voice = voice or self.voice
# 缓存检查
if not output_path:
h = hashlib.md5(text.encode()).hexdigest()[:12]
output_path = f"cache/{voice}_{h}.wav"
if Path(output_path).exists() and Path(output_path).stat().st_size > 1000:
print(f" [缓存命中] {output_path}")
return output_path
# ... 后续正常调用 ...
策略二:成本估算与预算告警
在批量生成前,先算一下这期节目要花多少钱:
"""成本估算:在批量生成前预估费用"""
def estimate_cost(script_path: str, price_per_1k_chars: float = 0.02):
script = json.loads(Path(script_path).read_text(encoding="utf-8"))
total_chars = sum(len(s["text"]) for s in script["segments"])
cost = total_chars / 1000 * price_per_1k_chars
print(f"总字符数: {total_chars}")
print(f"预估费用: ${cost:.2f} (按 ${price_per_1k_chars}/千字)")
return cost
把它加到 batch_generate.py 开头,跑之前先看一眼:
=== 成本预估 ===
总字符数: 1247
预估费用: $0.02 (按 $0.02/千字)
=== 开始生成: AI 时代的创作焦虑 (4 段) ===
策略三:长文本用异步 API(单价通常更低)
很多平台的异步/批量 API 比 同步 API 便宜 20-50%(因为服务器可以离线排队处理,成本更低)。如果你的内容不要求实时,优先走异步。
场景:一期 15 分钟的知识播客,4-6 个段落,全部用同一个旁白音色。
步骤:
scripts/ep03.json,5 个段落。python batch_generate.py scripts/ep03.json → 生成 5 个 wav 到 audio/ep03/。python quality_check.py audio/ep03/generation_report.json → 质检。python repair_failed.py audio/ep03/generation_report.json。ffmpeg -f concat -i filelist.txt -c copy ep03_full.wav(filelist.txt 里按顺序列出每段文件)。避坑:段落之间加 0.5 秒静音更自然。拼接时插入一段 silence_0.5.wav(用 ffmpeg -f lavfi -i anullsrc=r=24000:cl=mono -t 0.5 silence_0.5.wav 生成)。
场景:两个人物对话,旁白 + 角色 A + 角色 B 交替出现。
scripts/dialog_01.json:
{
"episode_id": "dialog01",
"title": "产品经理 vs 程序员",
"segments": [
{"id": "narr_01", "voice": "narrator_warm_v2", "text": "这天,产品经理走进了程序员的工作区。"},
{"id": "pm_01", "voice": "voice_pm_v1", "text": "嘿,我有个想法,加个按钮,就一个小功能。"},
{"id": "dev_01", "voice": "voice_dev_v1", "text": "你说的小功能,又要改三个接口加两张表。"},
{"id": "pm_02", "voice": "voice_pm_v1", "text": "那不是你的强项吗?"},
{"id": "dev_02", "voice": "voice_dev_v1", "text": "我的强项是拒绝这种需求。"},
{"id": "narr_02", "voice": "narrator_warm_v2", "text": "就这样,讨论持续了一个下午。"}
]
}
修改 batch_generate.py 的调用,让每段使用脚本里指定的 voice:
# 在 batch_generate 的循环里改为:
seg_voice = seg.get("voice", voice) # 优先用段内指定的 voice
audio_path = client.synthesize(
text=seg["text"],
voice=seg_voice,
output_path=str(out_dir / f"{seg['id']}.wav")
)
这样一段对话脚本里可以混合多个音色,生成后按 id 顺序拼接即可。
场景:一本 10 万字的有声书,按章节拆分,每章 3000-8000 字,用异步 API 批量生成。
"""批量提交异步任务,等待全部完成,再统一下载。"""
import json
from pathlib import Path
from config import *
from tts_client import TTSClient
def batch_async_generate(chapters: list, voice: str):
client = TTSClient(BASE_URL, API_KEY, voice=voice)
jobs = []
for ch in chapters:
print(f"提交: 第{ch['chapter']}章 ({len(ch['text'])} 字)")
try:
audio_path = client.synthesize_async(ch["text"], voice=voice)
jobs.append({"chapter": ch["chapter"], "audio": audio_path,
"status": "ok"})
except Exception as e:
jobs.append({"chapter": ch["chapter"], "audio": None,
"status": f"failed: {e}"})
return jobs
注意:异步 API 有并发限制(通常每账户 10-100 个并发任务)。如果 100 章一次性提交,可能触发 429。加一个简单的并发控制:
import time
# 每提交 10 章等待 5 秒
if i > 0 and i % 10 == 0:
print("等待 5 秒,避免触发限流...")
time.sleep(5)
把以下模板保存下来,换掉 API 地址和密钥就能用。
"""极简版:读 CSV → 逐行生成 → 存文件"""
import csv, requests, os
API_KEY = "你的密钥"
URL = "https://api.example-tts.com/v1/tts"
VOICE = "narrator_v1"
with open("scripts.csv", encoding="utf-8") as f:
for row in csv.DictReader(f):
resp = requests.post(URL,
headers={"Authorization": f"Bearer {API_KEY}"},
json={"text": row["text"], "voice": VOICE},
timeout=60)
with open(f"audio/{row['id']}.wav", "wb") as out:
out.write(resp.content)
print(f"完成: {row['id']}")
就是前面 tts_client.py + batch_generate.py 的组合。这是推荐的生产配置。
[ ] 1. generation_report.json 中 failed 数量为 0
[ ] 2. 质检脚本无 "warn" 级别问题
[ ] 3. 抽听首段和末段,确认音色一致
[ ] 4. 检查总时长是否与文本量匹配(粗算:字数 ÷ 4 ≈ 秒数)
[ ] 5. 用耳机听一遍拼接后的完整音频,确认段落衔接自然
[ ] 6. 确认音量均值在 -20 ~ -15 dB 之间(发布标准)
把这张清单贴在你显示器旁边。批量生成最大的风险不是"生成失败"——那个重试就行——而是"生成成功但质量不对",比如音色漂移、语速突变,这些只有人耳抽检才能发现。
Q1:我应该选同步还是异步 API?
短文本(一句话到一段话)用同步,实时拿结果。超过 1000 字或一次要生成很多段,用异步。不确定就用异步——它更稳,不会超时。
Q2:不同平台的 API 格式不一样怎么办?
本文的 TTSClient 是抽象层。换平台时只需要改 synthesize() 和 synthesize_async() 里的请求/响应解析逻辑,上层脚本(batch_generate.py、quality_check.py)不用动。这就是封装的价值。
Q3:缓存会不会导致我用旧音频?
会,如果你改了文本哪怕一个字,MD5 哈希就不同,不会命中缓存。但如果你改了音色参数(比如语速)但文本没变,哈希相同会命中旧缓存。解决办法:把参数也纳入哈希:
hash_input = f"{text}|{voice}|{speed}|{pitch}"
h = hashlib.md5(hash_input.encode()).hexdigest()[:12]
Q4:音色漂移(同一音色不同段落听起来不一样)怎么办?
这是 2026 年 TTS 仍然存在的痛点。对策:
Q5:API 突然涨价或停服怎么办?
这正是音色库版本管理 + 缓存的价值。你已经生成的音频在本地,不依赖 API 在线。如果换平台,参考音频在 voice_library/ 里,可以迁移到新平台重新创建音色。永远保留本地参考音频和生成结果的副本。
Q6:需要并发加速吗?
初学阶段不需要。串行逐段生成虽然慢,但简单、不容易限流、出问题好排查。等你一期节目超过 50 段、串行要跑半小时以上时,再考虑 concurrent.futures.ThreadPoolExecutor 控制在 3-5 并发。别一上来就开 20 并发——大概率触发 429。
TTS 引擎对标点敏感。连续的 。。。 或中英混排 AI时代 可能导致异常停顿。生成前做文本清洗:
import re
def clean_text(text: str) -> str:
text = re.sub(r"[。]{2,}", "。", text) # 多个句号合并
text = re.sub(r"[…]{2,}", "……", text) # 省略号统一
text = re.sub(r"([a-zA-Z])([一-龥])", r"\1 \2", text) # 英文中文间加空格
text = re.sub(r"([一-龥])([a-zA-Z])", r"\1 \2", text)
return text.strip()
不同段如果采样率不同(有的 16kHz,有的 24kHz),拼接后会出现变速或噪声。在质检里加采样率检查:
def ffprobe_sample_rate(path: str) -> int:
cmd = ["ffprobe", "-v", "quiet", "-select_streams", "a:0",
"-show_entries", "stream=sample_rate", "-of", "csv=p=0", path]
return int(subprocess.run(cmd, capture_output=True, text=True).stdout.strip())
统一用 ffmpeg -i input.wav -ar 24000 -y output.wav 重采样到一致。
大部分 TTS API 支持 SSML(Speech Synthesis Markup Language),可以精确控制停顿、重音、语速:
<speak version="1.0" xml:lang="zh-CN">
<voice name="narrator_warm_v2">
欢迎收听<break time="500ms"/>本期节目。
<prosody rate="slow" pitch="-2st">今天聊一个严肃话题。</prosody>
</voice>
</speak>
如果你的 API 支持 SSML,强烈建议用它代替纯文本。控制力强得多,尤其是需要停顿和语速变化的场景。
这是新手最常见的翻车点。用环境变量:
export TTS_API_KEY="sk-xxxxx"
python batch_generate.py scripts/ep01.json
或者在项目根目录放 .env(记得加到 .gitignore):
TTS_API_KEY=sk-xxxxx
TTS_BASE_URL=https://api.example-tts.com/v1
在生产脚本里加结构化日志,记录每次调用的文本、音色、耗时、状态:
import logging
logging.basicConfig(
filename="tts_generation.log",
level=logging.INFO,
format='{"time": "%(asctime)s", "level": "%(levelname)s", "msg": "%(message)s"}'
)
# 每段生成时:
logging.info(f"segment={seg_id}, voice={voice}, chars={len(text)}, "
f"duration={gen_time}s, status=ok")
出了问题(比如某期节目音色不对),翻日志就能找到原因。
本篇你掌握了:
下一级(L1B-08)我们将进入"图片生成工程化":同样的思路应用到 AI 绘图 API——批量出图、风格一致性管理、Prompt 版本控制、质检与重试。你会发现工程化的骨架是完全一样的:封装 → 批量 → 质检 → 重试 → 缓存 → 版本管理。换一个领域,同一套方法论。
如果你跳着看:从 L1B-08 开始也没问题,但建议先跑通本篇的 batch_generate.py 至少一次——亲手跑一次批量生成,你对"工程化"的体感会完全不同。
免责声明:本文 API 示例为通用教学用途,具体参数格式(请求体字段名、认证方式、响应结构)请以你实际使用的 TTS 平台官方文档为准。各平台 2026 年 8 月的 API 版本可能已有更新。
参考来源: