知识图谱构建实战：结合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)

运行后，您将看到最终结果：