streamguard 0.2.0

CPU-native realtime voice stream integrity middleware

streamguard

Realtime voice stream integrity middleware for AI voice pipelines.

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

---

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