GraphRAG 与 Neo4j 集成实战：数据导入与图谱可视化

批量数据导入流程

为了提高效率，我们采用分批导入策略。每次处理一定数量的行，避免内存溢出。

1. 批处理函数封装

def batched_import(statement, df, batch_size=1000):
    """
    使用批处理方式将 DataFrame 导入 Neo4j。
    :param statement: Cypher 查询语句
    :param df: 待导入的 Pandas DataFrame
    :param batch_size: 每批次处理的行数
    """
    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

2. 导入文档与文本单元

首先导入基础文档信息和分片（Text Units），建立文档与内容的关联。

# 导入文档
GRAPHRAG_FOLDER = "./output/artifacts"
doc_df = pd.read_parquet(f'{GRAPHRAG_FOLDER}/create_final_documents.parquet', columns=["id", "title"])

statement_doc = """
MERGE (d:__Document__ {id:value.id})
SET d += value {.title}
"""
batched_import(statement_doc, doc_df)

# 导入文本联系 (Chunks)
text_df = pd.read_parquet(f'{GRAPHRAG_FOLDER}/create_final_text_units.parquet',
                          columns=["id","text","n_tokens","document_ids"])

statement_chunk = """
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_chunk, text_df)

3. 导入实体与关系

实体是图谱的核心节点，关系则定义了它们之间的交互。导入时需处理标签动态添加和向量属性。

# 导入抽取的实体
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_statement = """
MERGE (e:__Entity__ {id:value.id})
SET e += value {.human_readable_id, "description": replace(value.description,'"','')}
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_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)

4. 导入社区与报告

GraphRAG 的层级结构通过 Community 节点体现，包含社区报告摘要和发现点。

# 导入社区
community_df = pd.read_parquet(f'{GRAPHRAG_FOLDER}/create_final_communities.parquet',
                               columns=["id", "level", "title", "text_unit_ids", "relationship_ids"])

statement_community = """
MERGE (c:__Community__ {community:value.id})
SET c += value {.level, .title}
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, 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_statement = """
MATCH (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)

结果展示与分析

完成上述步骤后，数据已成功存入 Neo4j。我们可以通过 Neo4j Browser 进行可视化探索。

1. 图谱概览

不同的节点颜色代表不同的类型（如 Entity, Document, Community）。通过缩放和平移，可以观察整体网络拓扑结构。例如，在《斗破苍穹》世界观案例中，可以看到角色、地点、事件之间复杂的关联网络。

2. 常用查询示例

• 查找特定实体的邻居：

  MATCH (e:__Entity__ {name:"萧炎"})-[*1..2]-(neighbor)
  RETURN e, neighbor
  

• 查看社区层级：

  MATCH (c:__Community__)-[:IN_COMMUNITY]->(e:__Entity__)
  RETURN c.level, count(e) AS entity_count
  ORDER BY c.level DESC
  

常见问题排查

1. 连接拒绝：检查 Docker 端口是否被占用，确认 neo4j://localhost 地址正确。
2. 权限错误：确保使用了正确的用户名和密码，且密码未过期。
3. 导入缓慢：增加 batch_size 参数或优化硬件资源；检查磁盘 I/O 性能。
4. 约束冲突：如果数据已存在且 ID 重复，需先清理旧数据或调整导入逻辑。

总结

本文详细演示了从 GraphRAG 导出结果到 Neo4j 入库的全流程。通过构建规范的数据模型和高效的导入脚本，我们可以将非结构化文本转化为可查询的知识图谱。这不仅有助于理解 GraphRAG 的内部机制，也为后续的大模型检索增强应用提供了坚实的结构化数据基础。在实际生产中，建议结合定期增量更新策略，保持图谱数据的时效性。