voxview 0.1.0

VoxView — observability and safety layer for voice AI agents. Per-turn latency, FSM task-completion auditing, and Panic Button guardrails across LiveKit, Vapi, and Retell.

VoxView

Observability and safety layer for voice AI agents — the metrics generic LLM-observability tools don't track.

Package name is voxview on PyPI; the importable module is agentlens (from agentlens import CallRecorder, Outcome) — internal history, not a typo.

What it is

A small Python package that records what actually happened in every voice AI call: per-turn latency broken down by stage, network health, barge-in detection, and task-completion auditing against a state machine you define — not a guess based on transcript text.

Native support for LiveKit agent sessions (live, in-process metrics). Webhook adapters for Vapi and Retell (see Platform Support below — these platforms don't expose the same data LiveKit does, and this README says exactly where the gaps are, not just where it works).

Platform Support

Capability	LiveKit	Vapi	Retell
Transcript + turn order	✅	✅	✅
Call outcome	✅	✅	✅
Per-turn STT/LLM-TTFT/TTS-TTFB latency	✅ (native session metrics)	❌ not exposed via webhook	❌ not exposed via webhook
Network health (packet loss/jitter/RTT)	✅	❌ not exposed via webhook	❌ not exposed via webhook
Barge-in detection	✅	✅ (user-interrupted event)	❌ not exposed via webhook
FSM task-completion auditing	✅ (tool calls)	✅ (tool-calls event, same principle)	⚠️ possible via transcript_with_tool_calls, not yet wired
Integration shape	live session hooks (attach())	stateful webhook accumulator (VapiCallAccumulator)	one-shot webhook parser (parse_retell_webhook())

Why the gaps exist: LiveKit gives you a live, in-process AgentSession with direct access to STT/LLM/TTS provider metrics as they happen. Vapi and Retell are hosted platforms — you only see what they choose to send in a webhook payload, and as of writing (2026-07), neither exposes granular per-stage latency there. If either platform adds this to their webhook or a detailed-metrics API in the future, the adapters here are the place to wire it in — the schema (LatencyMetrics) already has the fields, they're just left null for these two platforms today.

What it tracks

• Per-turn latency breakdown — stt_latency_ms, llm_ttft_ms, tts_ttfb_ms, eou_delay_ms, total_turn_latency_ms
• Network health — packet_loss_pct, jitter_ms, rtt_ms
• Barge-in detection — distinguishes pacing interjections ("mm-hmm") from real corrections
• Task-completion auditing — define an expected state sequence (FSM) per workflow; the recorder checks which states your agent's own tool calls actually reached, in order — ground truth from function calls, not transcript parsing
• Guardrail flagging — a place to record your own policy violations (e.g. an agent quoting an unapproved price) and have them roll up into hallucination_flagged
• Golden dataset candidates — calls that failed their task or had 3+ interruptions get auto-flagged, so you have a starting point for building an eval set

Why not just use Langfuse / LangSmith / Arize?

Those tools are excellent for text-based LLM chains. Voice agents have a different failure surface: latency is split across four distinct stages (not one LLM call), network conditions affect call quality independent of the model, and "did the conversation succeed" isn't answerable from the transcript alone — it needs a state-machine audit against real tool calls. VoxView is built specifically for that shape of problem, with 8 pre-built FSM starting points for the highest-volume voice AI verticals (staffing, insurance, property management, healthcare, financial services, retail, legal, home services) baked into schemas.py.

Install

pip install -e .

(PyPI package coming once the API stabilizes — for now, install from source.)

Usage

from agentlens import CallRecorder, Outcome

recorder = CallRecorder(
    call_id=ctx.room.name,
    client_id="acme-corp",
    niche="support",
    workflow_type="your_workflow_name",
    worker="agent-1",
    metadata={"ticket_id": "1234"},
)
recorder.attach(session)  # session = your LiveKit AgentSession

# From your agent's tool/function handlers, as real state transitions happen:
recorder.mark_state("call_connected")
recorder.mark_state("info_collected")

await session.wait_for_disconnect()
await recorder.flush(outcome=Outcome.COMPLETED)

Define your own workflows in agentlens/schemas.py:

FSM_DEFINITIONS: dict[str, list[str]] = {
    "your_workflow_name": [
        "call_connected",
        "info_collected",
        "action_confirmed",
    ],
}

See examples/usage_example.py for a fuller wiring example, and sql/schema.sql for the Postgres/Supabase tables CallRecorder.flush() writes to.

Vapi

Vapi sends multiple webhook messages across a call's lifetime, so you accumulate them per call and get a finished report when end-of-call-report arrives:

from agentlens.adapters.vapi import VapiCallAccumulator

# Keep one accumulator per in-flight call_id (e.g. in a dict keyed by call.id)
accumulator = VapiCallAccumulator(
    call_id=call_id, client_id="acme-corp",
    niche="support", workflow_type="your_workflow_name", worker="agent-1",
)

# In your Vapi Server URL webhook handler, for every incoming POST body:
report = accumulator.ingest(payload)
if report is not None:
    # This was the end-of-call-report message — report is finalized.
    # Write it to Supabase yourself (same call_logs/call_turns shape as
    # CallRecorder._write — copy that method if you want the same sink).
    ...

Retell

Retell sends the full call object on call_ended/call_analyzed, so parsing is one call, no accumulator needed:

from agentlens.adapters.retell import parse_retell_webhook

# In your Retell webhook handler, after verifying the signature:
if payload["event"] in ("call_ended", "call_analyzed"):
    report = parse_retell_webhook(
        payload, niche="support", workflow_type="your_workflow_name", worker="agent-1",
    )
    if report:
        ...  # write to your sink

Configuration

Set two environment variables:

SUPABASE_URL=https://your-project.supabase.co
SUPABASE_SERVICE_KEY=your-service-role-key

If unset, flush() logs a warning and drops telemetry instead of failing the call — a broken observability layer should never take down a live call.

Status

Early — built to solve a real problem in production, extracted here to be reusable. Expect the API to evolve. Issues and PRs welcome.

License

MIT