Um TTS próprio no LiveKit Agents: dê qualquer voz ao seu agente

Como funciona um plugin TTS

Um plugin herda de tts.TTS e devolve um ChunkedStream em synthesize. O método _run dele envia a requisição e passa os bytes de áudio a um AudioEmitter, que decodifica Ogg, mp3 ou wav sozinho e corta em quadros. Conferimos isso no LiveKit Agents 1.8: quando um TTS não recebe texto em streaming, a sessão o envolve num StreamAdapter que divide a resposta do modelo em frases. Cada frase vira uma requisição curta, e a primeira toca enquanto o modelo ainda escreve.

pip install livekit-agents aiohttp

O plugin

Salve como superclonica_tts.py ao lado do seu 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()

A requisição pede "stream": true, e a API responde com Ogg que chega enquanto a voz é gerada. O emissor decodifica e reamostra para os 44,1 kHz que a classe declara. Qualquer outra resposta gera APIStatusError: o framework repete 429 e erros do servidor conforme as opções de conexão e desiste na hora em erros como chave errada.

Use numa sessão

Passe a classe como tts ao criar a AgentSession, junto com os plugins de reconhecimento de fala e de modelo de linguagem que você já usa:

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

Defina SUPERCLONICA_KEY com uma chave da página da API de voz e VOICE_ID com o ID da página da voz escolhida. O resto do agente continua igual.

Limites para saber

A API usa HTTPS comum com áudio em streaming, não WebSocket, e não devolve marcações de tempo por palavra, então as transcrições alinhadas ficam desligadas. Ela fala 31 idiomas, mas ainda não chinês. Um plano dá segundos de fala em cada janela de 10 minutos para todas as chamadas juntas: a chave gratuita, um minuto por dia para desenvolver; o Inicial, US$ 5 por mês, até 3 minutos por janela; o Pro, 20 minutos, o bastante para várias chamadas movimentadas ao mesmo tempo.