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분, 월 5달러 스타트는 창마다 최대 3분, Pro는 20분으로 붐비는 통화 여러 개를 동시에 처리할 수 있습니다.