内容
# 流上下文
> **Kafka 与 AI 代理之间的语义网关。**
> 自描述事件流：每个主题、字段和关系都用自然语言解释，代理可查询 - 不仅限于人类阅读 Avro 模式。
[](https://github.com/jagguvarma15/streamcontext/actions/workflows/ci.yml)
[](LICENSE)
[](pyproject.toml)
[](docker-compose.yml)
> 文档中附带一个简短的演示视频 `docs/demo.mp4`；架构图在 [`docs/architecture.md`](docs/architecture.md) 中；目录深入了解 [`docs/catalog.md`](docs/catalog.md)。
---
## 为什么存在
LLM 代理擅长推理，但记忆力差。RAG 解决了静态知识的记忆问题 - 文档、wiki、代码库。但公司中最有趣的运营数据不是静态的：它现在正通过 Kafka 流动。订单、点击、警报、传感器读数、部署。
streamcontext 是缺失的中间层。三个协作进程：
- **摄入网关**：不断将 Kafka 消息嵌入到向量存储中，保留完整的 Kafka 元数据（主题、分区、偏移量、时间戳、头部、键）作为可筛选的有效载荷。
- **目录刷新器**：在 SQLite 文件中保存每个主题的语义目录 - 模式、示例消息、活动统计数据、推断描述、每个字段的含义和检测到的关系 - MCP 服务器读取。
- **MCP 服务器**：与代理主机（Claude Desktop、Cursor、Cline、自定义）一起运行，通过 MCP 工具公开数据。代理查询流就像查询其他知识库一样 - 但这个知识库始终是当前和自描述的。
图表在 [`docs/architecture.md`](docs/architecture.md)。目录详细信息在 [`docs/catalog.md`](docs/catalog.md)。
## 快速入门（10 分钟）
要求：Docker、Python 3.11+、[`uv`](https://github.com/astral-sh/uv) （或 pip）。
```bash
git clone https://github.com/jagguvarma15/streamcontext.git
cd streamcontext
cp .env.example .env
# 1. 启动 Kafka + Schema Registry + Qdrant + 摄入网关
docker compose up -d
# 2. 本地虚拟环境用于生产者/查询脚本和 MCP 服务器
uv venv && source .venv/bin/activate
uv pip install -e .
# 3. 生成合成电子商务订单
python examples/producer.py --rate 5
# 4. 在没有代理的情况下确认检索功能（健全性检查）
python examples/query.py "来自加利福尼亚州的高价值订单"
# 5. 填充语义目录（一次传递；--loop 连续运行）
python -m streamcontext.catalog.refresher
# 6. 连接您的代理（参见下面的“与 Claude Desktop 一起使用”）并开始提问
```
步骤 5 是可选的 - 确定性工具在没有它的情况下也能工作 - 但它解锁了目录支持的工具（`find_topics_by_purpose`、`get_topic_relationships`、`explain_field`，以及 `list_topics`/`describe_topic` 上的推断描述）。在运行刷新器之前设置 `SC_CATALOG_LLM_PROVIDER=anthropic` （或 `openai`，或 `local`），如果您想要基于 LLM 的描述；默认为 `disabled`，它保持目录严格确定。
在步骤 3 后几秒钟，您应该会看到排名靠前的匹配订单，每个订单都带有其 Kafka 坐标。Qdrant 仪表板在 <http://localhost:6333/dashboard> 上可视化这些点。
### 清理
```bash
docker compose down # 停止服务，保持卷
docker compose down -v # 还删除 Kafka/Qdrant 数据
```
## 与 Claude Desktop 一起使用
将以下内容添加到 `~/Library/Application Support/Claude/claude_desktop_config.json` （macOS）或 `%APPDATA%/Claude/claude_desktop_config.json` （Windows）：
```json
{
"mcpServers": {
"streamcontext": {
"command": "python",
"args": ["-m", "streamcontext.mcp_main"],
"env": {
"SC_QDRANT_URL": "http://localhost:6333",
"SC_QDRANT_COLLECTION": "streamcontext",
"SC_QDRANT_VECTOR_DIM": "384",
"SC_EMBEDDER_PROVIDER": "local",
"SC_EMBEDDER_MODEL": "all-MiniLM-L6-v2",
"SC_MCP_TOPIC_ALLOWLIST": "orders",
"SC_MCP_MAX_RESULTS": "50",
"SC_MCP_TOOL_TIMEOUT_SEC": "5",
"SC_LOG_JSON": "false"
}
}
}
}
```
重启 Claude Desktop。`streamcontext` 服务器出现在工具面板中。与 Cursor 和 Cline 的形状相同 - 请参阅 [`docs/mcp-setup.md`](docs/mcp-setup.md) 获取 IDE 特定注释和 SSE 传输。
## 可用的 MCP 工具
| 工具 | 功能 |
|---|---|
| `list_topics` | 列出已摄入的主题，带有近似计数、时间窗口和每个主题的目录推断描述。调用此工具时，用户首先引用不熟悉的数据。 |
| `describe_topic` | 模式（来自 Schema Registry 可达时）、计数、时间窗口、示例记录、目录的推断主题描述，以及每个字段的含义和置信度。 |
| `find_topics_by_purpose` | “我在哪里可以找到计费数据？”将输入嵌入并按主题描述的相似性排序。在推理未运行时，依赖于从字段名称合成的描述。 |
| `get_topic_relationships` | 主题之间检测到的关系（`shared_key`、`foreign_reference`、`event_chain`、`semantic`）及其置信度。在构建多主题查询之前使用。 |
| `explain_field` | 单个字段的推断含义以及来自样本的示例值。在对不熟悉的字段编写过滤谓词之前使用。 |
| `search_events` | 语义搜索，按与自然语言查询的相似性排名。支持结构化 `filters`、`topic`、`time_range_minutes`、`score_threshold` 和基于 MMR 的 `diverse` 模式，用于重复排除近似匹配。 |
| `find_similar_events` | 给定 `topic:partition:offset` 引用，类似于此事件。适用于事件调查。 |
每个结果都带有 Kafka 坐标 (`coord.topic:coord.partition:coord.offset:coord.timestamp_ms`)，以便代理可以准确引用记录的来源。
## 示例代理对话
工具调优的模式：
1. **“查找过去一小时内来自加利福尼亚州的高价值订单”** - `search_events` 带有 `topic="orders"`、`time_range_minutes=60` 和结构化过滤器 `region=US_WEST`、`total>=200`。
2. **“今天我们看到哪些交易失败了？”** - `describe_topic` 学习 `status` 是枚举，然后 `search_events` 带有过滤器 `[{field: status, in_values: [cancelled, refunded]}]`。
3. **“过去 5 分钟内有什么异常？”** - 嵌入意图（“不寻常、异常或意外事件”），设置 `diverse=true`，限制时间窗口。
4. **“我们有哪些客户数据流？”** - `list_topics` 查看推断描述，然后 `find_topics_by_purpose` 按相关性排序。
5. **“查找所有失败的付款及其关联的订单”** - `get_topic_relationships` 发现连接字段，然后两个 `search_events` 调用。
6. **“`risk_score` 字段实际上是什么意思？”** - `explain_field` 返回推断的含义和几个示例值。
在 [`docs/example-conversations.md`](docs/example-conversations.md) 中使用工具调用形状进行端到端演练。在该文件的底部有调优检查清单。
## 它如何组合
```
摄入过程 MCP 过程 目录刷新器
----------------- ----------- -----------------
Kafka -> 消费者 FastMCP 工具 ── 读取 ── 目录存储
-> 嵌入器 ^ （SQLite，WAL）
-> Qdrant ── 共享 ────/ ^
状态 写入
│
SchemaIntrospector、
MessageSampler、
ActivityProfiler、
InferenceEngine、
RelationshipDetector
```
完整的设计笔记 - 文本提取策略、批处理、偏移量语义、确定性点 ID、主题允许列表、有效载荷编辑、目录过程 - 在 [`docs/architecture.md`](docs/architecture.md) 中。目录深入了解 [`docs/catalog.md`](docs/catalog.md)。数据处理和提供商数据策略在 [`docs/data-handling.md`](docs/data-handling.md) 中。安全状况在 [`docs/security.md`](docs/security.md) 中。
## 配置
所有内容都由环境变量驱动。最有用的旋钮如下；请参阅 [`.env.example`](.env.example) 获取完整列表。
### 摄入网关
| 变量 | 默认值 | 注释 |
|---|---|---|
| `SC_KAFKA_BOOTSTRAP_SERVERS` | `localhost:9092` | |
| `SC_KAFKA_TOPICS` | `orders` | 以逗号分隔。 |
| `SC_SCHEMA_REGISTRY_URL` | `http://localhost:8081` | |
| `SC_EMBEDDER_PROVIDER` | `local` | `local`（sentence-transformers）或 `openai`。 |
| `SC_EMBEDDER_MODEL` | `all-MiniLM-L6-v2` | 与 `SC_QDRANT_VECTOR_DIM` 更改同步。 |
| `SC_QDRANT_VECTOR_DIM` | `384` | 必须与嵌入器匹配。在启动时验证。 |
| `SC_PAYLOAD_REDACT_FIELDS` | （空） | 要递归剥离的 csv 值字段名称。 |
| `SC_PAYLOAD_INCLUDE_HEADERS` | `false` | 头部通常携带身份验证令牌。 |
| `SC_PAYLOAD_INDEX_FIELDS` | （空） | 要在 Qdrant 中关键字索引的 csv 值字段名称。 |
| `SC_BATCH_SIZE` | `32` | 每条嵌入/插入调用的消息数。 |
### MCP 服务器
| 变量 | 默认值 | 注释 |
|---|---|---|
| `SC_MCP_TOPIC_ALLOWLIST` | （空） | 代理可以查询的 csv 主题。空日志警告并应用无过滤。 |
| `SC_MCP_MAX_RESULTS` | `100` | 按调用结果上限；超出限制的请求被限制（不拒绝）。 |
| `SC_MCP_MAX_TIME_RANGE_MINUTES` | `10080` | `time_range_minutes` 的硬上限（默认为 7 天）。 |
| `SC_MCP_TOOL_TIMEOUT_SEC` | `5.0` | 每个工具的时钟超时。 |
| `SC_MCP_RATE_LIMIT_PER_MINUTE` | `120` | 每个工具的令牌桶。`0` 禁用。 |
| `SC_MCP_EMBED_CACHE_SIZE` | `256` | 嵌入查询缓存的 LRU 大小。`0` 禁用。 |
| `SC_MCP_MAX_VALUE_BYTES` | `8192` | 每个结果的 `value` JSON 上限。较大的值被替换为截断的存根。 |
### 语义目录
| 变量 | 默认值 | 注释 |
|---|---|---|
| `SC_CATALOG_DB_PATH` | `/var/lib/streamcontext/catalog.sqlite` | 目录刷新器和 MCP 服务器之间共享的 SQLite 文件。 |
| `SC_CATALOG_TOPICS` | （空 → `SC_KAFKA_TOPICS`） | 目录管理的以逗号分隔的主题。 |
| `SC_CATALOG_SCHEMA_REFRESH_SEC` | `300` | 模式 TTL。 |
| `SC_CATALOG_SAMPLE_REFRESH_SEC` | `900` | 样本 TTL。 |
| `SC_CATALOG_STATS_REFRESH_SEC` | `60` | 统计 TTL（也是外部 `--loop` 间隔）。 |
| `SC_CATALOG_INFERENCE_REFRESH_SEC` | `3600` | LLM 推断 TTL。 |
| `SC_CATALOG_SAMPLE_COUNT` | `10` | 每个主题保留的近期消息。 |
| `SC_CATALOG_RETAIN_SAMPLES` | `true` | 设置 `false` 进行仅元数据持久性。 |
| `SC_CATALOG_LLM_PROVIDER` | `disabled` | `disabled`、`anthropic`、`openai` 或 `local`（Ollama）。 |
| `SC_CATALOG_LLM_MODEL` | `claude-haiku-4-5-20251001` | 选择最合适的模型。 |
| `SC_CATALOG_LLM_DAILY_CEILING_USD` | `1.0` | 硬性每日支出上限；一旦触发，推理将在 UTC 重置时禁用。 |
| `SC_CATALOG_PII_FIELDS` | （空） | 要从样本中删除的 csv 键（例如 `email,phone`）。 |
| `SC_CATALOG_PII_PATTERNS` | （空） | 要在字符串值中屏蔽的 csv 正则表达式。 |
## 安全
三个审计文档驱动安全立场：
- [`docs/audit-v0.1.md`](docs/audit-v0.1.md) - 摄入网关，在 MCP 工作开始之前。
- [`docs/audit-v0.2.md`](docs/audit-v0.2.md) - MCP 层，在 v0.2.0 版本之前。
- [`docs/audit-v0.3.md`](docs/audit-v0.3.md) - 语义目录，在 v0.3.0 版本之前。
威胁模型和推荐的生产环境设置在 [`docs/security.md`](docs/security.md) 中。数据表面和提供商数据策略在 [`docs/data-handling.md`](docs/data-handling.md) 中。标题：v0.3 仍假设本地/受信任主机部署，但 MCP 服务器现在提供 `authorize` 钩子，以便下游消费者可以在不分叉服务器的情况下插入真正的每个调用者身份验证。
## 路线图
- **v0.1 — 流式接收器。** Kafka、嵌入、Qdrant。批处理、编辑、确定性插入、停止失败语义。完成。
- **v0.2 — MCP 服务器。** `list_topics`、`describe_topic`、`search_events`、`find_similar_events`。主题允许列表、速率限制、嵌入缓存、值截断。完成。
- **v0.3 — 语义目录。** 模式内省、样本支持的推断描述、每个字段的含义、关系检测、每日 LLM 支出上限、持久性 PII 编辑。三个新的 MCP 工具：`find_topics_by_purpose`、`get_topic_relationships`、`explain_field`。此版本。
- **后续。** 双向流（代理将数据生成到 Kafka 并带有模式验证）。Kafka Connect 包装。多接收器（Pinecone、Weaviate、pgvector）。Jinja 文本提取模板。Prometheus `/metrics`。SASL/SSL Kafka、SR 身份验证、SSE 身份验证。
## 开发
```bash
uv pip install -e '.[dev]'
pytest -q # 单元测试（无需 Docker）
RUN_INTEGRATION=1 pytest -q # 集成测试（通过 testcontainers 启动 Kafka + Qdrant）
ruff check .
```
## 贡献
代码库故意设计得较小 - 30 分钟内可阅读，首日即可贡献。欢迎提出问题和 PR，尤其是：
- 新的 `VectorSink` 实现（Pinecone、pgvector、Weaviate）。
- 更好的文本提取策略（Jinja 模板、字段选择）。
- 生产者演示的新语言客户端。
- 更多 MCP 工具 - 过滤基元、聚合、异常启发式。
## 许可证
MIT - 请参阅 [LICENSE](LICENSE)。
