Acacian/aegis

What's New

Streaming-Aware Guardrail Engine

StreamingGuardrailEngine scans streaming LLM responses with automatic strategy selection:

• Windowed scan — configurable window_size for incremental guardrails (toxicity, keyword checks)
• Full-buffer mode — for guardrails where partial exposure is a violation (PII, credit cards)
• Auto-selection — if any guardrail sets requires_full_buffer=True, the engine switches to full-buffer mode automatically

from aegis.guardrails import GuardrailEngine, StreamingGuardrailEngine

engine = GuardrailEngine()
engine.add(my_pii_guardrail)  # requires_full_buffer=True

streaming = StreamingGuardrailEngine(engine, window_size=4)

async for chunk in streaming.scan_stream(llm_stream):
    if chunk.blocked:
        print("[BLOCKED]")
        break
    print(chunk.content, end="", flush=True)

This addresses a fundamental architectural gap in LLM frameworks where streaming bypasses output guardrails.

Playground: Streaming Guard Demo

Live split-screen comparison at acacian.github.io/aegis/playground:

• Left: LLM streams freely — PII leaks to the user
• Right: Aegis catches it mid-stream
• 5 scenarios including AI-powered semantic PII detection (Gemini Flash)

Full Changelog

See CHANGELOG.md