LAMBDA_RAG_PLAN.md 25 KB
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 查询分类器 (智能路由)
# 规则快速分类 (零成本)
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 表达式
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 数据结构
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 配置
# 统一检索配置
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：
# 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)
# 每次查询记录 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)
# 记录每种分类+策略的效果
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 维度四：分类校准
# 当用户纠正或回答不满意时，记录分类错误
# 例: "琼州海峡定线制" 被分为 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年，所有相对时间以此为基准
实现:
- 从文件名中提取日期（正则
\d{4}年\d+月\d+日） - 将日期注入 prompt：
文件发布日期: {date} - LLM 在摘要中将所有相对时间转为绝对时间
影响: temporal 类查询准确率预计从 82% → 92%+
11.2 Wiki 质量提升路径
| 优化 | 效果 | 依赖 |
|---|---|---|
| 时间规范化 | temporal 准确率 +10% | 修改 prompt |
| 换更强模型重编译 | 摘要质量、关系准确率全面提升 | 72B 模型或 API |
| 关系质量过滤 | 去除泛化关系（如"实行社会主义制度"） | 规则过滤 |
| 实体去重/合并 | "XX海事局"合并为具体名称 | 后处理脚本 |
十二、验证方式
- 用之前的 5 个标准问题做 A/B 对比
search_unified.py独立测试模式: 输出分类结果 + 选中策略 + 检索结果- Web 页面显示: 问题分类标签 + 使用的策略 + 耗时分解
- 降级测试: 关闭各组件验证 fallback 链
- 自增强验证: 连续问 20 个问题，观察图谱增长和缓存命中率