РУКОВОДСТВО

Свой TTS-сервис для Pipecat: дайте голосовому боту любой голос

Pipecat — open source фреймворк на Python для голосовых и мультимодальных ботов: конвейер обработчиков кадров ведёт звук собеседника через распознавание речи, языковую модель и озвучку текста обратно к нему. В нём есть сервисы для многих провайдеров речи, а новый требует одного небольшого класса.

В этом руководстве мы напишем такой класс для API озвучки SuperClonica, чтобы бот говорил любым из 190 голосов каталога, включая персонажей, дикторов и анонсеров, или голосом, обученным по вашей записи.

Получить API-ключ бесплатно Выбрать голос Или запустить движок у себя

Как устроен TTS-сервис в Pipecat

TTS-сервис наследует TTSService и реализует run_tts — асинхронный генератор, который получает кусок текста и выдаёт аудиокадры 16-битного PCM. Мы проверили это на 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 — 16-битный моно PCM 44,1 кГц за стандартным 44-байтным заголовком — и передаёт байты помощнику. Тот отбрасывает заголовок, пересэмплирует звук под выходную частоту конвейера, например 16 или 24 кГц, и выдаёт кадры. Если API отвечает ошибкой, например 429, когда окно тарифа исчерпано, сервис выдаёт 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 приходит целиком, когда озвучен; тексты длиннее примерно полуминуты речи API стримит. По нашим замерам речь начинается примерно через 300 мс плюс время сети. Тариф даёт секунды речи в каждом 10-минутном окне на все звонки вместе: бесплатный ключ — минуту в день для разработки, «Старт» за $5 в месяц — до 3 минут на окно, Pro — 20 минут. Supertonic 3 говорит на 31 языке, но пока не по-китайски, а API не возвращает тайминги слов.