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( model="gpt-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_tokens=800) for para in paragraphs: try: extracted = await extract_entities(para, client) # 添加实体（自动去重和合并） for entity in extracted["entities"]: kg.add_entity(entity, source_doc=doc.id, source_text=para) # 添加关系 for relation in extracted["relations"]: kg.add_relation(relation, source_doc=doc.id) except Exception as e: logging.warning(f"Entity 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 // 更新描述 """, id=entity["id"], name=entity["name"], type=entity["type"], description=entity.get("description", ""), source_doc=source_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 """, source=relation["source"], target=relation["target"], description=relation.get("description", ""), source_doc=source_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 """, name=entity_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_k=top_k) graph_tasks = [ self.kg.get_entity_neighborhood(entity, max_hops=2) 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( content=chunk.text, source="vector", relevance_score=chunk.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( content=graph_context, source="graph", relevance_score=0.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( + model="gpt-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_doc=doc.id, source_text=old_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*
