Ein eigenes TTS für LiveKit Agents: Geben Sie Ihrem Agenten jede Stimme

So funktioniert ein TTS-Plugin

Ein Plugin erbt von tts.TTS und gibt aus synthesize einen ChunkedStream zurück. Dessen Methode _run sendet die Anfrage und übergibt die Audiobytes an einen AudioEmitter, der Ogg, mp3 oder wav selbst dekodiert und in Frames schneidet. Wir haben das mit LiveKit Agents 1.8 geprüft: Nimmt ein TTS keinen Text als Stream an, packt die Sitzung es in einen StreamAdapter, der die Antwort des Modells in Sätze teilt. So wird jeder Satz eine kurze Anfrage, und der erste spielt, während das Modell noch schreibt.

pip install livekit-agents aiohttp

Das Plugin

Speichern Sie dies als superclonica_tts.py neben Ihrem Agenten:

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

Die Anfrage verlangt "stream": true, und die API antwortet mit Ogg, das schon während der Vertonung eintrifft. Der Emitter dekodiert es und rechnet es auf die von der Klasse angegebenen 44,1 kHz um. Jede andere Antwort löst APIStatusError aus: Das Framework wiederholt 429 und Serverfehler gemäß seinen Verbindungsoptionen und gibt bei Fehlern wie einem falschen Schlüssel sofort auf.

In einer Sitzung verwenden

Übergeben Sie die Klasse beim Erstellen der AgentSession als tts, zusammen mit den Plugins für Spracherkennung und Sprachmodell, die Sie schon nutzen:

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

Setzen Sie SUPERCLONICA_KEY auf einen Schlüssel von der Seite der Sprach-API und VOICE_ID auf die ID von der Seite der gewählten Stimme. Alles andere am Agenten bleibt gleich.

Grenzen, die Sie kennen sollten

Die API spricht einfaches HTTPS mit gestreamtem Audio, kein WebSocket, und liefert keine Wort-Zeitstempel, daher sind ausgerichtete Transkripte aus. Sie spricht 31 Sprachen, aber noch kein Chinesisch. Ein Tarif gibt Sekunden Sprache in jedem 10-Minuten-Fenster für alle Anrufe zusammen: der kostenlose Schlüssel eine Minute pro Tag für die Entwicklung, Start für 5 $ im Monat bis zu 3 Minuten pro Fenster, Pro 20 Minuten, genug für mehrere lebhafte Anrufe gleichzeitig.