Un service TTS sur mesure pour Pipecat : donnez n’importe quelle voix à votre bot

Comment fonctionne un service TTS Pipecat

Un service TTS hérite de TTSService et implémente run_tts, un générateur asynchrone qui reçoit un morceau de texte et produit des trames audio en PCM 16 bits. Nous l’avons vérifié avec Pipecat 1.14 : par défaut, le service regroupe le texte du modèle en phrases, donc run_tts est appelé une fois par phrase et la première se joue pendant que le modèle écrit encore. La classe de base fournit aussi une aide qui retire l’en-tête WAV et rééchantillonne au débit du pipeline.

pip install pipecat-ai aiohttp

Le service

Enregistrez ceci sous superclonica_pipecat.py à côté de votre 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

Le service demande à l’API du WAV, du PCM mono 16 bits à 44,1 kHz derrière un en-tête standard de 44 octets, et passe les octets à l’aide. Celle-ci retire l’en-tête, rééchantillonne au débit de sortie de votre pipeline, par exemple 16 ou 24 kHz, et produit les trames. Si l’API répond par une erreur, par exemple 429 quand la fenêtre de l’offre est épuisée, le service produit un ErrorFrame avec le statut et le message.

L’utiliser dans votre bot

Créez le service avec une session aiohttp et l’ID affiché sur la page de la voix choisie, puis placez-le dans votre Pipeline à la place de votre service TTS actuel :

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

Définissez SUPERCLONICA_KEY avec une clé de la page de l’API vocale. Le transport, la reconnaissance vocale et le modèle de langage de votre bot ne changent pas.

Latence et offres

Pour une phrase courte, le WAV arrive d’un bloc une fois lu ; au-delà d’environ une demi-minute de parole, l’API le diffuse en flux. Selon nos mesures, la parole démarre dès environ 300 ms, plus le temps réseau. Une offre donne des secondes de parole par fenêtre de 10 minutes pour tous les appels ensemble : la clé gratuite une minute par jour pour développer, Débutant à 5 $ par mois jusqu’à 3 minutes par fenêtre, Pro 20 minutes. Supertonic 3 parle 31 langues, mais pas encore le chinois, et l’API ne renvoie pas d’horodatage par mot.