简介：这份资源是一套基于模块图的检索增强生成（RAG）系统实现，核心为GraphRAG数据管道与转换套件，面向希望利用LLM从非结构化文本中抽取结构化知识、增强私有数据推理能力的开发者与研究者，适合具备一定Python与LLM应用基础的中高级读者。压缩包共654个文件，约10.26MB，以430个py源码为主体，辅以45个md文档、38个json配置、22个txt说明、18个yml与4个yaml流程定义，另有parquet、lance、manifest、txn等图索引与存储相关文件，以及少量ipynb示例、png图示和Dockerfile、sh脚本等部署辅助内容。资源围绕知识图谱存储结构增强LLM输出展开，包含完整的数据转换流程、索引构建模块与解决方案加速器示例，可帮助读者理解GraphRAG的端到端实现思路、成本控制要点与工程目录组织方式。目前已有125人学习，适合作为RAG与知识图谱结合方向的实践参考。
1. 模块图驱动的 RAG 系统：从检索瓶颈到结构化知识库的落地拆解
如果你已经用 LangChain 或 LlamaIndex 搭过一版 RAG，大概率遇到过这种场景：用户问“A 模块的接口变更会影响哪些下游服务”，向量检索返回了一堆语义相似但拓扑无关的文档片段，LLM 拿着这些碎片拼出一个看似合理实则错误的答案。问题不在 embedding 模型，也不在 LLM，而在于你把系统当成了扁平的文本库，忽略了模块之间天然存在的依赖关系。这份资源给出的方案，是用模块图（Module Graph）作为检索骨架，把 RAG 从“相似度匹配”升级为“结构感知检索”。它适合已经跑通基础 RAG、但被 rag 瓶颈卡住的工程师，也适合正在设计 kg 知识库与 rag 知识库混合架构的团队。核心思路一句话：先建图，再检索，最后生成。
2. 模块图怎么建：从代码仓库到结构化知识库的转换链路
2.1 为什么扁平向量库在模块依赖查询上会翻车
传统 RAG 的检索单元是 chunk，chunk 的边界由固定长度或语义分割决定，切完之后模块之间的调用关系、继承关系、数据流向全部丢失。当查询涉及“影响面分析”“调用链追溯”“变更传播”这类拓扑问题时，向量相似度几乎失效。我见过一个真实案例：某团队用纯向量方案做微服务文档问答，问“订单服务依赖哪些下游”，返回的是三篇都提到“订单”和“下游”的文档，但真正的依赖列表在另一篇没被召回的架构说明里。模块图的价值在于，它把“谁依赖谁”变成可遍历的边，检索时先定位节点，再沿边扩展，最后才交给 LLM 做归纳。这和 ontology rag 的思路一致：用本体或图结构约束检索范围，而不是让向量空间自由发挥。
2.2 从源码提取模块图的三个关键步骤
建图不是让你手动画 UML，而是从现有代码仓库自动抽取。常见做法是解析 import 语句、函数调用和配置文件。下面是一个 Python 脚本示例，用 ast 模块提取模块间依赖：
import ast import os from pathlib import Path def extract_imports(filepath): """解析单个 Python 文件，返回其导入的模块名列表""" with open(filepath, 'r', encoding='utf-8') as f: try: tree = ast.parse(f.read()) except SyntaxError: return [] imports = [] for node in ast.walk(tree): if isinstance(node, ast.Import): for alias in node.names: imports.append(alias.name.split('.')[0]) elif isinstance(node, ast.ImportFrom): if node.module: imports.append(node.module.split('.')[0]) return imports def build_module_graph(root_dir): """遍历项目目录，构建模块依赖图""" graph = {} root = Path(root_dir) for py_file in root.rglob('*.py'): module_name = py_file.stem deps = extract_imports(py_file) # 只保留项目内部模块，过滤标准库和第三方包 internal_deps = [d for d in deps if (root / f"{d}.py").exists() or (root / d).is_dir()] graph[module_name] = list(set(internal_deps)) return graph if __name__ == '__main__': g = build_module_graph('./your_project') for mod, deps in g.items(): print(f"{mod} -> {deps}")这段代码的逻辑说明：extract_imports用 AST 解析避免正则匹配的误判，能区分import x和from x import y；build_module_graph遍历目录时只保留项目内部依赖，防止把os、requests这类外部包混进图里。参数方面，root_dir指向你的源码根目录，如果项目是多层包结构，需要把module_name改成相对路径拼接，否则同名文件会覆盖。跑完之后你会得到一个邻接表，这就是模块图的雏形。
2.3 把图结构写入知识库：节点、边与属性的存储选型
拿到邻接表之后，下一步是决定存哪里。如果你只是做检索增强，不需要复杂图查询，用 NetworkX 在内存里维护，序列化成 JSON 挂到向量库的 metadata 里就够了。但如果要支持多跳推理和动态更新，建议上 Neo4j 或 NebulaGraph。我一般会这样设计 schema：
| 元素 | 属性 | 说明 |
|---|---|---|
| 节点 Module | name, path, loc, owner | loc 是代码行数，owner 从 CODEOWNERS 提取 |
| 边 DEPENDS_ON | weight, call_count | weight 可以是调用频次或耦合度 |
| 边 CONTAINS | order | 用于包与子模块的层级关系 |
写入 Neo4j 的 Cypher 语句示例：
// 创建模块节点 MERGE (m:Module {name: $name}) SET m.path = $path, m.loc = $loc // 创建依赖边 MATCH (a:Module {name: $src}), (b:Module {name: $dst}) MERGE (a)-[r:DEPENDS_ON]->(b) SET r.weight = $weight注意MERGE而不是CREATE，避免重复插入。参数$weight可以先用 1，后续根据调用次数动态调整。这一步做完，你的 rag 知识库就不再是纯文本集合，而是带拓扑信息的结构知识库。区分点在于：rag 知识库擅长语义召回，结构知识库擅长关系遍历，两者结合才能回答“改了 A 会影响谁”这类问题。
3. 检索层实战：图遍历与向量召回怎么协同
3.1 查询路由：先判断该走向量还是走图
不是所有查询都需要图遍历。我的做法是在检索入口加一个轻量分类器，用规则或小模型判断查询类型。规则可以这样写：
def route_query(query): """根据查询关键词决定检索策略""" graph_keywords = ['依赖', '调用', '影响', '下游', '上游', '变更', '传播'] if any(kw in query for kw in graph_keywords): return 'graph_first' return 'vector_first'逻辑说明：graph_first模式下，先用实体链接找到查询中的模块名，再从该节点出发做 1-2 跳遍历，把沿途节点对应的文档片段作为上下文。vector_first模式就是常规向量召回。参数上，跳数建议从 1 开始，超过 2 跳噪声会急剧增加。这个路由函数很粗糙，但实测能拦住大部分拓扑类查询，比纯向量方案准确率提升明显。
3.2 图遍历检索的代码实现与参数调优
假设你已经把图存进了 NetworkX 对象，下面是一个两阶段检索的完整示例：
import networkx as nx from sentence_transformers import SentenceTransformer model = SentenceTransformer('all-MiniLM-L6-v2') def graph_retrieve(graph, query, module_name, max_hops=2): """从指定模块出发，收集 max_hops 跳内的所有节点""" if module_name not in graph: return [] # 获取指定跳数内的邻居 neighbors = nx.single_source_shortest_path_length( graph, module_name, cutoff=max_hops ) # 按距离排序，近的优先 sorted_nodes = sorted(neighbors.items(), key=lambda x: x[1]) return [node for node, dist in sorted_nodes] def hybrid_search(graph, vector_store, query, module_name=None): """混合检索：图召回 + 向量召回""" candidates = [] if module_name: # 图召回 graph_nodes = graph_retrieve(graph, query, module_name) for node in graph_nodes: docs = vector_store.get_by_metadata({'module': node}) candidates.extend(docs) # 向量召回补充 query_emb = model.encode(query) vector_docs = vector_store.search(query_emb, top_k=5) candidates.extend(vector_docs) # 去重 seen = set() unique = [] for doc in candidates: if doc['id'] not in seen: seen.add(doc['id']) unique.append(doc) return unique[:10]逻辑说明：graph_retrieve用single_source_shortest_path_length拿到指定跳数内的所有节点，按距离排序保证近邻优先。hybrid_search先做图召回，再用向量召回兜底，最后去重截断。参数max_hops控制遍历深度，top_k控制向量召回数量，unique[:10]是最终送给 LLM 的上下文窗口。我一般会把max_hops设成 2，top_k设成 5，上下文总长度控制在 3000 token 以内。如果发现召回不足，优先调大top_k而不是max_hops，因为跳数增加带来的噪声比向量召回更不可控。
3.3 把图路径转成 LLM 能吃的上下文
图遍历拿到的是一堆节点名和边，直接丢给 LLM 效果很差。需要把路径转成自然语言描述。常见做法是模板化：
def path_to_text(graph, source, target, path): """把图路径转成自然语言""" edges = [] for i in range(len(path) - 1): edge_data = graph.get_edge_data(path[i], path[i+1]) weight = edge_data.get('weight', 1) if edge_data else 1 edges.append(f"{path[i]} 依赖 {path[i+1]}（权重 {weight}）") return f"从 {source} 到 {target} 的依赖路径：" + "，".join(edges)这样 LLM 拿到的是“A 依赖 B，B 依赖 C”的结构化描述，而不是一堆孤立文本。实测这种上下文对“影响面分析”类问题的回答准确率提升很大。注意模板不要写得太死，保留边属性字段，方便后续扩展。
4. 避坑与排查：模块图 RAG 落地时最容易翻车的五个点
4.1 现象：图建好了但检索时找不到模块名
原因：实体链接没做归一化。代码里的模块名可能是order_service，用户查询写的是“订单服务”，直接匹配不上。解决：建一个别名表，把中文名、驼峰、下划线、缩写都映射到标准节点名。我一般会在图节点上加aliases属性，检索前先做一轮别名替换。
4.2 现象：图遍历召回大量无关节点
原因：max_hops设太大，或者图里有超级节点（被大量模块依赖的公共库）。解决：限制单节点出度，超过阈值的边降权或截断；同时把max_hops降到 1 先试。另一个技巧是给边加方向权重，只沿DEPENDS_ON正向遍历，反向遍历单独处理。
4.3 现象：向量召回和图召回结果冲突，LLM 回答自相矛盾
原因：两路召回没有融合策略，直接拼接导致上下文里有矛盾信息。解决：在 prompt 里明确优先级，比如“以下信息中，图路径描述的结构关系优先于文本片段的语义描述”。或者用重排序模型对合并后的候选做统一打分，按分数截断。
4.4 现象：代码更新后图没同步，检索结果过时
原因：建图是离线批处理，没有增量更新机制。解决：把建图脚本挂到 CI 上，每次合并请求触发增量解析，只更新变更文件对应的节点和边。Neo4j 可以用MERGE做 upsert，NetworkX 需要自己写 diff 逻辑。我一般会在图节点上加last_modified时间戳，检索时过滤掉过期节点。
4.5 现象：Mac 上搭建时 Neo4j 内存不够，图查询超时
原因：Neo4j 默认堆内存配置偏小，模块图节点多了之后遍历变慢。解决：改neo4j.conf里的dbms.memory.heap.max_size，建议至少 2G。如果只是本地开发，用 NetworkX 加 JSON 序列化更轻量，没必要上图数据库。怎么在 mac 上搭建 rag 知识库这个问题，我的建议是先用内存图跑通流程，确认检索效果后再考虑持久化方案。
5. 进阶技巧：用图结构做检索结果的可解释性验证
5.1 给每个回答附上图路径证据
RAG 系统最被诟病的是黑匣子，LLM 说“A 影响 B”，但你不知道它从哪推出来的。模块图天然适合做可解释性：每次图召回都把遍历路径存下来，生成回答时一并返回。用户看到的不只是答案，还有“A -> B -> C”的推理链。实现上，在hybrid_search返回结果里加一个evidence_path字段，前端渲染成可展开的依赖链。这个功能对排查幻觉特别有用，我靠它定位过好几次 LLM 编造依赖关系的问题。
5.2 用图中心性指标做检索结果重排序
图不只是检索入口，还能给召回结果打分。常见做法是计算节点的 PageRank 或介数中心性，中心性高的模块在重排序时加权。代码示例：
import networkx as nx def rerank_by_centrality(graph, candidates): """按节点中心性对候选文档重排序""" pagerank = nx.pagerank(graph) scored = [] for doc in candidates: module = doc.get('module', '') score = pagerank.get(module, 0.0) scored.append((doc, score)) scored.sort(key=lambda x: x[1], reverse=True) return [doc for doc, _ in scored]逻辑说明：nx.pagerank返回每个节点的重要性分数，被依赖越多的模块分数越高。重排序时把图分数和向量相似度做加权融合，权重可以设成 0.3 图 + 0.7 向量，具体比例看你的查询类型分布。参数上，PageRank 的alpha默认 0.85，一般不用改。这个技巧在“核心模块变更影响分析”场景下效果很好，能把真正关键的节点排到前面。
5.3 验证检索质量：构造拓扑查询测试集
光靠感觉调参不靠谱，我一般会手工构造 20-30 条拓扑查询做回归测试。测试集格式：
| 查询 | 期望召回模块 | 期望路径 |
|---|---|---|
| 订单服务依赖哪些下游 | order_service, payment, inventory | order_service -> payment |
| 修改用户模块会影响谁 | user, auth, profile | user -> auth -> profile |
每次调整max_hops、top_k或重排序权重后，跑一遍测试集看召回率和路径准确率。这个习惯帮我避免了好几次“改了一个参数，修好一个 case 但弄坏三个”的翻车。从那以后我每次调检索参数都强制走一遍测试集，不再凭直觉改配置。希望这套模块图 RAG 的拆解能帮到你，少走一些我踩过的弯路。
本文还有配套的精品资源，点击获取
