Integracja z LiveKit

Ten przewodnik pokazuje, jak używać ElevenLabs Speech Engine jako warstwy głosowej pokoju LiveKit. Worker LiveKit Agents dołącza do pokoju jako uczestnik, subskrybuje ścieżkę audio użytkownika, otwiera WebSocket ze Speech Engine i publikuje zsyntetyzowane audio Speech Engine z powrotem w pokoju jako własną ścieżkę.

Architektura

Speech Engine obsługuje dwa rodzaje połączeń WebSocket:

• WebSocket mózgu, z którym łączy się API ElevenLabs. Twój serwer uruchamia go za pomocą SDK Speech Engine (engine.serve() / engine.attach()) i otrzymuje transkrypcje, na które ma odpowiadać.
• WebSocket rozmowy, z którym łączą się klienci. Przeglądarki łączą się przez token WebRTC; klienci spoza przeglądarki (np. worker LiveKit Agents) łączą się przez podpisany URL i strumieniują surowe audio PCM w obu kierunkach.

Worker LiveKit używa drugiego połączenia. Działa jako „klient” Speech Engine w imieniu uczestników pokoju LiveKit.

Serwer mózgu pozostaje bez zmian względem krótkiego wprowadzenia do Speech Engine — worker LiveKit zastępuje przeglądarkę jako źródło audio, ale logika LLM pozostaje taka sama.

Kiedy użyć tego wzorca

Skorzystaj z mostu LiveKit, gdy sam pokój jest częścią doświadczenia:

• Sesje z wieloma uczestnikami, podczas których użytkownicy rozmawiają z agentem razem
• Istniejące wdrożenia LiveKit, w których zmiana transportu zepsułaby klientów
• Agenci głosowi współdzielący pokój z udostępnianiem ekranu, wideo lub czatem tekstowym
• Kierowane połączenia SIP-to-LiveKit, które wymagają agenta AI na linii

Jeśli potrzebujesz tylko pętli głosowej przeglądarka–Speech Engine bez innych uczestników, klient WebRTC w krótkim wprowadzeniu do Speech Engine będzie prostszy — Speech Engine komunikuje się z przeglądarką bezpośrednio przez WebRTC, bez pokoju LiveKit.

Wymagania wstępne

• Projekt LiveKit (LiveKit Cloud lub serwer hostowany samodzielnie). Worker potrzebuje LIVEKIT_URL, LIVEKIT_API_KEY i LIVEKIT_API_SECRET.
• ElevenLabs Speech Engine. Postępuj zgodnie z krótkim wprowadzeniem do Speech Engine, aby go utworzyć i uruchomić serwer mózgu.
• Python 3.9+ lub Node.js 18+.

Skonfiguruj formaty audio Speech Engine

AudioStream LiveKit zmienia częstotliwość próbkowania przychodzących ścieżek Opus na dowolną żądaną częstotliwość PCM, więc możesz bezpośrednio dopasować wejście Speech Engine. Zaktualizuj Speech Engine, aby przyjmował PCM 16 kHz na wejściu ASR i generował PCM 24 kHz na wyjściu TTS.

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())

PCM Speech Engine ma w całym przepływie format 16-bitowy ze znakiem i kolejnością little-endian. Inne obsługiwane częstotliwości znajdziesz w dokumentacji formatów audio.

Zbuduj worker mostka

Worker to długotrwały proces, który łączy się z serwerem LiveKit, czeka na zadania, dołącza do przypisanych pokoi i przekazuje audio między pokojem a Speech Engine.

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

Wyślij workera do pokoju

Ponieważ worker ma agent_name, używa jawnego wysyłania — dołącza do pokoi tylko wtedy, gdy nakaże mu to backend. Najprostszy sposób to umieszczenie RoomAgentDispatch w tokenie dostępu LiveKit, którego przeglądarka używa do połączenia.

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)

Gdy przeglądarka użyje tego tokena, aby utworzyć pokój lub do niego dołączyć, LiveKit automatycznie wyśle workera mostka do tego samego pokoju.

Połącz z przeglądarki

Przeglądarka potrzebuje tylko standardowego klienta LiveKit — nie komunikuje się bezpośrednio ze 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>;
}

Po kliknięciu przycisku przeglądarka pobiera token LiveKit, dołącza do pokoju z włączonym mikrofonem i zaczyna odbierać ścieżkę audio agenta. Worker zostaje wysłany, otwiera sesję Speech Engine i przekazuje audio w obu kierunkach.

Informacje o formatach audio

Speech Engine obsługuje poniższe formaty audio. Skonfiguruj je w silniku za pomocą asr.user_input_audio_format i tts.agent_output_audio_format.

AudioStream i AudioSource w LiveKit obsługują resampling za ciebie — możesz zażądać dowolnej częstotliwości próbkowania od AudioStream, a SDK przekonwertuje ją z bazowej ścieżki Opus 48 kHz.

Kwestie produkcyjne

• Jawne wysyłanie: Zawsze ustawiaj agent_name / agentName w WorkerOptions. Automatyczne wysyłanie uruchamia workera dla każdego pokoju utworzonego w projekcie LiveKit, co rzadko jest tym, czego chcesz.
• Uwierzytelnianie serwera brain: Ustaw wspólny sekret w Speech Engine i weryfikuj go na serwerze brain, aby tylko Speech Engine mógł dotrzeć do endpointu:

  await elevenlabs.speech_engine.update(
      speech_engine_id="seng_8k3m9xr4hjnfg983brhmhkd98n6",
      speech_engine={"request_headers": {"x-api-key": os.environ["SHARED_SECRET"]}},
  )
  Serwer brain sprawdza następnie request.headers["x-api-key"] przed zaakceptowaniem przejścia na WebSocket.
• Serwer tokenów: Twórz tokeny LiveKit i Speech Engine po stronie serwera. Nigdy nie ujawniaj w przeglądarce LIVEKIT_API_SECRET ani ELEVENLABS_API_KEY.
• Higiena pętli zdarzeń: Nie uruchamiaj zadań intensywnie korzystających z CPU w pętli zdarzeń workera. AudioSource.capture_frame i iteracja AudioStream są wrażliwe na czas; długie synchroniczne wywołania opóźnią lub pominą zdarzenia przerwań. Do blokujących zadań używaj asyncio.to_thread() (Python) lub worker_threads (Node).
• Zamykanie: Zarejestruj ctx.add_shutdown_callback / ctx.addShutdownCallback, aby poprawnie zamknąć WebSocket ElevenLabs. Domyślnie pokój (i zadanie) jest kończony, gdy wyjdzie ostatni uczestnik niebędący agentem.

Kolejne kroki