指南

Pipecat 自定义 TTS 服务：让语音机器人使用任意声音

请注意：Supertonic 3 目前不支持中文，本文中的语音可朗读其支持的 31 种语言。Pipecat 是一个用于语音和多模态机器人的开源 Python 框架：由帧处理器组成的管道把来电者的音频依次送过语音识别、语言模型和语音合成，再送回去。它为许多语音服务商提供了服务类，新增一个只需写一个小类。

本指南为 SuperClonica 语音合成 API 编写这个类，让你的机器人能用目录中的 190 个声音说话（包括角色、旁白和播报员），也可以用你根据自己录音训练的声音。

免费获取 API 密钥 挑选声音 或自己运行引擎

Pipecat TTS 服务的工作原理

TTS 服务继承 TTSService 并实现 run_tts：这是一个异步生成器，接收一段文字，输出 16 位 PCM 音频帧。我们在 Pipecat 1.14 上验证过：默认情况下服务会把模型的文字按句子汇总，所以每句话调用一次 run_tts，模型还在写时第一句已经开始播放。基类还提供一个辅助方法，可以去掉 WAV 头并重采样到管道的采样率。

pip install pipecat-ai aiohttp

服务代码

把它保存为 superclonica_pipecat.py，放在机器人旁边：

import os
import aiohttp
from pipecat.frames.frames import ErrorFrame
from pipecat.services.settings import TTSSettings
from pipecat.services.tts_service import TTSService

URL = "https://superclonica.com/api/v1/speech"


class SuperClonicaTTSService(TTSService):
    def __init__(self, *, session: aiohttp.ClientSession, voice: str, language: str = "en", **kwargs):
        super().__init__(
            push_start_frame=True,
            push_stop_frames=True,
            settings=TTSSettings(model=None, voice=voice, language=language),
            **kwargs,
        )
        self._session = session
        self._voice, self._language = voice, language

    def can_generate_metrics(self):
        return True

    async def run_tts(self, text: str, context_id: str):
        headers = {"Authorization": "Bearer " + os.environ["SUPERCLONICA_KEY"]}
        body = {"text": text, "voice": self._voice, "language": self._language, "format": "wav"}
        async with self._session.post(URL, json=body, headers=headers) as answer:
            if answer.status != 200:
                yield ErrorFrame(error=f"SuperClonica: {answer.status} {await answer.text()}")
                return
            await self.start_tts_usage_metrics(text)
            async for frame in self._stream_audio_frames_from_iterator(
                answer.content.iter_chunked(self.chunk_size), strip_wav_header=True, context_id=context_id
            ):
                await self.stop_ttfb_metrics()
                yield frame

服务向 API 请求 WAV（标准 44 字节头之后是 44.1 kHz 的 16 位单声道 PCM），并把字节交给辅助方法。辅助方法去掉文件头，重采样到管道的输出采样率（如 16 或 24 kHz），然后输出音频帧。如果 API 返回错误，例如套餐窗口用完时的 429，服务会输出带有状态码和消息的 ErrorFrame。

在机器人中使用

用一个 aiohttp 会话和所选声音页面上的 ID 创建服务，并把它放进 Pipeline，替换你现在使用的 TTS 服务：

import os
import aiohttp
from superclonica_pipecat import SuperClonicaTTSService

session = aiohttp.ClientSession()
tts = SuperClonicaTTSService(session=session, voice=os.environ["VOICE_ID"], language="en")
# put tts in your Pipeline where your current TTS service is

把 SUPERCLONICA_KEY 设为语音合成 API 页面上创建的密钥。机器人的传输、语音识别和语言模型保持不变。

延迟与套餐

短句的 WAV 在合成完成后一次性返回；超过约半分钟语音的文本，API 会流式返回。按我们的测量，语音约 300 毫秒加网络时间即开始。套餐规定所有通话合计在每个 10 分钟窗口内的语音秒数：免费密钥每天 1 分钟，供开发使用；每月 5 美元的入门套餐每个窗口最多 3 分钟；Pro 为 20 分钟。Supertonic 3 支持 31 种语言，但暂不支持中文，API 也不返回逐词时间戳。