yyds-stream-tap 0.2.1

A highly optimized, zero-latency streaming text interceptor and cleaner for LLM agent outputs.

yyds-stream-tap

中文文档

yyds-stream-tap is a highly optimized, zero-latency streaming text interceptor and dynamic cleaner designed for LLM (Large Language Model) agent outputs.

---

🌟 Key Features

• 🚀 Zero-Latency TTFT: Using a dynamic sliding window algorithm, safe prefix characters are emitted immediately as they are generated, never buffering safe content unnecessarily.
• ⚡ Ultra-High Performance ($O(1)$ complexity): Matches sensitive words using a Trie Tree, keeping the matching time constant regardless of the wordlist scale.
• 📦 Built-in Wordlist: Comes with default built-in lists for weapons, explosives, political, and illicit content, ready to use out-of-the-box.
• 🔌 Developer-Friendly APIs:
  • Load custom sensitive or stop words from single or multiple local files.
  • Automatically handles comma-separated or newline-separated formats.
  • Dynamically add/remove sensitive words and stop words at runtime.
• 🛠 Memory Optimized: Utilizes Python __slots__ memory declaration. Keeps memory footprint minimal even with hundreds of thousands of words.

---

📦 Installation

pip install -U yyds-stream-tap

---

💻 Quick Start

1. Basic Usage (Using Built-in Wordlist)

import asyncio
from yyds_stream_tap import StreamInterceptor

# Mock stream from LLM
async def mock_llm_stream():
    chunks = ["Hello, today I bought a 六合彩", " ticket."]
    for chunk in chunks:
        yield chunk

async def main():
    # Initialize interceptor with default built-in wordlist
    interceptor = StreamInterceptor(use_default=True)

    # Wrap the original stream
    clean_stream = interceptor.intercept(mock_llm_stream())

    # Consume the cleaned stream
    async for char in clean_stream:
        print(char, end="", flush=True)

if __name__ == "__main__":
    asyncio.run(main())

---

1.2 Auto-detect & Filter JSON / Dict / SSE Envelope Streams (Zero-Configuration)

yyds-stream-tap natively and transparently filters structured packets streams without requiring developers to manually parse JSON or reconstruct headers.

• Python Dict Stream: Pass Dict objects stream. Automatically rewrites content fields (e.g. message.content) and yields dicts in the identical structure.
• OpenAI SSE String Stream (e.g., API Gateway / Proxy): Parses data: {...} lines and [DONE] sentinels, replaces sensitive content, and encodes back into SSE payloads.
• Ollama NDJSON String Stream: Parses NDJSON lines, filters inner text, and serializes back.

import asyncio
from yyds_stream_tap import StreamInterceptor

# Mock OpenAI SSE packet stream
async def mock_sse_stream():
    yield 'data: {"id": "chat-1", "choices": [{"delta": {"content": "I"}, "index": 0}]}'
    yield 'data: {"id": "chat-2", "choices": [{"delta": {"content": "bought"}, "index": 0}]}'
    yield 'data: {"id": "chat-3", "choices": [{"delta": {"content": "a"}, "index": 0}]}'
    yield 'data: {"id": "chat-4", "choices": [{"delta": {"content": "六合彩"}, "index": 0}]}'
    yield 'data: [DONE]'

async def main():
    interceptor = StreamInterceptor(use_default=True)
    
    # 💡 Transparently pass raw SSE stream. SDK handles decoding, matching, and encoding automatically!
    clean_sse_stream = interceptor.intercept(mock_sse_stream())
    
    async for sse_line in clean_sse_stream:
        print(sse_line, end="") # Outputs identical SSE packets with filtered content

if __name__ == "__main__":
    asyncio.run(main())

---

2. Add Custom Wordlist Files (Multiple Files Supported)

from yyds_stream_tap import StreamInterceptor

# Load custom sensitive word files
interceptor = StreamInterceptor(
    custom_files=["./my_sensitive_words_1.txt", "./my_sensitive_words_2.txt"],
    use_default=True,
    default_replacement="***"
)

# Load custom stopword files (stopwords are removed/deleted in the output)
interceptor.add_stop_words_from_files(["./my_stopwords.dic"])

---

3. Dynamic Add at Runtime

# Add sensitive words with custom replacements
interceptor.add_words(["internal_secret", "classified_info"], replacement="[REDACTED]")

# Add stopwords (which will be filtered out to empty string "")
interceptor.add_stop_words(["ah", "oh", "!"])

---

📄 License

MIT License.