前言：GraphRAG图谱导入Neo4j可视化分析实战

此前我们曾探讨过 GraphRAG 从原始文本中提取知识图谱并构建图结构的过程，最终生成的文件格式为 parquet，存储在类似下图的文件夹中：

本次任务十分明确：将这些图谱文件导入到 Neo4j 图数据库中，随后进行可视化分析。此外，还可以与我们此前构建的混合检索项目联动，实现一鱼多吃的效果。

一、准备工作（环境配置与依赖安装）

首先，创建一个 Python 脚本，例如命名为 graphrag_import.py，放置在项目根目录下（位置可自定义）。然后，指定 GraphRAG 生成的图谱目录路径：

  GRAPHRAG_FOLDER="artifacts"

若尚未安装 neo4j 驱动，请先执行安装命令：

  pip install --upgrade --quiet neo4j

接着，导入所需的 Python 库：

import pandas as pd
from neo4j import GraphDatabase
import time

配置 Neo4j 数据库的连接参数，包括 URI、用户名、密码及目标数据库：

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 空文件夹，并将下载的文件放入其中。

二、创建数据库约束与索引

首先，编写一个批量写入函数，用于将数据分批导入 Neo4j。该函数接收三个参数：要执行的 Cypher 查询语句、数据框（DataFrame）以及每批处理的行数（默认1000）。

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__。这些标签本身没有固定含义，完全取决于你的数据模型如何设计。

• __Entity__ 表示一个实体，例如公司、人物、地点等。

• __Document__ 表示文档或文件，例如一本书、一篇文章。

• __Chunk__ 表示文档的片段（文本块），例如段落或句子。

• __Community__ 表示图结构中的社区或聚类，例如社交网络中兴趣相投的用户群体。

• __Covariate__ 表示协变量，在统计模型中与其他变量一起使用，可能影响数据点的属性。

以下是一个简单的查询示例：

MATCH (e:Entity)-[:CONTAINS]->(d:Document)
WHERE e.type = 'Community' AND d.covariate = 'SomeValue'
RETURN e, d

该查询查找类型为 Community 的实体，以及它们所包含的、具有特定协变量值的文档。当然，实际用法因场景而异，关键在于根据具体需求设计合理的数据模型。关于 Neo4j 的查询细节，后续有机会再详细展开。

下面列出创建约束的 Cypher 语句，通过字符串分割后逐条执行：

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)

例如，第一条语句的含义是：为标签为 __Chunk__ 的节点设置 id 属性唯一约束，若约束已存在则忽略。执行成功后，会得到类似下方的结果：

三、导入文档与文本单元

首先，使用 pandas 加载文档的 parquet 文件，仅保留 id 和 title 两列。注意，此处无需 text_unit_ids 列，因为我们可以通过关系关联，且文本内容已包含在块（Chunk）中。

doc_df = pd.read_parquet(f'{GRAPHRAG_FOLDER}/create_final_documents.parquet', columns=["id", "title"])
doc_df.head(2)

接着，将文档数据批量写入 Neo4j：

# import documents
statement = """
MERGE (d:__Document__ {id:value.id})
SET d += value {.title}
"""
batched_import(statement, doc_df)

接下来处理文本单元（Text Unit）。为每个 id 创建一个 __Chunk__ 节点，设置文本内容和 token 数量，并将其与对应的 __Document__ 节点建立关联。

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__ 节点并设置属性；然后对每个相关的 document_id，找到对应的 __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,'"','')})
    // not necessary to merge on id as there is only one relationship per pair
    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}
/*
UNWIND value.text_unit_ids as text_unit_id
MATCH (t:__Chunk__ {id:text_unit_id})
MERGE (c)-[:HAS_CHUNK]->(t)
WITH distinct c, value
*/
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 的索引结果。这样一来，整个索引结果的结构与关联关系变得清晰可辨，分析效率得到了显著提升。