知识图谱构建实战：结合GraphRAG与Neo4j实现可视化分析
知识图谱构建实战：结合GraphRAG与Neo4j实现可视化分析
前言
在之前的讨论中，我们介绍了如何使用 GraphRAG 从原始文本中提取知识图谱并构建图结构。最终生成的知识图谱文件以 parquet 格式存储在以下文件夹中：
本节将深入探讨如何将生成的图谱文件导入到 Neo4j 图数据库中，以实现可视化分析，并与之前的项目进行结合检索。
一、准备工作
首先，创建一个新的 python 脚本文件，例如 graphrag_import.py，可以放在项目的根目录。接着设置 GraphRAG 生成的图谱文件所在目录：
GRAPHRAG_FOLDER="artifacts"
如果您之前已经安装了 neo4j，可以跳过以下安装步骤：
pip install --upgrade --quiet neo4j
接下来，导入所需的库：
import pandas as pd
from neo4j import GraphDatabase
import time
然后设置 Neo4j 数据库的地址、账户密码以及要导入的数据库名称：
NEO4J_URI="bolt://********:7687"
NEO4J_USERNAME="neo****"
NEO4J_PASSWORD="*****"
NEO4J_DATABASE="****"
driver = GraphDatabase.driver(NEO4J_URI, auth=(NEO4J_USERNAME, NEO4J_PASSWORD))
下载一个语料数据集，链接为 https://www.gutenberg.org/cache/epub/24022/pg24022.txt，并在根目录新建 /ragtest/input 文件夹，将下载的文本文件放入 input 目录中。
二、创建约束
接下来，我们定义一个批处理方法，以便将数据导入 Neo4j。
该方法的参数包括：statement（要执行的 Cypher 查询）、df（待导入的数据框）和 batch_size（每批导入的行数）。
def batched_import(statement, df, batch_size=1000):
total = len(df)
start_s = time.time()
for start in range(0,total, batch_size):
batch = df.iloc[start: min(start+batch_size,total)]
result = driver.execute_query("UNWIND $rows AS value " + statement,
rows=batch.to_dict('records'),
database_=NEO4J_DATABASE)
print(result.summary.counters)
print(f'{total} rows in { time.time() - start_s} s.')
return total
在 Neo4j 中，索引用于快速查找图形查询的起点。为避免重复，我们主要在实体类型的 ID 上创建约束。我们使用一些带有两个下划线的类型作为标签，以将它们与实际实体类型区分开。以下是一些常用标签及其含义：
__Entity__：通常代表现实世界中的某个对象、人物或地点等可识别的实体。__Document__：表示文档或文件，通常是文本数据的容器。__Chunk__：表示文档的某个片段，通常用于文本分块处理。__Community__：表示图结构中的聚类或群体，通常应用于社交网络分析。__Covariate__：代表与其他变量一起在统计模型中使用的变量。
以下是一个简单的 Neo4j 查询示例：
MATCH (e:Entity)-[:CONTAINS]->(d:Document)
WHERE e.type = 'Community' AND d.covariate = 'SomeValue'
RETURN e, d
以上查询将查找类型为 "Community" 的实体，并返回具有特定协变量值的文档。
创建约束的代码如下：
statements = """
create constraint chunk_id if not exists for (c:__Chunk__) require c.id is unique;
create constraint document_id if not exists for (d:__Document__) require d.id is unique;
create constraint entity_id if not exists for (c:__Community__) require c.community is unique;
create constraint entity_id if not exists for (e:__Entity__) require e.id is unique;
create constraint entity_title if not exists for (e:__Entity__) require e.name is unique;
create constraint entity_title if not exists for (e:__Covariate__) require e.title is unique;
create constraint related_id if not exists for ()-[rel:RELATED]->() require rel.id is unique;
""".split(";")
for statement in statements:
if len((statement or "").strip()) > 0:
print(statement)
driver.execute_query(statement)
通过将字符串分割为数组并循环执行，成功创建了相应的约束。在执行成功后，您将看到约束的结果：
三、导入文档
现在我们需要加载文档的 parquet 文件。使用 Python 的 pandas 库读取和处理数据，并根据其 id 创建节点，添加 title 属性。
以下是读取文档的代码：
doc_df = pd.read_parquet(f'{GRAPHRAG_FOLDER}/create_final_documents.parquet', columns=["id", "title"])
doc_df.head(2)
执行该代码后，您将看到文档数据的输出：
接着，将文档导入图数据库：
# import documents
statement = """
MERGE (d:__Document__ {id:value.id})
SET d += value {.title}
"""
batched_import(statement, doc_df)
导入完成后，您将得到以下结果：
然后，我们将加载文本单元，为每个 id 创建节点并设置文本和标记数量，并将这些单元连接到之前创建的文档。
text_df = pd.read_parquet(f'{GRAPHRAG_FOLDER}/create_final_text_units.parquet',
columns=["id","text","n_tokens","document_ids"])
text_df.head(2)
执行后，将看到以下结果：
接下来，将文本单元导入图数据库：
statement = """
MERGE (c:__Chunk__ {id:value.id})
SET c += value {.text, .n_tokens}
WITH c, value
UNWIND value.document_ids AS document
MATCH (d:__Document__ {id:document})
MERGE (c)-[:PART_OF]->(d)
"""
batched_import(statement, text_df)
该 cypher 语句的含义是创建或更新一个 __Chunk__ 节点，设置其属性，并为与该 Chunk 相关的每个 Document 创建 PART_OF 关系。运行该代码后，您将看到如下结果：
接下来，我们继续加载实体：
entity_df = pd.read_parquet(f'{GRAPHRAG_FOLDER}/create_final_entities.parquet',
columns=["name","type","description","human_readable_id","id","description_embedding","text_unit_ids"])
entity_df.head(2)
执行后将得到如下结果：
导入实体的代码如下：
entity_statement = """
MERGE (e:__Entity__ {id:value.id})
SET e += value {.human_readable_id, .description, name:replace(value.name,'"','')}
WITH e, value
CALL db.create.setNodeVectorProperty(e, "description_embedding", value.description_embedding)
CALL apoc.create.addLabels(e, case when coalesce(value.type,"") = "" then [] else [apoc.text.upperCamelCase(replace(value.type,'"',''))] end) yield node
UNWIND value.text_unit_ids AS text_unit
MATCH (c:__Chunk__ {id:text_unit})
MERGE (c)-[:HAS_ENTITY]->(e)
"""
batched_import(entity_statement, entity_df)
运行后，您将看到以下结果：
然后加载关系：
rel_df = pd.read_parquet(f'{GRAPHRAG_FOLDER}/create_final_relationships.parquet',
columns=["source","target","id","rank","weight","human_readable_id","description","text_unit_ids"])
rel_df.head(2)
得到结果如下：
导入关系的代码如下：
rel_statement = """
MATCH (source:__Entity__ {name:replace(value.source,'"','')})
MATCH (target:__Entity__ {name:replace(value.target,'"','')})
MERGE (source)-[rel:RELATED {id: value.id}]->(target)
SET rel += value {.rank, .weight, .human_readable_id, .description, .text_unit_ids}
RETURN count(*) as createdRels
"""
batched_import(rel_statement, rel_df)
运行后，您将获得如下结果：
接下来加载社区：
community_df = pd.read_parquet(f'{GRAPHRAG_FOLDER}/create_final_communities.parquet',
columns=["id","level","title","text_unit_ids","relationship_ids"])
community_df.head(2)
执行后，将看到结果：
导入社区的代码如下：
statement = """
MERGE (c:__Community__ {community:value.id})
SET c += value {.level, .title}
WITH *
UNWIND value.relationship_ids as rel_id
MATCH (start:__Entity__)-[:RELATED {id:rel_id}]->(end:__Entity__)
MERGE (start)-[:IN_COMMUNITY]->(c)
MERGE (end)-[:IN_COMMUNITY]->(c)
RETURN count(distinct c) as createdCommunities
"""
batched_import(statement, community_df)
执行后，您将看到以下结果：
最后，加载社区报告：
community_report_df = pd.read_parquet(f'{GRAPHRAG_FOLDER}/create_final_community_reports.parquet',
columns=["id","community","level","title","summary", "findings","rank","rank_explanation","full_content"])
community_report_df.head(2)
结果如下：
导入社区报告的代码如下：
# import communities
community_statement = """
MERGE (c:__Community__ {community:value.community})
SET c += value {.level, .title, .rank, .rank_explanation, .full_content, .summary}
WITH c, value
UNWIND range(0, size(value.findings)-1) AS finding_idx
WITH c, value, finding_idx, value.findings[finding_idx] as finding
MERGE (c)-[:HAS_FINDING]->(f:Finding {id:finding_idx})
SET f += finding
"""
batched_import(community_statement, community_report_df)
运行后，您将看到最终结果：
至此，我们已成功将 GraphRAG 生成的文件导入到 Neo4j 中，可以通过 Neo4j 浏览器界面进行可视化分析。
每个实体的各种关系可以直观地查看：
在查看社区时，可以看到关于某一事件的整合，并且关联了相关人物：
通过多种可视化分析方式，您可以查看文档、文本单元等。根据不同的输入检索文本，我们需要进行相应的数据分析，从而获得清晰直观的结果。
四、总结
通过将 GraphRAG 生成的图文件导入到 Neo4J，我们能够利用 Neo4J 进行可视化分析，从而更直观地理解 GraphRAG 的索引结果。
