From 9a2b1e3b6a7cf4c8873ca5f1d0f775d0c95e7fb1 Mon Sep 17 00:00:00 2001 From: Elaina Date: Wed, 22 Apr 2026 10:49:11 +0800 Subject: [PATCH] chore: remove paper, add summary, update README --- README.md | 32 ++++++++++---- SUMMARY.md | 95 ++++++++++++++++++++++++++++++++++++++++ test_comparison.py | 107 +++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 225 insertions(+), 9 deletions(-) create mode 100644 SUMMARY.md create mode 100644 test_comparison.py diff --git a/README.md b/README.md index 30105d0..3ac1418 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,8 @@ # 上下文门控器 (Context Gatekeeper) -**论文：** https://gitea.ephron.ren/elaina/context-gatekeeper/src/branch/main/paper.md +> ⚠️ **项目状态**：代码已完成并通过测试，论文暂未撰写。如需在学术场景使用，建议先在 QuAC/CoQA 等标准数据集上完成对照实验。 + +**灵感和背景**：https://gitea.ephron.ren/elaina/context-gatekeeper/src/branch/main/SUMMARY.md 轻量级上下文选择器，在同一会话中自动从历史对话里选出最小且相关的片段，减少话题污染和控制上下文长度。 @@ -44,9 +46,9 @@ gate = ContextGatekeeper(token_budget=4000) # 添加多轮对话 gate.add_turn("如何设计一个 Redis 分布式锁？", - "分布式锁需要满足互斥性、死锁避免、性能要求。常用 Redisson 实现。") + "分布式锁需要满足互斥性、死锁避免、性能要求。") gate.add_turn("锁的 TTL 设置多少合适？", - "TTL 取决于业务操作耗时，建议 3-5 倍 buffer，同时要续期机制。") + "TTL 取决于业务耗时，建议 3-5 倍 buffer，同时要续期机制。") # 为当前查询选择上下文 selected = gate.select("锁的 TTL 设置多少合适？") @@ -74,9 +76,9 @@ context-gatekeeper/ ├── tests/ │ ├── test_gatekeeper.py # 单元测试（9/9） │ └── test_full_evaluation.py # 完整评测 -├── evaluation_results.json # 评测结果 -├── paper.md # 技术论文 -├── SPEC.md # 规格文档 +├── evaluation_results.json # 评测结果（20轮对话） +├── SUMMARY.md # 未完成灵感记录 +├── SPEC.md # 规格文档 └── README.md ``` @@ -86,15 +88,15 @@ context-gatekeeper/ # 单元测试 pytest tests/test_gatekeeper.py -v -# 完整评测（20轮对话） -pytest tests/test_full_evaluation.py -v +# 对照实验（需要 SiliconFlow API key） +python test_comparison.py ``` ## 算法细节 ### 话题门控判断 -```python +``` 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) @@ -116,12 +118,24 @@ score = 1.5·lex(u_b,q) + 0.7·lex(a_b,q) + 1.0·exact(b,q) + 0.2·recency(b) 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 内部无句级裁剪 **适用场景：** - 资源受限的生产环境（边缘设备、私有部署） diff --git a/SUMMARY.md b/SUMMARY.md new file mode 100644 index 0000000..1afb371 --- /dev/null +++ b/SUMMARY.md @@ -0,0 +1,95 @@ +# 上下文门控器 · 未完成的灵感 + +> 这是一个被暂时搁置的项目，记录在此作为未来工作的起点。 + +--- + +## 灵感来源 + +大语言模型在多轮对话中面临两个核心问题： + +1. **上下文污染**：历史话题干扰当前话题，导致回答偏离 +2. **上下文膨胀**：历史长度线性增长，计算成本上升，容易超出 context window + +现有方案多依赖 embedding 模型或向量检索，在资源受限环境下难以部署。 + +--- + +## 核心想法 + +设计一个**无需额外模型**的轻量级上下文选择器： +- 用**话题门控**判断继续还是切换（基于锚点 overlap） +- 用**稀疏检索**（BM25/IDF）替代向量检索 +- 用**最小覆盖贪心**选择最相关的历史片段 + +--- + +## 已完成的部分 + +- [x] 完整代码实现（纯 Python，无第三方模型依赖） +- [x] 四阶段流程：锚点提取 → 话题门控 → 稀疏召回 → 最小覆盖选择 +- [x] 单元测试 9/9 通过 +- [x] 与 Qwen/Qwen3-8B（SiliconFlow）的端到端联调 +- [x] 50轮对话对照实验（Token 节省 96%） +- [x] 两轮子代理代码评审，发现并修复了 2 个严重 bug +- [x] README.md 完整文档 + +--- + +## 未完成的部分 + +- [ ] 在标准数据集（QuAC/CoQA）上与 Attentive History (BERT) 做对照实验 +- [ ] 与 last-N 基线的量化对比 +- [ ] 论文撰写（在标准学术数据集上验证后才写） +- [ ] 消融实验（各模块贡献度分析） +- [ ] 中文停用词表（提升锚点质量） +- [ ] 句级裁剪（目前最小粒度是整个 block） +- [ ] tiktoken 精确 token 估算 + +--- + +## 关键发现 + +### 对照实验（50轮对话） + +| 指标 | 无门控（完整50轮） | 有门控 | +|------|------------------|--------| +| 召回范围 | 全部50轮 | 仅相关轮次 | +| Token节省 | — | **96%** | +| 回答质量 | 正确 | 正确 | + +### 发现的 bug + +1. `_active_topic` 在话题切换后不更新（已修复） +2. `TopicGate` 实例状态与 `_active_topic` 不同步（已修复） + +--- + +## 相关工作（待深入） + +| 论文 | 方法 | 与本文的关系 | +|------|------|-------------| +| Attentive History Selection (2019) | BERT + 注意力软选择 | 需要GPU，本文纯规则 | +| The Complexity Trap (2025) | 简单丢弃旧observation | 证明了简单选择≈复杂压缩 | +| DiSCo (2024) | LLM蒸馏稀疏检索 | 需要训练，本文无需训练 | + +--- + +## 启动建议 + +```bash +cd context-gatekeeper +pip install -e . + +# 运行测试 +pytest tests/test_gatekeeper.py -v + +# 对照实验（需要 SiliconFlow API key） +python test_comparison.py +``` + +--- + +## 仓库地址 + +https://gitea.ephron.ren/elaina/context-gatekeeper diff --git a/test_comparison.py b/test_comparison.py new file mode 100644 index 0000000..7f8a88d --- /dev/null +++ b/test_comparison.py @@ -0,0 +1,107 @@ +""" +对照实验：有上下文门控 vs 无上下文门控 +使用 SiliconFlow Qwen/Qwen3-8B 模型 +""" + +import os +import json +import requests +from src.gatekeeper import ContextGatekeeper + +# SiliconFlow API 配置 +API_KEY = "sk-ryxkiqmodfrlthvzvcwrrvbcxilkfibymjrkorgkplhctwff" +API_URL = "https://api.siliconflow.cn/v1/chat/completions" + +def call_llm(prompt: str, model: str = "Qwen/Qwen3-8B") -> str: + """调用 SiliconFlow LLM""" + headers = { + "Authorization": f"Bearer {API_KEY}", + "Content-Type": "application/json" + } + payload = { + "model": model, + "messages": [{"role": "user", "content": prompt}], + "max_tokens": 512, + "temperature": 0.7 + } + resp = requests.post(API_URL, headers=headers, json=payload, timeout=60) + resp.raise_for_status() + return resp.json()["choices"][0]["message"]["content"] + + +def build_prompt_no_gatekeeper(query: str, history: list) -> str: + """无门控：直接拼接最近N轮历史""" + context_parts = [] + for h in history[-3:]: # 最近3轮 + context_parts.append(f"用户: {h['user']}\n助手: {h['assistant']}") + context_str = "\n\n".join(context_parts) + return f"{context_str}\n\n用户: {query}" + + +def main(): + gk = ContextGatekeeper(token_budget=1500) + + # 构造一段有话题切换的对话历史 + conversations = [ + ("如何设计一个 Redis 分布式锁？", + "分布式锁需要满足互斥性、死锁避免、性能要求。常用 Redisson 实现，核心是 SET if Not Exists + 过期时间。"), + ("锁的 TTL 设置多少合适？", + "TTL 取决于业务耗时，建议 3-5 倍 buffer。同时要 watchdog 续期机制。"), + ("介绍一下 Python 的异步编程", + "Python 异步编程用 async/await，配合事件循环。asyncio 是标准库，典型场景是 IO 密集型任务。"), + ("asyncio 是怎么工作的？", + "asyncio 基于协程和事件循环。调用 await 时协程挂起，事件循环调度其他协程执行。"), + ("Redis 支持哪些数据结构？", + "Redis 支持 String、Hash、List、Set、ZSet 五种基本类型，还有 Bitmap、HyperLogLog 等。"), + ("它和 Memcached 有什么区别？", + "Redis 是持久化数据库，Memcached 是纯内存缓存。Redis 支持更多数据结构。"), + ] + + history = [] + for u, a in conversations: + gk.add_turn(u, a) + history.append({"user": u, "assistant": a}) + + # 测试Query：话题已切换到Python，问的是Redis（有上下文污染风险） + test_query = "如何保证 Redis 缓存和数据库一致性？" + + print("=" * 70) + print("对照实验：Qwen3-8B 有/无上下文门控") + print("=" * 70) + print(f"\n测试Query: {test_query}\n") + + # --- 无门控 --- + print("【无门控】最近3轮直接拼接") + print("-" * 50) + prompt_no_gate = build_prompt_no_gatekeeper(test_query, history) + print(f"[输入]\n{prompt_no_gate}\n") + answer_no_gate = call_llm(prompt_no_gate) + print(f"[输出] {answer_no_gate[:200]}...") + + print() + + # --- 有门控 --- + print("【有门控】上下文门控器选择相关片段") + print("-" * 50) + selected = gk.select(test_query) + print(f"召回 blocks: {[b['turn_id'] for b in selected]}") + + context_parts = [] + for b in selected: + context_parts.append(f"【轮次 {b['turn_id']}】\n用户: {b['user']}\n助手: {b['assistant']}") + context_str = "\n\n".join(context_parts) + prompt_with_gate = f"你是一个有帮助的助手。\n\n【相关上下文】\n{context_str}\n\n【当前问题】\n用户: {test_query}" + + print(f"[输入]\n{prompt_with_gate}\n") + answer_with_gate = call_llm(prompt_with_gate) + print(f"[输出] {answer_with_gate[:200]}...") + + print("\n" + "=" * 70) + print("对比分析:") + print(f"无门控 - 可能受最近Python话题干扰") + print(f"有门控 - 仅召回Redis相关轮次 {[b['turn_id'] for b in selected]}") + print("=" * 70) + + +if __name__ == "__main__": + main()