Vapiの独自音声：30行でつくる自前のTTSサーバー

Vapiが送るものと期待するもの

Vapiの独自TTSのガイドによると、VapiはJSONのPOSTを送り、そのmessageオブジェクトにはtype voice-request、テキスト、希望するsampleRateが入っています。応答はヘッダーのない生のPCMで、モノラル、16ビット符号付き、リトルエンディアン、そのサンプルレートちょうど、Content-Typeはapplication/octet-streamである必要があります。インラインのsecretを設定すると、VapiはそれをX-VAPI-SECRETヘッダーで送ります。

サーバー

パッケージをインストールし、これをvapi_voice.pyとして保存します：

pip install fastapi uvicorn scipy superclonica-cpu

import math, os
import numpy as np
from fastapi import FastAPI, Header, HTTPException, Request, Response
from fastapi.concurrency import run_in_threadpool
from scipy.signal import resample_poly
from superclonica import Superclonica

VOICE = os.environ.get("TTS_VOICE", "F1")  # M1..M5, F1..F5 or a voice file from the catalog
LANGUAGE = os.environ.get("TTS_LANGUAGE", "en")
SECRET = os.environ["VAPI_SECRET"]
app = FastAPI()
tts = Superclonica()

def voice_audio(text):
    audio, rate = tts.synthesize(text, voice=VOICE, lang=LANGUAGE)
    return np.asarray(audio, dtype=np.float32).reshape(-1), rate

@app.post("/synthesize")
async def synthesize(request: Request, x_vapi_secret: str = Header(None)):
    if x_vapi_secret != SECRET:
        raise HTTPException(401)
    message = (await request.json())["message"]
    audio, rate = await run_in_threadpool(voice_audio, message["text"])
    target = int(message["sampleRate"])
    if target != rate:
        divisor = math.gcd(target, rate)
        audio = resample_poly(audio, target // divisor, rate // divisor)
    pcm = (np.clip(audio, -1, 1) * 32767).astype(np.int16).tobytes()
    return Response(pcm, media_type="application/octet-stream")

エンジンはワーカースレッドで44.1 kHzで読み上げ、scipyがVapiの求めるレートにリサンプリングし、サーバーは16ビットPCMを返します。正しいsecretのないリクエストには401を返します。uvicornで起動します：

uvicorn vapi_voice:app --host 0.0.0.0 --port 8000

アシスタントにつなぐ

Vapiが公開HTTPSでサーバーに届く必要があります。いつものリバースプロキシの後ろに置くか、テストならトンネルを使ってください。次に、たとえばVapi APIでアシスタントを更新するときに、アシスタントの声を設定します：

{
  "voice": {
    "provider": "custom-voice",
    "server": {
      "url": "https://your-server.example.com/synthesize",
      "secret": "the value of VAPI_SECRET",
      "timeoutSeconds": 30
    }
  }
}

サーバーのVAPI_SECRETにも同じ値を設定します。Vapiのガイドには、ダッシュボードで作成する認証情報や予備の声も説明されています。サーバーが落ちても通話が止まらないよう、予備の声を追加しておくとよいでしょう。

APIを使う同じサーバー

サーバーが小さくてモデルを置けないなら、SuperClonicaの音声合成APIで読み上げます。voice_audioをこの関数に置き換え、SUPERCLONICA_KEYとVOICE_IDを設定してください。リサンプリングと応答はそのままです。

import io, wave
import requests

API = "https://superclonica.com/api/v1/speech"
VOICE_ID = os.environ["VOICE_ID"]  # the ID on the page of a catalog voice

def voice_audio(text):
    answer = requests.post(API, timeout=60,
                           headers={"Authorization": "Bearer " + os.environ["SUPERCLONICA_KEY"]},
                           json={"text": text, "voice": VOICE_ID, "language": LANGUAGE, "format": "wav"})
    answer.raise_for_status()
    with wave.open(io.BytesIO(answer.content)) as file:
        rate = file.getframerate()
        pcm = np.frombuffer(file.readframes(file.getnframes()), dtype=np.int16)
    return pcm.astype(np.float32) / 32768, rate

カタログのどの声も、どのプランでもページに表示されるIDで使えます。スタートは月5ドルで、すべての通話を合わせて10分ごとの枠で最大3分、Proは20分です。

速度と声

サーバーは各リクエストを最後まで読み上げてから応答します。私たちの計測では、短い文の中央値は4スレッドのデスクトップRyzenで0.43秒、RTX 5070 Tiで0.12秒でした。アシスタントの文は短くしましょう。TTS_VOICEには内蔵の10の声（M1〜M5、F1〜F5）かカタログの声のファイルを指定します。Supertonic 3は日本語（TTS_LANGUAGE=ja）を含む31言語を話しますが、中国語にはまだ対応していません。