GraphRAG vs 传统 RAG 的本质区别 传统 RAG 的核心问题是"只见树木不见森林"——它能找到局部相关的文本块，但无法理解全局关系。GraphRAG 通过构建知识图谱，让系统具备全局视角和推理能力。
维度 传统 RAG GraphRAG 检索单元 文本块 实体+关系+文本块 全局理解 ❌ ✅ 社区摘要 多跳推理 ❌ ✅ 图遍历 可解释性 低 高（路径溯源） 构建成本 低 高 查询延迟 1-2s 3-10s 架构设计 ┌─────────────────────────────────────────────────────┐ │ GraphRAG 架构 │ ├─────────────────────────────────────────────────────┤ │ │ │ 离线构建 Pipeline │ │ ┌──────────┐ ┌───────────┐ ┌──────────────┐ │ │ │ 文档解析 │→│ 实体抽取 │→│ 关系抽取 │ │ │ └──────────┘ └───────────┘ └──────────────┘ │ │ ↓ │ │ ┌───────────┐ ┌───────────┐ ┌──────────────┐ │ │ │ 图谱构建 │←│ 社区检测 │←│ Embedding │ │ │ └───────────┘ └───────────┘ └──────────────┘ │ │ ↓ │ │ ┌─────────────┐ │ │ │ 索引持久化 │ │ │ └─────────────┘ │ │ │ │ 在线查询 Engine │ │ ┌──────────┐ ┌───────────┐ ┌──────────────┐ │ │ │ Query │→│ 路由决策 │→│ 双路检索 │ │ │ │ 理解 │ │ │ │ 向量+图谱 │ │ │ └──────────┘ └───────────┘ └──────────────┘ │ │ ↓ │ │ ┌──────────┐ ┌───────────┐ ┌──────────────┐ │ │ │ 生成+引用 │←│ 信息整合 │←│ 重排序 │ │ │ └──────────┘ └───────────┘ └──────────────┘ │ │ │ └─────────────────────────────────────────────────────┘ 离线构建 Pipeline 1. 实体与关系抽取 from pydantic import BaseModel from typing import List class Entity(BaseModel): name: str type: str # Person, Organization, Concept, Event, etc. description: str source_chunk_id: str class Relation(BaseModel): source_entity: str target_entity: str relation_type: str description: str confidence: float source_chunk_id: str ENTITY_EXTRACTION_PROMPT = """ 你是一个信息抽取专家。从以下文本中抽取实体和关系。 实体类型：Person, Organization, Concept, Technology, Event, Location 关系类型：works_at, created, related_to, part_of, located_in, depends_on, competes_with 文本： {text} 请以 JSON 格式输出： {{ "entities": [ {{"name": "...", "type": "...", "description": "..."}} ], "relations": [ {{"source": "...", "target": "...", "type": "...", "description": "...", "confidence": 0.0-1.0}} ] }} """ class EntityExtractor: def __init__(self, llm): self.llm = llm def extract(self, text: str, chunk_id: str): prompt = ENTITY_EXTRACTION_PROMPT.format(text=text) result = self.llm.generate(prompt, response_format="json") entities = [ Entity(**e, source_chunk_id=chunk_id) for e in result["entities"] ] relations = [ Relation(**r, source_chunk_id=chunk_id) for r in result["relations"] ] return entities, relations 2. 知识图谱构建 import networkx as nx from community_detection import LeidenAlgorithm class KnowledgeGraph: def __init__(self): self.graph = nx.DiGraph() self.entity_index = {} # name -> node_id def add_entities(self, entities: List[Entity]): for entity in entities: if entity.name not in self.entity_index: node_id = len(self.entity_index) self.entity_index[entity.name] = node_id self.graph.add_node( node_id, name=entity.name, type=entity.type, description=entity.description, source_chunks=[entity.source_chunk_id] ) else: # 合并描述 node_id = self.entity_index[entity.name] self.graph.nodes[node_id]["source_chunks"].append( entity.source_chunk_id ) def add_relations(self, relations: List[Relation]): for rel in relations: if rel.source_entity in self.entity_index and rel.target_entity in self.entity_index: src = self.entity_index[rel.source_entity] tgt = self.entity_index[rel.target_entity] if self.graph.has_edge(src, tgt): # 合并关系 existing = self.graph[src][tgt] existing["relations"].append({ "type": rel.relation_type, "description": rel.description, "confidence": rel.confidence }) else: self.graph.add_edge( src, tgt, relations=[{ "type": rel.relation_type, "description": rel.description, "confidence": rel.confidence }] ) def detect_communities(self): """使用 Leiden 算法进行社区检测""" undirected = self.graph.to_undirected() communities = LeidenAlgorithm().fit(undirected) # 为每个社区生成摘要 for comm_id, nodes in communities.items(): subgraph = self.graph.subgraph(nodes) summary = self._summarize_community(subgraph) for node in nodes: self.graph.nodes[node]["community_id"] = comm_id self.graph.graph.setdefault("community_summaries", {})[comm_id] = summary return communities def _summarize_community(self, subgraph): entities_info = [] for node_id in subgraph.nodes(): node = self.graph.nodes[node_id] entities_info.append(f"{node['name']} ({node['type']}): {node['description']}") relations_info = [] for u, v, data in subgraph.edges(data=True): for rel in data["relations"]: relations_info.append( f"{self.graph.nodes[u]['name']} --{rel['type']}--> {self.graph.nodes[v]['name']}" ) prompt = f""" 请总结以下知识图谱社区的关键信息： 实体： {chr(10).join(entities_info)} 关系： {chr(10).join(relations_info)} 请生成一段简洁的摘要，涵盖主要实体和它们之间的关系。 """ return self.llm.generate(prompt) 3. 索引持久化 class GraphRAGIndex: def __init__(self): self.graph = KnowledgeGraph() self.vector_store = MilvusIndex(dim=1024) self.community_store = CommunityStore() def build(self, documents: List[Document]): # 1. 文本分块与向量化 chunks = document_aware_chunk(documents) for chunk in chunks: embedding = embed_model.encode(chunk.text) self.vector_store.add(id=chunk.id, embedding=embedding, metadata={"text": chunk.text}) # 2. 实体关系抽取 all_entities = [] all_relations = [] for chunk in chunks: entities, relations = extractor.extract(chunk.text, chunk.id) all_entities.extend(entities) all_relations.extend(relations) # 3. 构建知识图谱 self.graph.add_entities(all_entities) self.graph.add_relations(all_relations) # 4. 社区检测与摘要 communities = self.graph.detect_communities() # 5. 持久化 self._persist() def _persist(self): # 图谱存 Neo4j 或 NetworkX pickle nx.write_gpickle(self.graph.graph, "graph.gpickle") # 向量索引已在 Milvus 中持久化 # 社区摘要存数据库 self.community_store.save(self.graph.graph.graph.get("community_summaries", {})) 在线查询引擎 class GraphRAGQueryEngine: def __init__(self, index: GraphRAGIndex, llm): self.index = index self.llm = llm def query(self, question: str) -> str: # 1. 判断查询类型 query_type = self._classify_query(question) if query_type == "global": # 全局型问题 → 社区检索 context = self._global_search(question) elif query_type == "specific": # 具体型问题 → 向量+图遍历 context = self._local_search(question) else: # 混合型 → 双路检索 context = self._hybrid_search(question) # 2. 生成答案 answer = self.llm.generate( prompt=ANSWER_PROMPT.format(question=question, context=context), citations=True ) return answer def _local_search(self, question: str): # 向量检索找到相关文本块 query_emb = embed_model.encode(question) vector_hits = self.index.vector_store.search(query_emb, top_k=10) # 从命中块中提取实体 entities = set() for hit in vector_hits: entities.update(self._extract_entities_from_chunk(hit)) # 图遍历扩展上下文 graph_context = [] for entity_name in entities: if entity_name in self.index.graph.entity_index: node_id = self.index.graph.entity_index[entity_name] # 1-hop 和 2-hop 邻居 neighbors = nx.single_source_shortest_path_length( self.index.graph.graph, node_id, cutoff=2 ) for neighbor_id, hops in neighbors.items(): if hops > 0: node = self.index.graph.graph.nodes[neighbor_id] graph_context.append({ "entity": node["name"], "type": node["type"], "hops": hops, "description": node["description"] }) return { "vector_context": vector_hits, "graph_context": graph_context } def _global_search(self, question: str): # 从社区摘要中检索 community_summaries = self.index.community_store.get_all() # 用 LLM 判断哪些社区相关 relevant = self.llm.generate( prompt=f""" 以下问题与哪些社区摘要相关？ 问题：{question} 社区摘要： {json.dumps(community_summaries, ensure_ascii=False)} 返回最相关的 3 个社区 ID。 """ ) return {"community_context": relevant} 生产运维 监控指标 @dataclass class GraphRAGMetrics: # 构建阶段 entity_extraction_rate: float # 每分钟抽取实体数 relation_extraction_rate: float # 每分钟抽取关系数 community_detection_time: float # 社区检测耗时 # 查询阶段 query_latency_p50: float query_latency_p99: float vector_recall: float # 向量检索召回率 graph_coverage: float # 图遍历覆盖率 community_hit_rate: float # 社区命中率 # 质量指标 answer_accuracy: float # 答案准确率 citation_rate: float # 引用覆盖率 hallucination_rate: float # 幻觉率 增量更新策略 class IncrementalUpdater: """GraphRAG 增量更新：只处理新增/修改的文档""" def update(self, new_documents: List[Document], deleted_ids: List[str]): # 1. 删除过期数据 for doc_id in deleted_ids: self._remove_document(doc_id) # 2. 处理新文档 for doc in new_documents: entities, relations = extractor.extract(doc) self.graph.add_entities(entities) self.graph.add_relations(relations) # 3. 局部社区检测（只重新检测受影响的社区） affected_communities = self._find_affected_communities(new_documents, deleted_ids) self._redetect_communities(affected_communities) # 4. 更新向量索引 self.vector_store.upsert(new_documents) 成本与效果 以 10 万篇技术文档为例：
...
