voxinfra 0.1.0

Production infrastructure for voice agents — memory, tool orchestration, and observability

VoxInfra

Production memory, tool safety, and call replay for LiveKit voice agents.

---

The Problem

• No memory — Your agent forgets users the moment the call ends, making every interaction feel cold and repetitive.
• Latency kills — A single CRM lookup timing out can freeze the audio stream and drop the call.
• No replay — When a call goes wrong, you have no structured trace to debug what the agent heard, recalled, or decided.

---

Install

pip install voxinfra

---

Quick Start

from dotenv import load_dotenv
from livekit.agents import Agent, AgentSession, JobContext, WorkerOptions, cli
from livekit.plugins import deepgram, elevenlabs, openai, silero
from voxinfra import VoxSession, vox_tool

load_dotenv()
vox = VoxSession()  # reads MEM0_API_KEY, SUPABASE_URL from env

@vox_tool(sla_ms=200, fallback={"name": "there"})
async def get_user_info(user_id: str) -> dict:
    return await crm.lookup(user_id)

class MyAgent(Agent):
    async def on_enter(self):
        call_id = str(__import__("uuid").uuid4())
        self.session.userdata["call_id"] = call_id
        vox.obs.start_call(call_id, self.session.userdata["user_id"])

    async def on_exit(self):
        await vox.end_call(self.session.userdata["call_id"])

    async def llm_node(self, chat_ctx, tools, model_settings):
        user_id = self.session.userdata["user_id"]
        call_id = self.session.userdata["call_id"]
        async with vox.turn(user_id=user_id, chat_ctx=chat_ctx, call_id=call_id) as turn:
            await turn.call_tool(get_user_info, user_id)
            return await turn.run_llm(tools, model_settings)

---

VoxMemory

Persistent cross-call memory backed by Mem0 (hosted) or Qdrant (self-hosted). At the start of every turn, VoxInfra semantically searches past conversations and injects the top results into the LLM context — automatically, in 3 lines or fewer to keep voice prompts lean.

# Automatic via vox.turn() — no extra code needed
async with vox.turn(user_id="u-123", chat_ctx=chat_ctx, call_id=call_id) as turn:
    # Memory was recalled and injected before this line
    return await turn.run_llm(tools, model_settings)

# Manual DPDP right-to-erasure
await vox.memory.delete_user("u-123")

---

VoxTools

A registry of async tools with per-tool SLA enforcement. If a tool exceeds its SLA, VoxInfra serves a cached result or fallback — the agent keeps talking and the call never drops.

@vox_tool(sla_ms=200, fallback={"status": "unavailable"})
async def get_appointments(user_id: str) -> dict:
    return await calendar_api.fetch(user_id)

# Inside llm_node
async with vox.turn(...) as turn:
    # If get_appointments takes >200ms, fallback is returned silently
    result = await turn.call_tool(get_appointments, user_id)

The @vox_tool decorator is transparent — it only attaches metadata. The function remains directly callable without going through VoxTools.

---

VoxObs

Structured call traces persisted to a Supabase table (vox_call_traces). Every turn records user transcript, memory recall latency, LLM latency, TTS latency, and all tool call results with SLA breach flags. Use vox.end_call() to finalise and flush the trace.

# In Agent.on_exit
await vox.end_call(self.session.userdata["call_id"])

The Supabase write is always fire-and-forget — it never blocks the call path.

---

Config

All configuration via environment variables or constructor kwargs.

Variable	Description	Required
MEM0_API_KEY	Mem0 hosted API key	Yes (mem0 backend)
QDRANT_URL	Qdrant server URL	Yes (qdrant backend)
QDRANT_API_KEY	Qdrant API key	No
SUPABASE_URL	Supabase project URL	No (obs disabled without it)
SUPABASE_SERVICE_KEY	Supabase service role key	No
VOXINFRA_API_KEY	Reserved for future cloud features	No

# From env (default)
vox = VoxSession()

# Explicit
vox = VoxSession(
    mem0_api_key="m0-...",
    supabase_url="https://xyz.supabase.co",
    supabase_key="service-key",
    default_tool_sla_ms=250,
)

# Qdrant backend (self-hosted)
vox = VoxSession(
    memory_backend="qdrant",
    qdrant_url="http://localhost:6333",
    qdrant_collection="my_agent_memories",
)

---

DPDP Compliance

For deployments subject to India's Digital Personal Data Protection Act, VoxInfra supports a fully self-hosted data path. Use the Qdrant backend pointed at a Mumbai-region instance to ensure all memory vectors stay within Indian jurisdiction. Call traces can be stored in a Supabase project provisioned in the ap-south-1 region. Users have the right to erasure under DPDP Section 13 — invoke it with a single call:

await vox.memory.delete_user(user_id)

This calls Mem0's delete_all or Qdrant's collection filter delete under the hood, removing all stored memories for that user permanently.

---

License

MIT © Ashu Singhania