ElevenLabs APIを使った多言語リアルタイム対話システム作ってみた

背景 / 目的

ちょっと思いつきで多言語のリアルタイムの対話ってどうやって技術的に実現できるだ？と思って作ったものを備忘録としてまとめる。

どうせOpenAI APIでもリアルタイム音声対話できるっしょ？と思ったが、目的の言語が対象外の可能性があった、中身がややブラックボックスで扱いにくかったため、少しハンドリングしやすいものを検証した背景がある。

結果として、ElevenLabs の Realtime STT + Streaming TTS, OpenAI API を Python で組み合わせ、対話型音声インターフェースの PoC を最小構成で形にする過程と得られた知見を共有する。

想定読者

• 音声インタラクション PoC を短期間で立ち上げたいエンジニア
• LLM + 音声ストリーミングの統合アーキテクチャを俯瞰したい人
• ElevenLabs / 他 STT・TTS サービス比較検討中の技術調査担当

全体アーキテクチャ概要

非同期タスク: mic 取得 / STT 受信 / 応答生成+TTS / 終了監視。

半二重: 応答中はユーザ発話を処理しない。

実装スクリプト

import os
import time
from typing import Any

import numpy as np
import sounddevice as sd
from dotenv import load_dotenv
from elevenlabs import stream as el_stream
from elevenlabs.client import ElevenLabs
from openai import OpenAI

load_dotenv()

OPENAI_API_KEY = os.environ["OPENAI_API_KEY"]
ELEVENLABS_API_KEY = os.environ["ELEVENLABS_API_KEY"]
VOICE_ID = os.getenv("ELEVENLABS_VOICE_ID", "21m00Tcm4TlvDq8ikWAM")
STT_MODEL_ID = os.getenv("ELEVENLABS_STT_MODEL", "scribe_v1")
STT_LANGUAGE_CODE = os.getenv("ELEVENLABS_STT_LANGUAGE_CODE")

# ==== オーディオ設定 ====
SR = 16000  # サンプリングレート
CH = 1  # モノラル
BLOCK_SEC = 0.5  # 録音ブロック長（秒）
RECORD_MAX_SEC = 15  # 1ターン最大録音長（秒）

# ==== クライアント ====
oai = OpenAI(api_key=OPENAI_API_KEY)
el = ElevenLabs(api_key=ELEVENLABS_API_KEY)


def record_until_silence(th_db=-38, min_sec=1.0, max_sec=RECORD_MAX_SEC):
    """
    スペース長押しなど“押してる間だけ録音”が最もシンプルですが、
    ここでは「喋っていない判定（無音）」で自動停止する簡易版。
    """
    print("🎙️ 話しかけてください（自動停止）。")
    buf = []
    start = time.time()

    def callback(indata, frames, time_info, status):
        buf.append(indata.copy())

    with sd.InputStream(samplerate=SR, channels=CH, dtype="float32", callback=callback):
        last_voice_ts = time.time()
        while True:
            sd.sleep(int(BLOCK_SEC * 1000))
            # 直近ブロックの音量で無音判定
            if buf:
                block = buf[-1]
                rms = np.sqrt(np.mean(block**2) + 1e-12)
                db = 20 * np.log10(rms + 1e-9)
                # 音があれば更新
                if db > th_db:
                    last_voice_ts = time.time()
            # 開始から一定時間未満は止めない
            if time.time() - start < min_sec:
                continue
            # 無音が1秒以上続いたら終了
            if time.time() - last_voice_ts > 1.0:
                break
            # マックス長
            if time.time() - start > max_sec:
                break

    audio = np.concatenate(buf, axis=0) if buf else np.zeros((0, CH), dtype=np.float32)
    # 16-bit PCM に変換
    pcm16 = (audio * 32767.0).astype(np.int16).tobytes()
    return pcm16


def stt_elevenlabs(pcm16_bytes):
    """
    ElevenLabs STTで文字起こし。raw PCMをそのまま送る。
    """
    import io

    if not pcm16_bytes:
        return ""

    payload = {
        "model_id": STT_MODEL_ID,
        "file": ("speech.pcm", io.BytesIO(pcm16_bytes), "application/octet-stream"),
        "file_format": "pcm_s16le_16",
    }
    if STT_LANGUAGE_CODE:
        payload["language_code"] = STT_LANGUAGE_CODE

    try:
        transcription = el.speech_to_text.convert(**payload)
    except Exception as exc:
        print("STT error:", exc)
        return ""

    text = _extract_transcript_text(transcription)
    if not text:
        print("（ElevenLabs STTの結果が空でした）")
    return text


def _extract_transcript_text(transcription: Any) -> str:
    if transcription is None:
        return ""

    text = getattr(transcription, "text", "") or ""
    if text.strip():
        return text.strip()

    transcripts = getattr(transcription, "transcripts", None)
    if transcripts:
        parts = [
            (getattr(chunk, "text", "") or "").strip()
            for chunk in transcripts
            if (getattr(chunk, "text", "") or "").strip()
        ]
        if parts:
            return " ".join(parts)

    message = getattr(transcription, "message", None)
    if message:
        print("STT info:", message)
    return ""


def chat_llm(
    user_text, system="あなたは有能な会話アシスタントです。簡潔に答えてください。"
):
    resp = oai.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": system},
            {"role": "user", "content": user_text},
        ],
        temperature=0.6,
    )
    return resp.choices[0].message.content.strip()


def tts_play_elevenlabs_stream(text, voice_id=VOICE_ID):
    """
    ElevenLabsのHTTPストリーミングで合成→即再生。
    SDKの `text_to_speech.stream` を使うとchunk単位で来るので、
    `elevenlabs.stream(audio_stream)` で再生できる。
    """
    try:
        audio_stream = el.text_to_speech.stream(
            voice_id=voice_id,
            optimize_streaming_latency=0,  # 0/1/2/3 小さいほど低レイテンシ
            model_id="eleven_multilingual_v2",  # プランに応じて
            text=text,
        )
        el_stream(audio_stream)
    except Exception as e:
        print("TTS stream error:", e)
        print(
            "※ 401の場合：APIキー/Voice ID/利用プラン/地域制限などを確認してください。"
        )


def main():
    print("=== リアルタイム会話(半二重) ===")
    print(
        "Enterを押すと録音開始。空白で話して、黙ると自動停止→返答音声が流れます。Ctrl+Cで終了。"
    )
    while True:
        try:
            input("\n[Enter] 録音開始 >> ")
            pcm16 = record_until_silence()
            if not pcm16:
                print("（音声なし）")
                continue

            user_text = stt_elevenlabs(pcm16)
            if not user_text:
                print("（認識結果なし）")
                continue
            print(f"🗣️ You: {user_text}")

            reply = chat_llm(user_text)
            print(f"🤖 Assistant: {reply}")

            tts_play_elevenlabs_stream(reply)

        except KeyboardInterrupt:
            print("\nbye!")
            break


if __name__ == "__main__":
    main()

実装概要

ループ構成（半二重）

• Enter 開始 → 録音（無音で自動停止） → STT → LLM → TTS 再生。
• 応答再生中は新規録音を受けない、シンプルな半二重フロー。

録音/停止のキモ

• 16kHz/モノラル、BLOCK_SEC=0.5s、最大 15s。
• 無音判定: RMS→dB、しきい値 -38dB。開始 1s は停止しない。

STT（ElevenLabs）

• raw PCM（s16le）をそのまま送信、model: scribe_v1。
• language_code 指定で精度/レイテンシを最適化。
• 結果が空ならスキップしてループ継続。

応答（LLM）

• gpt-4o-mini、temperature=0.6。
• システムプロンプトで「簡潔に」を明示し、短く自然な返答へ。

TTS（ElevenLabs）

• streaming API + optimize_streaming_latency=0 で先頭チャンク即再生。
• model: eleven_multilingual_v2、voice_id は環境変数で切替。
• 失敗時は要点ログ（鍵/プラン/地域制限）を出し継続。

UX とログ

• CLI に認識テキストと応答テキストを表示。
• エラーは握りつぶさず短く通知、処理は止めない。

よく触る調整パラメータ

• 録音: th_db, BLOCK_SEC, RECORD_MAX_SEC。
• STT: model_id, language_code。
• TTS: voice_id, model_id, optimize_streaming_latency。
• LLM: model, temperature。

まとめ

• これでOpenAI Reatime APIよりも多言語対応が柔軟にできるリアルタイム対話システムが構築できた。
• Vibe Codingありがてえ。
  • ただし、API動作検証 -> TTS -> STT -> ストリーミング処理 -> これらのオウム返し -> gpt-4o mini を使った対話という風に順を追って実装することで実現できた。
• 社内に英語、韓国語、日本語喋れるエンジニアがいたので技術検証が捗った。