RAG已经是标配了但如果你的知识库中有大量相互关联的概念、实体和关系传统的向量检索可能正在悄悄地丢掉很多重要信息。GraphRAG或者更准确地说基于知识图谱的检索增强正在成为下一个值得认真对待的技术方向。本文不讲GraphRAG的学术原理只讲实际落地。传统RAG在哪里失效先明确问题。传统RAG的核心逻辑是把文档切成chunk向量化用户提问时检索最相似的chunk塞给LLM。这个方案在以下场景会出明显问题多跳推理。用户问“哪些公司的CEO曾经在同一所大学读书”——这需要知道CEO是谁、他们的教育背景然后做关联推理。传统RAG检索到的chunk很难包含完整的关联链路。关系查询。“A产品依赖哪些第三方库这些库有没有已知漏洞”——这是典型的图结构查询向量相似度无法直接表达依赖关系。聚合问题。“过去一年我们的产品有多少类相似的Bug报告”——需要先分类、后统计单纯的向量检索做不到。跨文档关联。同一个实体比如某个API在十几个文档里被提及相关信息分散各处向量检索很难把这些碎片完整地拼起来。GraphRAG的核心思路GraphRAG的核心思路是在向量检索之外额外维护一个实体-关系图查询时同时利用向量相似度和图结构进行检索。用户查询 ↓ 实体识别从查询中提取关键实体 ↓ ├── 向量检索找相关文档块 └── 图遍历找相关实体及其关系 ↓ 结果融合 ↓ 上下文构建 → LLM生成 ## 知识图谱的构建 这是GraphRAG最耗工程量的部分也是最关键的。 ### 实体抽取 python from openai import AsyncOpenAI import json async def extract_entities(text: str, llm_client) - list[dict]: 从文本中抽取实体和关系 prompt f 从以下文本中提取所有命名实体和它们之间的关系。 文本 {text} 输出格式JSON {{ entities: [ {{id: 唯一标识, name: 实体名称, type: PERSON/COMPANY/PRODUCT/TECHNOLOGY/CONCEPT, description: 简短描述}} ], relations: [ {{source: 实体id, target: 实体id, relation: 关系类型, description: 关系描述}} ] }} 只提取文本中明确提到的实体和关系不要推断。 response await llm_client.chat.completions.create( modelgpt-4o, messages[{role: user, content: prompt}], response_format{type: json_object} ) return json.loads(response.choices[0].message.content) async def build_knowledge_graph(documents: list[Document]) - KnowledgeGraph: 从文档集合构建知识图谱 kg KnowledgeGraph() client AsyncOpenAI() for doc in documents: # 按段落切分每个段落独立抽取 paragraphs split_into_paragraphs(doc.content, max_tokens800) for para in paragraphs: try: extracted await extract_entities(para, client) # 添加实体自动去重和合并 for entity in extracted[entities]: kg.add_entity(entity, source_docdoc.id, source_textpara) # 添加关系 for relation in extracted[relations]: kg.add_relation(relation, source_docdoc.id) except Exception as e: logging.warning(fEntity extraction failed for paragraph: {e}) continue # 实体消歧合并指向同一现实对象的不同名称 kg.resolve_coreferences() return kg ### 知识图谱存储 用Neo4j或者内存图对于小规模场景 python from neo4j import AsyncGraphDatabase class Neo4jKnowledgeGraph: def __init__(self, uri, user, password): self.driver AsyncGraphDatabase.driver(uri, auth(user, password)) async def add_entity(self, entity: dict, source_doc: str, source_text: str): async with self.driver.session() as session: await session.run( MERGE (e:Entity {id: $id}) ON CREATE SET e.name $name, e.type $type, e.description $description, e.source_docs [$source_doc] ON MATCH SET e.source_docs e.source_docs [$source_doc], e.description $description // 更新描述 , identity[id], nameentity[name], typeentity[type], descriptionentity.get(description, ), source_docsource_doc) async def add_relation(self, relation: dict, source_doc: str): async with self.driver.session() as session: await session.run(f MATCH (s:Entity {{id: $source}}) MATCH (t:Entity {{id: $target}}) MERGE (s)-[r:{relation[relation]}]-(t) SET r.description $description, r.source_doc $source_doc , sourcerelation[source], targetrelation[target], descriptionrelation.get(description, ), source_docsource_doc) async def get_entity_neighborhood(self, entity_name: str, max_hops: int 2) - dict: 获取实体的邻域用于上下文构建 async with self.driver.session() as session: result await session.run(f MATCH (start:Entity {{name: $name}}) CALL apoc.path.subgraphAll(start, {{maxLevel: {max_hops}}}) YIELD nodes, relationships RETURN nodes, relationships , nameentity_name) record await result.single() if not record: return {} return { center: entity_name, nodes: [dict(n) for n in record[nodes]], relations: [ { source: r.start_node[name], type: r.type, target: r.end_node[name], description: r.get(description, ) } for r in record[relationships] ] } ## 查询时的图增强检索 python class GraphEnhancedRetriever: def __init__(self, vector_store: VectorStore, knowledge_graph: Neo4jKnowledgeGraph, llm_client): self.vector_store vector_store self.kg knowledge_graph self.llm llm_client async def retrieve(self, query: str, top_k: int 5) - list[RetrievedContext]: # 1. 从查询中识别关键实体 entities await self._extract_query_entities(query) # 2. 并行执行向量检索和图检索 vector_task self.vector_store.search(query, top_ktop_k) graph_tasks [ self.kg.get_entity_neighborhood(entity, max_hops2) for entity in entities ] vector_results, *graph_results await asyncio.gather( vector_task, *graph_tasks ) # 3. 融合结果 contexts [] # 向量检索结果 for chunk in vector_results: contexts.append(RetrievedContext( contentchunk.text, sourcevector, relevance_scorechunk.score )) # 图检索结果把邻域信息转成自然语言 for entity, graph_data in zip(entities, graph_results): if graph_data: graph_context self._graph_to_text(entity, graph_data) contexts.append(RetrievedContext( contentgraph_context, sourcegraph, relevance_score0.8 # 图检索结果的基础分 )) # 4. 去重和重排序 return self._deduplicate_and_rerank(contexts, query, top_k * 2) def _graph_to_text(self, entity: str, graph_data: dict) - str: 把图结构转成LLM可以理解的文本 lines [f关于「{entity}」的关联信息] for rel in graph_data.get(relations, []): lines.append( f- {rel[source]} {rel[type]} {rel[target]} (f{rel[description]} if rel.get(description) else ) ) return \n.join(lines) async def _extract_query_entities(self, query: str) - list[str]: 从查询中快速提取关键实体 response await self.llm.chat.completions.create( modelgpt-4o-mini, # 用小模型追求速度 messages[{ role: user, content: f从以下查询中提取关键实体名称只返回实体列表JSON数组\n{query} }], response_format{type: json_object} ) result json.loads(response.choices[0].message.content) return result.get(entities, []) ## 增量更新保持图谱与文档同步 知识图谱不是一次性构建就完事了需要随着文档更新而维护 python class GraphUpdateManager: def __init__(self, kg: Neo4jKnowledgeGraph, vector_store: VectorStore): self.kg kg self.vs vector_store self.change_tracker ChangeTracker() async def on_document_updated(self, doc: Document, old_doc: Document): 文档更新时增量更新知识图谱 # 1. 找出变化的段落 changed_paras self._diff_paragraphs(old_doc.content, doc.content) # 2. 删除旧段落抽取的实体和关系只删除仅来自该文档的 for old_para in changed_paras[removed]: await self.kg.remove_entities_by_source( source_docdoc.id, source_textold_para ) # 3. 添加新段落的实体和关系 for new_para in changed_paras[added]: extracted await extract_entities(new_para, self.llm) for entity in extracted[entities]: await self.kg.add_entity(entity, doc.id, new_para) for relation in extracted[relations]: await self.kg.add_relation(relation, doc.id) # 4. 同步更新向量存储 await self.vs.update_document(doc) ## 何时用GraphRAG何时用传统RAG 不是所有场景都需要GraphRAG它有额外的构建和维护成本。 **适合GraphRAG的场景** - 知识库有明确的实体和关系结构产品文档、人员关系、依赖图 - - 查询经常涉及跨文档的关联推理 - - 用户需要问X和Y有什么关系类型的问题 - - 文档间有大量交叉引用 **传统RAG就够的场景** - 文档主要是独立的问答对或操作手册 - - 查询主要是如何做X的过程性问题 - - 知识库更新频率很高图谱维护成本高 - - 团队没有足够资源维护图谱 **混合使用**推荐大多数企业知识库都适合混合检索——用向量检索处理大部分常规查询对涉及实体关系的查询启用图增强。 ## 工程踩坑记录 **实体消歧是最大的坑**。同一个实体可能有十几种不同的写法GPT-4、GPT4、gpt-4、OpenAI GPT-4如果不做消歧图谱会碎片化查询结果质量很差。解决方案维护一个实体别名词典并在抽取后做规范化处理。 **关系的粒度要合理**。不要抽取太细粒度的关系比如A提到了B这种信息量太低。要聚焦在有实际意义的关系依赖、竞争、包含、属于、参与等。 **图谱规模要控制**。超过10万个节点的图谱查询性能会明显下降。定期清理低价值节点只被一个文档引用、没有任何关系的孤立节点。 **不要盲目追求完整性**。图谱不需要包含所有信息它的价值在于表达关系专注于对查询最有价值的实体和关系类型。 --- *本文关键词GraphRAG、知识图谱、RAG优化、实体关系抽取、Neo4j*