手把手教你构建基于知识图谱的GraphRAG之结构化数据篇【LangChain Neo4j】

显示全部楼层

检索增强生成（RAG）是一种利用外部知识来增强大模型生成能力减少幻觉的主流方法，而对知识最常见的一种组织与索引的形式是向量化及基于向量相近性的检索。但除此之外，基于Graph图结构的知识图谱也是一种强大的知识组织工具，在很多场景下它可以实现更有意义的上下文检索并帮助模型输出更加准确的响应内容。我们将用实例来学习基于知识图谱的GraphRAG应用的构建：

预备知识：GraphRAG基础
构建GraphRAG：结构化数据
构建GraphRAG：非结构化数据
认识微软开源项目：GraphRAG

预备知识：GraphRAG基础

在开始之前，我们先快速了解图（Graph）、图数据库（GraphDB）、知识图谱（Knowledge Graph）以及GraphRAG的基础知识。

图（Graph）

图（Graph）是一种用来表示对象以及它们之间关系的数学结构。任何两个对象之间都可以直接发生联系，所以适合表达更复杂的关系信息。一个图结构的主要的组成是节点和边。

节点：用来表示一个对象。比如社交网络的一个用户。
边：用来表示对象之间的关系。比如用户之间的关系，如相互关注。

下面是一个关于明星、电影、公司这几种实体的一个Graph例子：

图数据库（GraphDB）

图数据库是一种专门用于存储和操作图结构数据的数据库管理系统。与关系型数据库不同，图数据库使用节点、边和属性来表示和存储数据。这使得它们非常适合处理高度连接的数据，提供高性能的复杂查询能力，用来遍历与发现有洞察力的数据关系。其最大特点是：

灵活的模型：可以方便地表示复杂的关系。
高效的查询：特别是多跳关系的查询，比关系数据库更高效。
可扩展性：能够处理大量节点和边。

图数据库通常具备强大的知识图谱的多跳检索能力：通过多次的关系跳跃来发现相关的信息，这在处理复杂查询与发现关系时特别有用。比如对上面的图提问：

“查找Tom的朋友主演的电影的制作公司的母公司”

这样的查询借助于图数据库的效率要远远高于传统的关系数据库的表连接。

最常见的图数据库管理系统有Neo4j、Amazon Neptune、OrientDB、TigerGraph等，被广泛应用于社交网络分析、推荐系统、金融欺诈检测等。

知识图谱与GraphRAG

知识图谱是一种基于图结构的语义网络，用于表示现实世界中的知识。知识图谱不仅包含实体和关系，还包含它们的语义信息。它通常使用图数据库来存储和管理数据，但增加了语义层次，以提供更高级的知识表示和推理能力。

GraphRAG就是一种对存储在图数据库中的知识图谱（而非存储在向量数据库中的知识向量）进行检索，获得关联知识并实现增强生成的LLM应用。这种方法可以更好的表示人类复杂的知识及其关系，并提供高效的检索能力，产生更加相关与丰富的上下文，让LLM生成最佳答案。

知识图谱更适合于需要精确表示实体间关系和进行复杂推理的场景。
向量化方案则更适合于基于相似性匹配和快速检索自然语言知识的应用。

GraphRAG在整体架构与传统RAG并无更大区别，区别在于检索的知识采用图结构的方式进行构建、存储并检索：

基于图数据库进行知识图谱检索的时候，又可以有两种常见方式（取决于图数据库的支持能力）：

借助Text-to-GraphQL：把自然语言的输入借助LLM与Text-to-GraphQL技术转换为图数据库的查询语言（比如Neo4j的Cypher语言），再使用图数据库的查询语言从知识图谱中检索出需要的知识。

借助Vector索引：在构建的Graph基础上进一步对其中的节点与关系创建向量索引，并通过向量相似性来检索出相关的节点和关系信息；还可以结合传统的关键词做混合检索（注意区分直接对原始知识做向量检索）。

Cypher 是 Neo4j 图数据库的查询语言，设计用于执行各种图数据操作。它的语法直观且类似于 SQL，可以理解为一种图数据库的SQL。

图数据库的强大之处在于你可以同时把结构化与非结构化的数据通过转化映射，以知识图谱的方式存储到单一的库中。通常来说，Text-to-GraphQL更适合对结构化数据生成的Graph进行查询，而Vector索引更适合对非结构化数据生成的Graph进行查询。

这里将首先展示如何从结构化的内容生成知识图谱并用于RAG查询生成。

构建GraphRAG：结构化数据

在这个例子中，我们把存储在MySQL关系型数据库中的一组信息转化为知识图谱存储到图数据库中，并基于此进行自然语言的提问，以验证基于结构化数据的GraphRAG的构建。

原始数据准备

原始数据以表的形式存储在MySQL数据库中，具有如下的主要实体和关系：

首先创建几个简单的表，借助工具生成必要的测试数据：

orders：订单表。包含客户id，产品id，数量，销售员，订单日期等。
custoemrs：客户信息表。包含客户id，姓名，电话，电子邮件，城市等。
products：产品信息表。包含产品ID，名称，单价等。
salespersons：销售员信息表。包含人员id，姓名，电话，所属部门等。
departs：部门信息表。包含部门id，部门名称等。

启动Neo4j图数据库

使用如下命令快速启动一个带有APOC插件的Neo4j数据库容器（APOC是一个强大的Neo4j增强功能插件）：

docker run \
-p 7474:7474 -p 7687:7687 \
-v $PWD/data:/data -v $PWD/plugins:/plugins \
--name neo4j-apoc \
-e NEO4J_apoc_export_file_enabled=true\
-e NEO4J_apoc_import_file_enabled=true\
-e NEO4J_apoc_import_file_use__neo4j__config=true\
-e NEO4JLABS_PLUGINS=\[\"apoc\"\] \
neo4j:latest

启动完成后，访问http://localhost:7474/即可管理Neo4j数据库（注意更改默认密码），在这里你可以更直观的查看数据库中的节点与关系信息，并发起一系列交互命令，比如Cypher查询。

生成Graph

我们需要把传统的RDBMS中的表数据转化为Graph的结构进行存储。参考上文的实体与关系图，这里借助Neo4j的SDK可以快速实现。按照如下顺序来进行：

1. 将数据表读取到本地的pandas的dataframe：

importpandas aspd
importmysql.connector
fromneo4j importGraphDatabase

deffetch_table_data(table_name):
cnx = mysql.connector.connect(
host='你的mysql主机',
user='你的mysql用户',
password='******',
database='sales'
)
cursor = cnx.cursor()
query = f"SELECT * FROM {table_name}"
cursor.execute(query)
rows = cursor.fetchall()
column_names = [desc[0] fordesc incursor.description]
df = pd.DataFrame(rows, columns=column_names)
cursor.close()
cnx.close()
returndf

# 读取到dataframe
orders_df = fetch_table_data("orders")
customers_df = fetch_table_data("customers")
products_df = fetch_table_data("products")
salespersons_df = fetch_table_data("salespersons")
departs_df = fetch_table_data("departs")

2. 创建Graph的节点。即把读取的数据转化为Neo4j中的Node，这里的技巧是读取dataframe的每条记录的key和value来构建节点的属性，并使用Cypher的CREATE语句来批量创建节点；为了在测试时能够不重复创建，这里通过指定的唯一键来防止重复创建（也可以通过CREATE CONSTRAINT给节点类型创建约束）：

defcreate_unique_nodes_from_dataframe(df, label, unique_id_property):
# 连接到Neo4j数据库
driver = GraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "******"))
# 创建一个会话来执行Cypher查询
withdriver.session() assession:
# 遍历DataFrame中的每一行
for_, row indf.iterrows():
# 从行中获取唯一id的值
unique_id_value = row[unique_id_property]
# 检查是否已经存在具有相同唯一id的节点
query = f"MATCH (n:{label}{{{unique_id_property}: '{unique_id_value}'}}) RETURN n"
result = session.run(query)
ifresult.single() isnotNone:
# 已经存在具有相同唯一id的节点，跳过创建新节点
continue
# 创建一个Cypher查询来创建具有给定标签和属性的节点
properties = ", ".join(f"{key}: '{value}'"forkey, value inrow.to_dict().items())
query = f"CREATE (n:{label}{{ {properties}}})"
# 执行Cypher查询
session.run(query)

# 为订单创建节点
create_unique_nodes_from_dataframe(orders_df, "Order","order_id")
create_unique_nodes_from_dataframe(customers_df, "Customer","customer_id")
create_unique_nodes_from_dataframe(products_df, "roduct","product_id")
create_unique_nodes_from_dataframe(salespersons_df, "Salesperson","salesperson_id")
create_unique_nodes_from_dataframe(departs_df, "Depart","depart_id")

3. 创建Graph的节点关系。关系是Graph最重要的特征，我们通过如下的方法来构建不同类型节点之间的关系，在创建关系时，节点与节点的关系通过指定连接条件（类似RDBMS中的表连接）与关系名称：

defcreate_relationships():
# 连接到Neo4j数据库
driver = GraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "******"))
# 创建一个会话来执行Cypher查询
withdriver.session() assession:
# 在顾客和订单之间创建关系
query = "MATCH (c:Customer), (o:Order) WHERE c.customer_id = o.customer_id MERGE (c)-[:ORDERED]->(o)"
session.run(query)

# 在销售人员和订单之间创建关系
query = "MATCH (s:Salesperson), (o:Order) WHERE s.salesperson_id = o.salesperson_id MERGE (s)-[:CREATED]->(o)"
session.run(query)

# 在产品和订单之间创建关系
query = "MATCH (product), (o:Order) WHERE p.product_id = o.product_id MERGE (p)-[:IS_ORDERED_IN]->(o)"
session.run(query)

# 在销售人员和部门之间创建关系
query = "MATCH (s:Salesperson), (depart) WHERE s.depart_id = d.depart_id MERGE (s)-[:BELONGS]->(d)"
session.run(query)

# 调用函数来创建关系
create_relationships()

这里的Cypher语句是不是有点类似SQL语句？注意创建关系时我们使用的是MERGE，这也是为了防止重复生成关系。

运行以上代码，如果一切正常，将会在Neo4j数据库中看到你创建所有的Node以及Relationship信息。可以在管理台看到类似下面的统计信息，能够看到创建的Node类型（用label标记）、Relationship类型以及具体数量：

基于Graph实现RAG

你可以借助LangChain中的GraphCypherQAChain组件来快速实现对Graph的检索与答案生成。GraphCypherQAChain的基本原理是把输入的自然语言借助LLM生成Graph查询的Cypher语句，然后获得查询结果，并把结果用于答案生成（Text-to-Cypher）。

使用如下代码（此处使用gpt-4o模型，你也可以尝试使用本地ollma模型），为了更好的控制生成与帮助理解，这里对LangChain的提示词做了显式设置：

fromlangchain.prompts importPromptTemplate
cypher_generation_template = """
任务：
为Neo4j图数据库生成Cypher查询。
说明：
仅使用提供的模式中的关系类型和属性。
不要使用任何未提供的其他关系类型或属性。
模式：
{schema}
注意：
在回答中不要包含任何解释或道歉。
不要回答任何可能要求你构建除Cypher语句之外的任何文本的问题。
确保查询中的关系方向是正确的。
确保正确地为实体和关系设置别名。
不要运行会向数据库添加或删除内容的任何查询。
确保将后续所有语句都设置别名为with语句。
如果需要除法运算，请确保将分母过滤为非零值。
问题是：
{question}
"""

cypher_generation_prompt = PromptTemplate(
input_variables=["schema", "question"], template=cypher_generation_template
)

qa_generation_template = """您是一个助手，根据Neo4j Cypher查询的结果生成人类可读的响应。
查询结果部分包含基于用户自然语言问题生成的Cypher查询的结果。
提供的信息是权威的，您绝不能怀疑它或尝试使用内部知识来纠正它。
使答案听起来像对问题的回答。
查询结果：
{context}
问题：
{question}
如果提供的信息为空，请说您不知道答案。
空信息的样子是这样的：[]
如果信息不为空，您必须使用结果提供答案。
如果问题涉及时间持续时间，请假设查询结果以天为单位，除非另有说明。
如果查询结果中有数据，永远不要说您没有正确的信息。始终使用查询结果中的数据。
有用的回答：
"""

qa_generation_prompt = PromptTemplate(
input_variables=["context", "question"], template=qa_generation_template
)

fromlangchain_community.graphs importNeo4jGraph
fromlangchain.chains importGraphCypherQAChain
fromlangchain_openai importChatOpenAI

graph = Neo4jGraph(
url="bolt://localhost:7687",
username="neo4j",
password="******",
)

graph.refresh_schema()

hospital_cypher_chain = GraphCypherQAChain.from_llm(
cypher_llm=ChatOpenAI(model='gpt-4o'),
qa_llm=ChatOpenAI(model='gpt-4o'),
graph=graph,
verbose=True,
qa_prompt=qa_generation_prompt,
cypher_prompt=cypher_generation_prompt,
validate_cypher=True,
top_k=100,
)

response = hospital_cypher_chain.invoke("黄彬伟订购了哪些产品?总金额是多少")
print(response['result'])

在控制台运行可以观察到如下的输出提示与结果，你可以看到生成的完整Cypher语句和运行结果，以及最后LLM的生成答案：

再测试一些常见的查询问题，都能得到满意的答案。比如：

response = hospital_cypher_chain.invoke("对比下不同产品的销售总额")
print(response['result'])

response = hospital_cypher_chain.invoke("销售金额最高的是哪一个产品？")
print(response['result'])

一个很有意思的尝试是构建一个基于关系型数据库和Text-to-SQL检索技术的RAG查询引擎，然后对两者查询的功能与性能进行对比。根据简单的测试，在关系比较简单的场景下，两者在功能上并没有太大的区别，大部分任务都可以完成；但是如果涉及较复杂的多跳查询，且在数据量较大（如100万以上）时，基于Graph的检索性能会更好。感兴趣的朋友可以自行做深入的研究。

以上我们介绍了如何把结构化的数据通过处理转化成基于Graph结构的知识图谱，并借助于Text-to-GraphQL技术进行RAG应用的检索与增强生成。我们将在下篇介绍如何把非结构化的文本知识抽取形成知识图谱，并构建GraphRAG。