跳到主要内容

第 1 章 · 索引流水线：文档怎么一步步变成图

这章讲什么： GraphRAG 的索引不是一个大函数，而是一串有名字的小步骤（workflow）顺序跑。本章讲这串步骤怎么被"工厂"按方法名组装、数据怎么在一堆表之间流、以及第一步"切块"的细节。抽图、聚类、写报告这三大步各有专章（02/03）。

1.1 一句话：索引 = 一串 workflow 顺序执行​

打开 packages/graphrag/graphrag/api/index.py 的 build_index，会发现它自己几乎不干活，只做三件事：

  1. 按"方法"造出一条流水线：PipelineFactory.create_pipeline(config, method)。
  2. 把流水线交给 run_pipeline 去驱动。
  3. 收集每步的结果 PipelineRunResult，回调通知开始/结束/出错。

所以理解索引 = 理解"有哪些 workflow、按什么顺序、各读写哪张表"。

1.2 工厂：把"方法名"翻译成一串步骤名​

PipelineFactory（packages/graphrag/graphrag/index/workflows/factory.py）是个纯注册表：

  • workflows: dict[str, WorkflowFunction] —— 步骤名 → 该步骤的函数。
  • pipelines: dict[str, list[str]] —— 方法名 → 步骤名的有序列表。

create_pipeline 的逻辑很直白：优先用配置里显式给的 config.workflows，否则按方法名查表，然后把"名字列表"映射成"函数列表"打包成 Pipeline：

# 示意，非源码。对应 factory.py create_pipeline
workflows = config.workflows or cls.pipelines.get(method, [])   # 名字列表
return Pipeline([(name, cls.workflows[name]) for name in workflows])  # 名字→函数

四种内置方法跑哪一串​

文件底部用 register_pipeline 注册了四条线（IndexingMethod 枚举见 config/enums.py）。它们共享大部分步骤，差别只在"抽图那一段"和"是否带增量更新尾巴"：

方法抽图方式是否增量适用
standardLLM 抽图（extract_graph）否质量优先、可承担 LLM 成本
fastNLP 名词短语抽图（extract_graph_nlp + prune_graph）否省钱/省时，质量略降
standard-updateLLM 抽图是（尾部加 *_update_workflows）已索引过、来了新文档
fast-updateNLP 抽图是同上但走快速路径

standard 方法展开后的完整步骤序列（_standard_workflows，前面再拼一个 load_input_documents）：

load_input_documents      读入原始文档
  └▶ create_base_text_units   切块 → text_units 表
  └▶ create_final_documents   定稿 documents 表
  └▶ extract_graph            LLM 抽实体/关系 (第 02 章)
  └▶ finalize_graph           算度数、定稿 entities/relationships
  └▶ extract_covariates       (可选) 抽"断言/协变量"claims
  └▶ create_communities       Leiden 聚类 → communities (第 03 章)
  └▶ create_final_text_units  把实体/关系 id 回填进 text_units
  └▶ create_community_reports  LLM 写社区报告 (第 03 章)
  └▶ generate_text_embeddings  文本向量化 → 向量库

fast 只是把 extract_graph 换成 extract_graph_nlp + prune_graph，并用 create_community_reports_text（基于文本块而非图）写报告。

1.3 运行时：workflow 之间靠"表"传数据​

索引期没有把整个数据集抱在内存里传来传去。相邻步骤通过一个共享的 table_provider 读写命名表（documents、text_units、entities…）来交接数据。

run_pipeline（packages/graphrag/graphrag/index/run/run_pipeline.py）就是个顺序驱动器 + 快照器：

# 示意，非源码。对应 _run_pipeline 的主循环
for name, workflow_function in pipeline.run():
    context.callbacks.workflow_start(name, None)
    result = await workflow_function(config, context)   # 步骤自己读表、写表
    context.callbacks.workflow_end(name, result)
    yield PipelineRunResult(workflow=name, result=result.result, ...)
    await _dump_stats_json(context)                     # 每步后落一次统计

每个 workflow 的签名统一是 run_workflow(config, context) -> WorkflowFunctionOutput；context 里带着 output_table_provider（读写表）、cache（LLM 缓存）、callbacks（进度）等。真实驱动见 _run_pipeline 中 for name, workflow_function in pipeline.run() 一段。

为什么这么设计（精华）： 步骤间用表交接，带来三个好处——(a) 每步产物落盘，失败可从中间续跑；(b) 内存占用只跟"当前一步"有关，能处理大语料；(c) 步骤是可插拔的纯函数，PipelineFactory.register 就能替换或加自定义步。

1.4 第一步细看：切块（create_base_text_units）​

为什么先讲切块：它是所有后续 LLM 处理的输入粒度，切得太大超上下文、太小丢关系。

create_base_text_units（packages/graphrag/graphrag/index/workflows/create_base_text_units.py）的核心 create_base_text_units 函数是一个流式读写循环：一行文档进、切成若干块、每块立刻写出，不把全部块堆在内存里。

# 示意，非源码。对应 create_base_text_units 内层循环
async for doc in documents_table:              # 逐篇文档流式读
    chunks = chunk_document(doc, chunker, prepend_metadata)
    for chunk_text in chunks:
        row = {"id": "", "document_id": doc["id"], "text": chunk_text,
               "n_tokens": len(tokenizer.encode(chunk_text))}
        row["id"] = gen_sha512_hash(row, ["text"])  # id = 文本内容的哈希
        await text_units_table.write(row)          # 立刻写出

三个值得记住的细节：

  • 块 id 是内容哈希（gen_sha512_hash(row, ["text"])）。内容不变 id 就不变——这是后面"增量更新/缓存命中"能省钱的地基。
  • 切块器可换：create_chunk_results / sentence_chunker 等在独立包 packages/graphrag-chunking/，通过 create_chunker(config.chunking, ...) 按配置选策略（token 切 / 句子切）。
  • 可选 prepend_metadata：把标题/日期等元数据拼到每块开头（chunk_document 里用 add_metadata），让下游抽取有更多上下文。

切完得到 text_units 表（列见 data_model/schemas.py(TEXT_UNITS_FINAL_COLUMNS)：id, text, n_tokens, document_id, entity_ids, relationship_ids, covariate_ids——后三个此刻为空，create_final_text_units 步会回填）。

1.5 数据在表之间怎么流（全景表）​

把 standard 方法各步的"读什么表 → 写什么表"列出来，就是整条数据流：

workflow主要读主要写
load_input_documents原始文件documents
create_base_text_unitsdocumentstext_units
extract_graphtext_unitsentities,relationships(+可选 raw)
finalize_graphrelationships定稿 entities,relationships(补 degree)
extract_covariatestext_unitscovariates(可选)
create_communitiesentities,relationshipscommunities
create_final_text_unitsentities,relationships,covariates回填 text_units
create_community_reportscommunities,entities,relationships,covariatescommunity_reports
generate_text_embeddings各文本列向量库

看懂这张表，就抓住了索引的骨架。接下来两章分别钻进最核心的两段：抽图（第 02 章，extract_graph / extract_graph_nlp）和聚类+报告（第 03 章，create_communities / create_community_reports）。

1.6 代码地图​

主题文件路径符号名
索引对外入口packages/graphrag/graphrag/api/index.pybuild_index
流水线工厂 / 四种方法注册packages/graphrag/graphrag/index/workflows/factory.pyPipelineFactory,register_pipeline,create_pipeline
方法枚举packages/graphrag/graphrag/config/enums.pyIndexingMethod
顺序驱动器 + 快照packages/graphrag/graphrag/index/run/run_pipeline.pyrun_pipeline,_run_pipeline
切块 workflowpackages/graphrag/graphrag/index/workflows/create_base_text_units.pycreate_base_text_units,chunk_document
块内容哈希 idpackages/graphrag/graphrag/index/utils/hashing.pygen_sha512_hash
切块器（独立包）packages/graphrag-chunking/graphrag_chunking/create_chunker,Chunker
输出表列定义packages/graphrag/graphrag/data_model/schemas.pyTEXT_UNITS_FINAL_COLUMNS 等