Python 调用 Sambert API：语音合成函数封装最佳实践

我们选择函数封装为主，支持扩展为类结构的设计思路。

2. 核心 API 接口分析

通过抓包分析 WebUI 请求，可得 Sambert 服务的 TTS 接口规范如下：

• URL: http://localhost:8080/tts
• Method: POST
• Content-Type: application/json
• Body 参数示例：

{
  "text": "今天天气真好",
  "voice": "zh-cn",
  "emotion": "happy",
  "speed": 1.0,
  "pitch": 1.0
}

• 返回结果：

{
  "status": "success",
  "audio_url": "/static/audio/xxx.wav"
}

注意：audio_url 是相对路径，需拼接完整地址下载音频。

3. 完整封装函数实现

import requests
import time
import os
from pathlib import Path
from typing import Literal, Optional

# -----------------------------
# 核心封装函数
# -----------------------------
def text_to_speech(
    text: str,
    output_path: str,
    server_url: str = "http://localhost:8080/tts",
    emotion: Literal["neutral", "happy", "sad", "angry", "surprised"] = "neutral",
    speed: float = 1.0,
    pitch: float = 1.0,
    voice: str = "zh-cn",
    timeout: int = 30,
    max_retries: int = 3,
    retry_delay: float = 1.0
) -> bool:
    """
    调用本地 Sambert-Hifigan 服务生成中文语音
    Args:
        text (str): 输入文本（建议≤500 字，超长自动分段）
        output_path (str): 输出 wav 文件路径
        server_url (str): TTS 服务 API 地址
        emotion (str): 情感类型，支持：neutral, happy, sad, angry, surprised
        speed (float): 语速比例，0.5~2.0
        pitch (float): 音调比例，0.5~2.0
        voice (str): 语音角色，固定为 zh-cn
        timeout (int): 请求超时时间（秒）
        max_retries (int): 最大重试次数
        retry_delay (float): 重试间隔（秒）
    Returns:
        bool: 成功返回 True，失败返回 False
    """
    # 参数合法性校验
    if not text.strip():
        print("❌ 错误：输入文本不能为空")
        return False
    if speed < 0.5 or speed > 2.0:
        print("⚠️ 警告：语速超出推荐范围 [0.5, 2.0]，已自动截断")
        speed = max(0.5, min(2.0, speed))
    if pitch < 0.5 or pitch > 2.0:
        print("⚠️ 警告：音调超出推荐范围 [0.5, 2.0]，已自动截断")
        pitch = max(0.5, min(2.0, pitch))

    # 构造请求数据
    payload = {
        "text": text.strip(),
        "voice": voice,
        "emotion": emotion,
        "speed": float(speed),
        "pitch": float(pitch)
    }
    headers = {"Content-Type": "application/json"}

    # 重试机制
    for attempt in range(max_retries):
        try:
            response = requests.post(
                server_url, json=payload, headers=headers, timeout=timeout
            )
            if response.status_code == 200:
                result = response.json()
                if result.get("status") == "success":
                    audio_url = result.get("audio_url")
                    if not audio_url:
                        print("❌ 响应缺少 audio_url 字段")
                        continue
                    # 拼接完整音频 URL
                    base_url = server_url.rsplit('/', 1)[0]
                    full_audio_url = f"{base_url}{audio_url}"
                    # 下载音频文件
                    return _download_audio(full_audio_url, output_path)
                else:
                    error_msg = result.get("message", "未知错误")
                    print(f"❌ 合成失败：{error_msg}")
            else:
                print(f"❌ HTTP {response.status_code}: {response.text}")
        except requests.exceptions.RequestException as e:
            print(f"🔁 第 {attempt + 1} 次请求失败：{e}")
            if attempt < max_retries - 1:
                time.sleep(retry_delay)
            else:
                print("❌ 所有重试均已失败")
                return False

# -----------------------------
# 辅助函数：下载音频
# -----------------------------
def _download_audio(audio_url: str, save_path: str) -> bool:
    """下载音频文件并保存"""
    try:
        response = requests.get(audio_url, timeout=15)
        if response.status_code == 200:
            Path(save_path).parent.mkdir(parents=True, exist_ok=True)
            with open(save_path, 'wb') as f:
                f.write(response.content)
            print(f"✅ 音频已保存至：{save_path}")
            return True
        else:
            print(f"❌ 下载失败，HTTP {response.status_code}")
            return False
    except Exception as e:
        print(f"❌ 下载异常：{e}")
        return False

4. 使用示例：一键生成带情感的语音

# 示例 1：基本调用
text_to_speech(
    text="欢迎使用 Sambert 语音合成服务，祝您工作愉快！",
    output_path="./output/greeting_happy.wav",
    emotion="happy",
    speed=1.1
)

# 示例 2：悲伤语境播报新闻
text_to_speech(
    text="昨日发生一起交通事故，造成三人受伤。",
    output_path="./output/news_sad.wav",
    emotion="sad",
    speed=0.9
)

输出日志：

✅ 音频已保存至：./output/greeting_happy.wav

进阶技巧：长文本分段合成与音频合并

当输入文本超过模型最大长度限制（约 500 汉字）时，需进行智能分句与音频拼接。

分段逻辑设计

import re
from pydub import AudioSegment

def split_chinese_text(text: str, max_len: int = 400) -> list:
    """按语义切分中文长文本"""
    sentences = re.split(r'[。！？；]', text)
    chunks = []
    current_chunk = ""
    for sent in sentences:
        sent = sent.strip()
        if not sent:
            continue
        if len(current_chunk + sent) <= max_len:
            current_chunk += sent + "。"
        else:
            if current_chunk:
                chunks.append(current_chunk)
            current_chunk = sent + "。"
    if current_chunk:
        chunks.append(current_chunk)
    return chunks

def long_text_to_speech(
    text: str,
    output_path: str,
    chunk_params: Optional[dict] = None
) -> bool:
    """
    支持长文本的语音合成（自动分段 + 拼接）
    需安装：pip install pydub
    """
    if chunk_params is None:
        chunk_params = {}
    chunks = split_chinese_text(text, max_len=400)
    temp_dir = Path("./temp_audio")
    temp_dir.mkdir(exist_ok=True)
    audio_segments = []
    for i, chunk in enumerate(chunks):
        temp_wav = temp_dir / f"part_{i:03d}.wav"
        success = text_to_speech(chunk, str(temp_wav), **chunk_params)
        if not success:
            print(f"❌ 第 {i+1} 段合成失败，终止处理")
            return False
        segment = AudioSegment.from_wav(str(temp_wav))
        audio_segments.append(segment)
    # 拼接所有音频
    final_audio = sum(audio_segments)
    final_audio.export(output_path, format="wav")
    print(f"✅ 长文本合成完成，总段数：{len(chunks)}，已保存至：{output_path}")
    # 清理临时文件（可选）
    # for p in temp_dir.glob("*.wav"): os.remove(p)
    return True

调用方式

long_text_to_speech(
    text="这是一段非常长的文字内容……（省略 500+ 字）",
    output_path="./output/long_story.wav",
    chunk_params={
        "emotion": "neutral",
        "speed": 1.0
    }
)

实践问题与优化建议

常见问题及解决方案

性能优化建议

1. 启用连接池：使用 requests.Session() 复用 TCP 连接
2. 异步调用：结合 asyncio + aiohttp 提升并发能力
3. 缓存机制：对重复文本生成结果做 MD5 哈希缓存
4. 本地代理层：在 Flask 服务前加 Nginx 反向代理，提升稳定性

总结：构建可落地的语音合成模块

本文围绕 Python 调用 Sambert API 展开，提出了一套完整的函数封装最佳实践方案：

• ✅ 稳定性优先：内置参数校验、异常捕获、自动重试机制
• ✅ 易用性强：接口简洁，支持情感、语速、音调调节
• ✅ 扩展性好：支持长文本分段合成与音频拼接
• ✅ 生产就绪：已在修复依赖冲突的稳定环境中验证通过

核心结论： 将 Sambert-Hifigan 服务封装为标准化函数模块，不仅能提升开发效率，更能保障线上系统的鲁棒性。建议将其作为企业级语音合成 SDK 的基础组件，进一步封装为微服务或集成进 RPA/AI Agent 系统中。

下一步学习建议

1. 学习 aiohttp 实现异步批量合成
2. 结合 gRPC 替代 HTTP 提升性能
3. 探索模型微调以适配特定声音风格
4. 集成 ASR 实现'语音对话闭环'

现在，你已经掌握了从零构建一个工业级中文语音合成调用模块的能力——是时候让它为你发声了。