使用Neo4j Graphrag SimpleKGPipeline时PipelineResult类型错误处理问询

1. 查看PipelineResult的可用属性/方法 #

由于PipelineResult属于experimental模块，官方文档缺失时，直接通过代码查看对象的内置成员：

result = await pipeline.run_async(...)
# 打印所有属性和方法
print(dir(result))

通常会包含nodes（节点列表）、relationships（关系列表）这类核心属性，直接对应知识图谱数据。

2. 转换为字典格式 #

如果PipelineResult基于Pydantic模型实现，直接调用模型方法导出字典：

# 适配不同版本的Pydantic
result_dict = result.model_dump()  # Pydantic v2
# 或者
result_dict = result.dict()  # Pydantic v1

如果没有内置转换方法，手动提取核心数据构造字典：

kg_dict = {
    "nodes": result.nodes,
    "relationships": result.relationships,
    # 其他可能的属性如metadata等按需添加
}

3. 提取数据存入Neo4j #

直接从PipelineResult中取出节点和关系，用Neo4j Driver执行Cypher导入：

from neo4j import AsyncGraphDatabase

async def import_kg_to_neo4j(nodes, relationships, uri, auth):
    driver = AsyncGraphDatabase.driver(uri, auth=auth)
    async with driver.session() as session:
        # 批量导入节点（推荐用UNWIND提升效率）
        await session.run(
            """
            UNWIND $nodes AS node
            CREATE (n:${node.label})
            SET n += node.properties
            """,
            nodes=nodes
        )
        # 批量导入关系
        await session.run(
            """
            UNWIND $rels AS rel
            MATCH (a:${rel.start_label} {id: rel.start_id}), (b:${rel.end_label} {id: rel.end_id})
            CREATE (a)-[r:${rel.type}]->(b)
            SET r += rel.properties
            """,
            rels=relationships
        )
    await driver.close()

# 使用示例
await import_kg_to_neo4j(
    result.nodes,
    result.relationships,
    uri="bolt://localhost:7687",
    auth=("neo4j", "your_password")
)

4. 转换为字符串用于日志/调试 #

直接调用str()方法，或自定义格式化输出：

# 基础转换
result_str = str(result)

# 自定义格式化，更清晰
formatted_str = (
    f"知识图谱统计：节点数={len(result.nodes)}, 关系数={len(result.relationships)}\n"
    f"节点示例：{result.nodes[:2]}\n"
    f"关系示例：{result.relationships[:2]}"
)

内容的提问来源于stack exchange，提问作者Aravind_Sagar