Свой TTS для LiveKit Agents: дайте голосовому агенту любой голос

Как устроен TTS-плагин

Плагин наследует tts.TTS и возвращает из synthesize объект ChunkedStream. Его метод _run отправляет запрос и передаёт байты звука в 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 кГц, заявленные классом. Любой другой ответ вызывает APIStatusError: фреймворк повторяет 429 и ошибки сервера по своим настройкам соединения и сразу сдаётся на ошибках вроде неверного ключа.

Подключение к сессии

Передайте класс как tts при создании AgentSession вместе с плагинами распознавания речи и языковой модели, которые вы уже используете:

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-минутном окне на все звонки вместе: бесплатный ключ — минуту в день для разработки, «Старт» за $5 в месяц — до 3 минут на окно, Pro — 20 минут, этого хватает на несколько оживлённых звонков одновременно.