Metadata-Version: 2.4
Name: voxview
Version: 0.1.0
Summary: 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.
License-Expression: MIT
Project-URL: Homepage, https://github.com/voxview-dev/voxview
Project-URL: Repository, https://github.com/voxview-dev/voxview
Project-URL: Issues, https://github.com/voxview-dev/voxview/issues
Keywords: voice-ai,observability,livekit,vapi,retell,telemetry,fsm
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Libraries
Classifier: Topic :: System :: Monitoring
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: pydantic>=2.0
Requires-Dist: aiohttp>=3.9
Dynamic: license-file

# 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](https://livekit.io) agent sessions (live, in-process metrics). Webhook adapters for [Vapi](https://vapi.ai) and [Retell](https://retellai.com) (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

```bash
pip install -e .
```

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

## Usage

```python
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`:

```python
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:

```python
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:

```python
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
