Metadata-Version: 2.4
Name: pipewatchdev
Version: 0.1.0
Summary: Observability for voice AI pipelines — trace the handoffs your dashboards can't see.
Author-email: Pipewatch <hello@pipewatch.dev>
License: MIT
Project-URL: Homepage, https://pipewatch.dev
Project-URL: Repository, https://github.com/kollaikal-rupesh/pipewatch
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Topic :: Software Development :: Libraries
Classifier: Programming Language :: Python :: 3
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: httpx>=0.25.0
Provides-Extra: dev
Requires-Dist: deepgram-sdk>=3.0.0; extra == "dev"
Requires-Dist: openai>=1.0.0; extra == "dev"
Requires-Dist: elevenlabs>=1.0.0; extra == "dev"
Requires-Dist: pytest>=7.0; extra == "dev"
Dynamic: license-file

# Pipewatch

**Observability for voice AI pipelines. See the failures nobody else can.**

> 80% of voice AI conversation failures happen in the handoffs between providers — not inside them. Pipewatch traces the invisible gaps.

## The Problem

Voice AI teams chain 3+ services for every conversation — ASR, LLM, TTS. Each provider says they responded in 200ms, but the user waited 2 seconds. The time disappears in the handoffs: buffering between Deepgram and OpenAI, WebSocket reconnects before Cartesia, state serialization nobody ever profiles. **These gaps are invisible to every existing observability tool.** Pipewatch turns on the lights.

## Setup

### Prerequisites

- Python 3.10+
- Node.js 18+
- API keys for at least one provider (Deepgram, OpenAI, Cartesia, or ElevenLabs)

### 1. Install the SDK

**From PyPI** (recommended for most users — instruments your app, ships traces to a Pipewatch backend):

```bash
pip install pipewatchdev
```

The package installs as `pipewatchdev` but imports as `pipewatch`:

```python
import pipewatch
pipewatch.init(endpoint="http://localhost:8000/api/traces")
```

**From source** (for running the full stack locally — SDK + backend + dashboard — or contributing):

```bash
git clone https://github.com/kollaikal-rupesh/pipewatch.git
cd pipewatch
python3 -m venv venv
source venv/bin/activate
pip install -e .
```

### 2. Start the backend

```bash
pip install fastapi uvicorn[standard] sqlalchemy[asyncio] aiosqlite pydantic python-dotenv python-multipart
PYTHONPATH=. python3 -m uvicorn server.main:app --host 127.0.0.1 --port 8000
```

The backend creates a SQLite database automatically on first run. No configuration needed.

### 3. Start the dashboard

```bash
cd dashboard
npm install
npm run dev
```

Open http://localhost:5173 — the dashboard is live.

### 4. Add your API keys

Create a `.env` file in the project root:

```
DEEPGRAM_API_KEY=your-key-here
OPENAI_API_KEY=your-key-here
CARTESIA_API_KEY=your-key-here
```

> These three are the providers Pipewatch has been tested end-to-end with. Any other supported provider (ElevenLabs, Silero via LiveKit, or a custom interceptor) works the same way — just drop its key into `.env` and it'll be auto-instrumented. See [Supported Providers](#supported-providers) for the full list.

### 5. Run the live demo

```bash
source venv/bin/activate
pip install deepgram-sdk openai cartesia
PYTHONPATH=. python3 scripts/demo_live.py
```

This runs a real 3-turn voice conversation through Deepgram (ASR) → OpenAI (LLM) → Cartesia (TTS). Every call is auto-instrumented. Open http://localhost:5173/call/live-demo-001 to see the trace with the call player.

## Instrument Your Own Pipeline

Add two lines to the top of your voice AI app:

```python
import pipewatch
pipewatch.init(endpoint="http://localhost:8000/api/traces")
```

Then wrap your pipeline in a trace:

```python
with pipewatch.trace("call-123") as t:
    transcript = dg.listen.v1.media.transcribe_file(request=audio, model="nova-3")
    reply = oai.chat.completions.create(model="gpt-4o", messages=[...])
    audio_chunks = cart.tts.bytes(model_id="sonic-2", transcript=reply, voice=voice, output_format=fmt)
```

That's it. Deepgram, OpenAI, ElevenLabs, and Cartesia calls are auto-traced. Handoff gaps between providers are computed automatically. Traces ship to the backend in a background thread — zero added latency.

### LiveKit Integration

If your pipeline uses LiveKit's agent framework, use the LiveKit adapter instead of the SDK monkey-patching:

```python
from pipewatch_livekit import register_pipewatch

# In your LiveKit agent entrypoint, after creating the AgentSession:
register_pipewatch(session, api_key="pw_test", endpoint="http://localhost:8000/api/traces", session_metadata={
    "agent_id": your_agent_id,
    "organization_id": your_org_id,
    "room_name": ctx.room.name,
})
```

This hooks into LiveKit's built-in metrics system (`STTMetrics`, `LLMMetrics`, `TTSMetrics`) and conversation events to capture per-component latency, transcripts, and tool calls. No monkey-patching needed — works with any provider plugin (Deepgram, OpenAI, Cartesia, ElevenLabs, Silero VAD).

### Audio Capture (Optional)

To capture raw audio bytes flowing through the pipeline:

```python
pipewatch.init(
    endpoint="http://localhost:8000/api/traces",
    capture_audio=True,              # opt-in
    audio_store_path="/tmp/pipewatch_audio",  # where to store locally
    audio_max_storage_mb=500,        # LRU eviction limit
)
```

When enabled, the SDK captures audio going into ASR (what the user said) and audio coming out of TTS (what the user heard). Audio files are stored locally and uploaded to the backend in the background.

## What You'll See

### Overview Dashboard
Top-level metrics: conversation count, average latency, handoff percentage, quality scores, error rate. Line charts for latency trends. Regression alerts when quality drops.

### Call Player (`/call/:id`)
The hero view. A synced call player with:
- **Waveform** with moving playhead (Space to play/pause, arrows to skip)
- **Transcript** with auto-scrolling chat bubbles (caller left, agent right)
- **Pipeline X-Ray** showing the waterfall breakdown for each turn — ASR, handoff, LLM, handoff, TTS with proportional bars and timing
- **Sidebar** with issues, pipeline stats, eval scores, system prompt

### Trace Detail (`/trace/:id`)
Waterfall timeline of every span in a conversation. Click any span for full details (input/output text, tokens, errors). Handoff analysis card showing total gap time and which handoff was longest.

### Evals (`/evals`)
Quality score distribution, trend over time, top failure reasons, low-quality conversations linking to trace detail.

## Key Features

- **Pipeline traces** — End-to-end latency breakdown across ASR → LLM → TTS with per-span timing, payloads, and errors
- **Handoff detection** — Automatically measures the invisible gaps between providers
- **Call player** — Synced audio playback with transcript and pipeline X-ray
- **Automated evals** — LLM-as-judge rates every conversation on quality, naturalness, and task completion
- **Regression alerts** — Instant detection when conversation quality drops >15% or latency increases >20%
- **Audio capture** — Opt-in recording of ASR input and TTS output with silence detection and clipping analysis
- **Provider-agnostic** — Works across Deepgram, OpenAI, ElevenLabs, Cartesia, and any custom provider

## Supported Providers

| Stage | Provider | SDK | LiveKit Plugin |
|-------|----------|-----|----------------|
| ASR | Deepgram | Yes | Yes |
| LLM | OpenAI | Yes | Yes |
| TTS | ElevenLabs | Yes | Yes |
| TTS | Cartesia | Yes | Yes |
| VAD | Silero | — | Yes |

Adding a new provider interceptor takes ~50 lines. See `pipewatch/interceptors/` for examples.

## Architecture

```
┌─────────────┐     ┌─────────────────┐     ┌───────────────┐
│  Your Voice  │     │   Pipewatch     │     │   Dashboard   │
│  AI App      │────▶│   Backend       │────▶│   (React)     │
│  + SDK       │     │   (FastAPI)     │     │               │
└─────────────┘     └─────────────────┘     └───────────────┘
```

**SDK** (`pipewatch/`) — Zero-config monkey-patching. Wraps provider calls at runtime. Computes handoff spans. Ships traces non-blocking in a background thread.

**LiveKit Adapter** (`pipewatch_livekit.py`) — Event-driven integration for LiveKit AgentSession. Captures STT/LLM/TTS metrics, transcripts, and tool calls without monkey-patching.

**Backend** (`server/`) — FastAPI + async SQLAlchemy. Trace ingestion with upsert support (handles incremental updates). Aggregate metrics, regression detection, LLM-as-judge evals, audio storage with auto-analysis.

**Dashboard** (`dashboard/`) — React + Tailwind + Recharts. Overview metrics, call player with synced waveform/transcript/X-ray, trace waterfall, eval tracking.

## API Endpoints

| Method | Endpoint | Description |
|--------|----------|-------------|
| POST | `/api/traces` | Ingest trace batches (supports upsert) |
| GET | `/api/traces` | List traces with filters and pagination |
| GET | `/api/traces/:id` | Full trace with spans and evals |
| GET | `/api/metrics` | Aggregate metrics (configurable time window) |
| GET | `/api/metrics/regression` | Regression detection (24h vs 7-day baseline) |
| POST | `/api/evals/run/:id` | Trigger LLM-as-judge evaluation |
| GET | `/api/evals` | List evaluations with filters |
| POST | `/api/audio/upload` | Upload audio blob with auto-analysis |
| GET | `/api/audio/:span_id/:type` | Stream audio for playback |
| GET | `/api/audio/:span_id/waveform` | 128-point amplitude array for visualization |

## Configuration

### Backend

| Env Var | Default | Description |
|---------|---------|-------------|
| `DATABASE_URL` | `sqlite+aiosqlite:///./pipewatch.db` | Database connection string |
| `CORS_ORIGINS` | `["http://localhost:3000","http://localhost:5173"]` | Allowed CORS origins (JSON array) |
| `AUDIO_STORAGE_PATH` | `./audio_storage` | Directory for audio blob storage |
| `ANTHROPIC_API_KEY` | — | Required for LLM-as-judge evals |

### SDK

```python
pipewatch.init(
    endpoint="http://localhost:8000/api/traces",  # Backend URL
    api_key="pw_...",               # API key (optional, no auth yet)
    capture_audio=False,            # Opt-in audio capture
    audio_store_path="/tmp/pipewatch_audio",
    audio_max_storage_mb=500,       # LRU eviction limit
    batch_size=10,                  # Traces per HTTP batch
    flush_interval_s=5.0,           # Max seconds between flushes
    debug=False,                    # Debug logging
)
```

## Status

Early development. Looking for design partners — if you deploy voice AI and want better visibility into your pipeline, reach out: **kollaikalrupesh@gmail.com**

## License

MIT
