LiveKit連携 | ElevenLabs Documentation

このガイドでは、ElevenLabs Speech EngineをLiveKitルームの音声レイヤーとして使用する方法を紹介します。LiveKit Agentsワーカーは参加者としてルームに参加し、ユーザーのオーディオトラックを購読してSpeech EngineへのWebSocketを開き、Speech Engineが合成したオーディオを独自のトラックとしてルームに公開します。

アーキテクチャ

Speech Engineは2種類のWebSocket接続を受け付けます。

ElevenLabs APIが接続するブレインWebSocket。サーバーでSpeech Engine SDK（engine.serve()／engine.attach()）を実行し、応答するための文字起こしを受け取ります。
クライアントが接続する会話WebSocket。ブラウザはWebRTCトークン経由で接続します。LiveKit Agentsワーカーなどの非ブラウザクライアントは、署名付きURL経由で接続し、未加工のPCMオーディオを双方向にストリーミングします。

LiveKitワーカーは2つ目の接続を使用します。LiveKitルーム内の参加者に代わって、Speech Engineの「クライアント」として動作します。

ブレインサーバーはSpeech Engineクイックスタートから変更不要です。LiveKitワーカーがブラウザの代わりにオーディオソースとなりますが、LLMロジックはそのままです。

このパターンを使う場面

ルーム自体が体験の一部である場合は、LiveKitブリッジを使用します。

複数のユーザーが同時にエージェントと会話するマルチ参加者セッション
トランスポートを切り替えるとクライアントが機能しなくなる、既存のLiveKitデプロイメント
画面共有、ビデオ、テキストチャットとルームを共有する音声エージェント
通話中にAIエージェントが必要な、SIPからLiveKitへディスパッチされる通話

他の参加者なしでブラウザからSpeech Engineへの音声ループだけが必要な場合は、Speech EngineクイックスタートのWebRTCクライアントのほうがシンプルです。Speech Engineがブラウザと直接WebRTCで通信するため、LiveKitルームは不要です。

前提条件

LiveKitプロジェクト（LiveKit Cloudまたはセルフホストサーバー）。ワーカーにはLIVEKIT_URL、LIVEKIT_API_KEY、LIVEKIT_API_SECRETが必要です。
ElevenLabs Speech Engine。Speech Engineクイックスタートに従って作成し、ブレインサーバーを実行してください。
Python 3.9+またはNode.js 18+。

Nodeブリッジワーカーは @livekit/rtc-nodeを使用します。これは現在 Developer Previewです。本番環境へのデプロイには、Pythonワーカーを推奨します。

Speech Engineのオーディオ形式を設定する

LiveKitのAudioStreamは、受信するOpusトラックを指定したPCMサンプルレートにリサンプリングします。そのため、Speech Engineの入力に直接合わせられます。Speech Engineを更新し、ASR入力には16 kHz PCMを受け入れ、TTS出力には24 kHz PCMを出力するようにします。

import asyncio
import os
from elevenlabs import AsyncElevenLabs
elevenlabs = AsyncElevenLabs(api_key=os.environ["ELEVENLABS_API_KEY"])
async def update_engine():
    await elevenlabs.speech_engine.update(
        speech_engine_id="seng_8k3m9xr4hjnfg983brhmhkd98n6",
        asr={"user_input_audio_format": "pcm_16000"},
        tts={"agent_output_audio_format": "pcm_24000"},
    )
asyncio.run(update_engine())

Speech EngineのPCMは、全体を通して符号付き16ビット・リトルエンディアンです。その他の対応レートについては、オーディオ形式リファレンスを参照してください。

ブリッジワーカーを構築する

ワーカーは、LiveKitサーバーに接続してジョブを待機し、割り当てられたルームに参加して、ルームとSpeech Engineの間でオーディオをブリッジする長時間実行プロセスです。

依存関係をインストールする

pip install "livekit-agents" "livekit-api" "elevenlabs" "aiohttp" "python-dotenv"

Speech Engineの署名付きURLを生成する

ワーカーはSpeech Engine会話WebSocket用の短期間有効な署名付きURLをリクエストします。署名付きURLにはエンジンIDと1回限りの署名が含まれるため、APIキーを公開せずにWebSocketを開けます。

from elevenlabs import AsyncElevenLabs
elevenlabs = AsyncElevenLabs(api_key=os.environ["ELEVENLABS_API_KEY"])
async def signed_url() -> str:
    response = await elevenlabs.conversational_ai.conversations.get_signed_url(
        agent_id=os.environ["SPEECH_ENGINE_ID"],
    )
    return response.signed_url

ワーカーのエントリーポイントを定義する

ワーカーがルームにディスパッチされるたびに、そのエントリーポイントが実行されます。エントリーポイントはルームに接続し、Speech Engine会話WebSocketを開き、2つのオーディオブリッジを開始します。1つはSpeech Engineに送る発信者オーディオ用、もう1つは戻ってくる合成オーディオ用です。

import asyncio
import base64
import json
import os
import aiohttp
from dotenv import load_dotenv
from elevenlabs import AsyncElevenLabs
from livekit import agents, rtc
from livekit.agents import JobContext, WorkerOptions, cli
load_dotenv()
elevenlabs = AsyncElevenLabs(api_key=os.environ["ELEVENLABS_API_KEY"])
SPEECH_ENGINE_ID = os.environ["SPEECH_ENGINE_ID"]
USER_INPUT_RATE = 16000
AGENT_OUTPUT_RATE = 24000
async def signed_url() -> str:
    response = await elevenlabs.conversational_ai.conversations.get_signed_url(
        agent_id=SPEECH_ENGINE_ID,
    )
    return response.signed_url
async def entrypoint(ctx: JobContext):
    el_ws_ready: asyncio.Future[aiohttp.ClientWebSocketResponse] = (
        asyncio.get_running_loop().create_future()
    )
    async def pump_user_audio(track: rtc.Track):
        el_ws = await el_ws_ready
        stream = rtc.AudioStream(
            track, sample_rate=USER_INPUT_RATE, num_channels=1,
        )
        async for event in stream:
            payload = base64.b64encode(bytes(event.frame.data)).decode()
            await el_ws.send_str(json.dumps({"user_audio_chunk": payload}))
    # Register the subscriber BEFORE ctx.connect() so we don't miss tracks
    # that get auto-subscribed during the connection handshake.
    @ctx.room.on("track_subscribed")
    def on_track_subscribed(track, publication, participant):
        if track.kind != rtc.TrackKind.KIND_AUDIO:
            return
        if participant.identity == ctx.room.local_participant.identity:
            return
        asyncio.create_task(pump_user_audio(track))
    await ctx.connect()
    # Publish a track for the agent's synthesized audio.
    source = rtc.AudioSource(sample_rate=AGENT_OUTPUT_RATE, num_channels=1)
    track = rtc.LocalAudioTrack.create_audio_track("elevenlabs-agent", source)
    await ctx.room.local_participant.publish_track(
        track,
        rtc.TrackPublishOptions(source=rtc.TrackSource.SOURCE_MICROPHONE),
    )
    # Open the Speech Engine conversation WebSocket.
    http = aiohttp.ClientSession()
    el_ws = await http.ws_connect(await signed_url())
    await el_ws.send_str(json.dumps({"type": "conversation_initiation_client_data"}))
    el_ws_ready.set_result(el_ws)
    async def el_to_room():
        async for msg in el_ws:
            if msg.type != aiohttp.WSMsgType.TEXT:
                continue
            event = json.loads(msg.data)
            etype = event.get("type")
            if etype == "audio":
                pcm = base64.b64decode(event["audio_event"]["audio_base_64"])
                samples_per_channel = len(pcm) // 2
                frame = rtc.AudioFrame(
                    pcm, AGENT_OUTPUT_RATE, 1, samples_per_channel,
                )
                await source.capture_frame(frame)
            elif etype == "interruption":
                source.clear_queue()
            elif etype == "ping":
                event_id = event.get("ping_event", {}).get("event_id")
                await el_ws.send_str(json.dumps({
                    "type": "pong", "event_id": event_id,
                }))
    pump_task = asyncio.create_task(el_to_room())
    async def cleanup():
        pump_task.cancel()
        await el_ws.close()
        await http.close()
    ctx.add_shutdown_callback(cleanup)
if __name__ == "__main__":
    cli.run_app(WorkerOptions(
        entrypoint_fnc=entrypoint,
        agent_name="elevenlabs-bridge",
    ))

ワーカーは、track_subscribedハンドラーでローカル参加者のIDと比較することで、自身が公開したオーディオを除外します。このチェックがない場合、ワーカーは自身が合成したオーディオをSpeech Engineへ送り返そうとしてしまいます。

正しく動作させるためには、順序に関する2つの重要なポイントがあります。

リスナーのタイミング：TrackSubscribedはctx.connect()の前に登録します。LiveKitは接続ハンドシェイク中に既存トラックを自動購読するため、その後に登録したリスナーはイベントを見逃す可能性があります。オーディオポンプはSpeech Engine WebSocketのFuture／Promiseを待機するため、すぐに購読でき、接続が開いた直後からオーディオを転送できます。
TypeScriptのみ：キャプチャの直列化：@livekit/rtc-nodeのAudioSource.captureFrameは、同時に呼び出すとInvalidStateをスローします。TypeScriptハンドラーでは、Promiseチェーンを使ってキャプチャを直列化します。Pythonの単一のasync for el_to_roomループは自然に順次実行されるため、これは不要です。

ワーカーを起動する

python bridge.py dev

devではホットリロードとカラー表示のログが有効になります。本番環境では、JSONログとグレースフルシャットダウンのためにstartを使用してください。

ワーカーはLiveKitサーバーに接続し、ジョブの割り当てを待機します。ディスパッチされるまで、どのルームにも参加しません。

ワーカーをルームにディスパッチする

ワーカーにはagent_nameがあるため、明示的なディスパッチを使用します。バックエンドから指示された場合にのみルームに参加します。最も簡単なパターンは、ブラウザが接続に使用するLiveKitアクセストークンにRoomAgentDispatchを含めることです。

import os
from dotenv import load_dotenv
from flask import Flask, jsonify, request
from livekit.api import AccessToken, RoomAgentDispatch, VideoGrants
load_dotenv()
app = Flask(**name**)
@app.route("/api/livekit-token")
def get_token():
room_name = request.args.get("room", "demo-room")
identity = request.args.get("identity", "web-user")
    token = (
        AccessToken(
            os.environ["LIVEKIT_API_KEY"],
            os.environ["LIVEKIT_API_SECRET"],
        )
        .with_identity(identity)
        .with_grants(VideoGrants(room_join=True, room=room_name))
        .with_room_config(
            room_configuration={
                "agents": [RoomAgentDispatch(agent_name="elevenlabs-bridge")],
            },
        )
    )
    return jsonify(token=token.to_jwt(), url=os.environ["LIVEKIT_URL"])
if **name** == "**main**":
app.run(port=3002)

ブラウザがこのトークンを使用してルームを作成または参加すると、LiveKitは同じルームにブリッジワーカーを自動的にディスパッチします。

ブラウザから接続する

ブラウザに必要なのは標準のLiveKitクライアントのみです。Speech Engineとは直接やり取りしません。

App.tsx

import { Room, RoomEvent, Track } from "livekit-client";
import { useCallback, useState } from "react";
export default function App() {
  const [room] = useState(() => new Room());
  const join = useCallback(async () => {
    const response = await fetch("/api/livekit-token");
    const { token, url } = await response.json();
    room.on(RoomEvent.TrackSubscribed, (track) => {
      if (track.kind === Track.Kind.Audio) {
        document.body.appendChild(track.attach());
      }
    });
    await room.connect(url, token);
    await room.localParticipant.setMicrophoneEnabled(true);
  }, [room]);
  return <button onClick={join}>Start conversation</button>;
}

ボタンをクリックすると、ブラウザはLiveKitトークンを取得し、マイクを有効にしてルームに参加し、エージェントのオーディオトラックの受信を開始します。ワーカーがディスパッチされてSpeech Engineセッションを開き、双方向にオーディオをブリッジします。

オーディオ形式リファレンス

Speech Engineは以下のオーディオ形式に対応しています。エンジンではasr.user_input_audio_formatとtts.agent_output_audio_formatで設定します。

形式	サンプルレート	エンコーディング	備考
`pcm_8000`	8 kHz	符号付き16ビットLE PCM	ASR入力のみ。
`pcm_16000`	16 kHz	符号付き16ビットLE PCM	LiveKitのユーザー入力に推奨。
`pcm_22050`	22.05 kHz	符号付き16ビットLE PCM
`pcm_24000`	24 kHz	符号付き16ビットLE PCM	LiveKitのエージェント出力に推奨。
`pcm_44100`	44.1 kHz	符号付き16ビットLE PCM	TTS出力にはIndependent Publisher tier以上が必要です。
`pcm_48000`	48 kHz	符号付き16ビットLE PCM	ASR入力のみ。
`ulaw_8000`	8 kHz	μ-law	Twilio Media Streamsで使用されます。

LiveKitのAudioStreamとAudioSourceはリサンプリングを自動的に処理します。AudioStreamには任意のサンプルレートをリクエストでき、SDKが基盤となる48 kHz Opusトラックから変換します。

本番運用時の考慮事項

明示的なディスパッチ：WorkerOptionsでは必ずagent_name／agentNameを設定してください。自動ディスパッチでは、LiveKitプロジェクトで作成されるすべてのルームに対してワーカーが起動します。これは通常、意図した動作ではありません。
ブレインサーバーの認証：Speech Engineに共有シークレットを設定し、ブレインサーバーで検証してください。これにより、Speech Engineだけがエンドポイントにアクセスできるようになります：
```
await elevenlabs.speech_engine.update(
    speech_engine_id="seng_8k3m9xr4hjnfg983brhmhkd98n6",
    speech_engine={"request_headers": {"x-api-key": os.environ["SHARED_SECRET"]}},
)
```
ブレインサーバーは、WebSocketアップグレードを受け入れる前にrequest.headers["x-api-key"]を確認します。
トークンサーバー：LiveKitとSpeech Engineのトークンはサーバー側で発行してください。LIVEKIT_API_SECRETやELEVENLABS_API_KEYをブラウザーに公開しないでください。
イベントループの健全性：CPU負荷の高い処理をワーカーのイベントループで実行しないでください。AudioSource.capture_frameとAudioStreamのイテレーションは時間に敏感です。長時間の同期呼び出しにより、割り込みイベントが遅延したり失われたりする可能性があります。ブロッキング処理にはasyncio.to_thread()（Python）またはworker_threads（Node）を使用してください。
シャットダウン：ctx.add_shutdown_callback／ctx.addShutdownCallbackを登録し、ElevenLabs WebSocketを適切に閉じてください。デフォルトでは、最後の非エージェント参加者が退出すると、ルーム（およびジョブ）は終了します。

次のステップ

Speech Engineクイックスタート

トランスクリプトに応答するブレインサーバーを構築します。

Pipecatインテグレーション

Speech Engineの背後にあるLLMパイプラインとしてPipecatを使用します。

Python SDKリファレンス

Speech Engine Python SDKのクラス、メソッド、イベント。

JavaScript SDKリファレンス

Speech Engine JavaScript SDKのクラス、メソッド、イベント。