Un TTS sur mesure pour LiveKit Agents : donnez n’importe quelle voix à votre agent

Comment fonctionne un plugin TTS

Un plugin hérite de tts.TTS et renvoie un ChunkedStream depuis synthesize. Sa méthode _run envoie la requête et transmet les octets audio à un AudioEmitter, qui décode lui-même Ogg, mp3 ou wav et les découpe en trames. Nous l’avons vérifié avec LiveKit Agents 1.8 : quand un TTS ne reçoit pas le texte en flux, la session l’enveloppe dans un StreamAdapter qui découpe la réponse du modèle en phrases. Chaque phrase devient une courte requête, et la première se joue pendant que le modèle écrit encore.

pip install livekit-agents aiohttp

Le plugin

Enregistrez ceci sous superclonica_tts.py à côté de votre agent :

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 requête demande "stream": true, et l’API répond par de l’Ogg qui arrive pendant la synthèse. L’émetteur le décode et le rééchantillonne aux 44,1 kHz déclarés par la classe. Toute autre réponse lève APIStatusError : le framework réessaie les 429 et les erreurs serveur selon ses options de connexion et abandonne tout de suite sur des erreurs comme une mauvaise clé.

L’utiliser dans une session

Passez la classe comme tts en créant l’AgentSession, avec les plugins de reconnaissance vocale et de modèle de langage que vous utilisez déjà :

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

Définissez SUPERCLONICA_KEY avec une clé de la page de l’API vocale et VOICE_ID avec l’ID affiché sur la page de la voix choisie. Le reste de votre agent ne change pas.

Les limites à connaître

L’API fonctionne en HTTPS classique avec audio en flux, pas en WebSocket, et ne renvoie pas d’horodatage par mot : les transcriptions alignées sont donc désactivées. Elle parle 31 langues, mais pas encore le chinois. Une offre donne des secondes de parole par fenêtre de 10 minutes pour tous les appels ensemble : la clé gratuite une minute par jour pour développer, Débutant à 5 $ par mois jusqu’à 3 minutes par fenêtre, Pro 20 minutes, de quoi tenir plusieurs appels animés à la fois.