152 lines
4.9 KiB
Markdown
152 lines
4.9 KiB
Markdown
# 上下文门控器 (Context Gatekeeper)
|
||
|
||
> ⚠️ **项目状态**：代码已完成并通过测试，论文暂未撰写。如需在学术场景使用，建议先在 QuAC/CoQA 等标准数据集上完成对照实验。
|
||
|
||
**灵感和背景**：https://gitea.ephron.ren/elaina/context-gatekeeper/src/branch/main/SUMMARY.md
|
||
|
||
轻量级上下文选择器，在同一会话中自动从历史对话里选出最小且相关的片段，减少话题污染和控制上下文长度。
|
||
|
||
## 特性
|
||
|
||
- 🚀 **纯 Python**，无需向量化模型依赖（无 embedding、reranker、分类器）
|
||
- 💻 **轻量运行**，2 核 2G 环境可流畅运行
|
||
- 🔍 **话题门控**，通过锚点 overlap + new_ratio 判断继续/切换，含指代词强制继承
|
||
- 📦 **稀疏召回**，BM25/IDF-overlap 评分，用户侧权重高于助手侧
|
||
- 🎯 **最小覆盖**，基于 IDF 加权集合覆盖的贪心选择
|
||
- ⚙️ **稳定约束区**，持久化用户偏好（语言/风格/禁用项）
|
||
|
||
## 核心流程
|
||
|
||
```
|
||
用户查询 q
|
||
↓
|
||
① 锚点提取（中文 2/3-gram、英文单词、代码标识符、版本号、引号短语）
|
||
↓
|
||
② 话题门控（overlap > 0.45 → 继续；overlap < 0.20 且 new_ratio > 0.70 → 切换；
|
||
有指代词 → 强制继续；中间地带默认继续）
|
||
↓
|
||
③ 稀疏召回（top-20，BM25/IDF-overlap + exact match + 新鲜度奖励）
|
||
↓
|
||
④ 最小覆盖选择（gain = ΣIDF(t) / cost^α，贪心选择达到 85% 覆盖停止）
|
||
```
|
||
|
||
## 安装
|
||
|
||
```bash
|
||
pip install -e .
|
||
```
|
||
|
||
## 快速开始
|
||
|
||
```python
|
||
from src.gatekeeper import ContextGatekeeper
|
||
|
||
# 初始化，token 预算 4000
|
||
gate = ContextGatekeeper(token_budget=4000)
|
||
|
||
# 添加多轮对话
|
||
gate.add_turn("如何设计一个 Redis 分布式锁？",
|
||
"分布式锁需要满足互斥性、死锁避免、性能要求。")
|
||
gate.add_turn("锁的 TTL 设置多少合适？",
|
||
"TTL 取决于业务耗时，建议 3-5 倍 buffer，同时要续期机制。")
|
||
|
||
# 为当前查询选择上下文
|
||
selected = gate.select("锁的 TTL 设置多少合适？")
|
||
|
||
for item in selected:
|
||
print(f"轮次 {item['turn_id']}: {item['user']}")
|
||
print(f"助手: {item['assistant']}\n")
|
||
|
||
# 构建完整 prompt（可直接发给 LLM）
|
||
prompt = gate.build_prompt("锁的 TTL 设置多少合适？")
|
||
print(prompt)
|
||
```
|
||
|
||
## 项目结构
|
||
|
||
```
|
||
context-gatekeeper/
|
||
├── src/
|
||
│ ├── anchor.py # 锚点提取（2/3-gram + IDF）
|
||
│ ├── block.py # Block 数据结构
|
||
│ ├── topic_gate.py # 话题门控（overlap + new_ratio + 指代词）
|
||
│ ├── sparse.py # 稀疏召回（BM25/IDF + exact + recency）
|
||
│ ├── selector.py # 最小覆盖选择（IDF加权贪心）
|
||
│ └── gatekeeper.py # 主模块（组合各子模块）
|
||
├── tests/
|
||
│ ├── test_gatekeeper.py # 单元测试（9/9）
|
||
│ └── test_full_evaluation.py # 完整评测
|
||
├── evaluation_results.json # 评测结果（20轮对话）
|
||
├── SUMMARY.md # 未完成灵感记录
|
||
├── SPEC.md # 规格文档
|
||
└── README.md
|
||
```
|
||
|
||
## 运行测试
|
||
|
||
```bash
|
||
# 单元测试
|
||
pytest tests/test_gatekeeper.py -v
|
||
|
||
# 对照实验（需要 SiliconFlow API key）
|
||
python test_comparison.py
|
||
```
|
||
|
||
## 算法细节
|
||
|
||
### 话题门控判断
|
||
|
||
```
|
||
overlap = Σ IDF(t) for t ∈ A(q)∩A(T) / Σ IDF(t) for t ∈ A(q)
|
||
new_ratio = Σ IDF(t) for t ∈ A(q)\A(T) / Σ IDF(t) for t ∈ A(q)
|
||
|
||
if overlap > 0.45: continue
|
||
elif overlap < 0.20 and new_ratio > 0.70: switch
|
||
elif has_deictic: continue # 指代词强制继承
|
||
else: continue # 中间地带默认继续
|
||
```
|
||
|
||
### 稀疏召回评分
|
||
|
||
```
|
||
score = 1.5·lex(u_b,q) + 0.7·lex(a_b,q) + 1.0·exact(b,q) + 0.2·recency(b)
|
||
```
|
||
|
||
### 最小覆盖 gain
|
||
|
||
```
|
||
gain(b|S) = Σ IDF(t) for t ∈ cov(b)\covered(S) / cost(b)^α, α=0.8
|
||
```
|
||
|
||
## 对照实验（50轮对话）
|
||
|
||
使用 SiliconFlow Qwen/Qwen3-8B 模型，50轮对话（前35轮Redis，中间10轮Python，最后5轮Redis）：
|
||
|
||
| 指标 | 无门控（完整50轮） | 有门控 |
|
||
|------|-----------------|--------|
|
||
| 召回范围 | 全部50轮 | 仅相关轮次 |
|
||
| Token节省 | — | **96%** |
|
||
|
||
有门控时 Query "Redis 的 GeoHash 用来做什么？" 仅召回轮次46（精确匹配），Python asyncio 轮次全部被过滤。
|
||
|
||
## 局限性与适用场景
|
||
|
||
**局限性：**
|
||
- 稀疏检索在语义相似但词形不同时召回率有限
|
||
- 中文锚点无停用词过滤，高频无意义词可能干扰 IDF
|
||
- Token 估算为粗略估算（字符数×1.5），与实际有 2-3 倍误差
|
||
- 最小粒度是整个 block，block 内部无句级裁剪
|
||
|
||
**适用场景：**
|
||
- 资源受限的生产环境（边缘设备、私有部署）
|
||
- 对延迟敏感的实时对话
|
||
- 中等复杂度对话（10-50轮）
|
||
|
||
**不适用：**
|
||
- 需要精确语义匹配的场景（建议用向量检索）
|
||
- 极长对话（>100轮，IDF 全量更新有偏）
|
||
|
||
## License
|
||
|
||
MIT
|