Twilio + Deepgram Voice Agent

[fidelzky-bot]

I am using Deepgram Agent Voice now on my Twilio App.

This is my code sample:

import asyncio
import base64
import json
import sys
import websockets
import ssl
import os

def sts_connect():
# you can run export DEEPGRAM_API_KEY="your key" in your terminal to set your API key.
api_key = os.getenv('DEEPGRAM_API_KEY')
if not api_key:
raise ValueError("DEEPGRAM_API_KEY environment variable is not set")

sts_ws = websockets.connect(
    "wss://agent.deepgram.com/v1/agent/converse",
    subprotocols=["token", api_key]
)
return sts_ws

async def twilio_handler(twilio_ws):
audio_queue = asyncio.Queue()
streamsid_queue = asyncio.Queue()

async with sts_connect() as sts_ws:
    config_message = {
        "type": "Settings",
        "audio": {
            "input": {
                "encoding": "mulaw",
                "sample_rate": 8000,
            },
            "output": {
                "encoding": "mulaw",
                "sample_rate": 8000,
                "container": "none",
            },
        },
        "agent": {
            "language": "en",
            "listen": {
                "provider": {
                    "type": "deepgram",
                    "model": "nova-3",
                    "keyterms": ["hello", "goodbye"]
                }
            },
            "think": {
                "provider": {
                    "type": "open_ai",
                    "model": "gpt-4o-mini",
                    "temperature": 0.7
                },
                "prompt": (
                    "You are a helpful voice assistant created by Deepgram for a law firm. Your role is to assist with new client call intakes. "
                    "Your responses should be friendly, human-like, and conversational. Always keep your answers concise, limited to 1–2 sentences and no more than 120 characters.\n\n"
                    "When responding to a user's message, follow these guidelines:\n"
                    "- Wait for the caller's response before asking the next question.\n"
                    "- Use their answers naturally in your replies to build a human-like connection.\n"
                    "- Avoid repeating the same phrase. Keep each response slightly unique.\n"
                    "- If a question is unclear, politely ask for clarification.\n"
                    "- Ask only one question at a time.\n"
                    "- If the user's message is empty or silent, respond with an empty message.\n"
                    "- If asked about your well-being, respond briefly and naturally.\n"
                    "- Remember that you have a voice interface — speak clearly and warmly.\n\n"
                    "☎️ Sample Conversation Script\n"
                    "AI: Thank you for calling [Your Law Firm Name]. Are you calling about a new legal matter?\n"
                    "If Caller says No: Alright, I'll transfer you to our main office line now.\n"
                    "If Caller says Yes: Great! I just need to get a few details from you. This won't take long—I'll make it easy.\n"
                    "AI: What's your name?\n"
                    "Caller: John Rivera\n"
                    "AI: Thanks, John. Nice to meet you!\n"
                    "AI: What's the best phone number to reach you?\n"
                    "Caller: 312-555-0199\n"
                    "AI: Got it. Thanks!\n"
                    "AI: What's your email address?\n"
                    "Caller: jrivera@email.com\n"
                    "AI: Thank you!\n"
                    "AI: When did the incident happen?\n"
                    "Caller: June 15\n"
                    "AI: Okay, noted.\n"
                    "AI: Were you injured? Can you tell me what happened?\n"
                    "Caller: I got hit by a car crossing the street.\n"
                    "AI: I'm really sorry to hear that. Thank you for sharing that with me.\n"
                    "AI: Did you go to the hospital or get medical treatment?\n"
                    "Caller: Yes, I went to the ER.\n"
                    "AI: Thank you. That helps.\n"
                    "AI: Do you know who was at fault?\n"
                    "Caller: The driver ran a red light.\n"
                    "AI: Understood.\n"
                    "AI: Was there a police report?\n"
                    "Caller: Yes, and I have a copy.\n"
                    "AI: That's great. Thanks for confirming.\n"
                    "AI: Do you know if the other party had insurance?\n"
                    "Caller: I think so, yes.\n"
                    "AI: Okay, got it.\n"
                    "AI: Have you signed anything from an insurance company or lawyer?\n"
                    "Caller: Not yet.\n"
                    "AI: Alright, thanks.\n"
                    "AI (Final Message): Thanks, John. I think we can help you. Please hold while I transfer you to an attorney."
                )
            },
            "speak": {
                "provider": {
                    "type": "deepgram",
                    "model": "aura-2-thalia-en"
                }
            },
            "greeting": "Thank you for calling The Illinois Hammer. Are you calling about a new case?"
    await sts_ws.send(json.dumps(config_message))

    async def sts_sender(sts_ws):
        print("sts_sender started")
        while True:
            chunk = await audio_queue.get()
            await sts_ws.send(chunk)

    async def sts_receiver(sts_ws):
        print("sts_receiver started")
        # we will wait until the twilio ws connection figures out the streamsid
        streamsid = await streamsid_queue.get()
        # for each sts result received, forward it on to the call
        async for message in sts_ws:
            if type(message) is str:
                print(message)
                # handle barge-in
                decoded = json.loads(message)
                if decoded['type'] == 'UserStartedSpeaking':
                    clear_message = {
                        "event": "clear",
                        "streamSid": streamsid
                    }
                    await twilio_ws.send(json.dumps(clear_message))

                continue

            print(type(message))
            raw_mulaw = message

            # construct a Twilio media message with the raw mulaw (see https://www.twilio.com/docs/voice/twiml/stream#websocket-messages---to-twilio)
            media_message = {
                "event": "media",
                "streamSid": streamsid,
                "media": {"payload": base64.b64encode(raw_mulaw).decode("ascii")},
            }

            # send the TTS audio to the attached phonecall
            await twilio_ws.send(json.dumps(media_message))

    async def twilio_receiver(twilio_ws):
        print("twilio_receiver started")
        # twilio sends audio data as 160 byte messages containing 20ms of audio each
        # we will buffer 20 twilio messages corresponding to 0.4 seconds of audio to improve throughput performance
        BUFFER_SIZE = 20 * 160

        inbuffer = bytearray(b"")
        async for message in twilio_ws:
            try:
                data = json.loads(message)
                if data["event"] == "start":
                    print("got our streamsid")
                    start = data["start"]
                    streamsid = start["streamSid"]
                    streamsid_queue.put_nowait(streamsid)
                if data["event"] == "connected":
                    continue
                if data["event"] == "media":
                    media = data["media"]
                    chunk = base64.b64decode(media["payload"])
                    if media["track"] == "inbound":
                        inbuffer.extend(chunk)
                if data["event"] == "stop":
                    break

                # check if our buffer is ready to send to our audio_queue (and, thus, then to sts)
                while len(inbuffer) >= BUFFER_SIZE:
                    chunk = inbuffer[:BUFFER_SIZE]
                    audio_queue.put_nowait(chunk)
                    inbuffer = inbuffer[BUFFER_SIZE:]
            except:
                break

    # the async for loop will end if the ws connection from twilio dies
    # and if this happens, we should forward an some kind of message to sts
    # to signal sts to send back remaining messages before closing(?)
    # audio_queue.put_nowait(b'')

    await asyncio.wait(
        [
            asyncio.ensure_future(sts_sender(sts_ws)),
            asyncio.ensure_future(sts_receiver(sts_ws)),
            asyncio.ensure_future(twilio_receiver(twilio_ws)),
        ]
    )

    await twilio_ws.close()

async def router(websocket, path):
print(f"Incoming connection on path: {path}")
if path == "/twilio":
print("Starting Twilio handler")
await twilio_handler(websocket)

def main():
# use this if using ssl
# ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
# ssl_context.load_cert_chain('cert.pem', 'key.pem')
# server = websockets.serve(router, '0.0.0.0', 443, ssl=ssl_context)

# use this if not using ssl
server = websockets.serve(router, "localhost", 5000)
print("Server starting on ws://localhost:5000")

asyncio.get_event_loop().run_until_complete(server)
asyncio.get_event_loop().run_forever()

if name == "main":
sys.exit(main() or 0)

Is there any way to make the AI respond faster? Right now, when I call my Twilio number with Deepgram's AI Agent, it takes around 4 to 5 seconds to reply. I’m hoping there’s a way to reduce the response time to 1 to 2 seconds or less.

I am using Fly.io to deploy my App.

[deepgram-community[bot]]

Thanks for asking your question. Please be sure to reply with as much detail as possible so the community can assist you efficiently.
_{Consider joining our Discord community for more opportunity to engage with your fellow Deepgram users. You can earn points which can be redeemed for cool stuff by being active in our communities!}

[deepgram-community[bot]]

Hey there! It looks like you haven't connected your GitHub account to your Deepgram account. You can do this at https://community.deepgram.com - being verified through this process will allow our team to help you in a much more streamlined fashion.

[deepgram-community[bot]]

It looks like we're missing some important information to help debug your issue. Would you mind providing us with the following details in a reply?

• The programming language you are working in (e.g. JavaScript, Python).
• A request ID that triggered your error or issue.