Custom LLM integration

$
pip install "elevenlabs" "aiohttp" "twilio" "python-dotenv"

1
from elevenlabs import AsyncElevenLabs
2
3
elevenlabs = AsyncElevenLabs(api_key=os.environ["ELEVENLABS_API_KEY"])
4
5
async def signed_url() -> str:
6
    response = await elevenlabs.conversational_ai.conversations.get_signed_url(
7
        agent_id=os.environ["SPEECH_ENGINE_ID"],
8
    )
9
    return response.signed_url

1
from aiohttp import web
2
from twilio.request_validator import RequestValidator
3
4
validator = RequestValidator(os.environ["TWILIO_AUTH_TOKEN"])
5
6
7
async def incoming_call(request: web.Request) -> web.Response:
8
    form = await request.post()
9
    signature = request.headers.get("X-Twilio-Signature", "")
10
    url = str(request.url)
11
    if not validator.validate(url, dict(form), signature):
12
        return web.Response(status=403, text="forbidden")
13
14
    host = request.headers.get("X-Forwarded-Host") or request.host
15
    twiml = (
16
        '<?xml version="1.0" encoding="UTF-8"?>'
17
        "<Response><Connect>"
18
        f'<Stream url="wss://{host}/media-stream"/>'
19
        "</Connect></Response>"
20
    )
21
    return web.Response(text=twiml, content_type="text/xml")

1
import asyncio
2
import json
3
4
import aiohttp
5
from aiohttp import web
6
7
8
async def media_stream(request: web.Request) -> web.WebSocketResponse:
9
    twilio_ws = web.WebSocketResponse()
10
    await twilio_ws.prepare(request)
11
12
    stream_sid: str | None = None
13
    el_session: aiohttp.ClientSession | None = None
14
    el_ws: aiohttp.ClientWebSocketResponse | None = None
15
    pump_task: asyncio.Task | None = None
16
17
    async def pump_el_to_twilio(el: aiohttp.ClientWebSocketResponse):
18
        async for msg in el:
19
            if msg.type != aiohttp.WSMsgType.TEXT:
20
                continue
21
            event = json.loads(msg.data)
22
            etype = event.get("type")
23
            if etype == "audio":
24
                await twilio_ws.send_str(json.dumps({
25
                    "event": "media",
26
                    "streamSid": stream_sid,
27
                    "media": {"payload": event["audio_event"]["audio_base_64"]},
28
                }))
29
            elif etype == "interruption":
30
                await twilio_ws.send_str(json.dumps({
31
                    "event": "clear",
32
                    "streamSid": stream_sid,
33
                }))
34
            elif etype == "ping":
35
                event_id = event.get("ping_event", {}).get("event_id")
36
                await el.send_str(json.dumps({
37
                    "type": "pong", "event_id": event_id,
38
                }))
39
40
    try:
41
        async for msg in twilio_ws:
42
            if msg.type != aiohttp.WSMsgType.TEXT:
43
                continue
44
            event = json.loads(msg.data)
45
46
            if event["event"] == "start":
47
                stream_sid = event["start"]["streamSid"]
48
                el_session = aiohttp.ClientSession()
49
                el_ws = await el_session.ws_connect(await signed_url())
50
                await el_ws.send_str(json.dumps({
51
                    "type": "conversation_initiation_client_data",
52
                }))
53
                pump_task = asyncio.create_task(pump_el_to_twilio(el_ws))
54
55
            elif event["event"] == "media" and el_ws is not None:
56
                await el_ws.send_str(json.dumps({
57
                    "user_audio_chunk": event["media"]["payload"],
58
                }))
59
60
            elif event["event"] == "stop":
61
                break
62
    finally:
63
        if pump_task:
64
            pump_task.cancel()
65
        if el_ws and not el_ws.closed:
66
            await el_ws.close()
67
        if el_session and not el_session.closed:
68
            await el_session.close()
69
70
    return twilio_ws

1
import os
2
3
from elevenlabs import AsyncElevenLabs
4
5
elevenlabs = AsyncElevenLabs(api_key=os.environ["ELEVENLABS_API_KEY"])
6
SHARED_SECRET = os.environ["SHARED_SECRET"]
7
8
9
async def brain_ws(request: web.Request) -> web.WebSocketResponse:
10
    if request.headers.get("x-api-key") != SHARED_SECRET:
11
        return web.Response(status=401, text="unauthorized")
12
13
    ws = web.WebSocketResponse()
14
    await ws.prepare(request)
15
16
    engine = await elevenlabs.speech_engine.get(os.environ["SPEECH_ENGINE_ID"])
17
    session = engine.create_session(ws)
18
19
    async def on_transcript(transcript):
20
        # Replace this with your own LLM call; see the quickstart.
21
        await session.send_response("Hello, you've reached the demo.")
22
23
    session.on("user_transcript", on_transcript)
24
    await session.run()
25
    return ws
26
27
28
def make_app() -> web.Application:
29
    app = web.Application()
30
    app.router.add_post("/incoming-call", incoming_call)
31
    app.router.add_get("/media-stream", media_stream)
32
    app.router.add_get("/ws", brain_ws)
33
    return app
34
35
36
if __name__ == "__main__":
37
    web.run_app(make_app(), port=3001)

1
await elevenlabs.speech_engine.update(
2
    speech_engine_id="seng_8k3m9xr4hjnfg983brhmhkd98n6",
3
    speech_engine={"ws_url": "wss://abc123.ngrok.io/ws"},
4
)