LiveKit इंटीग्रेशन

यह गाइड बताती है कि LiveKit रूम के लिए वॉइस लेयर के रूप में ElevenLabs स्पीच इंजन का इस्तेमाल कैसे करें। एक LiveKit Agents वर्कर पार्टिसिपेंट के रूप में रूम में शामिल होता है, यूज़र के ऑडियो ट्रैक को सब्सक्राइब करता है, स्पीच इंजन के लिए WebSocket खोलता है और स्पीच इंजन के सिंथेसाइज़ किए गए ऑडियो को अपने ट्रैक के रूप में वापस रूम में पब्लिश करता है।

आर्किटेक्चर

स्पीच इंजन दो तरह के WebSocket कनेक्शन स्वीकार करता है:

• ब्रेन WebSocket, जिससे ElevenLabs API कनेक्ट होता है। आपका सर्वर इसे स्पीच इंजन SDK (engine.serve() / engine.attach()) के साथ चलाता है और रिस्पॉन्ड करने के लिए ट्रांसक्रिप्ट प्राप्त करता है।
• कन्वर्सेशन WebSocket, जिससे क्लाइंट कनेक्ट होते हैं। ब्राउज़र WebRTC टोकन के ज़रिए कनेक्ट होते हैं; नॉन-ब्राउज़र क्लाइंट (जैसे LiveKit Agents वर्कर) साइन किए गए URL के ज़रिए कनेक्ट होते हैं और दोनों दिशाओं में रॉ PCM ऑडियो स्ट्रीम करते हैं।

LiveKit वर्कर दूसरे कनेक्शन का इस्तेमाल करता है। यह LiveKit रूम के पार्टिसिपेंट्स की ओर से स्पीच इंजन के “क्लाइंट” के रूप में काम करता है।

ब्रेन सर्वर स्पीच इंजन क्विकस्टार्ट से अपरिवर्तित रहता है — LiveKit वर्कर ऑडियो स्रोत के रूप में ब्राउज़र की जगह लेता है, लेकिन LLM लॉजिक वही रहता है।

इस पैटर्न का इस्तेमाल कब करें

जब रूम खुद अनुभव का हिस्सा हो, तब LiveKit ब्रिज इस्तेमाल करें:

• ऐसे मल्टी-पार्टिसिपेंट सेशन, जहां यूज़र एक-दूसरे के साथ एजेंट से बात करते हैं
• मौजूदा LiveKit डिप्लॉयमेंट, जहां ट्रांसपोर्ट बदलने से क्लाइंट्स काम करना बंद कर देंगे
• स्क्रीन शेयर, वीडियो या टेक्स्ट चैट के साथ रूम साझा करने वाले वॉइस एजेंट्स
• SIP-से-LiveKit डिस्पैच की गई कॉल्स, जिनमें लाइन पर AI एजेंट चाहिए

अगर आपको अन्य पार्टिसिपेंट्स के बिना सिर्फ़ ब्राउज़र-से-स्पीच-इंजन वॉइस लूप चाहिए, तो स्पीच इंजन क्विकस्टार्ट में WebRTC क्लाइंट आसान है — स्पीच इंजन सीधे ब्राउज़र से WebRTC पर बात करता है, किसी LiveKit रूम की ज़रूरत नहीं होती।

ज़रूरी शर्तें

• एक LiveKit प्रोजेक्ट (LiveKit Cloud या सेल्फ़-होस्टेड सर्वर)। वर्कर को LIVEKIT_URL, LIVEKIT_API_KEY और LIVEKIT_API_SECRET चाहिए।
• एक ElevenLabs स्पीच इंजन। इसे बनाने और ब्रेन सर्वर चलाने के लिए स्पीच इंजन क्विकस्टार्ट फ़ॉलो करें।
• Python 3.9+ या Node.js 18+।

स्पीच इंजन ऑडियो फ़ॉर्मैट कॉन्फ़िगर करें

LiveKit का AudioStream आने वाले Opus ट्रैक्स को आपके मांगे गए PCM सैंपल रेट पर री-सैंपल करता है, इसलिए आप इसे सीधे स्पीच इंजन के इनपुट से मैच कर सकते हैं। ASR इनपुट के लिए 16 kHz PCM स्वीकार करने और TTS आउटपुट के लिए 24 kHz PCM देने हेतु स्पीच इंजन अपडेट करें।

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 में हर जगह साइन किया हुआ 16-बिट लिटल-एंडियन फ़ॉर्मैट इस्तेमाल होता है। अन्य समर्थित रेट्स के लिए ऑडियो फ़ॉर्मैट रेफरेंस देखें।

ब्रिज वर्कर बनाएं

वर्कर एक लंबे समय तक चलने वाली प्रक्रिया है जो आपके LiveKit सर्वर से कनेक्ट होती है, जॉब्स का इंतज़ार करती है, असाइन किए गए रूम्स में शामिल होती है और रूम व 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

वर्कर को किसी रूम में भेजें

वर्कर के पास agent_name है, इसलिए यह एक्सप्लिसिट डिस्पैच का इस्तेमाल करता है — यह सिर्फ़ तब रूम्स में शामिल होता है, जब आपका बैकएंड इसे ऐसा करने के लिए कहता है। सबसे आसान तरीका है कि ब्राउज़र के कनेक्ट करने के लिए इस्तेमाल होने वाले LiveKit एक्सेस टोकन में RoomAgentDispatch शामिल करें।

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)

जब कोई ब्राउज़र इस टोकन का इस्तेमाल करके रूम बनाता है या उसमें शामिल होता है, तो LiveKit अपने-आप ब्रिज वर्कर को उसी रूम में भेज देता है।

ब्राउज़र से कनेक्ट करें

ब्राउज़र को सिर्फ़ स्टैंडर्ड LiveKit क्लाइंट चाहिए — यह सीधे 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>;
}

बटन पर क्लिक होने पर ब्राउज़र LiveKit टोकन फ़ेच करता है, माइक्रोफ़ोन चालू करके रूम में शामिल होता है और एजेंट का ऑडियो ट्रैक पाना शुरू करता है। वर्कर भेजा जाता है, अपना Speech Engine सेशन खोलता है और दोनों दिशाओं में ऑडियो ब्रिज करता है।

ऑडियो फ़ॉर्मैट रेफरेंस

Speech Engine नीचे दिए गए ऑडियो फ़ॉर्मैट्स को सपोर्ट करता है। इन्हें इंजन पर asr.user_input_audio_format और tts.agent_output_audio_format के ज़रिए कॉन्फ़िगर करें।

LiveKit में AudioStream और AudioSource आपके लिए री-सैंपलिंग संभालते हैं — आप AudioStream से कोई भी सैंपल रेट मांग सकते हैं और SDK अंतर्निहित 48 kHz Opus ट्रैक से कन्वर्ट कर देता है।

प्रोडक्शन से जुड़ी बातें

• एक्सप्लिसिट डिस्पैच: WorkerOptions पर हमेशा agent_name / agentName सेट करें। ऑटो-डिस्पैच आपके LiveKit प्रोजेक्ट में बनाए गए हर रूम के लिए वर्कर चलाता है, जो आमतौर पर आपकी ज़रूरत नहीं होती।
• ब्रेन सर्वर ऑथेंटिकेशन: Speech Engine पर एक शेयर किया गया सीक्रेट सेट करें और अपने ब्रेन सर्वर में इसे वेरिफ़ाई करें, ताकि सिर्फ़ Speech Engine ही आपके एंडपॉइंट तक पहुंच सके:

  await elevenlabs.speech_engine.update(
      speech_engine_id="seng_8k3m9xr4hjnfg983brhmhkd98n6",
      speech_engine={"request_headers": {"x-api-key": os.environ["SHARED_SECRET"]}},
  )
  फिर ब्रेन सर्वर WebSocket अपग्रेड स्वीकार करने से पहले request.headers["x-api-key"] की जांच करता है।
• टोकन सर्वर: LiveKit और Speech Engine टोकन सर्वर-साइड बनाएं। LIVEKIT_API_SECRET या ELEVENLABS_API_KEY को कभी ब्राउज़र के सामने न लाएं।
• इवेंट लूप हाइजीन: CPU-बाउंड काम को वर्कर के इवेंट लूप से अलग रखें। AudioSource.capture_frame और AudioStream इटरेशन समय के प्रति संवेदनशील हैं; लंबे सिंक्रोनस कॉल्स इंटरप्शन इवेंट्स को देर से पहुंचा सकते हैं या ड्रॉप कर सकते हैं। ब्लॉकिंग काम के लिए asyncio.to_thread() (Python) या worker_threads (Node) का इस्तेमाल करें।
• शटडाउन: ElevenLabs WebSocket को साफ़ तरीके से बंद करने के लिए ctx.add_shutdown_callback / ctx.addShutdownCallback रजिस्टर करें। डिफ़ॉल्ट रूप से, आख़िरी नॉन-एजेंट पार्टिसिपेंट के चले जाने पर रूम (और जॉब) समाप्त हो जाता है।

अगले चरण