Pipecat 사용자 TTS 서비스: 음성 봇에 원하는 목소리를

Pipecat TTS 서비스의 구조

TTS 서비스는 TTSService를 상속하고, 텍스트 조각을 받아 16비트 PCM 오디오 프레임을 내보내는 비동기 제너레이터 run_tts를 구현합니다. Pipecat 1.14에서 확인한 결과, 기본적으로 서비스가 모델의 텍스트를 문장 단위로 모으므로 run_tts는 문장마다 한 번 호출되고, 모델이 아직 쓰는 동안 첫 문장이 재생됩니다. 기본 클래스에는 WAV 헤더를 떼고 파이프라인 샘플레이트로 리샘플링하는 도우미도 있습니다.

pip install pipecat-ai aiohttp

서비스

이 파일을 봇 옆에 superclonica_pipecat.py로 저장하세요:

import os
import aiohttp
from pipecat.frames.frames import ErrorFrame
from pipecat.services.settings import TTSSettings
from pipecat.services.tts_service import TTSService

URL = "https://superclonica.com/api/v1/speech"


class SuperClonicaTTSService(TTSService):
    def __init__(self, *, session: aiohttp.ClientSession, voice: str, language: str = "en", **kwargs):
        super().__init__(
            push_start_frame=True,
            push_stop_frames=True,
            settings=TTSSettings(model=None, voice=voice, language=language),
            **kwargs,
        )
        self._session = session
        self._voice, self._language = voice, language

    def can_generate_metrics(self):
        return True

    async def run_tts(self, text: str, context_id: str):
        headers = {"Authorization": "Bearer " + os.environ["SUPERCLONICA_KEY"]}
        body = {"text": text, "voice": self._voice, "language": self._language, "format": "wav"}
        async with self._session.post(URL, json=body, headers=headers) as answer:
            if answer.status != 200:
                yield ErrorFrame(error=f"SuperClonica: {answer.status} {await answer.text()}")
                return
            await self.start_tts_usage_metrics(text)
            async for frame in self._stream_audio_frames_from_iterator(
                answer.content.iter_chunked(self.chunk_size), strip_wav_header=True, context_id=context_id
            ):
                await self.stop_ttfb_metrics()
                yield frame

서비스는 API에 WAV(표준 44바이트 헤더 뒤의 44.1 kHz 16비트 모노 PCM)를 요청하고 바이트를 도우미에 넘깁니다. 도우미는 헤더를 떼고 16 kHz나 24 kHz 같은 파이프라인 출력 레이트로 리샘플링해 프레임을 내보냅니다. 요금제 창을 다 쓴 경우의 429처럼 API가 오류를 돌려주면, 서비스는 상태와 메시지가 담긴 ErrorFrame을 내보냅니다.

봇에서 쓰기

aiohttp 세션과 고른 목소리 페이지에 표시된 ID로 서비스를 만들고, Pipeline에서 지금 쓰는 TTS 서비스 자리에 넣으세요:

import os
import aiohttp
from superclonica_pipecat import SuperClonicaTTSService

session = aiohttp.ClientSession()
tts = SuperClonicaTTSService(session=session, voice=os.environ["VOICE_ID"], language="en")
# put tts in your Pipeline where your current TTS service is

SUPERCLONICA_KEY에는 음성 합성 API 페이지에서 만든 키를 넣습니다. 봇의 전송, 음성 인식, 언어 모델은 그대로입니다.

지연과 요금제

짧은 문장은 합성이 끝나면 WAV가 한 번에 오고, 약 30초가 넘는 텍스트는 API가 스트리밍합니다. 저희 측정으로 말은 약 300 ms에 네트워크 시간을 더한 만큼 뒤에 시작됩니다. 요금제는 모든 통화를 합쳐 10분 창마다 쓸 수 있는 음성 초를 정합니다. 무료 키는 개발용으로 하루 1분, 월 5달러 스타트는 창마다 최대 3분, Pro는 20분입니다. Supertonic 3는 31개 언어를 말하지만 중국어는 아직 지원하지 않으며, API는 단어별 타임스탬프를 주지 않습니다.