简介一份基于模块图的检索增强生成RAG系统资源包源自微软GraphRAG项目面向AI开发者和研究人员旨在从非结构化文本中提取结构化数据并利用知识图谱存储结构增强LLM在私有数据上的推理表现。资源共654个文件压缩包约10.26MB其中包含430个Python源码、45个Markdown文档、38个JSON配置另有YAML、CSV、Parquet等数据文件分别承担核心实现、使用说明、参数配置和中间数据存储等角色。目前已有125人浏览学习适合希望上手GraphRAG但缺乏完整参考的开发者。包内附Dockerfile、多类输入输出样例及脚本可帮助读者快速梳理从文本解析、实体识别、图谱构建到检索融合的完整链路同时官方提示GraphRAG索引操作成本较高建议从小规模数据起步并先阅读全部文档。1. 基于模块图的 RAG 系统把检索链路从黑匣子变成一张可追踪的有向图跑过本地知识库问答项目的人大概率都经历过同一个场景Prompt 里明明拼了检索结果大模型还是答得离谱你也说不清是向量检索没召回、重排把相关文档裁掉、还是生成阶段在幻觉。传统 RAG 是一条线性流水线——切分脚本、向量库、拼接 Prompt环节之间全是黑匣子而基于模块图的 RAG 会把查询改写、意图路由、混合检索、重排序、生成全部拆成独立节点用一张有向图定义它们怎么串、怎么分支、失败怎么兜底。这套设计要解决的核心问题正是两个常见的 RAG 瓶颈知识割裂导致召回不全以及链路不可观测导致排障靠猜。适合正在做知识库 RAG 项目、对命中率和排查效率不满意的开发者。2. 模块图设计把 RAG 链路拆成四类节点和一张有向图2.1 为什么模块图优于线性流水线多数 RAG 项目是从线性链起跑的加载文档 → 切分 → 向量化 → 检索 → 拼 Prompt → 生成。这条链在演示环境里没问题一进真实知识库就暴露三个硬伤。第一单点故障不可见检索召回为空时后面的生成节点照样会把“空上下文”当成一种正常输入最后吐出一段看着合理、实则编造的答案。第二没有条件分支能力命中率低时应该改写查询重试检索为空时应该走兜底而不是一条路走到黑。第三状态没有接力路由判断、检索分数、证据出处都散在局部变量里后面的节点拿不到日志里也查不到。模块图的思路是把执行流程显式建模成一张有向图。每个节点只操作一个共享状态对象写完状态后由“边”决定下一个节点条件分支挂在边上日志里能复现完整执行轨迹。顺着这个思路再向前走一步就是 GraphRAG 那一类做法把知识索引建成图再在图上做多跳检索。模块图和 GraphRAG 并不冲突——你用 GraphRAG 建了索引查询侧的编排层依然需要这种可路由、可追踪的模块图。LangChain 生态里的 LangGraph 也在做类似抽象但如果你不想被框架绑定自实现一个百来行的执行器更透明出问题时定位也更直接。2.2 四类核心节点路由、变换、检索、生成我习惯把节点按职责分成四类这也是这套模块图 RAG 的基础抽象节点类型职责输入输出典型场景路由节点判断意图或条件决定下一跳query、意图字段route 字段区分产品参数问答和知识库问答变换节点改写、拆分、扩展查询querysub_queries 列表把对比问题拆成多个检索子查询检索节点从不同索引召回候选sub_queriesretrieved_docs、分数向量召回、BM25 召回并联生成节点组装证据并调用 LLMreranked_docs、routegeneration生成答案标注证据出处为了让节点能随意组合我给它们定义了一个很小的基类全部逻辑只有两个方法# node_base.py class BaseNode: 所有模块图节点的基类。 def __init__(self, name: str): self.name name def run(self, state: dict) - dict: # 读取 state执行自己的逻辑把结果写回 state raise NotImplementedError def next(self, state: dict) - str: # 返回下一个节点名返回空字符串表示走默认边 return run干实事next留给路由用。之所以把分支判断从run里的 if-else 挪到next是为了让执行轨迹在图上可见——每个节点“为什么走这条边”变成了可记录的状态而不是埋在大函数里的逻辑。四类节点继承基类后普通节点只实现run路由节点额外实现next。2.3 状态对象、边表和条件分支节点之间共享一个状态字典我叫它 GraphState。实际项目里我倾向直接用朴素 dict因为节点一多dataclass 的字段约束反而变成负担多人协作时再用 dataclass 也来得及。一个典型状态长这样# state.py from dataclasses import dataclass, field dataclass class GraphState: query: str intent: str sub_queries: list field(default_factorylist) retrieved_docs: list field(default_factorylist) reranked_docs: list field(default_factorylist) rerank_scores: dict field(default_factorydict) route: str vector generation: str guard_flag: str 这张图的默认边只有五跳start → query_rewrite → hybrid_retrieve → rerank → generate → end。条件分支只挂在两个位置query_rewrite 之后如果子查询拆分失败直接跳过检索走兜底rerank 之后如果最高分低于阈值走 empty_bridge 而不进生成。增删分支只需要改边表节点内部逻辑不用动这是模块图对比硬编码 if-else 的最大优势。我还会让每个节点写回 state 时带一个xx_meta后缀比如retrieve_meta、rerank_meta把这次执行用的参数、召回数量、耗时全部存下来。下面一章就基于这个抽象写一个能真正跑起来的图执行器。3. 落地实现用一个小图执行器把节点串成流程3.1 图执行器与核心执行循环执行器不关心节点内部逻辑只负责三件事按名字找节点、按边表推进、记录轨迹。最小实现长这样# graph_engine.py from collections import defaultdict class ModuleGraph: def __init__(self): self.nodes {} self.default_next {} self.condition_rules defaultdict(dict) def add_node(self, name, node, default_next): self.nodes[name] node self.default_next[name] default_next def add_condition(self, node_name, field, route_map): self.condition_rules[node_name][field] field self.condition_rules[node_name][map] route_map def execute(self, state): cur start trace [] while cur and cur ! end: node self.nodes[cur] state node.run(state) trace.append({ node: cur, state_snapshot: {k: v for k, v in state.items() if k ! trace} }) rule self.condition_rules.get(cur) if rule: field_val str(state.get(rule[field], )) cur rule[map].get(field_val, self.default_next[cur]) else: cur self.default_next[cur] state[trace] trace return state执行循环的要点是“节点只认 state不认上下文”。while 循环里每次只做两件事调当前节点的run然后按条件路由或默认边确定下一跳。state_snapshot在每个节点执行后打一次快照这是排错的主力数据字段名取自 state 自身所以以后加节点日志结构也不用变。add_condition的两个参数要说明一下field是确定路由依据的字段名route_map是字段值到目标节点名的映射。比如intentproduct时跳到structured_queryintentknowledge时跳到hybrid_retrieve这比在节点内部写 if-else 清晰得多。start 节点我通常会做成只做初始化的节点把原始 query 放进 state生成 request_id不承担任何业务。3.2 查询改写节点与混合检索节点先看查询改写。它的价值是解决对比型、多跳型问题“一条 Query 检索不全”的痛点# query_rewrite.py class QueryRewriteNode(BaseNode): def __init__(self, name, llm, max_sub_queries3): super().__init__(name) self.llm llm self.max_sub_queries max_sub_queries def run(self, state): prompt ( 把下面的问题拆成最多 {} 个独立的检索子查询 每个子查询必须语义完整适合向量检索。\n 问题{}\n只输出子查询每行一个不要编号。 ).format(self.max_sub_queries, state[query]) text self.llm.complete(prompt, temperature0) subs [ln.strip() for ln in text.strip().split(\n) if ln.strip()] state[sub_queries] subs[:self.max_sub_queries] state[rewrite_meta] {count: len(subs)} return statetemperature0在这里是必选项查询改写是确定性提取不需要模型发挥想象力。max_sub_queries我一般设 2~3设得太大容易把完整 Query 拆成碎片反而降低命中率。如果 LLM 输出带编号可以再加一层正则把开头的数字序号去掉。再看混合检索。它同时跑向量召回和 BM25 召回把两路结果融合再去重# retrievers.py class HybridRetrieverNode(BaseNode): def __init__(self, name, vector_store, bm25_index, top_k20, use_rrfTrue): super().__init__(name) self.vector_store vector_store self.bm25_index bm25_index self.top_k top_k self.use_rrf use_rrf def run(self, state): queries state.get(sub_queries) or [state[query]] vector_ranked [] keyword_ranked [] for q in queries: vector_ranked self.vector_store.search(q, top_kself.top_k) keyword_ranked self.bm25_index.search(q, top_kself.top_k) vector_ranked dedup_by_id(vector_ranked) keyword_ranked dedup_by_id(keyword_ranked) if self.use_rrf: merged rrf_fusion([vector_ranked, keyword_ranked], k60) else: merged merge_by_weight(vector_ranked, keyword_ranked, w(0.5, 0.5)) state[retrieved_docs] [doc_id for doc_id, _ in merged[:self.top_k]] state[retrieve_meta] { vector_count: len(vector_ranked), keyword_count: len(keyword_ranked), final_count: len(state[retrieved_docs]), } return stateuse_rrf是一个开关方便对比 RRF 融合和普通加权融合的差异。向量检索擅长语义近似和口语变体BM25 擅长精确术语——型号、SKU、接口名、报错码这类词一旦少了 BM25向量检索很容易把它们“语义化”成别的东西。两路并联后单通道的盲区被补上这是提高召回率的常用做法。3.3 路由节点、兜底节点和生成节点的组合路由节点负责判断用户问题走哪条通道。我一般是“LLM 意图分类 规则兜底”# route.py class IntentRouterNode(BaseNode): def __init__(self, name, llm): super().__init__(name) self.llm llm def run(self, state): prompt ( 判断用户的意图只输出一个词product / knowledge / chitchat。\n 用户问题{}.format(state[query]) ) state[intent] self.llm.complete(prompt, temperature0).strip() return state def next(self, state): mapping { product: structured_query, knowledge: hybrid_retrieve, chitchat: chitchat_bridge, } return mapping.get(state[intent], hybrid_retrieve)next是给执行器用的。执行器在execute循环里看到IntentRouterNode时会调用它的next拿到目标节点名然后跳过去。这跟在run里直接调另一个节点的写法有个关键区别执行轨迹里会明确记录一次路由决策图上是可见的边而不是代码里的函数调用排错时一眼就能看出走了哪条分支。兜底节点放在重排之后、生成之前防止无证据硬答# guard.py class EmptyGuardNode(BaseNode): def __init__(self, name, min_score0.3): super().__init__(name) self.min_score min_score def run(self, state): docs state.get(reranked_docs, []) scores state.get(rerank_scores, {}) if not docs or max(scores.values(), default0) self.min_score: state[guard_flag] empty state[generation] 知识库暂时没有找到可信依据建议换个说法再问。 return state def next(self, state): if state.get(guard_flag) empty: return end return min_score这个阈值有讲究不同重排模型的分数分布不一样凡是写死阈值的项目基本都会在换模型时翻车这一点第 5 章单列了一条。兜底节点的next返回空字符串执行器会自动落到默认边generate。生成节点负责把证据、路由信息、查询一起交给 LLM# generator.py class GeneratorNode(BaseNode): def __init__(self, name, llm, system_prompt): super().__init__(name) self.llm llm self.system_prompt system_prompt def run(self, state): evidence \n\n.join( f[{doc[id]}] {doc[content]} for doc in state[reranked_docs] ) prompt ( 请基于以下证据回答问题。证据不足时直接说明不要编造。\n 证据\n{}\n\n问题{}\n答案 ).format(evidence, state[query]) state[generation] self.llm.complete( prompt, systemself.system_prompt, temperature0.2 ) state[gen_meta] {evidence_count: len(state[reranked_docs])} return state生成节点最关键的一行是 prompt 里“证据不足时直接说明不要编造”。即便前面有兜底这句约束也要保留兜底拦截的只是“完全无证据”半证据、弱证据的情况只能靠生成节点自己把关。到这里四个核心节点都有了把它们拼成图只需要若干次add_node调用整个过程不到 150 行代码。4. 命中率提升查询改写、混合检索与重排序的三级配合4.1 查询改写先解决 Query 不适合检索的问题很多团队拿到 RAG 项目第一件事是调向量库的 top_k第二件事是换更大的模型第三件事是调 Prompt——结果往往不理想。我踩过一圈后的结论很一致先把 Query 本身改成适合检索的形式比什么都见效。比如“上海和北京分别用哪类传输工具”这种对比问题向量检索大概率只命中一边因为语义上它是个整体向量空间里很难同时照顾两个城市的检索需求。查询改写节点承担的就是这件事。它把一条复杂 Query 拆成多条独立的子查询每个子查询语义完整、可检索然后各查各的、最后融合。同时它还应该做同义扩展把“数据传输”扩展成“数据同步”“ETL”“CDC”这类词典优先LLM 只处理词典覆盖不到的写法。词典的好处是可控、零成本、不引入随机性LLM 改写则要承担“改了之后意思跑偏”的风险。但也不是所有 Query 都适合拆一句“你们的 API 限流策略是什么”拆成两个反而自损。我会先用意图模板筛掉简单问题只有对比、多跳、限定条件多的 Query 才进改写节点。改写节点还可以做一个保护动作比较改写前后 embedding 的余弦相似度变化太小时保留原 Query避免无效改写。4.2 混合检索和 RRF 融合为什么两路召回缺一不可混合检索的意思是“向量召回和 BM25 召回各自跑再用 RRF 合并排名”。RRF 只看名次不看分数绝对值天然解决两个通道分数尺度不一致的问题。实现很短# fusion.py def rrf_fusion(ranked_lists, k60): 把多组排序结果按 RRF 公式融合为一个排序。 score_map {} for ranked in ranked_lists: for rank, doc_id in enumerate(ranked): # rank 从 0 开始1 是为了避免除零 score_map[doc_id] score_map.get(doc_id, 0) 1.0 / (k rank 1) return sorted(score_map.items(), keylambda x: x[1], reverseTrue)k60是 RRF 的常见经验值文档在某个通道排名第 60对融合分数的贡献约等于另一通道排名第一的一半。实际项目里我会维持在 50~70不针对某个数据集反复调。RRF 还有个不容易察觉的好处两个通道中只要一个召回命中该文档就能以较低名次进入候选集正好补上“向量漏了但关键词能命中”的场景。融合之后接重排。重排使用 cross-encoder 模型把 query 和 document 成对输入输出一个相关性分数。我常用 bge-reranker-v2-m3 这一档中文和跨语言场景都够用。关键参数是top_n从 20 个候选里取 8 个进生成保留冗余让生成节点有得选候选质量足够稳定时再压到 5。重排节点的骨架如下# rerank.py class RerankNode(BaseNode): def __init__(self, name, reranker, top_n8): super().__init__(name) self.reranker reranker self.top_n top_n def run(self, state): pairs [(state[query], doc[content]) for doc in state[retrieved_docs]] scores self.reranker.compute_score(pairs) ordered sorted( zip(state[retrieved_docs], scores), keylambda x: x[1], reverseTrue )[:self.top_n] state[reranked_docs] [doc for doc, _ in ordered] state[rerank_scores] {doc[id]: float(score) for doc, score in ordered} return state这里有个容易忽略的点rerank_scores要保留到状态里不是用完就丢。后续的兜底节点和生成节点都要依赖它——分数是证据的一部分。4.3 用评估集验证命中率不量化就是玄学没有评估集谈命中率提升都是“玄学”。我会维护一个最小评估集挑 30~50 条真实用户问题逐条标注“这条问题对应哪个文档 ID”然后只看两个指标recallk检索召回集合里包含标注相关文档的比例k 取 20主要看检索环节。hitN重排后前 N 个结果里包含标注相关文档的比例N 取 5主要看精排环节。评估方式是全量遍历评估集把每个 query 的 trace 落盘再聚合成表格。评估集要固定版本否则换一轮 Embedding 模型两次结果的差异就解释不清了。我一般会把执行器包一层做成命令行工具跑python eval_runner.py --eval-set data/eval_v3.jsonl --graph-config configs/rag_graph.yaml --output results/eval_v3_result.json--graph-config控制节点参数方便做网格搜索——换个 top_k、换个 rerank top_n命中率变化在评估集上一目了然。做完一轮“查询改写 混合检索 重排”之后recall20 通常会有明显回升特别是多跳问题hit5 的变化很直观。具体数值因数据而异不要照抄别人的报告你的知识库分布不同唯一靠谱的动作是固定评估集、逐个参数验证。注意评估集一旦固定就不要随意改标注。改一次标注之前所有对比结论全部作废。我在这上面吃过亏现在评估集文件都有版本号每次只追加、不改旧行。5. 避坑与排查模块图 RAG 的六个翻车点5.1 状态被多个节点共享浅拷贝导致脏数据现象上一轮请求的reranked_docs混进了下一轮答案引用的证据根本不是本次检索出来的。 原因同一个 state 对象被多个节点复用某个节点用list.append原地改数据并发场景下两个请求共用一个对象数据互相污染。 解决执行器入口做一次copy.deepcopy(state)每个节点只对自己负责的 key 赋值。改列表前先复制state[retrieved_docs] state.get(retrieved_docs, [])[:]。我后来养成了习惯任何节点不能原地修改状态里的 list 或 dict要么整体替换要么复制后替换。5.2 条件路由嵌套太深图变成一团毛线现象边表越加越多分支分散在五六个节点上查问题时得扒着代码看半天才能确认走了哪条路径。 原因把节点内部可以做掉的判断全部做成了“图上的条件路由”控制流被过度铺开。 解决约束路由节点数量。我允许两到三个条件路由点分别放在检索前和精排后其余决策用状态字段表达。路由日志里只关心两三处分支、三五个可能值问题一下就好查了。记住图是给人看的不是给框架秀肌肉的。5.3 检索为空时不兜底生成节点硬编答案现象知识库根本没有相关内容模型却一本正经给出操作建议回答看着专业实际全是在编。 原因检索为空时仍走生成节点prompt 里拼进去的是空串模型只能自由发挥。 解决在重排后、生成前挂兜底节点。reranked_docs为空或最高分低于阈值时直接返回“知识库暂无依据”不再调用生成 LLM。这一步代价很小能把无效答案比例大幅压下来是模块图里最值得加的一个节点。5.4 重排阈值写死换模型立刻失效现象本地验证的阈值 0.3 放到生产后所有文档分数都低于阈值守卫形同虚设。 原因不同重排模型的分数分布差异极大。有的模型输出经过 sigmoid 归一到 0~1有的直接输出未归一化 logits范围能差一个数量级。 解决上线前先跑 50 条样本看分数分布再定阈值或者放弃绝对阈值改用相对策略——固定取前 N 名额外加一个“最高分低于整体分布 P10”的兜底判断。我在生产里更倾向后者抗模型替换的冲击。5.5 Embedding 模型版本混用检索效果莫名劣化现象代码没动命中率从稳定变成随机像中了邪。 原因查询侧加载的向量模型和索引侧不一致。有人重建过索引但查询侧加载的还是旧权重新旧向量不在同一向量空间检索等于随机取。 解决把模型名和版本写进索引 meta启动时做一次比对不一致直接拒绝启动。升级向量模型必须重建索引跑完回归评估才允许切线上流量。这套纪律能省掉很多“莫名其妙”的检索劣化问题。5.6 生成节点只拿到文档正文证据出处全部丢失现象答案像模像样用户追问“依据哪条文档”时系统给不出处也无法解释哪个低分证据被采用了。 原因拼 prompt 时只把文档正文放进去rerank_scores、doc_id、来源库名这些元数据全部丢掉了。 解决prompt 按固定格式渲染证据并附置信度只让模型采用高分证据低分证据降级为参考。同时把gen_meta里的证据 ID 清单和生成结果一起落日志事后可以逐条核对。6. 进阶给模块图 RAG 加可观测性和轻量自检回路6.1 结构化 trace每个节点干了什么一眼看到底模块图本身已经记录了轨迹进阶的一步是把它变成结构化日志。我在execute返回后把 trace 序列化进日志每条包含节点名、耗时的毫秒数、执行前后状态里关键字段的摘要。这样按 request_id 聚合就能复现一次问答的完整链路哪个节点耗时最长、哪个节点裁掉了多少候选、最终证据来自哪些文档。配合前面说的retrieve_meta、rerank_meta定位问题的效率比在代码里打断点高一个量级。给 trace 加耗时的做法很简单在execute循环里包一层计时即可。线上问题复盘时把 request_id 对应的 trace 捞出来几分钟就能确认是检索慢还是生成慢。我把这一步当作 RAG 上生产环境之前的必修课没有 trace 的 RAG 项目我都不敢接。6.2 轻量 LLM 自检答案不靠谱就自动走重试边模块图的另一个天然优势是支持 Agent 式的自检回路。我加了一个self_check节点放在生成之后把问题、答案、证据一起交给 LLM让它判断“答案是否被证据支持、有没有编造”。判断为 fail 时节点通过next返回query_rewrite让整个链路自动重跑一次只是这次改写策略更保守。这是 Agentic RAG 的一种轻量实现不依赖复杂计划器只利用图的边表加一圈重试。# self_check.py class SelfCheckNode(BaseNode): def __init__(self, name, llm): super().__init__(name) self.llm llm def run(self, state): prompt ( 判断答案是否被证据支持输出 pass 或 fail。\n 问题{}\n证据{}\n答案{} ).format(state[query], state.get(evidence_text, ), state[generation]) verdict self.llm.complete(prompt, temperature0).strip() state[self_check] verdict return state def next(self, state): if state.get(self_check) fail: return query_rewrite return 自检节点多消耗一次 LLM 调用但对高价值的问答场景很值。我一般只对“路由判定为 knowledge 且证据少于 3 条”的请求启用避免每个对话都翻倍成本。重试边不用多加代码原来的default_next已经指向end通过next返回query_rewrite就把图变成环了。这让我想起一次线上翻车用户反复问同一类产品参数问题命中率突然掉得厉害日志一查发现路由模型把所有问题都判定成了 chitchat直接走了闲聊旁路。因为路由模型对“参数”这个领域词不敏感而我又没在路由日志上做监控。从那以后我每次改动路由定义都会强制走一遍评估集回归并且要求路有结果必须落结构化日志评估集和 trace 成了这套模块图 RAG 的两条安全绳。希望这次的拆解能帮到你让你在 RAG 项目里少一点玄学多一点可复现的排障路径。本文还有配套的精品资源点击获取 模块图RAG：把检索链路从黑匣子变成可追踪的有向图
简介一份基于模块图的检索增强生成RAG系统资源包源自微软GraphRAG项目面向AI开发者和研究人员旨在从非结构化文本中提取结构化数据并利用知识图谱存储结构增强LLM在私有数据上的推理表现。资源共654个文件压缩包约10.26MB其中包含430个Python源码、45个Markdown文档、38个JSON配置另有YAML、CSV、Parquet等数据文件分别承担核心实现、使用说明、参数配置和中间数据存储等角色。目前已有125人浏览学习适合希望上手GraphRAG但缺乏完整参考的开发者。包内附Dockerfile、多类输入输出样例及脚本可帮助读者快速梳理从文本解析、实体识别、图谱构建到检索融合的完整链路同时官方提示GraphRAG索引操作成本较高建议从小规模数据起步并先阅读全部文档。1. 基于模块图的 RAG 系统把检索链路从黑匣子变成一张可追踪的有向图跑过本地知识库问答项目的人大概率都经历过同一个场景Prompt 里明明拼了检索结果大模型还是答得离谱你也说不清是向量检索没召回、重排把相关文档裁掉、还是生成阶段在幻觉。传统 RAG 是一条线性流水线——切分脚本、向量库、拼接 Prompt环节之间全是黑匣子而基于模块图的 RAG 会把查询改写、意图路由、混合检索、重排序、生成全部拆成独立节点用一张有向图定义它们怎么串、怎么分支、失败怎么兜底。这套设计要解决的核心问题正是两个常见的 RAG 瓶颈知识割裂导致召回不全以及链路不可观测导致排障靠猜。适合正在做知识库 RAG 项目、对命中率和排查效率不满意的开发者。2. 模块图设计把 RAG 链路拆成四类节点和一张有向图2.1 为什么模块图优于线性流水线多数 RAG 项目是从线性链起跑的加载文档 → 切分 → 向量化 → 检索 → 拼 Prompt → 生成。这条链在演示环境里没问题一进真实知识库就暴露三个硬伤。第一单点故障不可见检索召回为空时后面的生成节点照样会把“空上下文”当成一种正常输入最后吐出一段看着合理、实则编造的答案。第二没有条件分支能力命中率低时应该改写查询重试检索为空时应该走兜底而不是一条路走到黑。第三状态没有接力路由判断、检索分数、证据出处都散在局部变量里后面的节点拿不到日志里也查不到。模块图的思路是把执行流程显式建模成一张有向图。每个节点只操作一个共享状态对象写完状态后由“边”决定下一个节点条件分支挂在边上日志里能复现完整执行轨迹。顺着这个思路再向前走一步就是 GraphRAG 那一类做法把知识索引建成图再在图上做多跳检索。模块图和 GraphRAG 并不冲突——你用 GraphRAG 建了索引查询侧的编排层依然需要这种可路由、可追踪的模块图。LangChain 生态里的 LangGraph 也在做类似抽象但如果你不想被框架绑定自实现一个百来行的执行器更透明出问题时定位也更直接。2.2 四类核心节点路由、变换、检索、生成我习惯把节点按职责分成四类这也是这套模块图 RAG 的基础抽象节点类型职责输入输出典型场景路由节点判断意图或条件决定下一跳query、意图字段route 字段区分产品参数问答和知识库问答变换节点改写、拆分、扩展查询querysub_queries 列表把对比问题拆成多个检索子查询检索节点从不同索引召回候选sub_queriesretrieved_docs、分数向量召回、BM25 召回并联生成节点组装证据并调用 LLMreranked_docs、routegeneration生成答案标注证据出处为了让节点能随意组合我给它们定义了一个很小的基类全部逻辑只有两个方法# node_base.py class BaseNode: 所有模块图节点的基类。 def __init__(self, name: str): self.name name def run(self, state: dict) - dict: # 读取 state执行自己的逻辑把结果写回 state raise NotImplementedError def next(self, state: dict) - str: # 返回下一个节点名返回空字符串表示走默认边 return run干实事next留给路由用。之所以把分支判断从run里的 if-else 挪到next是为了让执行轨迹在图上可见——每个节点“为什么走这条边”变成了可记录的状态而不是埋在大函数里的逻辑。四类节点继承基类后普通节点只实现run路由节点额外实现next。2.3 状态对象、边表和条件分支节点之间共享一个状态字典我叫它 GraphState。实际项目里我倾向直接用朴素 dict因为节点一多dataclass 的字段约束反而变成负担多人协作时再用 dataclass 也来得及。一个典型状态长这样# state.py from dataclasses import dataclass, field dataclass class GraphState: query: str intent: str sub_queries: list field(default_factorylist) retrieved_docs: list field(default_factorylist) reranked_docs: list field(default_factorylist) rerank_scores: dict field(default_factorydict) route: str vector generation: str guard_flag: str 这张图的默认边只有五跳start → query_rewrite → hybrid_retrieve → rerank → generate → end。条件分支只挂在两个位置query_rewrite 之后如果子查询拆分失败直接跳过检索走兜底rerank 之后如果最高分低于阈值走 empty_bridge 而不进生成。增删分支只需要改边表节点内部逻辑不用动这是模块图对比硬编码 if-else 的最大优势。我还会让每个节点写回 state 时带一个xx_meta后缀比如retrieve_meta、rerank_meta把这次执行用的参数、召回数量、耗时全部存下来。下面一章就基于这个抽象写一个能真正跑起来的图执行器。3. 落地实现用一个小图执行器把节点串成流程3.1 图执行器与核心执行循环执行器不关心节点内部逻辑只负责三件事按名字找节点、按边表推进、记录轨迹。最小实现长这样# graph_engine.py from collections import defaultdict class ModuleGraph: def __init__(self): self.nodes {} self.default_next {} self.condition_rules defaultdict(dict) def add_node(self, name, node, default_next): self.nodes[name] node self.default_next[name] default_next def add_condition(self, node_name, field, route_map): self.condition_rules[node_name][field] field self.condition_rules[node_name][map] route_map def execute(self, state): cur start trace [] while cur and cur ! end: node self.nodes[cur] state node.run(state) trace.append({ node: cur, state_snapshot: {k: v for k, v in state.items() if k ! trace} }) rule self.condition_rules.get(cur) if rule: field_val str(state.get(rule[field], )) cur rule[map].get(field_val, self.default_next[cur]) else: cur self.default_next[cur] state[trace] trace return state执行循环的要点是“节点只认 state不认上下文”。while 循环里每次只做两件事调当前节点的run然后按条件路由或默认边确定下一跳。state_snapshot在每个节点执行后打一次快照这是排错的主力数据字段名取自 state 自身所以以后加节点日志结构也不用变。add_condition的两个参数要说明一下field是确定路由依据的字段名route_map是字段值到目标节点名的映射。比如intentproduct时跳到structured_queryintentknowledge时跳到hybrid_retrieve这比在节点内部写 if-else 清晰得多。start 节点我通常会做成只做初始化的节点把原始 query 放进 state生成 request_id不承担任何业务。3.2 查询改写节点与混合检索节点先看查询改写。它的价值是解决对比型、多跳型问题“一条 Query 检索不全”的痛点# query_rewrite.py class QueryRewriteNode(BaseNode): def __init__(self, name, llm, max_sub_queries3): super().__init__(name) self.llm llm self.max_sub_queries max_sub_queries def run(self, state): prompt ( 把下面的问题拆成最多 {} 个独立的检索子查询 每个子查询必须语义完整适合向量检索。\n 问题{}\n只输出子查询每行一个不要编号。 ).format(self.max_sub_queries, state[query]) text self.llm.complete(prompt, temperature0) subs [ln.strip() for ln in text.strip().split(\n) if ln.strip()] state[sub_queries] subs[:self.max_sub_queries] state[rewrite_meta] {count: len(subs)} return statetemperature0在这里是必选项查询改写是确定性提取不需要模型发挥想象力。max_sub_queries我一般设 2~3设得太大容易把完整 Query 拆成碎片反而降低命中率。如果 LLM 输出带编号可以再加一层正则把开头的数字序号去掉。再看混合检索。它同时跑向量召回和 BM25 召回把两路结果融合再去重# retrievers.py class HybridRetrieverNode(BaseNode): def __init__(self, name, vector_store, bm25_index, top_k20, use_rrfTrue): super().__init__(name) self.vector_store vector_store self.bm25_index bm25_index self.top_k top_k self.use_rrf use_rrf def run(self, state): queries state.get(sub_queries) or [state[query]] vector_ranked [] keyword_ranked [] for q in queries: vector_ranked self.vector_store.search(q, top_kself.top_k) keyword_ranked self.bm25_index.search(q, top_kself.top_k) vector_ranked dedup_by_id(vector_ranked) keyword_ranked dedup_by_id(keyword_ranked) if self.use_rrf: merged rrf_fusion([vector_ranked, keyword_ranked], k60) else: merged merge_by_weight(vector_ranked, keyword_ranked, w(0.5, 0.5)) state[retrieved_docs] [doc_id for doc_id, _ in merged[:self.top_k]] state[retrieve_meta] { vector_count: len(vector_ranked), keyword_count: len(keyword_ranked), final_count: len(state[retrieved_docs]), } return stateuse_rrf是一个开关方便对比 RRF 融合和普通加权融合的差异。向量检索擅长语义近似和口语变体BM25 擅长精确术语——型号、SKU、接口名、报错码这类词一旦少了 BM25向量检索很容易把它们“语义化”成别的东西。两路并联后单通道的盲区被补上这是提高召回率的常用做法。3.3 路由节点、兜底节点和生成节点的组合路由节点负责判断用户问题走哪条通道。我一般是“LLM 意图分类 规则兜底”# route.py class IntentRouterNode(BaseNode): def __init__(self, name, llm): super().__init__(name) self.llm llm def run(self, state): prompt ( 判断用户的意图只输出一个词product / knowledge / chitchat。\n 用户问题{}.format(state[query]) ) state[intent] self.llm.complete(prompt, temperature0).strip() return state def next(self, state): mapping { product: structured_query, knowledge: hybrid_retrieve, chitchat: chitchat_bridge, } return mapping.get(state[intent], hybrid_retrieve)next是给执行器用的。执行器在execute循环里看到IntentRouterNode时会调用它的next拿到目标节点名然后跳过去。这跟在run里直接调另一个节点的写法有个关键区别执行轨迹里会明确记录一次路由决策图上是可见的边而不是代码里的函数调用排错时一眼就能看出走了哪条分支。兜底节点放在重排之后、生成之前防止无证据硬答# guard.py class EmptyGuardNode(BaseNode): def __init__(self, name, min_score0.3): super().__init__(name) self.min_score min_score def run(self, state): docs state.get(reranked_docs, []) scores state.get(rerank_scores, {}) if not docs or max(scores.values(), default0) self.min_score: state[guard_flag] empty state[generation] 知识库暂时没有找到可信依据建议换个说法再问。 return state def next(self, state): if state.get(guard_flag) empty: return end return min_score这个阈值有讲究不同重排模型的分数分布不一样凡是写死阈值的项目基本都会在换模型时翻车这一点第 5 章单列了一条。兜底节点的next返回空字符串执行器会自动落到默认边generate。生成节点负责把证据、路由信息、查询一起交给 LLM# generator.py class GeneratorNode(BaseNode): def __init__(self, name, llm, system_prompt): super().__init__(name) self.llm llm self.system_prompt system_prompt def run(self, state): evidence \n\n.join( f[{doc[id]}] {doc[content]} for doc in state[reranked_docs] ) prompt ( 请基于以下证据回答问题。证据不足时直接说明不要编造。\n 证据\n{}\n\n问题{}\n答案 ).format(evidence, state[query]) state[generation] self.llm.complete( prompt, systemself.system_prompt, temperature0.2 ) state[gen_meta] {evidence_count: len(state[reranked_docs])} return state生成节点最关键的一行是 prompt 里“证据不足时直接说明不要编造”。即便前面有兜底这句约束也要保留兜底拦截的只是“完全无证据”半证据、弱证据的情况只能靠生成节点自己把关。到这里四个核心节点都有了把它们拼成图只需要若干次add_node调用整个过程不到 150 行代码。4. 命中率提升查询改写、混合检索与重排序的三级配合4.1 查询改写先解决 Query 不适合检索的问题很多团队拿到 RAG 项目第一件事是调向量库的 top_k第二件事是换更大的模型第三件事是调 Prompt——结果往往不理想。我踩过一圈后的结论很一致先把 Query 本身改成适合检索的形式比什么都见效。比如“上海和北京分别用哪类传输工具”这种对比问题向量检索大概率只命中一边因为语义上它是个整体向量空间里很难同时照顾两个城市的检索需求。查询改写节点承担的就是这件事。它把一条复杂 Query 拆成多条独立的子查询每个子查询语义完整、可检索然后各查各的、最后融合。同时它还应该做同义扩展把“数据传输”扩展成“数据同步”“ETL”“CDC”这类词典优先LLM 只处理词典覆盖不到的写法。词典的好处是可控、零成本、不引入随机性LLM 改写则要承担“改了之后意思跑偏”的风险。但也不是所有 Query 都适合拆一句“你们的 API 限流策略是什么”拆成两个反而自损。我会先用意图模板筛掉简单问题只有对比、多跳、限定条件多的 Query 才进改写节点。改写节点还可以做一个保护动作比较改写前后 embedding 的余弦相似度变化太小时保留原 Query避免无效改写。4.2 混合检索和 RRF 融合为什么两路召回缺一不可混合检索的意思是“向量召回和 BM25 召回各自跑再用 RRF 合并排名”。RRF 只看名次不看分数绝对值天然解决两个通道分数尺度不一致的问题。实现很短# fusion.py def rrf_fusion(ranked_lists, k60): 把多组排序结果按 RRF 公式融合为一个排序。 score_map {} for ranked in ranked_lists: for rank, doc_id in enumerate(ranked): # rank 从 0 开始1 是为了避免除零 score_map[doc_id] score_map.get(doc_id, 0) 1.0 / (k rank 1) return sorted(score_map.items(), keylambda x: x[1], reverseTrue)k60是 RRF 的常见经验值文档在某个通道排名第 60对融合分数的贡献约等于另一通道排名第一的一半。实际项目里我会维持在 50~70不针对某个数据集反复调。RRF 还有个不容易察觉的好处两个通道中只要一个召回命中该文档就能以较低名次进入候选集正好补上“向量漏了但关键词能命中”的场景。融合之后接重排。重排使用 cross-encoder 模型把 query 和 document 成对输入输出一个相关性分数。我常用 bge-reranker-v2-m3 这一档中文和跨语言场景都够用。关键参数是top_n从 20 个候选里取 8 个进生成保留冗余让生成节点有得选候选质量足够稳定时再压到 5。重排节点的骨架如下# rerank.py class RerankNode(BaseNode): def __init__(self, name, reranker, top_n8): super().__init__(name) self.reranker reranker self.top_n top_n def run(self, state): pairs [(state[query], doc[content]) for doc in state[retrieved_docs]] scores self.reranker.compute_score(pairs) ordered sorted( zip(state[retrieved_docs], scores), keylambda x: x[1], reverseTrue )[:self.top_n] state[reranked_docs] [doc for doc, _ in ordered] state[rerank_scores] {doc[id]: float(score) for doc, score in ordered} return state这里有个容易忽略的点rerank_scores要保留到状态里不是用完就丢。后续的兜底节点和生成节点都要依赖它——分数是证据的一部分。4.3 用评估集验证命中率不量化就是玄学没有评估集谈命中率提升都是“玄学”。我会维护一个最小评估集挑 30~50 条真实用户问题逐条标注“这条问题对应哪个文档 ID”然后只看两个指标recallk检索召回集合里包含标注相关文档的比例k 取 20主要看检索环节。hitN重排后前 N 个结果里包含标注相关文档的比例N 取 5主要看精排环节。评估方式是全量遍历评估集把每个 query 的 trace 落盘再聚合成表格。评估集要固定版本否则换一轮 Embedding 模型两次结果的差异就解释不清了。我一般会把执行器包一层做成命令行工具跑python eval_runner.py --eval-set data/eval_v3.jsonl --graph-config configs/rag_graph.yaml --output results/eval_v3_result.json--graph-config控制节点参数方便做网格搜索——换个 top_k、换个 rerank top_n命中率变化在评估集上一目了然。做完一轮“查询改写 混合检索 重排”之后recall20 通常会有明显回升特别是多跳问题hit5 的变化很直观。具体数值因数据而异不要照抄别人的报告你的知识库分布不同唯一靠谱的动作是固定评估集、逐个参数验证。注意评估集一旦固定就不要随意改标注。改一次标注之前所有对比结论全部作废。我在这上面吃过亏现在评估集文件都有版本号每次只追加、不改旧行。5. 避坑与排查模块图 RAG 的六个翻车点5.1 状态被多个节点共享浅拷贝导致脏数据现象上一轮请求的reranked_docs混进了下一轮答案引用的证据根本不是本次检索出来的。 原因同一个 state 对象被多个节点复用某个节点用list.append原地改数据并发场景下两个请求共用一个对象数据互相污染。 解决执行器入口做一次copy.deepcopy(state)每个节点只对自己负责的 key 赋值。改列表前先复制state[retrieved_docs] state.get(retrieved_docs, [])[:]。我后来养成了习惯任何节点不能原地修改状态里的 list 或 dict要么整体替换要么复制后替换。5.2 条件路由嵌套太深图变成一团毛线现象边表越加越多分支分散在五六个节点上查问题时得扒着代码看半天才能确认走了哪条路径。 原因把节点内部可以做掉的判断全部做成了“图上的条件路由”控制流被过度铺开。 解决约束路由节点数量。我允许两到三个条件路由点分别放在检索前和精排后其余决策用状态字段表达。路由日志里只关心两三处分支、三五个可能值问题一下就好查了。记住图是给人看的不是给框架秀肌肉的。5.3 检索为空时不兜底生成节点硬编答案现象知识库根本没有相关内容模型却一本正经给出操作建议回答看着专业实际全是在编。 原因检索为空时仍走生成节点prompt 里拼进去的是空串模型只能自由发挥。 解决在重排后、生成前挂兜底节点。reranked_docs为空或最高分低于阈值时直接返回“知识库暂无依据”不再调用生成 LLM。这一步代价很小能把无效答案比例大幅压下来是模块图里最值得加的一个节点。5.4 重排阈值写死换模型立刻失效现象本地验证的阈值 0.3 放到生产后所有文档分数都低于阈值守卫形同虚设。 原因不同重排模型的分数分布差异极大。有的模型输出经过 sigmoid 归一到 0~1有的直接输出未归一化 logits范围能差一个数量级。 解决上线前先跑 50 条样本看分数分布再定阈值或者放弃绝对阈值改用相对策略——固定取前 N 名额外加一个“最高分低于整体分布 P10”的兜底判断。我在生产里更倾向后者抗模型替换的冲击。5.5 Embedding 模型版本混用检索效果莫名劣化现象代码没动命中率从稳定变成随机像中了邪。 原因查询侧加载的向量模型和索引侧不一致。有人重建过索引但查询侧加载的还是旧权重新旧向量不在同一向量空间检索等于随机取。 解决把模型名和版本写进索引 meta启动时做一次比对不一致直接拒绝启动。升级向量模型必须重建索引跑完回归评估才允许切线上流量。这套纪律能省掉很多“莫名其妙”的检索劣化问题。5.6 生成节点只拿到文档正文证据出处全部丢失现象答案像模像样用户追问“依据哪条文档”时系统给不出处也无法解释哪个低分证据被采用了。 原因拼 prompt 时只把文档正文放进去rerank_scores、doc_id、来源库名这些元数据全部丢掉了。 解决prompt 按固定格式渲染证据并附置信度只让模型采用高分证据低分证据降级为参考。同时把gen_meta里的证据 ID 清单和生成结果一起落日志事后可以逐条核对。6. 进阶给模块图 RAG 加可观测性和轻量自检回路6.1 结构化 trace每个节点干了什么一眼看到底模块图本身已经记录了轨迹进阶的一步是把它变成结构化日志。我在execute返回后把 trace 序列化进日志每条包含节点名、耗时的毫秒数、执行前后状态里关键字段的摘要。这样按 request_id 聚合就能复现一次问答的完整链路哪个节点耗时最长、哪个节点裁掉了多少候选、最终证据来自哪些文档。配合前面说的retrieve_meta、rerank_meta定位问题的效率比在代码里打断点高一个量级。给 trace 加耗时的做法很简单在execute循环里包一层计时即可。线上问题复盘时把 request_id 对应的 trace 捞出来几分钟就能确认是检索慢还是生成慢。我把这一步当作 RAG 上生产环境之前的必修课没有 trace 的 RAG 项目我都不敢接。6.2 轻量 LLM 自检答案不靠谱就自动走重试边模块图的另一个天然优势是支持 Agent 式的自检回路。我加了一个self_check节点放在生成之后把问题、答案、证据一起交给 LLM让它判断“答案是否被证据支持、有没有编造”。判断为 fail 时节点通过next返回query_rewrite让整个链路自动重跑一次只是这次改写策略更保守。这是 Agentic RAG 的一种轻量实现不依赖复杂计划器只利用图的边表加一圈重试。# self_check.py class SelfCheckNode(BaseNode): def __init__(self, name, llm): super().__init__(name) self.llm llm def run(self, state): prompt ( 判断答案是否被证据支持输出 pass 或 fail。\n 问题{}\n证据{}\n答案{} ).format(state[query], state.get(evidence_text, ), state[generation]) verdict self.llm.complete(prompt, temperature0).strip() state[self_check] verdict return state def next(self, state): if state.get(self_check) fail: return query_rewrite return 自检节点多消耗一次 LLM 调用但对高价值的问答场景很值。我一般只对“路由判定为 knowledge 且证据少于 3 条”的请求启用避免每个对话都翻倍成本。重试边不用多加代码原来的default_next已经指向end通过next返回query_rewrite就把图变成环了。这让我想起一次线上翻车用户反复问同一类产品参数问题命中率突然掉得厉害日志一查发现路由模型把所有问题都判定成了 chitchat直接走了闲聊旁路。因为路由模型对“参数”这个领域词不敏感而我又没在路由日志上做监控。从那以后我每次改动路由定义都会强制走一遍评估集回归并且要求路有结果必须落结构化日志评估集和 trace 成了这套模块图 RAG 的两条安全绳。希望这次的拆解能帮到你让你在 RAG 项目里少一点玄学多一点可复现的排障路径。本文还有配套的精品资源点击获取