Um serviço TTS próprio no Pipecat: dê qualquer voz ao seu bot

Como funciona um serviço TTS do Pipecat

Um serviço TTS herda de TTSService e implementa run_tts, um gerador assíncrono que recebe um pedaço de texto e produz quadros de áudio PCM de 16 bits. Conferimos isso no Pipecat 1.14: por padrão o serviço junta o texto do modelo em frases, então run_tts é chamado uma vez por frase, e a primeira toca enquanto o modelo ainda escreve. A classe base também tem uma função auxiliar que tira o cabeçalho WAV e reamostra para a taxa do pipeline.

pip install pipecat-ai aiohttp

O serviço

Salve como superclonica_pipecat.py ao lado do seu bot:

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

O serviço pede WAV à API, PCM mono de 16 bits a 44,1 kHz atrás de um cabeçalho padrão de 44 bytes, e passa os bytes à função auxiliar. Ela tira o cabeçalho, reamostra para a taxa de saída do seu pipeline, como 16 ou 24 kHz, e produz os quadros. Se a API responder com erro, por exemplo 429 quando a janela do plano acaba, o serviço produz um ErrorFrame com o status e a mensagem.

Use no seu bot

Crie o serviço com uma sessão aiohttp e o ID da página da voz escolhida e coloque-o no seu Pipeline no lugar do serviço TTS atual:

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

Defina SUPERCLONICA_KEY com uma chave da página da API de voz. Transporte, reconhecimento de fala e modelo de linguagem do bot continuam iguais.

Latência e planos

Para uma frase curta, o WAV chega inteiro assim que é gerado; textos com mais de meio minuto de fala a API transmite em streaming. Pelas nossas medições, a fala começa a partir de uns 300 ms mais o tempo de rede. Um plano dá segundos de fala em cada janela de 10 minutos para todas as chamadas juntas: a chave gratuita, um minuto por dia para desenvolver; o Inicial, US$ 5 por mês, até 3 minutos por janela; o Pro, 20 minutos. O Supertonic 3 fala 31 idiomas, mas ainda não chinês, e a API não devolve marcações de tempo por palavra.