7.0 KiB
7.0 KiB
上下文门控器 (Context Gatekeeper)
轻量级上下文选择器，在同一会话中自动从历史对话里选出最小且相关的片段，减少话题污染和控制上下文长度。
特性
- 🚀 纯 Python，无需向量化模型依赖（无 embedding、reranker、分类器）
- 💻 资源消耗极低，依赖极少，普通的私有部署环境都能跑
- 🔍 话题门控，通过锚点 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% 覆盖停止）
安装
pip install -e .
快速开始
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
运行测试
# 单元测试
pytest tests/test_gatekeeper.py -v
# 对照实验（需要 SiliconFlow API key）
python test_comparison.py
算法细节
锚点提取
从文本中提取有检索价值的关键词单元，支持：
- 中文：2-gram 和 3-gram（如"分布式锁"、"跨进程通信"）
- 英文：单词形态
- 代码：标识符、版本号（如
v1.2.3） - 引号短语：完整的技术术语
规则驱动，无需分词库，响应速度极快。
话题门控判断
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)
lex(u_b,q)：用户轮次与 Query 的 BM25/IDF 重叠（权重 1.5）lex(a_b,q)：助手轮次与 Query 的 BM25/IDF 重叠（权重 0.7，助手侧信息量通常更小）exact(b,q)：完全匹配奖励（精确命中关键词）recency(b)：新鲜度奖励，越近的轮次权重越高
取 top-20 进入下一步。
最小覆盖 gain
gain(b|S) = Σ IDF(t) for t ∈ cov(b)\covered(S) / cost(b)^α, α=0.8
覆盖率达到 85% 或 token 预算耗尽时停止。
为什么用 IDF 加权：高频词（如"数据"、"系统"）区分度低，低频词（如"GeoHash"、"分布式锁"）才是真正的语义锚点。用 IDF 加权确保选择的是真正有信息量的片段，而不是反复覆盖高频通用词。
对照实验（50轮对话）
使用 SiliconFlow Qwen/Qwen3-8B 模型，50轮对话（前35轮Redis，中间10轮Python，最后5轮Redis）：
| 指标 | 无门控（完整50轮） | 有门控 |
|---|---|---|
| 召回范围 | 全部50轮 | 仅相关轮次 |
| Token节省 | — | 96% |
有门控时 Query "Redis 的 GeoHash 用来做什么？" 仅召回轮次46（精确匹配），Python asyncio 轮次全部被过滤。
完整伪代码：
function select(q, turns):
# 1. 锚点提取
anchors_q = extract_anchors(q)
active_topic = get_active_topic()
# 2. 话题门控
overlap = compute_overlap(anchors_q, active_topic)
new_ratio = compute_new_ratio(anchors_q, active_topic)
if overlap < 0.20 and new_ratio > 0.70:
active_topic = create_new_topic(anchors_q) # 切换
elif has_deictic(q):
inherit_recent(2) # 指代词，强制继承最近2轮
# 否则继续当前话题
# 3. 稀疏召回
candidates = []
for each turn i:
score_i = 1.5 * bm25(user_i, q) + 0.7 * bm25(assistant_i, q) + \
1.0 * exact_match(i, q) + 0.2 * recency(i)
candidates.append((score_i, i))
top20 = top_k(candidates, k=20)
# 4. 最小覆盖贪心选择
selected = []
covered = empty_set()
for each block b in top20 sorted by gain:
new_anchors = extract_anchors(b) \ covered
if len(new_anchors) == 0: continue
gain_b = sum(IDF(t) for t in new_anchors) / cost(b)^0.8
selected.append((gain_b, b))
covered.update(new_anchors)
if coverage(covered) >= 0.85: break
return selected
局限性与适用场景
局限性：
- 稀疏检索依赖词形匹配，语义相近但词形不同的情况容易漏召
- Token 估算为粗略估算（字符数×1.5），与实际有 2-3 倍误差
- 最小粒度是整个 block，block 内部无句级裁剪，边界粗糙
- 没有在 QuAC 这类标准学术数据集上做对照实验，无法跟 Attentive History 这类基于注意力机制的方法直接对比
适用场景：
- 资源受限的生产环境（边缘设备、私有部署）
- 对延迟敏感的实时对话
- 中等复杂度对话（10-50轮）
不适用：
- 需要精确语义匹配的场景（建议用向量检索）
- 极长对话（>100轮，IDF 全量更新有偏）
License
MIT