LiveKit-Integration

Dieser Leitfaden zeigt, wie Sie ElevenLabs Speech Engine als Sprachschicht für einen LiveKit-Raum verwenden. Ein LiveKit-Agents-Worker tritt dem Raum als Teilnehmer bei, abonniert die Audiospur des Benutzers, öffnet einen WebSocket zu Speech Engine und veröffentlicht die synthetisierten Audiodaten von Speech Engine als eigene Spur zurück im Raum.

Architektur

Speech Engine akzeptiert zwei Arten von WebSocket-Verbindungen:

• Den Brain-WebSocket, mit dem sich die ElevenLabs API verbindet. Ihr Server führt diesen mit dem Speech Engine SDK (engine.serve() / engine.attach()) aus und erhält Transkripte, auf die er reagieren kann.
• Den Konversations-WebSocket, mit dem sich Clients verbinden. Browser verbinden sich über ein WebRTC-Token; Nicht-Browser-Clients (wie ein LiveKit-Agents-Worker) verbinden sich über eine signierte URL und streamen rohe PCM-Audiodaten in beide Richtungen.

Der LiveKit-Worker verwendet die zweite Verbindung. Er fungiert im Namen der Teilnehmer im LiveKit-Raum als „Client“ von Speech Engine.

Der Brain-Server bleibt gegenüber dem Speech Engine Quickstart unverändert – der LiveKit-Worker ersetzt den Browser als Audioquelle, die LLM-Logik bleibt jedoch gleich.

Wann Sie dieses Muster verwenden sollten

Nutzen Sie die LiveKit-Bridge, wenn der Raum selbst Teil der Erfahrung ist:

• Sitzungen mit mehreren Teilnehmern, in denen Benutzer gemeinsam mit dem Agenten sprechen
• Bestehende LiveKit-Deployments, bei denen ein Wechsel des Transports Clients beeinträchtigen würde
• Sprachagenten, die einen Raum mit Bildschirmfreigabe, Video oder Textchat teilen
• Über SIP an LiveKit weitergeleitete Anrufe, die einen KI-Agenten in der Leitung benötigen

Wenn Sie nur eine Browser-zu-Speech-Engine-Sprachschleife ohne weitere Teilnehmer benötigen, ist der WebRTC-Client im Speech Engine Quickstart einfacher – Speech Engine kommuniziert direkt per WebRTC mit dem Browser, ein LiveKit-Raum ist nicht erforderlich.

Voraussetzungen

• Ein LiveKit-Projekt (LiveKit Cloud oder ein selbst gehosteter Server). Der Worker benötigt LIVEKIT_URL, LIVEKIT_API_KEY und LIVEKIT_API_SECRET.
• Eine ElevenLabs Speech Engine. Folgen Sie dem Speech Engine Quickstart, um eine zu erstellen und den Brain-Server auszuführen.
• Python 3.9+ oder Node.js 18+.

Audioformate für Speech Engine konfigurieren

LiveKits AudioStream passt eingehende Opus-Spuren an jede von Ihnen angeforderte PCM-Abtastrate an. So können Sie sie direkt auf Speech Engine abstimmen. Aktualisieren Sie Speech Engine, um 16-kHz-PCM für ASR-Eingaben zu akzeptieren und 24-kHz-PCM für TTS-Ausgaben zu erzeugen.

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 ist durchgehend vorzeichenbehaftetes 16-Bit-Little-Endian. Weitere unterstützte Abtastraten finden Sie in der Referenz zu Audioformaten.

Bridge-Worker erstellen

Der Worker ist ein lang laufender Prozess, der sich mit Ihrem LiveKit-Server verbindet, auf Aufträge wartet, zugewiesenen Räumen beitritt und Audiodaten zwischen dem Raum und Speech Engine überträgt.

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

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

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",
    ))

python bridge.py dev

Worker an einen Raum weiterleiten

Da der Worker einen agent_name hat, verwendet er eine explizite Weiterleitung – er tritt Räumen nur bei, wenn Ihr Backend ihn dazu auffordert. Das einfachste Muster besteht darin, einen RoomAgentDispatch in das LiveKit-Zugriffstoken aufzunehmen, das der Browser für die Verbindung verwendet.

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)

Wenn ein Browser dieses Token verwendet, um einen Raum zu erstellen oder ihm beizutreten, leitet LiveKit den Bridge-Worker automatisch in denselben Raum weiter.

Verbindung über den Browser herstellen

Der Browser benötigt nur den Standard-LiveKit-Client – er interagiert nicht direkt mit Speech Engine.

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>;
}

Wenn auf die Schaltfläche geklickt wird, ruft der Browser ein LiveKit-Token ab, tritt dem Raum mit aktiviertem Mikrofon bei und beginnt, die Audiospur des Agenten zu empfangen. Der Worker wird weitergeleitet, öffnet seine Speech-Engine-Sitzung und überträgt Audio in beide Richtungen.

Referenz für Audioformate

Speech Engine unterstützt die folgenden Audioformate. Konfigurieren Sie diese in der Engine über asr.user_input_audio_format und tts.agent_output_audio_format.

AudioStream und AudioSource in LiveKit übernehmen das Resampling für Sie — Sie können von AudioStream jede Abtastrate anfordern, und das SDK konvertiert sie aus dem zugrunde liegenden Opus-Track mit 48 kHz.

Hinweise für den Produktionseinsatz

• Explizite Zuweisung: Setzen Sie auf WorkerOptions immer agent_name / agentName. Bei der automatischen Zuweisung wird der Worker für jeden Raum gestartet, der in Ihrem LiveKit-Projekt erstellt wird. Das ist selten erwünscht.
• Authentifizierung des Brain-Servers: Legen Sie ein gemeinsames Secret für die Speech Engine fest und prüfen Sie es in Ihrem Brain-Server, damit nur die Speech Engine Ihren Endpunkt erreichen kann:

  await elevenlabs.speech_engine.update(
      speech_engine_id="seng_8k3m9xr4hjnfg983brhmhkd98n6",
      speech_engine={"request_headers": {"x-api-key": os.environ["SHARED_SECRET"]}},
  )
  Der Brain-Server prüft dann request.headers["x-api-key"], bevor er das WebSocket-Upgrade akzeptiert.
• Token-Server: Erstellen Sie LiveKit- und Speech-Engine-Tokens serverseitig. Geben Sie LIVEKIT_API_SECRET oder ELEVENLABS_API_KEY niemals im Browser preis.
• Saubere Event Loop-Nutzung: Halten Sie CPU-intensive Aufgaben vom Event Loop des Workers fern. AudioSource.capture_frame und die Iteration von AudioStream sind zeitkritisch. Lange synchrone Aufrufe verzögern oder verwerfen Unterbrechungsereignisse. Verwenden Sie für blockierende Aufgaben asyncio.to_thread() (Python) oder worker_threads (Node).
• Herunterfahren: Registrieren Sie ctx.add_shutdown_callback / ctx.addShutdownCallback, um den ElevenLabs-WebSocket sauber zu schließen. Standardmäßig wird der Raum (und der Job) beendet, wenn der letzte Teilnehmer, der kein Agent ist, den Raum verlässt.

Nächste Schritte