nano-graphrag 源码深度解析 v0.0.8.2 · gusye1234/nano-graphrag

nano-graphrag 是微软 GraphRAG 的极简重实现：约 1100 行核心代码（不含测试与 Prompt），保留了图索引 + 社区报告 + 多模式检索的全部核心能力。

整个系统围绕「三类存储 + 两条管线」构建：写入时把文本炼成知识图谱与社区报告，查询时从图与报告中组装上下文交给 LLM 作答。

源码地图

graphrag.py:52 — 一个 @dataclass 承担了「配置中心 + 依赖容器 + 流程编排」三重角色，__post_init__ 完成全部装配。

配置分组一览

__post_init__ 装配流程

整个项目的可扩展性支点：三个抽象基类 + 一套生命周期回调，任何后端只要实现这几个 async 方法即可接入。

4.1 三类存储接口

生命周期回调

所有存储继承 StorageNameSpace，带三个钩子：index_start_callback（插入前，Neo4j 用它建索引）、index_done_callback（插入后统一 commit：JSON 落盘、graphml 写出、向量库存盘）、query_done_callback。GraphRAG 用 asyncio.gather 并发驱动所有存储的回调——这就是「内存中改，结束时一次性写盘」的批式持久化模型。

4.2 默认实现与产物文件

graphrag.py:277 ainsert — 七步流水线，每步幂等、可重入。

5.1 文本分块：两种策略

chunking_by_token_size（默认）：滑动窗口硬切，步长 = max_token_size - overlap_token_size。简单、与语义无关，胜在可预测。

# _op.py:31 — 滑动窗口
for start in range(0, len(tokens), max_token_size - overlap_token_size):
    chunk_token.append(tokens[start : start + max_token_size])
    lengths.append(min(max_token_size, len(tokens) - start))

chunking_by_seperators：先用 SeparatorSplitter 按多级分隔符（段落 → 换行 → 中英文句号/感叹/问号 → 空白，见 prompt.py 的 default_text_separator）切出原子片段，再贪心合并到 chunk_size，并对相邻 chunk 强制注入 overlap——切点落在自然边界，对下游抽取更友好。分隔符本身也被 encode 成 token 序列在 token 空间匹配。

5.2 实体关系抽取 extract_entities

这是与 LLM 交互最密集的环节，设计上有四个值得注意的机制：

① 约定式文本协议（而非 JSON）

Prompt 要求 LLM 输出用三种分隔符编码的记录流：

("entity"<|>"ALEX"<|>"PERSON"<|>"Alex is a character who...")##
("relationship"<|>"ALEX"<|>"TAYLOR"<|>"Alex is affected by..."<|>7)<|COMPLETE|>

# tuple_delimiter=<|>  record_delimiter=##  completion_delimiter=<|COMPLETE|>

元组首字段区分实体/关系；解析时正则 r"\((.*)\)" 取括号内内容再按 tuple_delimiter 切分。相比 JSON，这种协议对开源模型更鲁棒（不易被截断成非法 JSON）。

② Gleaning（补抽循环）

首轮抽取后，追加发送「MANY entities were missed...」继续提示，让 LLM 补充遗漏；并可选地再问一句「还有遗漏吗？YES/NO」决定是否继续。entity_extract_max_gleaning 控制最大补抽轮数（默认 1）。多轮对话历史通过 pack_user_ass_to_openai_messages 拼接维护。

③ 全 chunk 并发 + 全局聚合

# 所有 chunk 的抽取在限流器约束下并发
results = await asyncio.gather(*[_process_single_content(c) for c in ordered_chunks])

# 跨 chunk 聚合：同名实体收拢；无向图——边键统一 sorted(tuple)
maybe_edges[tuple(sorted(k))].extend(v)

④ 记录级校验

_handle_single_entity_extraction 要求至少 4 个字段且首字段为 "entity"；关系要求 5 字段，权重字段用 is_float_regex 校验，不合法则回退 1.0。实体名/类型统一 clean_str + upper() 保证图内实体命名一致（这是合并正确性的基石）。

5.3 合并与去重：_merge_nodes / _merge_edges

合并完成后，实体以 ent-md5 写入 entities_vdb，content = entity_name + description——向量检索时语义信号同时来自名称与描述。

5.4 Leiden 社区聚类

NetworkX 路径（gdb_networkx.py:230）：

community_schema() 随后重建社区视图：遍历所有节点的 clusters，聚合每个社区的节点集、边集、chunk_ids；occurrence = 社区 chunk 数 / 全图最大社区 chunk 数（衡量社区覆盖语料的广度）；跨层通过节点集包含关系推导 sub_communities。Neo4j 版本则用 GDS 的 gds.leiden.write 把 communityIds 写回节点属性，逻辑同构。

5.5 社区报告生成：层级化 + 预算打包

_op.py:625 generate_community_report 是全文最精巧的函数之一。

按层级从细到粗生成

for level in sorted(set(levels), reverse=True):   # level 2 → 1 → 0（细 → 粗）
    this_level_reports = await asyncio.gather(*[
        _form_single_community_report(c, community_datas) for c in this_level])
    community_datas.update(...)  # 细社区的报告会喂给粗社区当输入

粗社区的描述里可以引用子社区的报告摘要——形成自下而上的信息浓缩层级。

_pack_single_community_describe 的预算分配术

报告 Schema

LLM 以 JSON 返回：title / summary / rating(0-10) / rating_explanation / findings(5-10 条 summary+explanation)。_community_report_json_to_str 再把 JSON 渲染成 Markdown 报告字符串，两份都存进 community_reports KV。rating 在 global 查询里充当过滤与排序信号。

QueryParam.mode 决定三种检索路径，共享同一套 token 预算与「CSV 上下文注入」的回答风格。

6.1 local 模式：以实体为锚的四源上下文组装

_op.py:844 _build_local_query_context — 经典的 GraphRAG 局部检索：先向量召回 top_k（默认20）个相关实体，再沿图结构扩展出四类证据。

三个扩展函数各自藏着排序智慧：

• _find_most_related_community_from_entities：统计命中实体指向的社区（level ≤ param.level），按命中频次 → 报告 rating 二级排序截断。
• _find_most_related_edges_from_entities：取全部实体的一跳边去重，按 (edge_degree, weight) 降序截断。
• _find_most_related_text_unit_from_entities：从实体 source_id 回溯原始分块；关键创新是计算每个分块与一跳邻居共享的关联数（relation_counts），排序键 (order, -relation_counts)——既保召回实体的原文优先，也让「多方共同佐证」的分块胜出。

四路内容最终渲染为四段 CSV（Reports / Entities / Relationships / Sources）塞进 local_rag_response 系统提示词。CSV 用 list_of_list_to_csv（逗号+tab 分隔、字段加引号并转义）——token 高效且 LLM 容易理解。

6.2 global 模式：社区报告上的 Map-Reduce

与官方实现的差异：官方是「每个社区一个分析师 + 投票排序」的多分析师框架；nano 简化为「按 token 分组的并行 map + 全局 score 排序 reduce」，调用次数更少、逻辑更直白，效果在多数场景足够。

6.3 naive 模式：最朴素的向量 RAG

chunks_vdb 向量检索 top_k 分块 → 取回原文 → token 截断 → 用 --New Chunk-- 拼接 → naive_rag_response 提示词作答。价值在于提供一条无图基线：同一语料下对比 local/global/naive 的效果，量化知识图谱带来的增益（examples/benchmarks 里即用此思路）。

QueryParam 预算总表

设计模式：local 的三路预算 4000+4800+3200 ≈ 12000 tokens，加上实体表与提示词模板正好填满一个 16k 级模型的舒适区——预算是显式工程，不是碰运气。

对多家供应商的统一 async 封装，核心是三件事：单例客户端、指数退避重试、基于内容哈希的响应缓存。

统一函数签名

async def openai_complete_if_cache(model, prompt, system_prompt=None, history_messages=[], **kwargs) -> str

所有 complete 函数遵循同一签名：prompt + system_prompt + history_messages → str。kwargs 里弹出 hashing_kv（由 GraphRAG 在装配时 partial 注入）。基于此，create_amazon_bedrock_complete_function(model_id) 用工厂模式为任意 Bedrock 模型动态生成符合签名的函数——这正是「换 LLM 只改一个字段」的实现基础。

LLM 缓存机制

# 1. 组装 messages 后算哈希
args_hash = compute_args_hash(model, messages)          # md5(str((model, messages)))
if_cache_return = await hashing_kv.get_by_id(args_hash)
if if_cache_return is not None:
    return if_cache_return["return"]                    # 命中：零成本返回

# 2. 未命中：调 API，写回缓存并立即落盘
response = await openai_async_client.chat.completions.create(...)
await hashing_kv.upsert({args_hash: {"return": ..., "model": model}})
await hashing_kv.index_done_callback()

缓存粒度是「完整 messages 列表」——相同的抽取 prompt、相同的社区内容在重复 insert 时完全免费。这也解释了为什么「丢弃旧社区报告后重跑」在语料不变时成本趋近于零。

重试与容错

• tenacity 装饰器：RateLimitError / APIConnectionError 时最多重试 5 次，指数退避 4~10 秒。
• 全局单例客户端（get_openai_async_client_instance 等）避免重复建连。
• Embedding 函数用 wrap_embedding_func_with_attrs 附带 embedding_dim / max_token_size 元数据——存储层构造时读取维度初始化索引，无需用户重复声明。

305 行里藏着六个支撑全局的设计，逐一拆解：

520 行 Prompt 全部移植自微软 GraphRAG，集中在一个 PROMPTS 字典里，运行时可整体覆盖——魔改行为最经济的入口。

Prompt 设计的三个共性

• Grounding 规则反复强调：「Do not include information where the supporting evidence for it is not provided」「Do not make anything up」在每个回答类提示里至少出现两次（目标段前后各一次）——抗幻觉的提示层手段。
• 目标长度/格式二次注入：{response_type} 在提示头尾各出现一次，强化格式遵循。
• 分隔符即协议：DEFAULT_TUPLE_DELIMITER=<|>、RECORD=##、COMPLETION=<|COMPLETE|>、GRAPH_FIELD_SEP=<SEP>——四个常量贯穿抽取、解析、合并、存储全链路。

entity_extraction/ — 可选的「类型化抽取」替代路径：用 DSPy Signature + Pydantic 模型取代文本协议解析。

DSPy 版抽取后仍复用 _merge_nodes_then_upsert / _merge_edges_then_upsert——抽取器可替换、合并语义保持统一，这正是 base 层抽象的价值。另有 generate_dataset 可把抽取结果固化为 dspy.Example 数据集用于后续编译优化。

通读全部源码后，值得带走的八个设计决策：

已知局限

• 社区报告不支持增量更新，每次 insert 全量重建（有 LLM 缓存兜底，但聚类本身会重算）。
• covariates（claim 追踪）只有 Prompt 残留，功能未实现。
• NetworkX 全内存模型适合中小图；超大图需换 Neo4j 后端。
• JSON KV 存储整读整写，多 GB 语料时建议换嵌入式数据库实现。
• node2vec 节点嵌入是预留接口，local 查询实际依赖实体文本向量而非图结构嵌入。

如果你要亲自读一遍，推荐这条 90 分钟路径：