RAG 知识图谱增强：结构化关系 + 非结构化文本的联合检索-平芜编程栈

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_factory=dict) @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_factory=list) # 绑定的实体 # ========== 知识图谱 ========== 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, hop=1) 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(key=lambda x: x["score"], reverse=True) 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(key=lambda x: x[1], reverse=True) 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, hop=1) 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_k=5) 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()