Integração com LiveKit

Este guia mostra como usar o Speech Engine da ElevenLabs como a camada de voz de uma sala do LiveKit. Um worker do LiveKit Agents entra na sala como participante, assina a faixa de áudio do usuário, abre um WebSocket para o Speech Engine e publica o áudio sintetizado pelo Speech Engine de volta na sala como sua própria faixa.

Arquitetura

O Speech Engine aceita dois tipos de conexões WebSocket:

• O WebSocket do cérebro, ao qual a API da ElevenLabs se conecta. Seu servidor o executa com o SDK do Speech Engine (engine.serve() / engine.attach()) e recebe transcrições para responder.
• O WebSocket da conversa, ao qual os clientes se conectam. Os navegadores se conectam por um token WebRTC; clientes que não são navegadores (como um worker do LiveKit Agents) se conectam por uma URL assinada e transmitem áudio PCM bruto em ambas as direções.

O worker do LiveKit usa a segunda conexão. Ele atua como um “cliente” do Speech Engine em nome dos participantes da sala do LiveKit.

O servidor do cérebro não muda em relação ao guia rápido do Speech Engine — o worker do LiveKit substitui o navegador como fonte de áudio, mas a lógica do LLM continua a mesma.

Quando usar este padrão

Use a ponte com o LiveKit quando a própria sala fizer parte da experiência:

• Sessões com vários participantes, em que os usuários conversam com o agente ao mesmo tempo
• Implantações existentes do LiveKit, nas quais trocar o transporte interromperia os clientes
• Agentes de voz que compartilham uma sala com compartilhamento de tela, vídeo ou chat de texto
• Chamadas SIP para LiveKit distribuídas que precisam de um agente de IA na chamada

Se você só precisa de um loop de voz do navegador para o Speech Engine, sem outros participantes, o cliente WebRTC no guia rápido do Speech Engine é mais simples — o Speech Engine se comunica diretamente com o navegador via WebRTC, sem precisar de uma sala do LiveKit.

Pré-requisitos

• Um projeto do LiveKit (LiveKit Cloud ou um servidor auto-hospedado). O worker precisa de LIVEKIT_URL, LIVEKIT_API_KEY e LIVEKIT_API_SECRET.
• Um Speech Engine da ElevenLabs. Siga o guia rápido do Speech Engine para criar um e executar o servidor do cérebro.
• Python 3.9+ ou Node.js 18+.

Configure os formatos de áudio do Speech Engine

O AudioStream do LiveKit reamostra faixas Opus recebidas para qualquer taxa de amostragem PCM que você solicitar, então é possível corresponder diretamente à entrada do Speech Engine. Atualize o Speech Engine para aceitar PCM de 16 kHz como entrada de ASR e gerar PCM de 24 kHz como saída 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())

O PCM do Speech Engine é sempre assinado, de 16 bits e little-endian. Consulte a referência de formatos de áudio para ver outras taxas compatíveis.

Crie o worker de ponte

O worker é um processo de longa execução que se conecta ao seu servidor LiveKit, aguarda tarefas, entra nas salas atribuídas e faz a ponte de áudio entre a sala e o 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

Envie o worker para uma sala

Como o worker tem um agent_name, ele usa envio explícito — só entra em salas quando seu backend solicita. O padrão mais simples é incluir um RoomAgentDispatch no token de acesso do LiveKit usado pelo navegador para se conectar.

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 um navegador usa esse token para criar ou entrar em uma sala, o LiveKit envia automaticamente o worker de ponte para a mesma sala.

Conecte-se pelo navegador

O navegador só precisa do cliente padrão do LiveKit — ele não interage diretamente com o 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 o botão é clicado, o navegador busca um token do LiveKit, entra na sala com o microfone ativado e começa a receber a faixa de áudio do agente. O worker é enviado, abre sua sessão do Speech Engine e faz a ponte de áudio nas duas direções.

Referência de formatos de áudio

O Speech Engine é compatível com os seguintes formatos de áudio. Configure-os no mecanismo usando asr.user_input_audio_format e tts.agent_output_audio_format.

AudioStream e AudioSource no LiveKit fazem a reamostragem para você — é possível solicitar qualquer taxa de amostragem ao AudioStream, e o SDK converte a partir da faixa Opus subjacente de 48 kHz.

Considerações para produção

• Despacho explícito: Sempre defina agent_name / agentName em WorkerOptions. O despacho automático aciona o worker para cada sala criada no seu projeto LiveKit, o que raramente é o desejado.
• Autenticação do servidor de raciocínio: Defina um segredo compartilhado no Speech Engine e verifique-o no seu servidor de raciocínio para que apenas o Speech Engine possa acessar seu endpoint:

  await elevenlabs.speech_engine.update(
      speech_engine_id="seng_8k3m9xr4hjnfg983brhmhkd98n6",
      speech_engine={"request_headers": {"x-api-key": os.environ["SHARED_SECRET"]}},
  )
  Em seguida, o servidor de raciocínio verifica request.headers["x-api-key"] antes de aceitar a atualização do WebSocket.
• Servidor de tokens: Gere tokens do LiveKit e do Speech Engine no servidor. Nunca exponha LIVEKIT_API_SECRET nem ELEVENLABS_API_KEY no navegador.
• Higiene do loop de eventos: Mantenha tarefas que exigem CPU fora do loop de eventos do worker. A iteração de AudioSource.capture_frame e AudioStream é sensível ao tempo; chamadas síncronas longas atrasarão ou descartarão eventos de interrupção. Use asyncio.to_thread() (Python) ou worker_threads (Node) para tarefas bloqueantes.
• Encerramento: Registre ctx.add_shutdown_callback / ctx.addShutdownCallback para fechar corretamente o WebSocket da ElevenLabs. Por padrão, a sala (e o trabalho) é encerrada quando o último participante que não é agente sai.

Próximas etapas