Intégration LiveKit

Ce guide explique comment utiliser ElevenLabs Speech Engine comme couche vocale pour une salle LiveKit. Un worker LiveKit Agents rejoint la salle en tant que participant, s’abonne à la piste audio de l’utilisateur, ouvre un WebSocket vers Speech Engine et publie l’audio synthétisé par Speech Engine dans la salle sous la forme de sa propre piste.

Architecture

Speech Engine accepte deux types de connexions WebSocket :

• Le WebSocket brain, auquel l’API ElevenLabs se connecte. Votre serveur l’exécute avec le SDK Speech Engine (engine.serve() / engine.attach()) et reçoit les transcriptions auxquelles répondre.
• Le WebSocket de conversation, auquel les clients se connectent. Les navigateurs se connectent via un jeton WebRTC ; les clients non basés sur un navigateur, comme un worker LiveKit Agents, se connectent via une URL signée et diffusent de l’audio PCM brut dans les deux sens.

Le worker LiveKit utilise la seconde connexion. Il agit comme un « client » de Speech Engine pour le compte des participants de la salle LiveKit.

Le serveur brain reste identique à celui du guide de démarrage rapide de Speech Engine, le worker LiveKit remplace le navigateur comme source audio, mais la logique LLM reste la même.

Quand utiliser ce modèle

Utilisez le pont LiveKit lorsque la salle fait elle-même partie de l’expérience :

• Sessions multiparticipants où les utilisateurs parlent avec l’agent en même temps
• Déploiements LiveKit existants pour lesquels changer de transport perturberait les clients
• Agents vocaux partageant une salle avec le partage d’écran, la vidéo ou le chat textuel
• Appels distribués de SIP vers LiveKit qui nécessitent un agent IA en ligne

Si vous avez uniquement besoin d’une boucle vocale entre le navigateur et Speech Engine, sans autres participants, le client WebRTC du guide de démarrage rapide de Speech Engine est plus simple : Speech Engine communique directement avec le navigateur via WebRTC, sans nécessiter de salle LiveKit.

Prérequis

• Un projet LiveKit, soit LiveKit Cloud, soit un serveur auto-hébergé. Le worker nécessite LIVEKIT_URL, LIVEKIT_API_KEY et LIVEKIT_API_SECRET.
• Un ElevenLabs Speech Engine. Suivez le guide de démarrage rapide de Speech Engine pour en créer un et exécuter le serveur brain.
• Python 3.9+ ou Node.js 18+.

Configurer les formats audio de Speech Engine

AudioStream de LiveKit rééchantillonne les pistes Opus entrantes à la fréquence d’échantillonnage PCM demandée, ce qui vous permet de les faire correspondre directement à l’entrée de Speech Engine. Mettez à jour Speech Engine pour accepter du PCM 16 kHz en entrée ASR et produire du PCM 24 kHz en sortie 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())

Le PCM de Speech Engine est partout au format 16 bits signé little-endian. Consultez la référence des formats audio pour connaître les autres fréquences prises en charge.

Créer le worker de pont

Le worker est un processus de longue durée qui se connecte à votre serveur LiveKit, attend les tâches, rejoint les salles attribuées et transfère l’audio entre la salle et 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

Envoyer le worker dans une salle

Puisque le worker possède un agent_name, il utilise un envoi explicite : il rejoint uniquement les salles lorsque votre backend le lui demande. Le modèle le plus simple consiste à inclure un RoomAgentDispatch dans le jeton d’accès LiveKit que le navigateur utilise pour se connecter.

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)

Lorsqu’un navigateur utilise ce jeton pour créer ou rejoindre une salle, LiveKit envoie automatiquement le worker de pont dans cette même salle.

Se connecter depuis le navigateur

Le navigateur n’a besoin que du client LiveKit standard, il n’interagit pas directement avec 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>;
}

Lorsque le bouton est sélectionné, le navigateur récupère un jeton LiveKit, rejoint la salle avec le microphone activé et commence à recevoir la piste audio de l’agent. Le worker est envoyé, ouvre sa session Speech Engine et transfère l’audio dans les deux sens.

Référence des formats audio

Speech Engine prend en charge les formats audio suivants. Configurez-les sur le moteur via asr.user_input_audio_format et tts.agent_output_audio_format.

AudioStream et AudioSource dans LiveKit gèrent le rééchantillonnage pour vous : vous pouvez demander n’importe quelle fréquence d’échantillonnage à AudioStream, et le SDK effectue la conversion à partir de la piste Opus sous-jacente de 48 kHz.

Points à considérer pour la production

• Envoi explicite : définissez toujours agent_name / agentName dans WorkerOptions. L’envoi automatique déclenche le worker pour chaque salle créée dans votre projet LiveKit, ce qui est rarement souhaitable.
• Authentification du serveur brain : définissez un secret partagé sur Speech Engine et vérifiez-le dans votre serveur brain afin que seul Speech Engine puisse atteindre votre point de terminaison :

  await elevenlabs.speech_engine.update(
      speech_engine_id="seng_8k3m9xr4hjnfg983brhmhkd98n6",
      speech_engine={"request_headers": {"x-api-key": os.environ["SHARED_SECRET"]}},
  )
  Le serveur brain vérifie ensuite request.headers["x-api-key"] avant d’accepter la mise à niveau WebSocket.
• Serveur de jetons : créez les jetons LiveKit et Speech Engine côté serveur. N’exposez jamais LIVEKIT_API_SECRET ou ELEVENLABS_API_KEY au navigateur.
• Hygiène de la boucle d’événements : évitez les tâches gourmandes en CPU dans la boucle d’événements du worker. AudioSource.capture_frame et l’itération de AudioStream sont sensibles au temps ; des appels synchrones longs retarderont ou ignoreront des événements d’interruption. Utilisez asyncio.to_thread() (Python) ou worker_threads (Node) pour les tâches bloquantes.
• Arrêt : enregistrez ctx.add_shutdown_callback / ctx.addShutdownCallback pour fermer proprement le WebSocket ElevenLabs. Par défaut, la salle, et la tâche, sont terminées lorsque le dernier participant non agent quitte la salle.

Étapes suivantes