# LambdaRAG: 统一知识检索方案 ## Context 综合 RAG v1 (BM25)、RAG v2 (Hybrid)、RAG Fusion (Multi-query)、Wiki (知识图谱) 四种方案的优势,结合 lambdagentpaas 平台的三层抽象 (Skill → Pattern → Orchestration) 和 Lambda DSL,设计一个统一的自适应知识检索系统。 核心理念:**不同问题用不同策略,由 Orchestrator 智能路由,利用 entity 关系图谱增强跨文档推理。** --- ## 一、架构总览:三层抽象 ``` ┌─────────────────────────────────────────────────────────┐ │ Layer 3: Orchestration (编排层) │ │ │ │ QueryClassifier >> Route({ │ │ fact → Pipeline(keyword_search), │ │ compare → Pipeline(graph_expand >> fusion_search), │ │ relation → Pipeline(graph_walk >> wiki_query), │ │ synthesis→ fan_out_merge(wiki, fusion, graph), │ │ temporal → Pipeline(time_filter >> keyword_search), │ │ }) >> KnowledgeFeedback │ │ │ ├─────────────────────────────────────────────────────────┤ │ Layer 2: Pattern (组合模式层) │ │ │ │ Pipeline: skill1 >> skill2 >> skill3 │ │ fan_out_merge: Par(s1, s2, s3) >> merge │ │ map_reduce: split >> Par(process...) >> reduce │ │ review: generate >> check >> (retry or pass) │ │ graph_expand: entity_extract >> bfs_walk >> expand │ │ │ ├─────────────────────────────────────────────────────────┤ │ Layer 1: Skill (原子能力层) │ │ │ │ bm25_search 向量检索 embedding_search │ │ query_expand rerank graph_walk │ │ wiki_page_read entity_extract relation_lookup │ │ time_filter citation_check fact_extract │ │ answer_generate merge_results cost_track │ │ │ └─────────────────────────────────────────────────────────┘ ``` --- ## 二、Layer 1: Skill 原子能力 每个 Skill 是一个独立的 Lambda Term (Lam 或 Tool),单一职责。 ### 检索类 Skill | Skill | 类型 | Lambda | 输入→输出 | 成本 | |-------|------|--------|----------|------| | `bm25_search` | Tool | λq. BM25(q, top_k) | str → [chunk] | 0, <10ms | | `embedding_search` | Tool | λq. cosine(embed(q), vectors) | str → [chunk] | $0.0001, ~100ms | | `wiki_page_read` | Tool | λq. read_wiki_pages(q) | str → [page_content] | 0, <50ms | | `relation_lookup` | Tool | λe. graph.neighbors(e) | entity → [relation] | 0, <10ms | ### 处理类 Skill | Skill | 类型 | Lambda | 说明 | |-------|------|--------|------| | `query_expand` | Lam | λq. LLM(q) → [q1..qN] | 用本地 vLLM 扩展查询 | | `entity_extract` | Lam | λq. LLM(q) → [entities] | 从查询中提取实体 | | `rerank` | Tool | λ(q, docs). API(q, docs) → [ranked_docs] | SiliconFlow rerank API | | `time_filter` | Tool | λ(q, docs). filter(docs, date) | 按时效性过滤 | | `rrf_merge` | Tool | λ([list1, list2..]) → [merged] | RRF 倒数排名融合 | | `merge_results` | Tool | λ(wiki_r, rag_r) → combined | 合并多源结果 | ### 生成类 Skill | Skill | 类型 | Lambda | 说明 | |-------|------|--------|------| | `answer_generate` | Lam | λ(q, context). LLM → answer | 基于上下文生成回答 | | `citation_check` | Lam | λ(answer, sources). LLM → verdict | 核查引用准确性 | | `fact_extract` | Lam | λanswer. LLM → [facts, relations] | 从回答中提取新知识 | ### 图谱类 Skill | Skill | 类型 | Lambda | 说明 | |-------|------|--------|------| | `graph_walk` | Tool | λ(entities, n_hops). BFS → [entities, relations, paths] | N跳图遍历 | | `graph_expand` | Tool | λ(query, entities). query + related_terms | 用图谱扩展查询范围 | --- ## 三、Layer 2: Pattern 组合模式 ### Pattern 1: keyword_search (快速精确) ``` λq. bm25_search(q) >> answer_generate(q, _) ``` 适用: fact 类问题 ("2019年船员注册人数") ### Pattern 2: hybrid_search (语义增强) ``` λq. Par(bm25_search(q), embedding_search(q)) >> rrf_merge >> rerank(q, _) ``` 适用: 需要语义理解但单一查询够用的问题 ### Pattern 3: fusion_search (多视角) ``` λq. query_expand(q) → [q1..qN] >> map(qi → hybrid_search(qi)) >> rrf_merge >> rerank(q, _) ``` 适用: 复杂对比类问题,表达方式影响检索结果 ### Pattern 4: graph_retrieval (关系增强) ``` λq. entity_extract(q) → [entities] >> graph_walk(entities, 2) → [expanded_entities, relations] >> graph_expand(q, expanded_entities) → expanded_query >> hybrid_search(expanded_query) ``` 适用: 关系类问题 ("海事局的上级部门是谁") ### Pattern 5: wiki_deep (编译知识) ``` λq. wiki_page_read(q) → [wiki_pages] >> If(sufficient(wiki_pages), then_= answer_generate(q, wiki_pages), else_= gap_fill(q, wiki_pages)) ``` 其中 gap_fill: ``` λ(q, wiki). missing_topics(q, wiki) >> fusion_search(missing) >> merge_results(wiki, fusion) ``` 适用: 综合性问题,wiki 已有部分知识但不完整 ### Pattern 6: synthesis (全量综合) ``` λq. fan_out_merge( wiki_deep(q), graph_retrieval(q), fusion_search(q) ) >> merge_deduplicate >> review(citation_check) ``` 适用: 跨文档综合分析 ("总结海事局近年政策方向") ### Pattern 7: temporal_search (时效性) ``` λq. extract_time_range(q) >> bm25_search(q) >> time_filter(results, time_range) >> answer_generate ``` 适用: "琼州海峡定线制最新版本何时施行" --- ## 四、Layer 3: Orchestration 编排 ### 4.1 查询分类器 (智能路由) ```python # 规则快速分类 (零成本) def fast_classify(query): if re.search(r'\d{4}年.*多少|数量|人数|统计', query): return "fact" if re.search(r'异同|比较|区别|对比|不同', query): return "compare" if re.search(r'关系|隶属|上级|下级|依据|谁.*管', query): return "relation" if re.search(r'最新|修订|变化|趋势|演变', query): return "temporal" if re.search(r'总结|综合|梳理|分析.*体系|框架', query): return "synthesis" return None # 无法快速判断 # LLM 精确分类 (有成本但准确) classifier_lam = Lam("classify", """ 分类用户问题为以下之一: - fact: 事实查找 (具体数字/日期/名称/条款) - compare: 对比分析 (两个以上事物的异同) - relation: 关系查询 (实体间的关系、层级、依据) - synthesis: 综合分析 (跨文档总结、体系梳理) - temporal: 时效相关 (最新版本、变化趋势、修订历史) 同时提取问题中的关键实体。 输出JSON: {"type":"...", "entities":["..."], "confidence": 0.9} """) # 自适应分类: 先试规则,不确定时用 LLM adaptive_classifier = If( cond=Tool("try_fast", fast_classify), # 返回非 None 表示成功 then_=Tool("identity", lambda x: x), # 用快速结果 else_=classifier_lam # 回退 LLM ) ``` ### 4.2 问题类型 → 策略映射 | 问题类型 | 策略 (Pattern) | 为什么 | 示例 | |---------|---------------|-------|------| | **fact** | keyword_search | 精确匹配,快速,零成本 | "2019年船员注册总数" | | **compare** | graph_expand >> fusion_search | 图谱找到两个实体的关联文档,多视角检索覆盖两侧 | "海上交通安全法与水污染防治法异同" | | **relation** | graph_walk >> wiki_query | 图谱直接回答关系,wiki 补充细节 | "海事局的职责和上级部门" | | **synthesis** | fan_out_merge(wiki, fusion, graph) | 三路并行获取最大信息量 | "总结船员管理法规体系" | | **temporal** | time_filter >> keyword_search | 时间过滤确保获取最新版本 | "琼州海峡定线制最新规定" | ### 4.3 完整编排 Lambda 表达式 ```python unified_retrieval = Compose( # Step 0: 分类 adaptive_classifier, # Step 1: 路由到对应策略 Route( classifier=identity, routes={ "fact": Compose(bm25_search, answer_generate), "compare": Compose( entity_extract, graph_walk, graph_expand, fusion_search, Guard(answer_generate, citation_check, retry=1) ), "relation": Compose( entity_extract, graph_walk, Par(wiki_page_read, relation_lookup), merge_results, answer_generate ), "synthesis": Compose( Par( Compose(wiki_page_read, answer_generate), Compose(entity_extract, graph_walk, fusion_search), ), merge_results, Guard(answer_generate, citation_check, retry=1) ), "temporal": Compose( time_filter, bm25_search, answer_generate ), }, default=Compose(fusion_search, answer_generate), # 兜底 ), # Step 2: 知识反馈 (仅多源回答触发) If( cond=Tool("has_multi_source", lambda r: r.count("[来源:") >= 2), then_=Compose(fact_extract, update_wiki, update_relations), else_=identity ), ) ``` ### 4.4 效果类型注解 ``` fact: io (纯 BM25) compare: llm · io · (io ∥ io) · llm (classify + graph + par_search + generate) relation: llm · io · (io ∥ io) · llm (classify + graph + par(wiki,relation) + generate) synthesis: llm · ((io · llm) ∥ (llm · io · io)) · llm (最复杂) temporal: io · io · llm (filter + search + generate) ``` --- ## 五、Entity 关系图谱核心设计 ### 5.1 数据结构 ```python class EntityGraph: # 邻接表: entity → [(relation_type, target, source_doc, description)] adjacency: Dict[str, List[Tuple]] # 反向邻接: entity → [(relation_type, source_entity, source_doc)] reverse: Dict[str, List[Tuple]] # 实体元数据: entity → {type, refs_count, wiki_page_path} entities: Dict[str, Dict] def walk(self, seeds, max_hops=2, max_nodes=20): """BFS N-hop 遍历,返回子图""" def find_entities(self, text): """从文本中模糊匹配已知实体""" def shortest_path(self, entity_a, entity_b): """两实体间最短关系路径""" def subgraph(self, entities): """提取包含指定实体的子图""" ``` ### 5.2 关系增强检索流程 ``` Q: "海上交通安全法与水污染防治法在船舶管理方面有何异同?" Step 1: entity_extract → ["海上交通安全法", "水污染防治法", "船舶管理"] Step 2: graph_walk(entities, hops=2) 海上交通安全法 --规定--> 船舶航行安全 海上交通安全法 --发布机构--> 全国人大常委会 海上交通安全法 --涉及--> 船员管理 水污染防治法 --规定--> 船舶排放标准 水污染防治法 --发布机构--> 全国人大常委会 水污染防治法 --涉及--> 海洋环境保护 船舶管理 --相关法规--> 防治船舶污染海洋环境管理条例 Step 3: graph_expand → 扩展查询增加: "船舶航行安全", "船舶排放标准", "船员管理", "海洋环境保护" Step 4: fusion_search (原始query + 扩展terms) → 现在能检索到法律原文,而不仅是行政检查清单 Step 5: answer_generate with relations context → 回答中自然包含关系链信息 ``` ### 5.3 关系在回答中的作用 ``` Prompt 增强: ## 实体关系 (来自知识图谱) - 海上交通安全法 --发布机构--> 全国人大常委会 - 海上交通安全法 --规定--> 船舶航行安全 - 水污染防治法 --发布机构--> 全国人大常委会 - 水污染防治法 --规定--> 船舶排放标准 - 两法 --共同涉及--> 船舶管理 ## 检索到的文档片段 [doc1: 海上交通安全法原文] ... [doc2: 水污染防治法原文] ... ## 问题 海上交通安全法与水污染防治法在船舶管理方面有何异同? ``` --- ## 六、新增/修改文件清单 ### 新增文件 | 文件 | 位置 | 说明 | |------|------|------| | `graph_engine.py` | `qaagent67/scripts/` | EntityGraph 类 + graph_walk/expand | | `search_engine_v2.py` | `qaagent67/scripts/` | 混合检索 (BM25+Embedding+Rerank) | | `search_engine_fusion.py` | `qaagent67/scripts/` | RAG Fusion (Multi-query) | | `search_unified.py` | `qaagent67/scripts/` | 统一入口: 分类→路由→策略→反馈 | | `build_vector_index.py` | `qaagent67/scripts/` | 向量索引构建 | ### 修改文件 | 文件 | 改动 | |------|------| | `agent-config.yml` | 新增 retrieval 配置段 | | `qademo/app.py` | import 改为 `from search_unified import search` | ### 不动的文件 | 文件 | 说明 | |------|------| | `search_engine.py` | BM25 v1,被 v2 内部调用 | | `wiki_compile.py` | Wiki 编译,已有关系提取 | | `rebuild_index.py` | 旧索引构建,保留参考 | --- ## 七、agent-config.yml 配置 ```yaml # 统一检索配置 retrieval: mode: unified # Layer 1: Skills skills: bm25: enabled: true topK: 10 embedding: enabled: true model: bge-m3 backend: npu # 本地 NPU 推理 (免费) baseUrl: http://127.0.0.1:11435/api/embed batchSize: 32 dimensions: 1024 rerank: enabled: true backend: vllm # 用本地 vLLM qwen2.5-32b 做 rerank baseUrl: http://127.0.0.1:8000/v1 graph: enabled: true relationsFile: wiki/relations.json maxHops: 2 maxExpand: 20 wiki: enabled: true dir: wiki/ # Layer 2: Patterns patterns: keyword_search: skills: [bm25] hybrid_search: skills: [bm25, embedding] merge: rrf rrfK: 60 fusion_search: skills: [bm25, embedding] merge: rrf queryExpansion: true expansionCount: 4 graph_retrieval: skills: [graph, hybrid_search] wiki_deep: skills: [wiki, hybrid_search] gapFill: true # Layer 3: Orchestration orchestration: classifier: fastClassify: true # 规则优先 llmFallback: true # 规则不确定时用 LLM routing: fact: keyword_search compare: [graph_retrieval, fusion_search] relation: [graph_retrieval, wiki_deep] synthesis: [wiki_deep, fusion_search, graph_retrieval] temporal: [keyword_search] default: hybrid_search feedback: enabled: true minSources: 2 cost: budgetPerQuery: 0.05 fallback: keyword_search ``` --- ## 八、实施顺序 | 阶段 | 内容 | 依赖 | |------|------|------| | 1 | `graph_engine.py` — 实体关系图谱 | relations.json (Wiki 编译产出) | | 2 | `search_engine_v2.py` — 混合检索 | SiliconFlow API + numpy | | 3 | `build_vector_index.py` — 构建向量索引 | search_engine_v2 | | 4 | `search_engine_fusion.py` — RAG Fusion | search_engine_v2 + vLLM | | 5 | `search_unified.py` — 统一编排 | 以上全部 | | 6 | `app.py` 改 1 行 import | search_unified | 每阶段独立可用,逐层叠加。 --- ## 九、预期效果 | 问题类型 | 当前 (BM25) | 统一方案 | 提升原因 | |---------|------------|---------|---------| | "2019年船员注册人数" | ✅ 准确 | ✅ 准确 (fact→BM25快速路径) | 不退化 | | "海上交通安全法与水污染防治法异同" | ❌ 检索到行政清单 | ✅ 检索到法律原文 | graph_expand + fusion | | "海事局的职责" | ⚠️ 部分 | ✅ 完整 (含上下级关系) | relation_lookup + wiki | | "船员管理法规体系梳理" | ❌ 碎片化 | ✅ 体系化 | synthesis 三路并行 | | "琼州海峡定线制最新版本" | ⚠️ 新旧混淆 | ✅ 最新版 | temporal_filter | --- ## 十、自增强学习机制 (越用越好) 核心理念:**每次用户查询都是一次学习机会**,系统从使用中积累知识、优化检索、改进路由。 ### 10.1 四维自增强闭环 ``` 用户查询 → 检索 → LLM 回答 → 反馈信号 │ ┌───────────────────────┼───────────────────────┐ ↓ ↓ ↓ 知识积累 检索优化 路由优化 │ │ │ ① Wiki 更新 ③ Query-Doc 对 ⑤ 路由权重 ② 关系图谱扩展 ④ 同义词表 ⑥ 策略成功率 新实体发现 BM25 Boost 分类校准 ``` ### 10.2 维度一:知识积累 (Wiki Feedback Loop) 每次**多源回答**自动提取新知识写回 Wiki: ```python # Lambda: If(has_multi_source, Compose(fact_extract, update_wiki, update_graph), identity) def knowledge_feedback(query, answer, sources): # 触发条件: 回答引用了 2+ 来源 if answer.count("[来源:") < 2: return # 跳过单源简单回答 # ① 新实体发现: 回答中出现但图谱中不存在的实体 new_entities = extract_entities(answer) - graph.known_entities # → 写入 wiki/entities/ # ② 新关系发现: 从回答中提取实体间关系 # "海上交通安全法由全国人大常委会制定" # → (海上交通安全法, 制定机构, 全国人大常委会) new_relations = extract_relations(answer) # → 追加 relations.json + 更新 entity 页面 # ③ 跨文档综合: 多源综合的新知识 synthesis = summarize_cross_doc(query, answer, sources) # → 写入 wiki/analyses/{query_hash}.md ``` **效果**: 图谱越丰富 → graph_walk 覆盖越广 → 检索越准 → 回答越好 → 产生更多新知识 ### 10.3 维度二:检索优化 (Relevance Learning) ```python # 每次查询记录 query-doc 关联 query_log = { "query": "海上交通安全法与水污染防治法", "type": "compare", "strategy": "graph_retrieval", "top_docs": ["海上交通安全法.txt", "水污染防治法.txt"], "scores": [0.85, 0.82], "response_quality": "good", # 可选: 用户反馈/LLM 自评 "timestamp": "2026-04-06T13:00:00" } ``` **用途:** | 机制 | 说明 | Lambda 表达 | |------|------|------------| | **热门缓存** | 相同问题直接返回 | `If(in_cache, cache_hit, full_search)` | | **同义词表** | "异同"↔"区别"↔"对比" 从 query_expand 积累 | 扩展 BM25 tokenizer | | **BM25 Boost** | 高频 query-doc 对加权 | 修改 IDF 分数 | | **失败记录** | 检索失败的 query 触发 Wiki 补充 | 异步 wiki_compile | ### 10.4 维度三:路由优化 (Strategy Learning) ```python # 记录每种分类+策略的效果 routing_stats = { "fact → keyword_search": {"total": 50, "success": 48, "avg_time": 0.5}, "compare → graph_retrieval": {"total": 20, "success": 18, "avg_time": 3.2}, "compare → fusion_search": {"total": 10, "success": 6, "avg_time": 5.1}, } # 动态调整路由权重 def adaptive_route(query_type): candidates = routing_stats.get_strategies(query_type) # 选成功率最高 + 耗时最短的策略 return max(candidates, key=lambda s: s.success_rate * 0.7 + (1/s.avg_time) * 0.3) ``` **效果**: 系统自动发现哪种策略对哪类问题最有效,逐渐优化路由决策。 ### 10.5 维度四:分类校准 ```python # 当用户纠正或回答不满意时,记录分类错误 # 例: "琼州海峡定线制" 被分为 fact 但实际需要 temporal correction_log = { "query": "琼州海峡定线制", "predicted_type": "fact", "correct_type": "temporal", # 从回答质量推断 } # 积累后: 更新规则分类器的模式 # re.search(r'定线制', q) → 'temporal' (新增规则) ``` ### 10.6 存储结构 ``` knowledge/maritime/ feedback/ query_log.jsonl # 查询日志 (append-only, 每次追加) synonym_map.json # 同义词映射 (从 query_expand 积累) query_cache.json # 热门查询缓存 (LRU, max 1000) routing_stats.json # 路由策略统计 (per type × strategy) classification_rules.json # 学习到的分类规则 ``` ### 10.7 增强触发时机 | 触发条件 | 动作 | Lambda 表达 | |---------|------|------------| | 回答引用 2+ 来源 | 提取新知识→Wiki | `If(multi_source, feedback, identity)` | | 每次查询完成 | 记录 query log | `Compose(search, log_query)` | | 相同查询出现 3+ 次 | 加入热门缓存 | `Tool("cache_check")` | | 检索结果为空 | 触发补充索引 | `If(empty_results, trigger_reindex, pass)` | | 每 100 次查询 | 更新路由统计 | `Tool("update_routing_stats")` | | Wiki 编译完成 | 刷新图谱缓存 | `Tool("reload_graph")` | ### 10.8 实现优先级 | 机制 | 难度 | 影响 | 优先级 | |------|------|------|--------| | 查询日志 (JSONL) | 低 | 数据基础 | **P0** | | Wiki 新关系写回 | 低 | 图谱持续增长 | **P0** | | 热门查询缓存 | 低 | 重复查询加速 | **P1** | | 路由成功率统计 | 中 | 自动调优路由 | **P1** | | 同义词表积累 | 中 | 扩展检索覆盖 | **P2** | | 分类规则学习 | 中 | 减少 LLM 分类 | **P2** | | BM25 Boost | 高 | 检索精度提升 | **P3** | ### 10.9 飞轮效应 ``` 更多查询 → 更多 query log → 更好的路由 → 更多知识反馈 → 更丰富的 Wiki + 图谱 → 更准的检索 → 更好的回答 → 用户更多使用 → 更多查询 ... ``` --- ## 十一、Wiki 编译优化项 ### 11.1 时间规范化 (Temporal Normalization) **问题**: 文档中大量相对时间表达("近年来"、"本规定自发布之日起施行"、"去年"),导致 temporal 类查询无法精确匹配。 **方案**: 在 wiki_compile.py 的 LLM prompt 中增加时间规范化指令: ``` ## 时间规范化要求 文档中所有相对时间表述,必须根据文件的发布日期转换为绝对时间: - "自发布之日起施行" → "自{文件发布年份}起施行" - "近年来" → "{发布年份前2-3年}至{发布年份}" - "去年" → "{发布年份-1}年" - "现行有效" → "截至{文件日期}现行有效" - "经最新修订" → "经{具体修订年份}年修订" 文件名中通常包含施行日期,如: "2017年8月31日起施行--交通运输部关于..." → 发布日期=2017年,所有相对时间以此为基准 ``` **实现**: 1. 从文件名中提取日期(正则 `\d{4}年\d+月\d+日`) 2. 将日期注入 prompt:`文件发布日期: {date}` 3. LLM 在摘要中将所有相对时间转为绝对时间 **影响**: temporal 类查询准确率预计从 82% → 92%+ ### 11.2 Wiki 质量提升路径 | 优化 | 效果 | 依赖 | |------|------|------| | 时间规范化 | temporal 准确率 +10% | 修改 prompt | | 换更强模型重编译 | 摘要质量、关系准确率全面提升 | 72B 模型或 API | | 关系质量过滤 | 去除泛化关系(如"实行社会主义制度") | 规则过滤 | | 实体去重/合并 | "XX海事局"合并为具体名称 | 后处理脚本 | --- ## 十二、验证方式 1. 用之前的 5 个标准问题做 A/B 对比 2. `search_unified.py` 独立测试模式: 输出分类结果 + 选中策略 + 检索结果 3. Web 页面显示: 问题分类标签 + 使用的策略 + 耗时分解 4. 降级测试: 关闭各组件验证 fallback 链 5. 自增强验证: 连续问 20 个问题,观察图谱增长和缓存命中率