Metadata-Version: 2.4
Name: yyds-stream-tap
Version: 0.2.1
Summary: A highly optimized, zero-latency streaming text interceptor and cleaner for LLM agent outputs.
Home-page: https://github.com/yyds-fast/yyds-stream-tap
Author: yyds-fast
Author-email: yyds.fast@gmail.com
License: MIT
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.7
Classifier: Programming Language :: Python :: 3.8
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Operating System :: OS Independent
Classifier: License :: OSI Approved :: MIT License
Requires-Python: >=3.7
Description-Content-Type: text/markdown
Dynamic: author
Dynamic: author-email
Dynamic: classifier
Dynamic: description
Dynamic: description-content-type
Dynamic: home-page
Dynamic: license
Dynamic: requires-python
Dynamic: summary

# yyds-stream-tap

[![PyPI version](https://img.shields.io/pypi/v/yyds-stream-tap.svg)](https://pypi.org/project/yyds-stream-tap/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)

[中文文档](README_CN.md)

`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

```bash
pip install -U yyds-stream-tap
```

---

## 💻 Quick Start

### 1. Basic Usage (Using Built-in Wordlist)

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

```python
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)

```python
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

```python
# 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.
