Un TTS propio para LiveKit Agents: da a tu agente de voz cualquier voz

Cómo funciona un plugin TTS

Un plugin hereda de tts.TTS y devuelve un ChunkedStream desde synthesize. Su método _run envía la petición y pasa los bytes de audio a un AudioEmitter, que decodifica Ogg, mp3 o wav por sí mismo y los corta en tramas. Lo comprobamos con LiveKit Agents 1.8: si un TTS no recibe el texto en streaming, la sesión lo envuelve en un StreamAdapter que divide la respuesta del modelo en frases. Así cada frase es una petición corta y la primera suena mientras el modelo sigue escribiendo.

pip install livekit-agents aiohttp

El plugin

Guárdalo como superclonica_tts.py junto a tu agente:

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()

La petición pide "stream": true y la API responde con Ogg que llega mientras se genera. El emisor lo decodifica y lo remuestrea a los 44,1 kHz que declara la clase. Cualquier otra respuesta lanza APIStatusError: el framework reintenta los 429 y los errores del servidor según sus opciones de conexión y abandona enseguida ante errores como una clave incorrecta.

Úsalo en una sesión

Pasa la clase como tts al crear la AgentSession, junto con los plugins de reconocimiento de voz y de modelo de lenguaje que ya usas:

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"),
)

Define SUPERCLONICA_KEY con una clave de la página de la API de voz y VOICE_ID con el ID de la página de la voz elegida. Todo lo demás de tu agente sigue igual.

Límites que conviene saber

La API funciona con HTTPS normal y audio en streaming, no con WebSocket, y no devuelve marcas de tiempo por palabra, así que las transcripciones alineadas quedan desactivadas. Habla 31 idiomas, pero todavía no chino. Un plan da segundos de voz en cada ventana de 10 minutos para todas las llamadas juntas: la clave gratuita, un minuto al día para desarrollar; Inicio, 5 $ al mes, hasta 3 minutos por ventana; Pro, 20 minutos, suficiente para varias llamadas activas a la vez.