Deepgram Voice Agent web search

import asyncio
import json
import os

import httpx
import websockets

FUNCTIONS = [
    {
        "name": "search_web",
        "description": (
            "Search the live web for a fact the caller asks about, such as news, prices, opening hours, "
            "schedules, scores or who holds a role. Returns the answer, the page it was read on and the top results."
        ),
        "parameters": {
            "type": "object",
            "properties": {
                "q": {
                    "type": "string",
                    "description": "The question in everyday words, with every name and place spelled out",
                },
            },
            "required": ["q"],
        },
    },
    {
        "name": "read_page",
        "description": "Read the main text of one web page the caller names, or one that search_web returned.",
        "parameters": {
            "type": "object",
            "properties": {
                "url": {"type": "string", "description": "The full address of the page, starting with https://"},
            },
            "required": ["url"],
        },
    },
]

specstream = httpx.AsyncClient(
    base_url="https://api.specstream.ai",
    headers={"Authorization": f"Bearer {os.environ['SPECSTREAM_API_KEY']}"},
    timeout=20,
)


async def search_web(q: str) -> dict:
    r = await specstream.get("/v1/web-search", params={"q": q, "num": 5})
    r.raise_for_status()
    data = r.json()["data"]
    answer = data.get("answer") or {}
    return {
        "answer": answer.get("value"),
        "answer_page": answer.get("page"),
        "results": [
            {"title": s.get("title"), "site": s.get("site"), "snippet": s.get("snippet"), "url": s["url"]}
            for s in data.get("sources", [])[:5]
        ],
    }


async def read_page(url: str) -> dict:
    r = await specstream.post("/v1/extract", json={"urls": [url], "format": "text"}, timeout=40)
    r.raise_for_status()
    page = r.json()["data"]["results"][0]
    if page["status"] != "ok":
        return {"status": page["status"], "next": "Tell the caller the page could not be read."}
    return {"title": page.get("title"), "text": (page.get("text") or "")[:4000]}

HANDLERS = {"search_web": search_web, "read_page": read_page}
running = {}  # FunctionCallRequest id -> its task, so a cancelled call can be stopped


async def answer(ws, call: dict):
    try:
        result = await HANDLERS[call["name"]](**json.loads(call["arguments"]))
    except httpx.TimeoutException:
        result = {"error": "No reply in time. Tell the caller and offer to try again."}
    except httpx.HTTPStatusError as e:
        result = {"error": e.response.text[:300]}
    except httpx.HTTPError as e:
        result = {"error": str(e)}
    finally:
        running.pop(call["id"], None)
    reply = {"type": "FunctionCallResponse", "id": call["id"], "name": call["name"], "content": json.dumps(result)}
    if call.get("thought_signature"):
        reply["thought_signature"] = call["thought_signature"]
    await ws.send(json.dumps(reply))

SETTINGS = {
    "type": "Settings",
    "audio": {
        "input": {"encoding": "linear16", "sample_rate": 16000},
        "output": {"encoding": "linear16", "sample_rate": 16000, "container": "none"},
    },
    "agent": {
        "language": "en",
        "listen": {"provider": {"type": "deepgram", "model": "nova-3"}},
        "think": {
            "provider": {"type": "open_ai", "model": "gpt-4o-mini"},
            "prompt": (
                "You are a voice assistant with live web search. For anything that can change, such as news, "
                "prices, opening hours, schedules, scores or who holds a role, call search_web before you answer. "
                "Answer in one or two short sentences and say which site the answer came from. "
                "Never read a web address aloud. When the caller names a page, call read_page."
            ),
            "functions": FUNCTIONS,
        },
        "speak": {"provider": {"type": "deepgram", "model": "aura-2-thalia-en"}},
        "greeting": "Hello. What would you like to know?",
    },
}


async def keep_alive(ws):
    while True:  # no microphone in this test, so the socket needs a KeepAlive every 8 seconds
        await asyncio.sleep(8)
        await ws.send(json.dumps({"type": "KeepAlive"}))


async def type_to_agent(ws):
    while True:
        text = await asyncio.to_thread(input, "you> ")
        await ws.send(json.dumps({"type": "InjectUserMessage", "content": text}))


async def main():
    async with websockets.connect(
        "wss://agent.deepgram.com/v1/agent/converse",
        additional_headers={"Authorization": f"Token {os.environ['DEEPGRAM_API_KEY']}"},
    ) as ws:
        await ws.send(json.dumps(SETTINGS))
        async for message in ws:
            if isinstance(message, bytes):
                continue  # the agent's voice; play it as Deepgram's Build a Voice Agent guide does
            event = json.loads(message)
            if event["type"] == "SettingsApplied":
                asyncio.create_task(keep_alive(ws))
                asyncio.create_task(type_to_agent(ws))
            elif event["type"] == "ConversationText":
                print(f"{event['role']}: {event['content']}")
            elif event["type"] == "FunctionCallRequest":
                for call in event["functions"]:
                    if call["client_side"]:
                        running[call["id"]] = asyncio.create_task(answer(ws, call))
            elif event["type"] == "FunctionCallCancelled":
                for call in event["functions"]:
                    task = running.pop(call["id"], None)
                    if task:
                        task.cancel()


asyncio.run(main())