Pipecatの独自TTSサービス：音声ボットに好きな声を

PipecatのTTSサービスのしくみ

TTSサービスはTTSServiceを継承し、run_tts（テキストの断片を受け取り、16ビットPCMの音声フレームを返す非同期ジェネレーター）を実装します。Pipecat 1.14で確認したところ、既定ではモデルのテキストを文ごとにまとめるので、run_ttsは1文に1回呼ばれ、モデルが書いている間に最初の文が再生されます。基底クラスには、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日1分、月5ドルのスタートは枠ごとに最大3分、Proは20分です。Supertonic 3は31言語を話しますが中国語にはまだ対応しておらず、APIは単語ごとのタイムスタンプを返しません。