Metadata-Version: 2.4
Name: llm-stream-guard
Version: 0.1.0
Summary: Streaming keyword guard for LLM output.
Author-email: zjding <1095245867@qq.com>
License-Expression: MIT
License-File: LICENSE
Keywords: aho-corasick,content-safety,guardrails,llm,sensitive-words,streaming
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: pyahocorasick>=2.0.0
Description-Content-Type: text/markdown

# LLM Stream Guard

Python 流式敏感词拦截 SDK。它不封装模型厂商 SDK，只接收任意 `AsyncIterable[str]` 文本流，在文本发给客户端前做确定性敏感词检测、跨 chunk 拦截和安全前缀输出。

License: MIT

## 安装

从 PyPI 安装：

```bash
pip install llm-stream-guard
```

从本地 wheel 安装：

```bash
uv pip install ./dist/llm_stream_guard-0.1.0-py3-none-any.whl
```

或：

```bash
python -m pip install ./dist/llm_stream_guard-0.1.0-py3-none-any.whl
```

## 词库

业务侧维护自己的词库文件，例如：

```text
# block_words.txt
hydrangea
violet comet
forbidden nebula
```

格式规则：

- 一行一个 block 词。
- 空行会被忽略。
- `#` 开头的行会被忽略。

## 基础用法

```python
from llm_stream_guard import BlockedEvent, DeltaEvent, StreamGuard

guard = StreamGuard.from_file("block_words.txt")

async for event in guard.wrap(model_stream, on_block=cancel_model):
    if isinstance(event, DeltaEvent):
        yield event.text

    if isinstance(event, BlockedEvent):
        yield {"type": "blocked", "word": event.word}
        break
```

## 模型流接入

SDK 只要求模型输出是 `AsyncIterable[str]`：

```python
from collections.abc import AsyncIterable


async def model_stream() -> AsyncIterable[str]:
    async for chunk in native_model_sdk_stream:
        yield extract_text_delta(chunk)
```

然后交给 guard：

```python
guard = StreamGuard.from_file("block_words.txt")

async for event in guard.wrap(model_stream(), on_block=cancel_model):
    ...
```

## 跨 Chunk 拦截

如果词库中有：

```text
hydrangea
```

模型输出被拆成：

```text
chunk1: "hello hydr"
chunk2: "angea world"
```

SDK 会先输出安全前缀：

```text
hello 
```

并在第二个 chunk 命中后返回 `BlockedEvent`，不会把 `hydrangea` 泄漏给客户端。

## 目录词库

可以读取目录下所有 `.txt` 文件：

```python
guard = StreamGuard.from_directory("/path/to/Vocabulary", min_word_length=2)
```

如果使用 `konsheng/Sensitive-lexicon`：

```bash
git clone https://github.com/konsheng/Sensitive-lexicon.git /tmp/Sensitive-lexicon
```

```python
guard = StreamGuard.from_directory(
    "/tmp/Sensitive-lexicon/Vocabulary",
    min_word_length=2,
)
```

注意：通用敏感词库容易包含短词和高误杀词。生产环境更建议筛选后写入自己的 `block_words.txt`。

## 归一化

如果需要匹配大小写、全半角、符号插入等变体：

```python
guard = StreamGuard.from_file(
    "block_words.txt",
    drop_separators=True,
)
```

例如词库有：

```text
violetcomet
```

可以命中：

```text
VIOLET-comet
```

## 本地验证

```bash
uv run pytest python_tests
uv run python examples/basic_usage.py
uv run python examples/agno_terminal_chat.py
uv run python scripts/test_wheel_package.py
```

`scripts/test_wheel_package.py` 会重新构建 wheel，在临时 venv 里安装，并运行安装后的调用测试。

## 打包

```bash
uv build
```

产物：

```text
dist/llm_stream_guard-0.1.0-py3-none-any.whl
dist/llm_stream_guard-0.1.0.tar.gz
```

## 发布到 PyPI

先创建 PyPI API token，然后发布：

```bash
uv publish --token "$PYPI_TOKEN"
```

也可以用环境变量：

```bash
export UV_PUBLISH_TOKEN="pypi-..."
uv publish
```

建议先发 TestPyPI 验证：

```bash
uv publish \
  --publish-url https://test.pypi.org/legacy/ \
  --token "$TEST_PYPI_TOKEN"
```

正式发布后，同事即可：

```bash
pip install llm-stream-guard
```

注意：

- PyPI 的包名必须唯一；我检查时 `llm-stream-guard` 当前返回 404，表示暂未被占用。
- 同一个版本号不能重复上传。修复后需要提升版本号，例如 `0.1.1`。
- wheel 不包含你的业务词库，同事需要自己提供 `block_words.txt` 或其他词库路径。

## 发布到私有源

如果公司有私有 pip 源：

```bash
uv publish \
  --publish-url "https://your-private-index.example.com/legacy/" \
  --token "$PRIVATE_PYPI_TOKEN"
```

同事安装：

```bash
pip install \
  --index-url "https://your-private-index.example.com/simple/" \
  llm-stream-guard
```

## Anthropic-Compatible 体验脚本

当前仓库带一个真实模型体验脚本：

```bash
uv run python scripts/chat_anthropic_compatible.py
```

环境变量：

```env
ANTHROPIC_COMPATIBLE_API_KEY=
ANTHROPIC_COMPATIBLE_BASE_URL=https://open.bigmodel.cn/api/anthropic
ANTHROPIC_COMPATIBLE_MODEL=glm-5.1
```

这个脚本使用 Anthropic Messages 兼容协议：`POST /v1/messages`、`anthropic-version` header、`content_block_delta` SSE。SDK 核心不绑定 Anthropic、OpenAI 或 BigModel。

## Agno 终端对话示例

如果要直接使用当前项目 `.env` 中的 `BIGMODEL_*` 配置：

```bash
uv sync --group dev
uv run python examples/agno_current_env_chat.py
```

该脚本读取：

```env
BIGMODEL_API_KEY=
BIGMODEL_BASE_URL=https://open.bigmodel.cn/api/anthropic
BIGMODEL_MODEL=glm-5.1
```

如果业务使用 Agno 的 `agent.arun(..., stream=True)`，可以运行：

```bash
uv sync --group dev
uv run python examples/agno_terminal_chat.py
```

环境变量：

```env
AGNO_OPENAI_API_KEY=
AGNO_OPENAI_BASE_URL=
AGNO_MODEL_ID=gpt-4o-mini
```

如果没有设置 `AGNO_OPENAI_API_KEY`，示例会 fallback 到 `OPENAI_API_KEY`。
该示例依赖 `agno` 和 `openai`，它们只放在 dev 依赖里，不会进入 SDK 的运行时依赖。

核心适配逻辑是把 Agno 事件转成文本流：

```python
async def agno_text_stream(agent, prompt):
    async for event in agent.arun(prompt, stream=True):
        text = getattr(event, "content", None)
        if isinstance(text, str) and text:
            yield text
```

然后交给：

```python
guard.wrap(agno_text_stream(agent, prompt), on_block=cancel_agent_stream)
```

## 项目结构

```text
llm_stream_guard/              SDK 源码
python_tests/                  行为测试
examples/basic_usage.py        最小调用示例
examples/agno_current_env_chat.py 当前 .env 的 Agno 对话示例
examples/agno_terminal_chat.py Agno 终端对话示例
scripts/test_wheel_package.py  wheel 安装验证
scripts/chat_anthropic_compatible.py
block_words.txt                本地测试词库
```
