🌊 LLM Stream Processor

A callback-driven, prefix-safe, lazy LLM stream sanitization library.

Real-time filtering, redaction, and control of streaming LLM outputs with sub-microsecond overhead.

✨ Features

• 🔒 Prefix-Safe Pattern Matching — Uses Aho-Corasick automaton to ensure no partial sensitive content leaks before full match confirmation
• ⚡ Ultra-Low Latency — Target <5μs per-token overhead, designed for real-time streaming
• 🔄 Sync & Async Support — Works seamlessly with both synchronous and asynchronous LLM SDKs
• 🎯 Flexible Actions — PASS, DROP, REPLACE, HALT, or CONTINUE_DROP/PASS based on pattern matches
• 📊 History Tracking — Optional input/output/action history for debugging and analytics
• 🔌 Runtime Updates — Dynamically register/deregister patterns without restarting streams

📦 Installation

pip install llm-stream-processor

For development:

git clone https://github.com/DevOpRohan/llm_stream_processor.git
cd llm_stream_processor
pip install -e .

🚀 Quickstart

from stream_processor import KeywordRegistry, llm_stream_processor, replace, halt

# Create a registry and register pattern callbacks
reg = KeywordRegistry()
reg.register("secret", lambda ctx: replace("[REDACTED]"))
reg.register("STOP", halt)  # Halt stream on this keyword

@llm_stream_processor(reg, yield_mode="token")
def generate_response():
    yield "The secret password is hidden. "
    yield "Do not STOP here."
    yield "This won't be seen."

# Consume the filtered stream
for token in generate_response():
    print(token, end="", flush=True)
# Output: The [REDACTED] password is hidden. Do not 

📖 API Reference

Core Classes

Decorator

@llm_stream_processor(registry, yield_mode='token', record_history=True)

Action Helpers

🎯 Use Cases

PII Redaction

import re
from stream_processor import KeywordRegistry, llm_stream_processor, replace

reg = KeywordRegistry()

# Redact email-like patterns (register common domains)
for domain in ["@gmail.com", "@yahoo.com", "@outlook.com"]:
    reg.register(domain, lambda ctx: replace("@[REDACTED]"))

# Redact SSN patterns
reg.register("SSN:", lambda ctx: replace("SSN: [REDACTED]"))

Content Moderation (Drop Segments)

from stream_processor import KeywordRegistry, llm_stream_processor, continuous_drop, continuous_pass

reg = KeywordRegistry()

# Drop internal "thinking" blocks
reg.register("<think>", continuous_drop)
reg.register("</think>", continuous_pass)

@llm_stream_processor(reg, yield_mode="token")
def llm_stream():
    yield "Hello! <think>internal reasoning here</think>Here's my response."

print("".join(llm_stream()))
# Output: Hello! </think>Here's my response.

Async Streaming (OpenAI Pattern)

import asyncio
from stream_processor import KeywordRegistry, llm_stream_processor, replace

reg = KeywordRegistry()
reg.register("API_KEY", lambda ctx: replace("[HIDDEN]"))

@llm_stream_processor(reg, yield_mode="token")
async def stream_chat():
    # Simulating async LLM response chunks
    chunks = ["Your ", "API_KEY ", "is safe."]
    for chunk in chunks:
        yield chunk
        await asyncio.sleep(0.1)

async def main():
    async for token in stream_chat():
        print(token, end="", flush=True)

asyncio.run(main())

🏗️ Architecture

Token Generator (sync/async)
         │
         ▼
@llm_stream_processor    ← Decorator API
         │
         ▼
   StreamProcessor       ← Character-level engine
   ┌──────────────┐
   │ Aho-Corasick │
   │  Automaton   │
   └──────────────┘
         │
         ▼
  Lazy Buffer + Callbacks
         │
         ▼
   Re-packer (char/token/chunk)
         │
         ▼
      Consumer

For detailed architecture, see docs/ARCHITECTURE.md.

🧪 Development

# Install in editable mode
pip install -e .

# Run tests
python -m pytest tests/ -v

# Run the example
python -m examples.example

📚 Documentation

• Problem Statement: docs/PROBLEM.md
• Architecture & Design: docs/ARCHITECTURE.md
• Contributing Guide: CONTRIBUTING.md

🤝 Contributing

Contributions are welcome! Please read our Contributing Guide for details on:

• Code of Conduct
• Development setup
• Submitting pull requests

📄 License

This project is licensed under the MIT License - see the LICENSE file for details.

🙏 Acknowledgments

• Aho-Corasick algorithm for efficient multi-pattern matching
• Inspired by the need for real-time LLM output sanitization in production systems