Metadata-Version: 2.4
Name: streamguard
Version: 0.2.0
Summary: CPU-native realtime voice stream integrity middleware
Author: streamguard contributors
License-Expression: MIT
Project-URL: Homepage, https://github.com/jempf/streamguard
Project-URL: Documentation, https://github.com/jempf/streamguard#readme
Project-URL: Repository, https://github.com/jempf/streamguard
Project-URL: Issues, https://github.com/jempf/streamguard/issues
Keywords: tts,voice,guardrail,speech,realtime,audio,streaming,middleware
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Topic :: Multimedia :: Sound/Audio :: Speech
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: numpy>=1.24
Requires-Dist: scipy>=1.10
Provides-Extra: speaker
Requires-Dist: resemblyzer; extra == "speaker"
Provides-Extra: semantic
Requires-Dist: openai-whisper; extra == "semantic"
Requires-Dist: langdetect; extra == "semantic"
Provides-Extra: all
Requires-Dist: resemblyzer; extra == "all"
Requires-Dist: openai-whisper; extra == "all"
Requires-Dist: langdetect; extra == "all"
Provides-Extra: dev
Requires-Dist: pytest>=7; extra == "dev"
Requires-Dist: pytest-asyncio>=0.23; extra == "dev"
Requires-Dist: pytest-cov; extra == "dev"
Requires-Dist: ruff; extra == "dev"
Requires-Dist: build; extra == "dev"
Requires-Dist: twine; extra == "dev"
Dynamic: license-file

<div align="center">

# streamguard

**Realtime voice stream integrity middleware for AI voice pipelines.**

CPU-native · async-first · zero model downloads · works anywhere

[![PyPI version](https://img.shields.io/pypi/v/streamguard.svg)](https://pypi.org/project/streamguard/)
[![Python 3.10+](https://img.shields.io/badge/python-3.10+-blue.svg)](https://www.python.org/)
[![License: MIT](https://img.shields.io/badge/License-MIT-green.svg)](LICENSE)

</div>

---

## What it does

LLM-based TTS systems can silently produce broken audio:

| Problem | Example |
|---|---|
| Clipping / silence collapse | Audio rail-saturates or drops out mid-utterance |
| Spectral corruption | Codec artifact, noise burst, garbled synthesis |
| Speaker identity drift | Voice clone drifts mid-session or switches speaker |
| Duplicate chunks | Same audio frame sent twice in a streaming pipeline |
| Stream discontinuity | Abrupt spectral jump between chunks (reset, decoder collapse) |
| Language / accent mismatch | English prosody on German text *(optional)* |
| Gibberish / unintelligible | CER too high against expected text *(optional)* |

Existing tools catch **what** is said (LLM observers, text filters).  
**streamguard** catches **how** it is said — before the audio reaches the speaker.

---

## Install

```bash
pip install streamguard && streamguard doctor
```

| Profile | Command | What you get |
|---------|---------|----------------|
| **Core** (recommended) | `pip install streamguard` | Clipping, spectral, speaker (MFCC), ~10 ms inspect — no downloads |
| **Speaker+** | `pip install "streamguard[speaker]"` | ECAPA embeddings via resemblyzer |
| **Semantic** | `pip install "streamguard[semantic]"` | Whisper ASR + language mismatch |
| **Full** | `pip install "streamguard[all]"` | All optional checkers |

```bash
streamguard doctor   # verify install + smoke inspect
streamguard demo     # PASS vs silence in cascading mode
streamguard init --integration kokoro   # streamguard.toml + integration snippet
python -m streamguard doctor
```

Also works with **uv**: `uv add streamguard` then `uv run streamguard doctor`.

```bash
streamguard init --integration livekit   # LiveKit hook snippet
streamguard init --integration pipecat   # Pipecat processor snippet
streamguard init --integration kokoro --policy cascading --sample-rate 24000
```

---

## Quick start

### Choose your integration mode

| Mode | When to use | API |
|------|-------------|-----|
| **Cascading** | Each TTS call returns a full buffer (`kokoro.create()`, WAV file) | `AudioGuard.for_utterance()` + `inspect()` / `inspect_utterance()` |
| **Streaming** | Chunks are sequential slices of one live synthesis | `AudioGuard.for_streaming()` + `inspect_chunk()` with shared `StreamState` |

Do **not** reuse `StreamState` across unrelated `kokoro.create()` calls — continuity will false-positive on sentence boundaries.

### Cascading pipeline (offline / sentence-at-a-time TTS)

```python
from streamguard import AudioGuard, Action

guard = AudioGuard.for_utterance(sample_rate=24_000)  # Kokoro native rate

audio = your_tts(text)                          # numpy float32 array
decision = guard.inspect(audio, expected_text=text)

if decision.action == Action.PASS:
    play(audio)
elif decision.action == Action.RETRY:
    play(your_tts(text))                        # regenerate
elif decision.action == Action.HALT:
    alert("synthesis failed")

print(decision.format_summary())  # per-checker scores + fused score
print(decision.triggering_signals)  # e.g. ["clipping"] when CRITICAL
```

See `examples/kokoro_cascading.py` for a minimal runnable example.

### LiveKit Agents (streaming hook)

```python
from streamguard import AudioGuard, StreamState
from streamguard.integrations.livekit import LiveKitGuardHook

guard = AudioGuard.for_streaming(sample_rate=24_000)
hook = LiveKitGuardHook(guard, StreamState.fresh(sample_rate=24_000))

async for chunk in hook(tts_chunk_generator(), expected_text=text):
    await play(chunk)
```

Runnable demo: `python examples/livekit_minimal.py`

### Pipecat (between TTS and transport output)

```python
from streamguard.integrations.pipecat import create_pipecat_processor

processor = create_pipecat_processor()  # pip install pipecat-ai
pipeline = Pipeline([..., tts, processor, transport.output()])
```

Or inspect one frame without a processor: `inspect_pipecat_chunk(guard, state, frame.audio, sample_rate=frame.sample_rate)`  
Runnable demo: `python examples/pipecat_minimal.py`

### Full-duplex streaming (async)

```python
from streamguard import AudioGuard, Action, StreamState

guard = AudioGuard.for_streaming()
state = StreamState.fresh()                     # one per call / connection

async for chunk in tts_stream(text):
    decision = await guard.inspect_chunk(chunk, state, expected_text=text)

    if decision.action == Action.PASS:
        await speaker.write(chunk)
    elif decision.action == Action.HALT:
        break
```

### Enroll a reference speaker

```python
from streamguard.identity.embedding import embed_utterance

# CPU-native, no downloads — runs in ~0.5 ms
reference = embed_utterance(reference_audio)    # numpy float32 (T,)

guard = AudioGuard.default(speaker_reference=reference)
# or: state = StreamState.fresh(speaker_reference=reference)
```

---

## Architecture

```
audio chunk
     │
     ▼
┌─────────────────────────────────────────────────────────┐
│  Checkers  (async, parallel, per-checker timeout)        │
│                                                          │
│  integrity/   ClippingChecker     ~1 ms   CPU            │
│               SpectralChecker     ~2 ms   CPU            │
│                                                          │
│  streaming/   ContinuityChecker   ~3 ms   CPU            │
│               (duplicates, spectral jumps, resets)       │
│                                                          │
│  identity/    SpeakerDriftChecker ~5 ms   CPU            │
│               (MFCC default · ECAPA with [speaker])      │
│                                                          │
│  extras/      ASRChecker          ~200 ms GPU/CPU opt.   │
│  (optional)   LanguageChecker     ~10 ms  CPU            │
└────────────────────┬────────────────────────────────────┘
                     │
              WeightedFusion
                     │
               Policy engine
                     │
             DecisionResult
          .action   PASS | RETRY | FALLBACK | HALT
          .fused_score   0.0 – 1.0
          .confidence    decreases when checkers time out
          .degraded      True when ≥1 checker unavailable
          .missing_signals  ["asr", ...]
          .reason        human-readable explanation
```

### Degraded mode

If a checker times out or its optional dependency is missing, it is skipped and
`DecisionResult.confidence` is reduced proportionally. The guard **always returns
a decision** — it never blocks your pipeline.

```python
decision.degraded          # True
decision.missing_signals   # ["asr"]  — Whisper timed out
decision.confidence        # 0.7  — reduced, but still actionable
```

---

## Checkers

### Core (zero extra deps)

| Checker | Module | What it catches | Latency |
|---|---|---|---|
| `ClippingChecker` | `integrity.clipping` | Hard/soft clipping, silence collapse | ~1 ms |
| `SpectralChecker` | `integrity.spectral` | Noise bursts, spectral flatness, codec corruption | ~2 ms |
| `ContinuityChecker` | `streaming.continuity` | Duplicate chunks, spectral discontinuity, stream resets | ~3 ms |
| `SpeakerDriftChecker` | `identity.speaker` | Voice identity drift via MFCC fingerprinting | ~5 ms |

### Optional extras

| Checker | Extra | What it catches |
|---|---|---|
| `ASRChecker` | `[semantic]` | Gibberish, unintelligibility (CER vs. expected text) |
| `LanguageChecker` | `[semantic]` | Language / accent mismatch |
| ECAPA embeddings | `[speaker]` | Higher-quality speaker identity (resemblyzer GE2E) |

---

## Policy presets

```python
from streamguard.policies.presets import (
    VoiceAssistantPolicy,   # low-latency streaming, tolerates minor warnings
    CascadingTtsPolicy,     # offline / sentence-at-a-time TTS (Kokoro, file chunks)
    AudiobookPolicy,        # high quality bar, HALT on critical
    NPCDialoguePolicy,      # very tolerant, only blocks hard failures
    CustomerSupportPolicy,  # strict on speaker identity, FALLBACK on critical
)

guard = AudioGuard.default(policy=AudiobookPolicy())
```

Or bring your own:

```python
from streamguard.policies.base import Policy
from streamguard.signals import Action, DecisionResult

class MyPolicy(Policy):
    def evaluate(self, signals, missing, fusion) -> DecisionResult:
        ...
```

---

## Signal dataclass

Every checker returns a `Signal`:

```python
@dataclass
class Signal:
    name: str          # "clipping", "speaker", "continuity", ...
    score: float       # 0.0 (bad) – 1.0 (perfect)
    confidence: float  # how much weight fusion should give this signal
    severity: Severity # INFO | WARNING | CRITICAL
    latency_ms: float
    metadata: dict     # checker-specific details
```

Inspect individual checker results:

```python
for sig in decision.signals:
    print(f"{sig.name:12s}  score={sig.score:.2f}  {sig.severity.name}")
    print(f"              {sig.metadata}")
```

---

## Dual-mode speaker embeddings

**Default (MFCC, CPU-native):**
- numpy + scipy only, ~0.5 ms, no downloads
- 40-dimensional mean+std MFCC pooling
- Good enough for drift detection in streaming sessions

**Upgraded (ECAPA via resemblyzer):**
```bash
pip install "streamguard[speaker]"
```
```python
guard = AudioGuard.default(use_ecapa=True)
# silently falls back to MFCC if resemblyzer isn't installed
```

---

## Custom checkers

Build your own checker and drop it into the pipeline:

```python
from streamguard._base import BaseChecker
from streamguard.signals import Signal, Severity

class PitchRangeChecker(BaseChecker):
    name = "pitch"
    timeout_ms = 30.0

    async def check(self, chunk, state, internals) -> Signal:
        # your analysis here
        return Signal(
            name=self.name,
            score=1.0,
            confidence=0.8,
            severity=Severity.INFO,
            latency_ms=0.0,
        )

guard = AudioGuard(
    checkers=[
        *AudioGuard.default().checkers,
        PitchRangeChecker(),
    ]
)
```

---

## Documentation

| Doc | Purpose |
|-----|---------|
| [docs/CHECKER_ACCURACY.md](docs/CHECKER_ACCURACY.md) | False positive / false negative notes per checker |
| [docs/DOGFOODING.md](docs/DOGFOODING.md) | Before/after glitch-rate methodology + JSONL logging |
| [docs/PRODUCT_ROADMAP.md](docs/PRODUCT_ROADMAP.md) | Standalone product roadmap |

### Dogfooding metrics

```bash
# voice-loop (reference integrator)
uv run voice_loop_mac.py --guard --guard-log tmp/guard-on.jsonl

# summarize two arms
python scripts/summarize_guard_log.py tmp/guard-off.jsonl tmp/guard-on.jsonl
```

---

## Deployment targets

Because the core is CPU-native and has no mandatory ML downloads:

- LiveKit Agents (`integrations/livekit.py`, `examples/livekit_minimal.py`)
- Pipecat (`integrations/pipecat.py`, `examples/pipecat_minimal.py`)
- OpenAI Realtime API pipelines
- Twilio / Vonage voice systems
- WebRTC-adjacent Python servers
- Cartesia / ElevenLabs / Fish-Speech / F5-TTS / CosyVoice
- Game NPC dialogue systems
- Edge / embedded inference
- Browser-adjacent voice stacks

---

## Roadmap

See [docs/PRODUCT_ROADMAP.md](docs/PRODUCT_ROADMAP.md) for the full standalone-product plan. Highlights:

- [x] Cascading vs streaming modes, LiveKit + Pipecat recipes, JSONL telemetry
- [ ] Golden WAV regression suite + published dogfood glitch-rate table
- [ ] OpenTelemetry / Prometheus exporters
- [ ] Per-vendor threshold presets (Kokoro, Cartesia, ElevenLabs)
- [ ] Hosted inspect API (optional SaaS)

---

## Development

```bash
git clone https://github.com/jempf/streamguard
cd streamguard
pip install -e ".[dev]"
pytest                   # 50 tests, no ML deps required
python examples/livekit_minimal.py
python examples/pipecat_minimal.py
```

---

## License

MIT
