Skip to main content

CPU-native realtime voice stream integrity middleware

Project description

streamguard

Realtime voice stream integrity middleware for AI voice pipelines.

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

PyPI version Python 3.10+ License: MIT


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

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
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.

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)

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)

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)

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)

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

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.

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

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:

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:

@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:

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):

pip install "streamguard[speaker]"
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:

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 False positive / false negative notes per checker
docs/DOGFOODING.md Before/after glitch-rate methodology + JSONL logging
docs/PRODUCT_ROADMAP.md Standalone product roadmap

Dogfooding metrics

# 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 for the full standalone-product plan. Highlights:

  • 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

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

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

streamguard-0.2.0.tar.gz (38.8 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

streamguard-0.2.0-py3-none-any.whl (39.7 kB view details)

Uploaded Python 3

File details

Details for the file streamguard-0.2.0.tar.gz.

File metadata

  • Download URL: streamguard-0.2.0.tar.gz
  • Upload date:
  • Size: 38.8 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.12.2

File hashes

Hashes for streamguard-0.2.0.tar.gz
Algorithm Hash digest
SHA256 9ff1d8ec66337248ec6a0aac706354bdd9b5dee888477e214861bc24fbea3ae9
MD5 1f77a578117cf5dae323fe9f121f4f31
BLAKE2b-256 a67ba52410d0502f5bcd01b128fe02be65ed947dc1ba20b0674879f0a914b352

See more details on using hashes here.

File details

Details for the file streamguard-0.2.0-py3-none-any.whl.

File metadata

  • Download URL: streamguard-0.2.0-py3-none-any.whl
  • Upload date:
  • Size: 39.7 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.12.2

File hashes

Hashes for streamguard-0.2.0-py3-none-any.whl
Algorithm Hash digest
SHA256 2e10ada69f7e3915fd8108aabaffd7c25ebd6f4217fd010048d14392e5972581
MD5 b220c7259db937969247ac802b5e058b
BLAKE2b-256 e1210be9a6ac693b7be33e95505c5fa4ba8238045761ebd75b98a7e0bb6632a1

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page