Appearance
知识图谱
知识图谱是一种结构化的知识表示方法，通过节点和边来表示实体及其关系。将知识图谱与RAG技术结合，可以显著提高系统的推理能力和知识表示能力。本章节将详细介绍知识图谱的概念、构建方法以及与RAG的集成。
1. 知识图谱基础
核心概念
- 实体（Entity）：现实世界中的事物，如人物、组织、地点等
- 关系（Relationship）：实体之间的联系，如"属于"、"位于"、"创立"等
- 属性（Attribute）：实体的特征，如人物的年龄、组织的成立时间等
- 三元组（Triple）：知识图谱的基本单元，由（主语，谓语，宾语）组成
知识图谱示例
(张三, 是, 学生)
(张三, 就读于, 北京大学)
(北京大学, 位于, 北京)
(北京, 是, 中国的首都)2. 知识图谱构建
数据来源
- 结构化数据：数据库、CSV文件等
- 半结构化数据：XML、JSON等
- 非结构化数据：文本、网页等
构建步骤
- 实体识别：从文本中识别实体
- 关系抽取：识别实体之间的关系
- 属性抽取：提取实体的属性
- 知识融合：合并不同来源的知识
- 知识验证：验证知识的正确性
工具与库
- Stanford CoreNLP：自然语言处理工具包
- spaCy：Python自然语言处理库
- OpenIE：开放信息抽取系统
- DeepDive：知识图谱构建系统
3. 知识图谱存储
图数据库
- Neo4j：最流行的图数据库
python
from py2neo import Graph, Node, Relationship
# 连接Neo4j
graph = Graph("bolt://localhost:7687", auth=("neo4j", "password"))
# 创建节点
person = Node("Person", name="张三", age=25)
school = Node("School", name="北京大学", founded=1898)
graph.create(person)
graph.create(school)
# 创建关系
studies_at = Relationship(person, "STUDIES_AT", school, since=2020)
graph.create(studies_at)
# 查询
query = """
MATCH (p:Person)-[:STUDIES_AT]->(s:School)
WHERE s.name = '北京大学'
RETURN p.name
"""
results = graph.run(query)
for record in results:
print(record["p.name"])RDF存储
- Apache Jena：开源的RDF框架
- Virtuoso：高性能的RDF存储
python
from rdflib import Graph, Namespace, Literal, URIRef
from rdflib.namespace import RDF, RDFS
# 创建RDF图
g = Graph()
# 定义命名空间
ex = Namespace("http://example.org/")
# 添加三元组
g.add((ex.张三, RDF.type, ex.Person))
g.add((ex.张三, ex.name, Literal("张三")))
g.add((ex.张三, ex.studiesAt, ex.北京大学))
# 序列化
print(g.serialize(format="turtle"))4. 知识图谱与RAG集成
架构设计
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ 用户查询 │────>│ 查询理解 │────>│ 图谱查询 │
└─────────────┘ └─────────────┘ └─────────────┘
│
▼
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ 生成回答 │<────│ LLM │<────│ 图谱结果 │
└─────────────┘ └─────────────┘ └─────────────┘
▲ │
└────────────────────────────────────┘
向量检索结果（补充）实现示例
python
class KnowledgeGraphRAG:
def __init__(self, graph_db, vector_store, llm):
self.graph = graph_db
self.vector_store = vector_store
self.llm = llm
def query(self, question):
"""执行知识图谱增强的RAG查询"""
# 1. 从问题中提取实体
entities = self.extract_entities(question)
# 2. 在知识图谱中查询相关子图
kg_results = self.query_knowledge_graph(entities)
# 3. 向量检索补充信息
vector_results = self.vector_store.similarity_search(question, k=3)
# 4. 融合信息并生成回答
context = self.fuse_context(kg_results, vector_results)
answer = self.generate_answer(question, context)
return {
"answer": answer,
"kg_results": kg_results,
"vector_results": vector_results
}
def extract_entities(self, text):
"""提取实体"""
# 使用NER模型或规则提取实体
# 这里简化处理
import jieba
words = jieba.lcut(text)
# 返回名词作为实体
return [w for w in words if len(w) > 1]
def query_knowledge_graph(self, entities):
"""查询知识图谱"""
results = []
for entity in entities:
query = f"""
MATCH (n)-[r]->(m)
WHERE n.name CONTAINS '{entity}'
RETURN n.name, type(r), m.name
LIMIT 10
"""
data = self.graph.run(query).data()
results.extend(data)
return results
def fuse_context(self, kg_results, vector_results):
"""融合知识图谱和向量检索结果"""
context_parts = []
# 添加知识图谱信息
if kg_results:
kg_text = "知识图谱信息：\n"
for result in kg_results:
kg_text += f"- {result['n.name']} {result['type(r)']} {result['m.name']}\n"
context_parts.append(kg_text)
# 添加向量检索信息
if vector_results:
vector_text = "相关文档：\n"
for doc in vector_results:
vector_text += f"- {doc.page_content[:200]}...\n"
context_parts.append(vector_text)
return "\n\n".join(context_parts)
def generate_answer(self, question, context):
"""生成回答"""
prompt = f"""基于以下信息回答问题：
{context}
问题：{question}
请提供准确且详细的回答："""
return self.llm.generate(prompt)5. 知识图谱增强检索
多跳推理
python
def multi_hop_query(graph, start_entity, hops=2):
"""多跳查询"""
query = f"""
MATCH path = (start)-[*1..{hops}]->(end)
WHERE start.name = '{start_entity}'
RETURN path
LIMIT 20
"""
return graph.run(query).data()路径排序
python
def rank_paths(paths, query_embedding):
"""根据与查询的相关性排序路径"""
path_scores = []
for path in paths:
# 将路径转换为文本
path_text = path_to_text(path)
# 计算与查询的相似度
path_embedding = embed_text(path_text)
similarity = cosine_similarity(query_embedding, path_embedding)
path_scores.append((path, similarity))
# 按相似度排序
return sorted(path_scores, key=lambda x: x[1], reverse=True)6. 应用场景
企业知识管理
python
class EnterpriseKGRAG(KnowledgeGraphRAG):
def __init__(self):
super().__init__()
# 定义企业知识图谱schema
self.schema = {
"entities": ["员工", "部门", "项目", "产品", "客户"],
"relations": ["属于", "负责", "参与", "使用", "服务"]
}
def answer_org_question(self, question):
"""回答组织架构相关问题"""
# 识别问题类型
if "谁负责" in question:
return self.query_responsibility(question)
elif "属于哪个部门" in question:
return self.query_department(question)
# ...医疗问答
python
class MedicalKGRAG(KnowledgeGraphRAG):
def __init__(self):
super().__init__()
self.schema = {
"entities": ["疾病", "症状", "药物", "检查", "科室"],
"relations": ["表现为", "治疗", "需要", "属于"]
}
def answer_medical_question(self, question):
"""回答医疗相关问题"""
# 提取症状
symptoms = self.extract_symptoms(question)
# 查询可能的疾病
diseases = self.query_diseases_by_symptoms(symptoms)
return diseases7. 最佳实践
知识图谱构建建议
- 从小规模开始：先构建核心领域的知识图谱
- 持续迭代：根据使用反馈不断完善
- 质量保证：建立知识验证机制
- 版本管理：对知识图谱进行版本控制
性能优化
python
# 使用缓存加速图谱查询
from functools import lru_cache
class CachedGraphRAG(KnowledgeGraphRAG):
@lru_cache(maxsize=1000)
def query_knowledge_graph(self, entity_tuple):
"""缓存图谱查询结果"""
entities = list(entity_tuple)
return super().query_knowledge_graph(entities)