Ein eigener TTS-Dienst für Pipecat: Geben Sie Ihrem Sprachbot jede Stimme

So funktioniert ein Pipecat-TTS-Dienst

Ein TTS-Dienst erbt von TTSService und implementiert run_tts, einen asynchronen Generator, der ein Stück Text erhält und Audio-Frames aus 16-Bit-PCM liefert. Wir haben das mit Pipecat 1.14 geprüft: Standardmäßig sammelt der Dienst den Text des Modells zu Sätzen, run_tts wird also einmal pro Satz aufgerufen, und der erste spielt, während das Modell noch schreibt. Die Basisklasse hat außerdem eine Hilfsfunktion, die einen WAV-Header entfernt und auf die Rate der Pipeline umrechnet.

pip install pipecat-ai aiohttp

Der Dienst

Speichern Sie dies als superclonica_pipecat.py neben Ihrem 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

Der Dienst fordert von der API WAV an, 16-Bit-Mono-PCM mit 44,1 kHz hinter einem normalen 44-Byte-Header, und übergibt die Bytes der Hilfsfunktion. Sie entfernt den Header, rechnet auf die Ausgaberate Ihrer Pipeline um, etwa 16 oder 24 kHz, und liefert die Frames. Antwortet die API mit einem Fehler, etwa 429, wenn ein Tariffenster aufgebraucht ist, liefert der Dienst einen ErrorFrame mit Status und Meldung.

Im Bot verwenden

Erstellen Sie den Dienst mit einer aiohttp-Sitzung und der ID von der Seite der gewählten Stimme und setzen Sie ihn in Ihrer Pipeline an die Stelle des bisherigen TTS-Dienstes:

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

Setzen Sie SUPERCLONICA_KEY auf einen Schlüssel von der Seite der Sprach-API. Transport, Spracherkennung und Sprachmodell Ihres Bots bleiben gleich.

Latenz und Tarife

Bei einem kurzen Satz kommt das WAV als Ganzes, sobald es vertont ist; Texte über etwa eine halbe Minute Sprache streamt die API. Nach unseren Messungen beginnt die Sprache ab etwa 300 ms plus Netzwerkzeit. Ein Tarif gibt Sekunden Sprache in jedem 10-Minuten-Fenster für alle Anrufe zusammen: der kostenlose Schlüssel eine Minute pro Tag für die Entwicklung, Start für 5 $ im Monat bis zu 3 Minuten pro Fenster, Pro 20 Minuten. Supertonic 3 spricht 31 Sprachen, aber noch kein Chinesisch, und die API liefert keine Wort-Zeitstempel.