Integrazione LiveKit

Questa guida spiega come usare ElevenLabs Speech Engine come livello vocale per una stanza LiveKit. Un worker LiveKit Agents entra nella stanza come partecipante, si iscrive alla traccia audio dell’utente, apre un WebSocket verso Speech Engine e pubblica l’audio sintetizzato da Speech Engine nella stanza come propria traccia.

Architettura

Speech Engine accetta due tipi di connessioni WebSocket:

• Il WebSocket brain a cui si connette l’API ElevenLabs. Il tuo server lo esegue con l’SDK Speech Engine (engine.serve() / engine.attach()) e riceve trascrizioni a cui rispondere.
• Il WebSocket di conversazione a cui si connettono i client. I browser si connettono tramite un token WebRTC; i client non browser (come un worker LiveKit Agents) si connettono tramite un URL firmato e trasmettono audio PCM raw in entrambe le direzioni.

Il worker LiveKit usa la seconda connessione. Agisce come “client” di Speech Engine per conto dei partecipanti nella stanza LiveKit.

Il server brain rimane invariato rispetto alla guida rapida di Speech Engine: il worker LiveKit sostituisce il browser come sorgente audio, ma la logica LLM resta la stessa.

Quando usare questo schema

Usa il bridge LiveKit quando la stanza stessa fa parte dell’esperienza:

• Sessioni con più partecipanti in cui gli utenti parlano con l’agente insieme
• Implementazioni LiveKit esistenti in cui cambiare trasporto interromperebbe i client
• Agenti vocali che condividono una stanza con condivisione schermo, video o chat testuale
• Chiamate inviate da SIP a LiveKit che richiedono un agente IA in linea

Se ti serve solo un loop vocale dal browser a Speech Engine senza altri partecipanti, il client WebRTC nella guida rapida di Speech Engine è più semplice: Speech Engine comunica direttamente via WebRTC con il browser, senza bisogno di una stanza LiveKit.

Prerequisiti

• Un progetto LiveKit (LiveKit Cloud o un server ospitato autonomamente). Il worker richiede LIVEKIT_URL, LIVEKIT_API_KEY e LIVEKIT_API_SECRET.
• Un ElevenLabs Speech Engine. Segui la guida rapida di Speech Engine per crearne uno ed eseguire il server brain.
• Python 3.9+ o Node.js 18+.

Configura i formati audio di Speech Engine

AudioStream di LiveKit ricampiona le tracce Opus in entrata a qualsiasi frequenza di campionamento PCM richiesta, quindi puoi adattarla direttamente all’input di Speech Engine. Aggiorna Speech Engine affinché accetti PCM a 16 kHz per l’input ASR ed emetta PCM a 24 kHz per l’output 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())

Il PCM di Speech Engine è sempre little-endian con segno a 16 bit. Consulta il riferimento dei formati audio per le altre frequenze supportate.

Crea il worker bridge

Il worker è un processo a lunga esecuzione che si connette al tuo server LiveKit, attende i job, entra nelle stanze assegnate e fa da ponte per l’audio tra la stanza e 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

Invia il worker a una stanza

Poiché il worker ha un agent_name, usa l’invio esplicito: entra nelle stanze solo quando il tuo backend glielo indica. Lo schema più semplice consiste nell’includere un RoomAgentDispatch nel token di accesso LiveKit che il browser usa per connettersi.

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)

Quando un browser usa questo token per creare o entrare in una stanza, LiveKit invia automaticamente il worker bridge nella stessa stanza.

Connettiti dal browser

Il browser necessita solo del client LiveKit standard: non interagisce direttamente con 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>;
}

Quando si fa clic sul pulsante, il browser recupera un token LiveKit, entra nella stanza con il microfono abilitato e inizia a ricevere la traccia audio dell’agente. Il worker viene inviato, apre la sua sessione Speech Engine e fa da ponte per l’audio in entrambe le direzioni.

Riferimento dei formati audio

Speech Engine supporta i seguenti formati audio. Configurali sul motore tramite asr.user_input_audio_format e tts.agent_output_audio_format.

AudioStream e AudioSource in LiveKit gestiscono il ricampionamento per te: puoi richiedere qualsiasi frequenza di campionamento a AudioStream e l’SDK converte dalla traccia Opus sottostante a 48 kHz.

Considerazioni per la produzione

• Invio esplicito: imposta sempre agent_name / agentName su WorkerOptions. L’invio automatico attiva il worker per ogni stanza creata nel tuo progetto LiveKit, cosa che raramente desideri.
• Autenticazione del server brain: imposta un segreto condiviso su Speech Engine e verificalo nel tuo server brain, così solo Speech Engine può raggiungere il tuo endpoint:

  await elevenlabs.speech_engine.update(
      speech_engine_id="seng_8k3m9xr4hjnfg983brhmhkd98n6",
      speech_engine={"request_headers": {"x-api-key": os.environ["SHARED_SECRET"]}},
  )
  Il server brain verifica quindi request.headers["x-api-key"] prima di accettare l’upgrade WebSocket.
• Server dei token: genera i token LiveKit e Speech Engine lato server. Non esporre mai LIVEKIT_API_SECRET o ELEVENLABS_API_KEY al browser.
• Igiene dell’event loop: non eseguire operazioni vincolate alla CPU nell’event loop del worker. AudioSource.capture_frame e l’iterazione di AudioStream sono sensibili ai tempi; lunghe chiamate sincrone ritarderanno o scarteranno gli eventi di interruzione. Usa asyncio.to_thread() (Python) o worker_threads (Node) per le operazioni bloccanti.
• Arresto: registra ctx.add_shutdown_callback / ctx.addShutdownCallback per chiudere correttamente il WebSocket ElevenLabs. Per impostazione predefinita, la stanza (e il job) viene terminata quando l’ultimo partecipante non agente esce.

Passaggi successivi