LiveKit Agents 自定义 TTS：让语音智能体使用任意声音

TTS 插件的工作原理

插件继承 tts.TTS，并在 synthesize 中返回一个 ChunkedStream。它的 _run 方法发送请求，并把音频字节交给 AudioEmitter；AudioEmitter 会自行解码 Ogg、mp3 或 wav 并切成音频帧。我们在 LiveKit Agents 1.8 上验证过：如果 TTS 不接收流式文本，会话会用 StreamAdapter 包装它，把模型的回答按句子拆分。每句话成为一次短请求，模型还在写后面的内容时，第一句已经开始播放。

pip install livekit-agents aiohttp

插件代码

把它保存为 superclonica_tts.py，放在智能体旁边：

import os
import aiohttp
from livekit.agents import DEFAULT_API_CONNECT_OPTIONS, APIConnectOptions, APIStatusError, tts, utils

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


class SuperClonicaTTS(tts.TTS):
    def __init__(self, voice: str, language: str = "en"):
        super().__init__(capabilities=tts.TTSCapabilities(streaming=False), sample_rate=44100, num_channels=1)
        self.voice, self.language = voice, language
        self._session = None

    def synthesize(self, text: str, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS):
        return SuperClonicaStream(tts=self, input_text=text, conn_options=conn_options)

    def session(self):
        if self._session is None:
            key = os.environ["SUPERCLONICA_KEY"]
            self._session = aiohttp.ClientSession(headers={"Authorization": "Bearer " + key})
        return self._session

    async def aclose(self):
        if self._session:
            await self._session.close()


class SuperClonicaStream(tts.ChunkedStream):
    async def _run(self, output_emitter: tts.AudioEmitter):
        engine = self._tts
        body = {"text": self.input_text, "voice": engine.voice, "language": engine.language, "stream": True}
        async with engine.session().post(URL, json=body) as answer:
            if answer.status != 200:
                raise APIStatusError(await answer.text(), status_code=answer.status)
            output_emitter.initialize(
                request_id=utils.shortuuid(), sample_rate=44100, num_channels=1, mime_type="audio/ogg"
            )
            async for chunk in answer.content.iter_chunked(4096):
                output_emitter.push(chunk)
            output_emitter.flush()

请求指定 "stream": true，API 在合成过程中就返回 Ogg 音频。发射器将其解码并重采样为类中声明的 44.1 kHz。其他任何响应都会抛出 APIStatusError：框架会按连接选项重试 429 和服务器错误，遇到密钥错误之类的问题则立即放弃。

在会话中使用

创建 AgentSession 时，把这个类作为 tts 传入，同时使用你现有的语音识别和语言模型插件：

import os
from livekit.agents import AgentSession
from superclonica_tts import SuperClonicaTTS

session = AgentSession(
    stt=...,  # your speech recognition plugin
    llm=...,  # your language model plugin
    tts=SuperClonicaTTS(voice=os.environ["VOICE_ID"], language="en"),
)

把 SUPERCLONICA_KEY 设为语音合成 API 页面上创建的密钥，把 VOICE_ID 设为所选声音页面上显示的 ID。智能体的其他部分保持不变。

需要了解的限制

API 使用普通 HTTPS 流式传输音频，而不是 WebSocket，也不返回逐词时间戳，因此对齐字幕是关闭的。它支持 31 种语言，但暂不支持中文。套餐规定所有通话合计在每个 10 分钟窗口内可用的语音秒数：免费密钥每天 1 分钟，供开发使用；每月 5 美元的入门套餐每个窗口最多 3 分钟；Pro 为 20 分钟，足以同时应付多个繁忙的通话。