GraphRAG项目中的JSON格式输出问题分析与解决方案

问题背景

在GraphRAG项目中，用户在使用--emit json参数运行索引命令时遇到了一个关键问题。当指定输出为JSON格式时，系统仍然尝试查找并加载Parquet格式文件，导致流程中断。这一现象揭示了项目中文件格式处理逻辑存在的不一致性。

问题详细分析

问题的核心在于GraphRAG的代码实现中，部分函数对文件格式的处理采用了硬编码方式。具体表现为：

1. 在run.py文件中，load_table_from_storage和inject_workflow_data_dependencies等函数默认查找.parquet后缀的文件
2. 虽然系统成功生成了JSON格式的输出文件（如create_base_text_units.json），但后续流程仍尝试加载不存在的Parquet文件
3. 这种不一致性导致整个索引流程在创建基础文本单元后立即失败

技术原理

GraphRAG作为一个基于图的检索增强生成框架，其数据处理流程涉及多种文件格式：

• Parquet：列式存储格式，适合大规模数据分析，具有高效的压缩和查询性能
• JSON：轻量级数据交换格式，易于人类阅读和编写，但解析效率较低

在数据处理流程中，系统需要保持输入输出格式的一致性。当用户指定JSON格式输出时，整个流程的各环节都应适配这种格式，而非混合使用不同格式。

解决方案

针对这一问题，社区贡献者提出了两种解决方案：

核心代码修改方案

1. 修改load_table_from_storage函数：

async def load_table_from_storage(name: str) -> pd.DataFrame:
    if not await storage.has(name):
        msg = f"Could not find {name} in storage!"
        raise ValueError(msg)
    try:
        log.info("read table from storage: %s", name)
        # 读取JSON数据而非Parquet
        content = await storage.get(name, encoding='utf-8')
        json_data = [json.loads(line) for line in content.splitlines() if line.strip()]
        return pd.DataFrame(json_data)
    except Exception:
        log.exception("error loading table from storage: %s", name)
        raise

2. 更新inject_workflow_data_dependencies函数：

async def inject_workflow_data_dependencies(workflow: Workflow) -> None:
    workflow.add_table(DEFAULT_INPUT_NAME, dataset)
    deps = workflow_dependencies[workflow.name]
    log.info("dependencies for %s: %s", workflow.name, deps)
    for id in deps:
        workflow_id = f"workflow:{id}"
        # 加载JSON文件而非Parquet
        table = await load_table_from_storage(f"{id}.json")
        workflow.add_table(workflow_id, table)

查询功能适配方案

对于查询功能，也需要相应修改以支持JSON格式：

1. 本地搜索功能修改：

def run_local_search(data_dir: str | None, root_dir: str | None, ...):
    # 使用JSON文件读取替代Parquet
    def read_json_file(file_path):
        with open(file_path, 'r') as f:
            return pd.DataFrame([json.loads(line) for line in f if line.strip()])
    
    final_nodes = read_json_file(data_path / "create_final_nodes.json")
    final_community_reports = read_json_file(data_path / "create_final_community_reports.json")
    # 其他文件读取...

2. 全局搜索功能修改：

def run_global_search(data_dir: str | None, root_dir: str | None, ...):
    # 同样采用JSON文件读取
    def read_json_file(file_path):
        with open(file_path, 'r') as f:
            return pd.DataFrame([json.loads(line) for line in f if line.strip()])
    
    final_nodes = read_json_file(data_path / "create_final_nodes.json")
    final_entities = read_json_file(data_path / "create_final_entities.json")
    # 其他处理逻辑...

项目维护方向

值得注意的是，项目官方最终决定统一使用Parquet格式作为标准输出格式。这一决策可能基于以下考虑：

1. 性能因素：Parquet在大规模数据处理中具有明显优势
2. 维护成本：支持多种格式会增加代码复杂性和测试负担
3. 使用场景：GraphRAG主要面向生产环境，JSON的易读性优势在此场景下不明显

总结与建议

对于GraphRAG用户，建议：

1. 遵循项目官方规范，使用Parquet格式进行数据处理
2. 如需JSON格式，可参考社区贡献的修改方案，但需注意这些修改可能不被官方版本支持
3. 在自定义修改前，充分评估格式选择对系统性能和维护成本的影响

这一问题的出现和解决过程，体现了开源项目中功能需求与技术决策之间的平衡，也为开发者提供了关于数据格式选择的实际参考案例。