TTS 与语音合成应用

文字转语音（TTS）技术让 AI 应用从纯文字互动跨越到自然语音输出，广泛应用于有声书、语音助手、无障碍服务和内容创作。

TTS 技术演进

生产级 TTS Pipeline

import hashlib
import json
import time
from dataclasses import dataclass, field
from enum import Enum
from pathlib import Path
from typing import Optional
class TTSProvider(Enum):
OPENAI = "openai"          # OpenAI TTS (tts-1 / tts-1-hd)
ELEVENLABS = "elevenlabs"  # ElevenLabs
AZURE_TTS = "azure"        # Azure Cognitive Speech
COQUI = "coqui"            # 开源 XTTS (本地)
EDGE_TTS = "edge_tts"      # Microsoft Edge TTS (免费)
class AudioFormat(Enum):
MP3 = "mp3"
WAV = "wav"
OPUS = "opus"
AAC = "aac"
@dataclass
class TTSRequest:
"""TTS 合成请求"""
text: str
voice: str               # 音色 ID（如 "nova", "zh-CN-XiaoxiaoNeural"）
provider: TTSProvider = TTSProvider.OPENAI
speed: float = 1.0       # 0.25–4.0
format: AudioFormat = AudioFormat.MP3
language: str = "zh"
def __post_init__(self):
self.text = self.text.strip()
if len(self.text) > 4096:
raise ValueError("TTS 文本超过 4096 字符限制")
if not 0.25 <= self.speed <= 4.0:
raise ValueError("速度参数须在 0.25–4.0 之间")
@property
def cache_key(self) -> str:
content = f"{self.text}{self.voice}{self.provider.value}{self.speed}"
return hashlib.md5(content.encode()).hexdigest()[:16]
def split_chunks(self, max_chars: int = 500) -> list[str]:
"""将长文本拆分为合适的段落（按句断句）"""
import re
sentences = re.split(r'(?<=[。！？.!?])', self.text)
chunks = []
current = ""
for sent in sentences:
if len(current) + len(sent) <= max_chars:
current += sent
else:
if current:
chunks.append(current.strip())
current = sent
if current:
chunks.append(current.strip())
return [c for c in chunks if c]
@dataclass
class TTSResult:
"""TTS 合成结果"""
request_id: str
audio_path: Path
duration_sec: float
file_size_bytes: int
generation_time_ms: int
cost_usd: float = 0.0
cached: bool = False
class TTSPipeline:
"""生产级 TTS Pipeline"""
COST_PER_1K_CHARS = {
TTSProvider.OPENAI: {"tts-1": 0.015, "tts-1-hd": 0.030},
TTSProvider.ELEVENLABS: {"default": 0.018},
TTSProvider.AZURE_TTS: {"default": 0.016},
TTSProvider.EDGE_TTS: {"default": 0.000},  # 免费
}
def __init__(self, cache_dir: Path | None = None):
self.cache_dir = cache_dir or Path("./tts_cache")
self.cache_dir.mkdir(parents=True, exist_ok=True)
self._total_chars = 0
self._total_cost = 0.0
def synthesize(self, request: TTSRequest) -> TTSResult:
"""合成语音"""
# 检查缓存
cached_path = self.cache_dir / f"{request.cache_key}.{request.format.value}"
if cached_path.exists():
return TTSResult(
request_id=request.cache_key,
audio_path=cached_path,
duration_sec=len(request.text) / 5.0,  # 估算
file_size_bytes=cached_path.stat().st_size,
generation_time_ms=0,
cached=True,
)
start_ms = int(time.time() * 1000)
if request.provider == TTSProvider.OPENAI:
audio_bytes = self._call_openai_tts(request)
elif request.provider == TTSProvider.EDGE_TTS:
audio_bytes = self._call_edge_tts(request)
else:
audio_bytes = b""  # 其他 Provider 示意
elapsed = int(time.time() * 1000) - start_ms
# 保存文件
cached_path.write_bytes(audio_bytes)
cost = len(request.text) / 1000 * self.COST_PER_1K_CHARS.get(
request.provider, {}
).get("default", 0.015)
self._total_chars += len(request.text)
self._total_cost += cost
return TTSResult(
request_id=request.cache_key,
audio_path=cached_path,
duration_sec=len(request.text) / 5.0,  # 约 5 字/秒（中文）
file_size_bytes=len(audio_bytes),
generation_time_ms=elapsed,
cost_usd=cost,
)
def _call_openai_tts(self, request: TTSRequest) -> bytes:
"""调用 OpenAI TTS API"""
from openai import OpenAI
client = OpenAI()
response = client.audio.speech.create(
model="tts-1",
voice=request.voice,  # alloy, echo, fable, onyx, nova, shimmer
input=request.text,
speed=request.speed,
response_format=request.format.value,
)
return response.content
def _call_edge_tts(self, request: TTSRequest) -> bytes:
"""调用 Edge TTS（免费，需 pip install edge-tts）"""
import asyncio
import edge_tts
async def _synthesize():
communicate = edge_tts.Communicate(
text=request.text,
voice=request.voice,  # zh-CN-XiaoxiaoNeural
rate=f"+{int((request.speed - 1.0) * 100)}%",
)
chunks = []
async for chunk in communicate.stream():
if chunk["type"] == "audio":
chunks.append(chunk["data"])
return b"".join(chunks)
return asyncio.run(_synthesize())
def synthesize_long_text(self, text: str, voice: str, **kwargs) -> list[TTSResult]:
"""分段合成长文本"""
request = TTSRequest(text=text, voice=voice, **kwargs)
chunks = request.split_chunks(max_chars=500)
results = []
for i, chunk in enumerate(chunks):
chunk_request = TTSRequest(text=chunk, voice=voice, **kwargs)
result = self.synthesize(chunk_request)
results.append(result)
print(f"[TTS] 合成段落 {i+1}/{len(chunks)}: {len(chunk)} 字")
return results
# 使用示例
pipeline = TTSPipeline()
# 短文本合成（中文）
req = TTSRequest(
text="欢迎使用智能语音助手，今天有什么可以帮您的吗？",
voice="zh-CN-XiaoxiaoNeural",
provider=TTSProvider.EDGE_TTS,
speed=1.1,
)
print(f"文本长度: {len(req.text)} 字")
print(f"缓存键: {req.cache_key}")
chunks = req.split_chunks(max_chars=20)
print(f"分段结果: {chunks}")

TTS 提供商对比

本章小结

• Edge TTS 免费且质量不错——开发阶段和非商业项目的首选
• 长文本必须分段合成——每段控制在 500 字以内，防止截断
• 缓存是降本的关键——相同文本无需二次调用 API
• 中文优先用 Azure 或 ElevenLabs——OpenAI TTS 中文质感稍差
• 声音克隆需要版权许可——商业使用必须确认声源授权

下一章：多模态RAG与检索