RAG 知识图谱增强结构化关系 非结构化文本的联合检索一、退货政策和退货流程是两个文档用户需要同时知道用户问买到假货怎么退货RAG 系统从知识库检索出了《退换货政策》文档给出了退货条件的回答。但用户接着问具体怎么操作——系统又检索了一次返回了《退货操作流程》文档。两个文档明明高度关联但向量检索把它们当成了独立的文本块。传统 RAG 的局限它只知道文档和问题的语义相似度不知道文档之间的关系。知识图谱补充了这块——它用结构化的关系如退货政策→规定→退货流程补上了向量检索的孤立性。二、知识图谱增强 RAG 的架构核心思路两路召回 → 融合排序 → 送给模型。flowchart TD Q[用户问题] -- E[查询向量化] E -- V[向量检索: 语义相似] E -- NER[实体识别: 提取关键实体] V -- VK[Top-K 文档片段] NER -- KG[知识图谱查询] KG -- KG1[实体关系: 1-hop 邻居] KG -- KG2[关联文档: 关系指向的文档] KG1 -- KG3[图上下文: 结构化关系描述] KG2 -- KG4[关联文本: 图谱节点绑定的文本] VK -- Merge[多路融合排序] KG3 -- Merge KG4 -- Merge Merge -- Context[拼接上下文] Context -- LLM[模型推理] LLM -- R[回答]关键设计知识图谱不只是词网络的图而是实体-关系-实体的三元组集合。比如(退货政策, 引用_法规, 消费者权益法)、(退货流程, 前置步骤, 退货政策)。这些关系帮助模型理解跨文档的信息组织方式。三、Python 实现图谱增强检索import json from dataclasses import dataclass, field from typing import Optional from collections import defaultdict import numpy as np # 知识图谱数据结构 dataclass class Entity: 知识图谱实体 id: str name: str entity_type: str # policy, process, product, service properties: dict field(default_factorydict) dataclass class Relation: 知识图谱关系三元组 subject_id: str # 头实体 predicate: str # 关系类型 object_id: str # 尾实体 weight: float 1.0 dataclass class Document: 文档片段 id: str content: str entity_ids: list[str] field(default_factorylist) # 绑定的实体 # 知识图谱 class KnowledgeGraph: 知识图谱存储与查询 def __init__(self): self.entities: dict[str, Entity] {} self.relations: list[Relation] [] # 索引实体 → 关联文档 self.entity_docs: dict[str, list[str]] defaultdict(list) # 索引文档 → 实体 self.doc_entities: dict[str, list[str]] defaultdict(list) def add_entity(self, entity: Entity): self.entities[entity.id] entity def add_relation(self, relation: Relation): self.relations.append(relation) def bind_document(self, entity_id: str, doc_id: str): 将文档绑定到实体 self.entity_docs[entity_id].append(doc_id) self.doc_entities[doc_id].append(entity_id) def get_neighbors(self, entity_id: str, hop: int 1) - list[dict]: 获取实体的 N 跳邻居 neighbors [] # 1-hop直接关系 for rel in self.relations: if rel.subject_id entity_id: obj self.entities.get(rel.object_id) if obj: neighbors.append({ entity: obj, relation: rel.predicate, hop: 1, direction: out, }) elif rel.object_id entity_id: subj self.entities.get(rel.subject_id) if subj: neighbors.append({ entity: subj, relation: f被{rel.predicate}, hop: 1, direction: in, }) return neighbors def get_entity_by_name(self, name: str) - Optional[Entity]: 按名称查找实体 for entity in self.entities.values(): if entity.name name: return entity return None # 图谱增强检索器 class GraphEnhancedRetriever: 知识图谱增强的检索器 def __init__(self, kg: KnowledgeGraph, documents: list[Document]): self.kg kg self.documents {d.id: d for d in documents} # 简化的向量模拟实际项目用向量数据库 self.doc_vectors: dict[str, np.ndarray] {} def retrieve(self, query: str, top_k: int 5) - list[dict]: 融合检索向量相似 图谱关系 返回增强后的文档列表每条包含来源标注。 results [] seen_doc_ids set() # 第 1 路向量语义检索 vector_docs self._vector_search(query, top_k) for doc_id, score in vector_docs: if doc_id not in seen_doc_ids: results.append({ doc_id: doc_id, content: self.documents[doc_id].content, score: score, source: vector, relation: 语义相似, }) seen_doc_ids.add(doc_id) # 第 2 路知识图谱实体检索 # 从 query 中识别实体 query_entities self._extract_entities(query) for entity in query_entities: # 获取实体的关联文档 for doc_id in self.kg.entity_docs.get(entity.id, []): if doc_id not in seen_doc_ids: doc self.documents[doc_id] results.append({ doc_id: doc_id, content: doc.content, score: 0.8, # 图谱来源的基础分 source: kg, relation: f实体关联: {entity.name}, }) seen_doc_ids.add(doc_id) # 获取邻居实体的文档 neighbors self.kg.get_neighbors(entity.id, hop1) for neighbor in neighbors: for doc_id in self.kg.entity_docs.get(neighbor[entity].id, []): if doc_id not in seen_doc_ids: doc self.documents[doc_id] results.append({ doc_id: doc_id, content: doc.content, score: 0.7, source: kg, relation: ( f{entity.name} f→{neighbor[relation]} f→{neighbor[entity].name} ), }) seen_doc_ids.add(doc_id) # 第 3 路排序 results.sort(keylambda x: x[score], reverseTrue) return results[:top_k] def _vector_search(self, query: str, top_k: int) - list[tuple[str, float]]: 模拟向量检索实际用向量数据库 # 简化返回所有文档按名称匹配度排序 results [] for doc_id, doc in self.documents.items(): overlap len(set(query) set(doc.content)) / max(len(query), 1) results.append((doc_id, overlap)) results.sort(keylambda x: x[1], reverseTrue) return results[:top_k] def _extract_entities(self, query: str) - list[Entity]: 实体识别简化版字符串匹配 entities [] for entity in self.kg.entities.values(): if entity.name in query: entities.append(entity) return entities def build_graph_context(self, entity: Entity) - str: 构建图谱上下文实体关系描述 neighbors self.kg.get_neighbors(entity.id, hop1) if not neighbors: return lines [f关于「{entity.name}」以下信息可能有帮助] for n in neighbors[:5]: # 限制数量 lines.append( f- {entity.name} {n[relation]} 「{n[entity].name}」 ) return \n.join(lines) # 使用示例 def demo(): # 1. 构建知识图谱 kg KnowledgeGraph() policy_entity Entity(e1, 退货政策, policy) process_entity Entity(e2, 退货流程, process) law_entity Entity(e3, 消费者权益保护法, law) kg.add_entity(policy_entity) kg.add_entity(process_entity) kg.add_entity(law_entity) kg.add_relation(Relation(e1, 引用_法规, e3)) kg.add_relation(Relation(e2, 前置步骤, e1)) # 2. 文档绑定 doc1 Document(d1, 退货政策购买后7天内可无理由退货..., [e1]) doc2 Document(d2, 退货流程1.提交申请 2.等待审核 3.寄回商品..., [e2]) doc3 Document(d3, 消费者权益保护法第25条..., [e3]) kg.bind_document(e1, d1) kg.bind_document(e2, d2) kg.bind_document(e3, d3) # 3. 检索 retriever GraphEnhancedRetriever(kg, [doc1, doc2, doc3]) query 买到假货怎么退货、需要走什么流程 results retriever.retrieve(query, top_k5) print(f查询: {query}\n) print(检索结果:) for i, r in enumerate(results, 1): print(f {i}. [{r[source]}] {r[relation]}) print(f {r[content][:60]}...\n) if __name__ __main__: demo()四、知识图谱增强的边界与权衡图谱建设需要额外投入。向量检索可以自动搞定但知识图谱需要人工或半自动化构建。对于知识体量小 100 个文档的场景图谱的收益可能抵不上构建成本。当文档数量上千且关系复杂时图谱才真正体现价值。实体识别的准确率影响检索质量。如果 NER 把苹果识别为水果而非科技公司图谱召回就偏了。实体识别模型需要根据领域做微调。两路融合的权重需要调优。纯向量检索的结果语义相似和图谱召回的结果结构关联如何加权建议从 0.6:0.4 开始根据用户反馈点赞/踩动态调整。图谱需要持续更新。知识库更新时图谱也需要同步更新实体和关系。如果更新不同步图谱召回的文档可能指向已过期的知识。五、总结知识图谱增强 RAG 的核心是补上向量检索看不到的结构关系。实施路径先用向量检索跑起来 → 分析高频查询中检索到了但不完整的案例 → 针对性构建图谱关系 → 两路融合排序。不要一上来就建万级别的全量图谱——从 20-50 个核心实体开始验证图召回对回答质量的提升效果后再扩大。