LiveKit Agentsの独自TTS：音声エージェントに好きな声を

TTSプラグインのしくみ

プラグインはtts.TTSを継承し、synthesizeからChunkedStreamを返します。その_runメソッドがリクエストを送り、音声のバイト列をAudioEmitterに渡します。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 kHzにリサンプリングします。それ以外の応答ではAPIStatusErrorを発生させます。フレームワークは429とサーバーエラーを接続オプションに従って再試行し、キーの誤りなどではすぐにあきらめます。

セッションで使う

AgentSessionをつくるときに、すでに使っている音声認識と言語モデルのプラグインと一緒に、このクラスをttsとして渡します：

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はWebSocketではなく通常のHTTPSでストリーミング音声を返し、単語ごとのタイムスタンプは返さないため、整列済みの字幕はオフになります。31言語を話しますが、中国語にはまだ対応していません。プランは10分ごとの枠で、すべての通話を合わせた音声の秒数を決めます。無料キーは開発用に1日1分、月5ドルのスタートは枠ごとに最大3分、Proは20分で、にぎやかな通話を同時にいくつか処理できます。