Integración con LiveKit

Esta guía muestra cómo usar ElevenLabs Speech Engine como capa de voz para una sala de LiveKit. Un worker de LiveKit Agents se une a la sala como participante, se suscribe a la pista de audio del usuario, abre un WebSocket con Speech Engine y publica el audio sintetizado de Speech Engine de vuelta en la sala como su propia pista.

Arquitectura

Speech Engine acepta dos tipos de conexiones WebSocket:

• El WebSocket del cerebro al que se conecta la API de ElevenLabs. Tu servidor lo ejecuta con el SDK de Speech Engine (engine.serve() / engine.attach()) y recibe transcripciones a las que responder.
• El WebSocket de conversación al que se conectan los clientes. Los navegadores se conectan mediante un token WebRTC; los clientes que no son navegadores (como un worker de LiveKit Agents) se conectan mediante una URL firmada y transmiten audio PCM sin procesar en ambas direcciones.

El worker de LiveKit utiliza la segunda conexión. Actúa como un “cliente” de Speech Engine en nombre de los participantes de la sala de LiveKit.

El servidor del cerebro no cambia respecto a la guía rápida de Speech Engine: el worker de LiveKit sustituye al navegador como fuente de audio, pero la lógica del LLM sigue siendo la misma.

Cuándo usar este patrón

Usa el puente de LiveKit cuando la sala forme parte de la experiencia:

• Sesiones con varios participantes en las que usuarios hablan entre sí junto con el agente
• Implementaciones existentes de LiveKit en las que cambiar de transporte rompería los clientes
• Agentes de voz que comparten una sala con pantalla compartida, vídeo o chat de texto
• Llamadas SIP a LiveKit enviadas a un worker que necesitan un agente de IA en la línea

Si solo necesitas un bucle de voz entre el navegador y Speech Engine sin otros participantes, el cliente WebRTC de la guía rápida de Speech Engine es más sencillo: Speech Engine se comunica directamente mediante WebRTC con el navegador, sin necesidad de una sala de LiveKit.

Requisitos previos

• Un proyecto de LiveKit (LiveKit Cloud o un servidor autoalojado). El worker necesita LIVEKIT_URL, LIVEKIT_API_KEY y LIVEKIT_API_SECRET.
• Un Speech Engine de ElevenLabs. Sigue la guía rápida de Speech Engine para crear uno y ejecutar el servidor del cerebro.
• Python 3.9+ o Node.js 18+.

Configura los formatos de audio de Speech Engine

El AudioStream de LiveKit remuestrea las pistas Opus entrantes a la frecuencia de muestreo PCM que solicites, por lo que puedes ajustarla directamente a la entrada de Speech Engine. Actualiza Speech Engine para aceptar PCM de 16 kHz como entrada de ASR y emitir PCM de 24 kHz como salida de 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())

El PCM de Speech Engine utiliza enteros con signo de 16 bits en formato little-endian en todo el proceso. Consulta la referencia de formatos de audio para conocer otras frecuencias compatibles.

Crea el worker puente

El worker es un proceso de larga duración que se conecta a tu servidor de LiveKit, espera trabajos, se une a las salas asignadas y conecta el audio entre la sala y 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

Asigna el worker a una sala

Como el worker tiene un agent_name, utiliza una asignación explícita: solo se une a salas cuando tu backend se lo indica. El patrón más sencillo es incluir un RoomAgentDispatch en el token de acceso de LiveKit que usa el navegador para conectarse.

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)

Cuando un navegador usa este token para crear o unirse a una sala, LiveKit asigna automáticamente el worker puente a la misma sala.

Conéctate desde el navegador

El navegador solo necesita el cliente estándar de LiveKit; no interactúa directamente 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>;
}

Al hacer clic en el botón, el navegador obtiene un token de LiveKit, se une a la sala con el micrófono activado y empieza a recibir la pista de audio del agente. Se asigna el worker, que abre su sesión de Speech Engine y conecta el audio en ambas direcciones.

Referencia de formatos de audio

Speech Engine admite los siguientes formatos de audio. Configúralos en el motor mediante asr.user_input_audio_format y tts.agent_output_audio_format.

AudioStream y AudioSource de LiveKit gestionan el remuestreo por ti: puedes solicitar cualquier frecuencia de muestreo a AudioStream y el SDK la convierte desde la pista Opus subyacente de 48 kHz.

Consideraciones para producción

• Asignación explícita: Configura siempre agent_name / agentName en WorkerOptions. La asignación automática activa el worker para cada sala creada en tu proyecto de LiveKit, algo que rara vez querrás.
• Autenticación del servidor brain: Configura un secreto compartido en Speech Engine y verifícalo en tu servidor brain, para que solo Speech Engine pueda acceder a tu ruta:

  await elevenlabs.speech_engine.update(
      speech_engine_id="seng_8k3m9xr4hjnfg983brhmhkd98n6",
      speech_engine={"request_headers": {"x-api-key": os.environ["SHARED_SECRET"]}},
  )
  El servidor brain comprueba entonces request.headers["x-api-key"] antes de aceptar la actualización a WebSocket.
• Servidor de tokens: Genera los tokens de LiveKit y Speech Engine en el servidor. No expongas nunca LIVEKIT_API_SECRET ni ELEVENLABS_API_KEY al navegador.
• Higiene del bucle de eventos: Mantén el trabajo intensivo de CPU fuera del bucle de eventos del worker. La llamada a AudioSource.capture_frame y la iteración de AudioStream son sensibles al tiempo; las llamadas síncronas largas retrasarán o descartarán eventos de interrupción. Usa asyncio.to_thread() (Python) o worker_threads (Node) para trabajo bloqueante.
• Cierre: Registra ctx.add_shutdown_callback / ctx.addShutdownCallback para cerrar correctamente el WebSocket de ElevenLabs. De forma predeterminada, la sala (y el trabajo) se termina cuando se va el último participante que no es un agente.

Próximos pasos