图检索的成本账：从 GraphRAG 的 Leiden 重建到 LightRAG 的增量并集

图检索的成本账：从 GraphRAG 的 Leiden 重建到 LightRAG 的增量并集

本文从 WildGraphBench 那组数字（naive vector 46.15%、多事实问题掉到 16.67%、图方法 83.33%）出发，用互信息的交叉项把向量 RAG 失败的根因形式化：chunk 独立编码在索引阶段就把拓扑置零。接着拆解 GraphRAG 五段流水线，讲清模块化度的定义如何决定 Leiden 是全局算法、新文档进来只能重建。成本部分自建模型把 500 页语料逐级算成美元，做社区数增长的敏感性分析并给出可运行计算器，指出 50~200 美元是 2024 年定价口径、真正的成本在查询侧。工程部分给出抽取与双层检索两份完整脚本、增量并合的最小实现、消歧四级流水线、图退化检测与存储选型，并列出六种不该上图的场景。

这篇把图检索（GraphRAG / LightRAG）的成本结构和增量更新工程讲透。先用 WildGraphBench 那组数字说明向量 RAG 在多跳与全局归纳上为什么崩（naive vector 46.15%，多事实问题掉到 16.67%，图方法 83.33%），从 chunk 独立向量丢失拓扑讲清根因；然后拆 GraphRAG 全流程，重点讲清 Leiden 为什么是全局算法、因此新文档进来必须重建；接着自己建一个成本模型，把 500 页语料逐级算成美元，并对社区数量随语料规模的增长率做敏感性分析，给出一份可运行的成本计算器；最后是 LightRAG 双层检索与五种查询模式、增量并集合并算法与自写最小实现、实体消歧为什么是真正的瓶颈、图退化检测，以及 Neo4j / NetworkX / PostgreSQL+AGE 的选型。读完你能自己算出这套方案值不值，以及什么情况下根本不该上图。

关键词：GraphRAG、LightRAG、Leiden、知识图谱、实体消歧、增量更新、RAG 成本、Neo4j、Apache AGE

一、写在前面：为什么现在必须聊这件事

2026 年上半年我接手了一个企业知识库项目，客户最开始的诉求很朴素：“我们有个内部的问答，用向量检索做的，用户问’张三负责的项目里，哪几个用了我们自研的中间件’这种问题，答不上来。”

这类问题的特点是：答案不在任何一个 chunk 里，而在 chunk 之间的关系里。 向量检索的第一步就是把文档切成互不相关的块，每块单独编码成一个向量——从这一步开始，拓扑信息就已经丢了。你再怎么调 embedding 模型、加 reranker，都找不回来，因为信息在索引阶段就被扔掉了。

我当时做了两套方案对比：一套是官方 GraphRAG，一套是 LightRAG。两周下来最大的感受不是效果差别（两者在这个语料上差 3 个点以内），而是成本结构的差别：GraphRAG 索引一次花了大约 40 分钟和一笔不便宜的 API 费用，查询一次要 20 秒；LightRAG 索引 3 分钟，查询 30 毫秒。更关键的是第二个星期——客户那边每天有新文档进来，GraphRAG 这套架构要付一次全量重建的钱。

先放五条结论：

二、技术背景与现状梳理

2.1 那组数字：向量检索到底差多少

WildGraphBench 是这两年被引用最多的一组图检索对比数据。我把关键几行抄在这里：

真正刺眼的是拆开看的那一行：在需要跨多个事实的问题上，naive vector RAG 的准确率掉到 16.67%，而图方法能维持在 83.33%。 差了 5 倍。

这个 5 倍不是"图方法更聪明"，是"向量方法在这个问题形态下根本没有可用的信号"。后面 3.1 节会把根因形式化。

另外补充一组 2026 年的独立旁证：Salesforce 的 HERB benchmark（EMNLP 2025 Industry Track）在一个模拟 530 人企业、39190 份异构材料的环境里测出，标准混合检索（dense + BM25）只能拿 20.61 分，GPT-4o 的 ReAct agent 拿 32.96 分，而检索被明确识别为主要瓶颈。有意思的是同一份 benchmark 里 GraphRAG 变体的分数低于朴素混合检索——这说明图不是万灵药，构建质量才是关键。

2.2 四条路线：按写入模式分，不按算法分

2026 年看这类框架，我觉得最有用的分类维度不是"用了什么算法"，而是语料变化时它做什么。这个维度直接决定了你的长期账单：

选型的判断顺序我建议是：先看语料会不会变，再看要不要全局归纳，最后才看图算法本身。 大部分人把顺序搞反了，先被"图能做多跳推理"吸引，上线之后才发现每周付一次全量索引的钱。

微软自己出了 LazyGraphRAG 这个变体，思路是把社区摘要的成本推迟到查询时再做——官方要出一个变体来绕开自己架构的成本问题，这件事本身就说明了问题有多真实。

三、核心原理拆解

3.1 根因：chunk 独立向量丢失了拓扑

把这件事形式化一下。设语料被切成 C 个 chunk：{c₁, …, c_C}，embedding 函数 f 把每个 chunk 独立映射成向量 v_i = f(c_i)。检索时给定查询 q，取 top-k argmax sim(f(q), v_i)。

注意这里发生的事情：f 的输入只有 c_i 自己。 如果两个实体 A 和 B 的关系只存在于"c₃ 提到了 A，c₁₇ 提到了 B，c₃ 和 c₁₇ 都提到了同一个项目 P"这种跨块结构里，那么 sim(f(q), v₃) 和 sim(f(q), v₁₇) 都不包含这个三跳关系的信息——因为编码 v₃ 的时候根本没看过 c₁₇。

用信息论的语言说：令 R 是语料里的关系集合（图的边），I(R; v_i) 是单个 chunk 向量里包含的关系信息。因为 f 独立作用于 c_i，

I(R; v₁, …, v_C) = Σ_i I(R; v_i) 只在关系完全落在单 chunk 内时成立

一旦关系跨 chunk，联合编码能拿到而独立编码拿不到的那部分，就是互信息的交叉项：

I(R; v₁, …, v_C) = Σ_i I(R; v_i) + I_cross

I_cross 这一项，独立 chunk 编码永远是 0。图检索做的全部事情，就是先花一次索引成本把 I_cross 显式地抽出来存成边，然后在查询时沿着边走。

这就解释了三组现象：

• 为什么 top-k 调大没用。 你召回 50 个 chunk 只是把 Σ_i I(R; v_i) 这一项变大，I_cross 还是 0。召回越多，噪声越大，甚至可能稀释答案。
• 为什么 reranker 只能救一半。 reranker 在候选集内重排序，它挑得出"最相关的一个 chunk"，但挑不出"三个各占三分之一相关性的 chunk 的组合"。
• 为什么多事实问题掉得最狠。 16.67% 那一行对应的正是 I_cross 占比最高的问题形态。

反过来，这也给出了判断标准：如果你的问题形态里 I_cross ≈ 0，图检索就没有增量价值。 这是 9.2 节"什么情况不该上图"的理论依据。

3.2 GraphRAG 全流程拆解

GraphRAG 的索引是一条五段流水线：

原始文档
│
├─(1) 切 chunk（1200 token，overlap 100）
▼
chunks ──(2) LLM 抽取实体与关系（含 N 轮 gleaning 追问）──► (实体, 类型, 描述) / (源, 目标, 描述, 权重)
│
├─(3) 实例归一化：同名实体合并，描述拼接
▼
实体表 / 关系表 ──(4) LLM 对每个实体与关系生成摘要描述 ──► 带描述的图
│
├─(5a) Leiden 社区检测（层次结构 level 0/1/2）
├─(5b) 对每个社区生成 community report（map-reduce）
▼
图 + 层次社区 + 社区摘要

查询侧两条路：

• Local Search：从查询里抽实体 → 在图上定位 → 取 1-hop 邻域的实体描述、关系描述、原始 chunk、以及该实体所属社区的报告 → 一次生成。适合"具体事实"类问题。
• Global Search：map-reduce。map 阶段把所有社区报告分批喂给 LLM，每批产出一个带评分的中间答案；reduce 阶段把高分中间答案汇总成最终回答。适合"这个语料整体在讲什么"这类问题。

Global Search 就是那个 61 万 token 的来源，3.5 节算。

3.3 Leiden 为什么是全局算法，因此必须重建

这是全文最关键的一节，也是网上资料讲得最含糊的一节。

Leiden 算法优化的目标是模块化度（modularity）Q：

Q = (1 / 2m) · Σ_{i,j} [ A_ij − (k_i · k_j) / (2m) ] · δ(c_i, c_j)

其中 A 是邻接矩阵，m 是总边数，k_i 是节点 i 的度，c_i 是节点 i 所属社区，δ 在两者同社区时为 1。

看这个式子的三个性质，就知道它为什么是全局的：

所以链路是这样的：

新增文档 → 新实体新边 → 图的度分布变化 → Leiden 重新优化 Q
→ 底层社区边界整体漂移 → 上层所有社区成员变化 → 全部社区摘要失效 → 全部重生成

"新文档进来必须重建"不是实现偷懒，是模块化度这个目标函数的定义决定的。 局部更新一个社区的划分会破坏 Q 的全局最优性，而社区摘要又强依赖划分结果。唯一正确的操作就是重跑。

Leiden 还有两个工程上必须知道的性质：

• 它有分辨率极限（resolution limit）。 模块化度优化在大图上倾向于把小社区吞进大社区。这意味着你的语料越大，单个社区的语义越杂，社区摘要越容易变成一锅粥。这是社区摘要质量随规模下降的原因之一。
• 它几乎是免费的（算力上）。 Leiden 本身是 CPU 算法，跑一遍几十秒。GraphRAG 索引贵不贵，跟 Leiden 没关系，贵在抽取调用和社区摘要的 LLM 费用。这一点经常被搞混——有人说"社区检测太慢"，实际上慢的是 LLM。

3.4 LightRAG 的双层检索与五种查询模式

LightRAG 保留了图，扔掉了社区层次。它的图只有两层抽象：

• low level（实体层）：具体实体节点及其 1-hop 邻域。回答"张三负责哪个项目"这类问题。
• high level（主题层）：从关系中归纳出的主题/概念。回答"这个语料里反复出现的架构问题是什么"这类问题。

查询时 LLM 先从问题里抽出两类关键词：low-level keywords（具体实体名）和 high-level keywords（抽象主题）。两类关键词分别去图上锚定，取回子图后再从向量库里捞对应的原始 chunk，拼成带结构的上下文。

WildGraphBench 上 71.16% 那一行就是 hybrid 模式。我的默认也是 hybrid：local 单独用会漏掉跨主题的联系，global 单独用会丢掉具体事实，两个合并起来几乎没有短板。

3.5 查询侧的 61 万 token 从哪来

这个数字需要拆清楚，因为社区里传得已经变形了。

GraphRAG 的 global search 是 map-reduce：

map : 把 N_comm 个社区报告按批次喂给 LLM，每批产出 (答案片段, 有用性评分)
reduce: 取评分最高的若干片段，拼起来生成最终答案

token 消耗 ≈ N_comm × 每份报告的 token 数。论文里 Legal 数据集那次是610 个 level-2 社区，每份报告约 1000 token，所以 610 × 1000 ≈ 61 万。

这里有个必须纠正的点：61 万不是通用常数，它是那个特定语料规模下的值。 按我 4.2 节的模型，500 页的语料大约产生 80~110 个 level-2 社区，单次全局查询是 8~11 万 token，不是 61 万。把 61 万当成"GraphRAG 每次查询的成本"是错的，但数量级关系（图方法遍历全部社区 vs LightRAG 取有界邻域）是真实的，这个差距在任何规模下都存在。

LightRAG 那边论文给的是"<100 token + 1 次 API 调用"。也要说清楚：这 100 token 指的是检索阶段图查询本身的开销（构造查询、定位节点、取邻域），不含最终喂给生成模型的上下文。 别理解成整个查询只花 100 token。真实的差距在"检索开销"这一项上是三个数量级，加上生成上下文之后总成本的差距会收敛到几倍到几十倍——但仍然是一个方向的。

根本原因是遍历集合的形态不同：

"有界"这两个字是全部。 有界邻域的检索成本与语料总量无关，遍历全量社区摘要的成本与语料总量线性（或次线性）相关。语料翻十倍，前者的查询成本不变，后者翻几倍。

四、环境与依赖准备

python -m venv ~/venv/kg && source ~/venv/kg/bin/activate
pip install networkx==3.4 numpy==1.26.4 openai==1.68.2 tiktoken==0.9.0
pip install python-dotenv==1.0.1 tenacity==9.0.0 pydantic==2.10.6
# 生产图库三选一，见 8.3 节
pip install neo4j==5.28.1 # 路线 A：Neo4j
# 路线 B：NetworkX（已在上面）
# 路线 C：psycopg[binary]==3.2.3 # PostgreSQL + Apache AGE

python -c "import networkx, numpy, openai; print('ok')"

硬件和运维门槛：

一条硬性建议：抽取侧的 LLM 和回答侧的 LLM 分开选。 抽取是机械劳动，用便宜模型（甚至本地 14B）就行，占索引成本的 90% 以上；回答侧才需要好模型。把这两件事用同一个模型做，是最常见的浪费。

五、完整实操

5.1 成本模型：把 500 页逐级算成美元

先给参数表，后面所有数字都从这张表来：

社区规模随语料增长这件事，公开资料里没有统一公式。 我按 comm_size = 8 × (pages/100)^β 建模，β = 0 表示社区大小固定（社区数线性增长），β = 0.25 表示社区数按 N^0.75 增长，β = 0.4 表示 N^0.6。真实情况大概率在 β ∈ [0.2, 0.4] 之间，所以后面三个都算。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""graph_rag_cost.py —— GraphRAG / LightRAG 的索引与查询成本模型。
自己改参数，直接跑：python graph_rag_cost.py
输出三张表：索引成本随语料规模、社区数敏感性、查询成本对比。
"""
import math
from dataclasses import dataclass

@dataclass
class Prices:
name: str
in_per_mtok: float # 每百万 input token 的美元
out_per_mtok: float # 每百万 output token 的美元

CHEAP = Prices("2026 中端模型", 0.15, 0.60)
MID = Prices("2026 高端小模型", 0.30, 1.20)
FRONTIER = Prices("2026 前沿模型", 1.25, 10.00)
GPT4O_2024 = Prices("2024 GPT-4o 口径", 2.50, 10.00)
GPT4_2024 = Prices("2024 GPT-4 口径", 30.00, 60.00) # 社区流传的 50~200 美元出处

TOK_PER_PAGE = 667
CHUNK, OVERLAP = 1200, 100
ENT_PER_CHUNK, DUP_FACTOR, REL_PER_ENT = 8.0, 1.8, 1.6
PROMPT_TOK = 800 # 抽取 prompt 模板本身的开销
EXTRACT_OUT = 600 # 单次抽取的输出 token
GLEANS = 1.5 # GraphRAG 的 gleaning 循环带来的平均调用倍数
ENT_DESC_IN, ENT_DESC_OUT = 1200, 150
REL_DESC_IN, REL_DESC_OUT = 800, 100
REPORT_CTX, REPORT_OUT = 8000, 1000

def corpus_stats(pages):
tokens = pages * TOK_PER_PAGE
chunks = math.ceil(tokens / (CHUNK – OVERLAP))
entities = math.ceil(chunks * ENT_PER_CHUNK / DUP_FACTOR)
relations = math.ceil(entities * REL_PER_ENT)
return tokens, chunks, entities, relations

def community_count(pages, beta):
"""社区数 = 实体数 / 平均社区规模；社区规模按 (pages/100)^beta 增长。"""
_, _, entities, _ = corpus_stats(pages)
size = 8.0 * (max(pages, 10) / 100.0) ** beta
return max(1, math.ceil(entities / size)), entities, size

def graphrag_index(pages, beta=0.25):
_, chunks, entities, relations = corpus_stats(pages)
calls = chunks * GLEANS
tin = calls * (PROMPT_TOK + CHUNK)
tout = calls * EXTRACT_OUT
# 实体 / 关系描述摘要
tin += entities * ENT_DESC_IN + relations * REL_DESC_IN
tout += entities * ENT_DESC_OUT + relations * REL_DESC_OUT
# 社区报告（只算 level 2 这一层）
ncomm, _, _ = community_count(pages, beta)
tin += ncomm * REPORT_CTX
tout += ncomm * REPORT_OUT
return {"in": tin, "out": tout, "calls": calls + entities + relations + ncomm,
"communities": ncomm}

def lightrag_index(pages, beta=0.25):
_, chunks, entities, _ = corpus_stats(pages)
tin = chunks * (PROMPT_TOK + CHUNK)
tout = chunks * EXTRACT_OUT
# 没有社区层次；实体/关系描述靠拼接去重，不再单独调 LLM
return {"in": tin, "out": tout, "calls": chunks, "communities": 0}

def usd(tok_in, tok_out, p: Prices):
return tok_in / 1e6 * p.in_per_mtok + tok_out / 1e6 * p.out_per_mtok

def table_index():
print("=" * 96)
print("表 1：索引成本随语料规模（β=0.25，价格 = 2026 中端模型）")
print("=" * 96)
print(f"{'页数':>7}{'chunk':>8}{'实体':>8}{'社区':>7}"
f"{'GraphRAG 索引 Mtok':>20}{'LightRAG 索引 Mtok':>20}{'比值':>7}{'GraphRAG $':>12}{'LightRAG $':>12}")
for pages in [50, 100, 500, 2000, 10000]:
g = graphrag_index(pages)
l = lightrag_index(pages)
_, chunks, ents, _ = corpus_stats(pages)
gt = (g["in"] + g["out"]) / 1e6
lt = (l["in"] + l["out"]) / 1e6
print(f"{pages:>7}{chunks:>8}{ents:>8}{g['communities']:>7}"
f"{gt:>20.2f}{lt:>20.2f}{gt/lt:>7.1f}"
f"{usd(g['in'], g['out'], CHEAP):>12.2f}{usd(l['in'], l['out'], CHEAP):>12.2f}")

def table_sensitivity():
print("\\n" + "=" * 96)
print("表 2：社区数随语料增长的敏感性（β 越大 = 社区越大 = 社区数增长越慢）")
print("=" * 96)
print(f"{'页数':>8}{'实体':>9}" + "".join(f"{'β='+str(b):>16}" for b in [0.0, 0.25, 0.4]))
for pages in [100, 500, 2000, 10000]:
_, _, ents, _ = corpus_stats(pages)
row = f"{pages:>8}{ents:>9}"
for b in [0.0, 0.25, 0.4]:
n, _, sz = community_count(pages, b)
row += f"{n:>9}(size {sz:.0f})"
print(row)
print("\\n同一语料下 GraphRAG 索引成本对 β 的敏感性（2026 中端模型）：")
for pages in [500, 2000, 10000]:
row = f" {pages:>6} 页: "
for b in [0.0, 0.25, 0.4]:
g = graphrag_index(pages, b)
row += f"β={b}: ${usd(g['in'], g['out'], CHEAP):>7.2f} "
print(row)

def table_prices():
print("\\n" + "=" * 96)
print("表 3：同一套 500 页索引，在不同定价口径下的费用 —— 这就是 50~200 美元的出处")
print("=" * 96)
g = graphrag_index(500)
l = lightrag_index(500)
for p in [CHEAP, MID, FRONTIER, GPT4O_2024, GPT4_2024]:
print(f" {p.name:<20} GraphRAG ${usd(g['in'], g['out'], p):>8.2f} "
f"LightRAG ${usd(l['in'], l['out'], p):>8.2f}")

def table_query():
print("\\n" + "=" * 96)
print("表 4：单次查询成本（图检索开销，不含最终生成上下文）")
print("=" * 96)
print(f"{'页数':>7}{'社区数':>8}{'GraphRAG global Ktok':>22}"
f"{'LightRAG hybrid tok':>22}{'倍数':>9}")
for pages in [100, 500, 5000, 10000]:
n, _, _ = community_count(pages, 0.25)
gq = n * REPORT_OUT # 查询时读的是报告本身（约 REPORT_OUT token），不是生成它时的 8000 上下文
lq = 100 # 论文口径：检索阶段 <100 token + 1 次调用
print(f"{pages:>7}{n:>8}{gq/1000:>22.1f}{lq:>22}{gq/lq:>9.0f}x")
print("\\n 注：论文那次 610K token = 610 个 level-2 社区 × 每份报告约 1000 token。")
print(" 61 万不是通用常数；真实值 = 你的社区数 × 报告长度。")
print(" 用本模型反推，约 5000 页语料才会到 610 个社区、61 万 token 的量级。")

if __name__ == "__main__":
table_index()
table_sensitivity()
table_prices()
table_query()

跑一遍的关键输出我先贴在这里（2026 中端模型定价）：

同一套 500 页索引换到不同定价口径下，是这个样子：

先说最重要的一条：在 2026 年的模型定价下，500 页语料的 GraphRAG 索引只要 1.26 美元，不是 50~200 美元。 那组广为流传的 50~200，用同一套参数在 2024 GPT-4 定价下跑出来是 $204——略高于区间上沿（我的 gleaning 参数和描述摘要长度取得偏保守），但量级完全对得上。所以"GraphRAG 索引太贵"这个骂名，有一半是 2024 年定价留下的历史包袱。

但另一半是真问题，而且降不下来：

• 增量成本。 上面都是"一次全量索引"的账。语料每周变一次的话，GraphRAG 每周付一次 $1.26（500 页）或 $22.96（10000 页），LightRAG 只付新增文档那一小部分。一年 52 周，10000 页语料下这个差距是 $1194 对 $20 量级。
• 查询成本。 这是架构性的，改模型价格救不了。

第二条敏感性结论：β 对总成本的影响远小于语料规模的线性项。 500 页时 β 从 0 到 0.4，索引成本在 $1.22 ~ $1.36 之间浮动，只有 11%；10000 页时是 $22.00 ~ $27.11。原因是社区报告只占总 token 的 17% 左右，大头在实体/关系描述摘要上。想省 GraphRAG 的索引钱，该动的是描述摘要那一层和 gleaning 轮数，不是社区检测。

第三条，关于那个 61 万 token：按这个模型反推，要到 5000 页左右才会出现 610 个 level-2 社区、单查 61 万 token 的量级。 500 页的语料只有 114 个社区、单查 11.4 万 token。别把 61 万当成 GraphRAG 的通用常数，它是那个特定数据集规模下的值。

5.2 用 LLM 抽取实体关系的 ingestion 脚本

下面这份是我实际在用的 ingestion 脚本。三个设计点值得说：gleaning 循环（问模型"还有漏的吗"，能显著提召回，但也翻倍成本）、断点续跑（大语料跑到一半挂了不能从头再来）、置信度过滤（Adobe 那篇 KG-RAG 的做法，低于 0.6 的三元组直接丢，图噪声降得非常明显）。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""kg_ingest.py —— 用 LLM 从文档抽取实体与关系，产出图与向量索引。
设计点：gleaning 追问循环 / 断点续跑 / 置信度过滤 / 实体名归一化前置。

用法:
export OPENAI_API_KEY=sk-xxx
export OPENAI_BASE_URL=https://api.xxx.com/v1 # 可选，兼容任意 OpenAI 风格接口
python kg_ingest.py –input ./docs –out ./kg_store –model gpt-4o-mini \\
–chunk 1200 –overlap 100 –glean 1 –min-conf 0.6
"""
import argparse, hashlib, json, os, re, time
from pathlib import Path
from collections import defaultdict

from openai import OpenAI
from tenacity import retry, stop_after_attempt, wait_exponential

EXTRACT_SYS = """你是一个知识图谱抽取器。从给定文本中抽取实体与关系。
只输出 JSON，不要解释。格式：
{"entities":[{"name":"规范化名称","type":"Person|Org|Product|Event|Concept|Location",
"description":"一句话描述","confidence":0.0}],
"relations":[{"src":"实体A","dst":"实体B","description":"一句话描述关系",
"weight":1,"confidence":0.0}]}
规则：
1. 实体名用最简短的规范形式，不要带冠词和修饰语。
2. 关系描述要能独立理解，不要把代词带进去。
3. confidence 是你对这个三元组确实来自文本的把握，0.0~1.0。
4. 不要抽取你在文本里看不到的东西。"""

GLEAN_SYS = """上一轮已经抽取了一批实体关系。请只补充**遗漏**的，格式相同。
如果确实没有遗漏，输出 {"entities":[],"relations":[]}。"""

def chunk_text(text, size, overlap):
"""按 token 近似切分（中文按字符、英文按 4 字符/token 估）。"""
unit = 2 if sum('\\u4e00' <= c <= '\\u9fff' for c in text[:2000]) > 200 else 4
step, win = (size – overlap) * unit, size * unit
out, i = [], 0
while i < len(text):
out.append(text[i:i + win])
i += step
return out

def norm_name(s):
"""实体名归一化：消歧的第一道防线，越早做越省事。"""
s = s.strip().lower()
s = re.sub(r"[（）()\\[\\]【】\\"'“”‘’。，,、；;：:]", " ", s)
s = re.sub(r"\\b(the|a|an|inc|ltd|llc|corp|co|有限公司|股份有限公司|集团)\\b", " ", s)
s = re.sub(r"\\s+", " ", s).strip()
return s

@retry(stop=stop_after_attempt(4), wait=wait_exponential(multiplier=1, min=1, max=20))
def call_llm(client, model, sys_p, usr_p, max_tok=1600):
r = client.chat.completions.create(
model=model,
messages=[{"role": "system", "content": sys_p},
{"role": "user", "content": usr_p}],
temperature=0.0, max_tokens=max_tok,
response_format={"type": "json_object"},
)
return r.choices[0].message.content

def parse_json(raw):
try:
return json.loads(raw)
except Exception:
m = re.search(r"\\{.*\\}", raw, re.S)
return json.loads(m.group(0)) if m else {"entities": [], "relations": []}

def main():
ap = argparse.ArgumentParser()
ap.add_argument("–input", required=True, help="目录（递归找 .txt/.md）")
ap.add_argument("–out", required=True)
ap.add_argument("–model", default="gpt-4o-mini")
ap.add_argument("–chunk", type=int, default=1200)
ap.add_argument("–overlap", type=int, default=100)
ap.add_argument("–glean", type=int, default=1, help="追问轮数，0 关闭")
ap.add_argument("–min-conf", type=float, default=0.6)
ap.add_argument("–sleep", type=float, default=0.0)
args = ap.parse_args()

client = OpenAI(base_url=os.getenv("OPENAI_BASE_URL") or None)
outdir = Path(args.out); outdir.mkdir(parents=True, exist_ok=True)
cache_f = outdir / "chunks.jsonl"
graph_f = outdir / "graph.jsonl"
done = set()
if cache_f.exists(): # 断点续跑：已处理过的 chunk 跳过
done = {json.loads(l)["cid"] for l in cache_f.open(encoding="utf-8")}

files = sorted([p for p in Path(args.input).rglob("*")
if p.suffix.lower() in (".txt", ".md")])
print(f"发现 {len(files)} 个文件，已完成 {len(done)} 个 chunk")

ents, rels = {}, {}
fw_e = graph_f.open("a", encoding="utf-8")
nchunk = 0
for fp in files:
text = fp.read_text(encoding="utf-8", errors="ignore")
for idx, ck in enumerate(chunk_text(text, args.chunk, args.overlap)):
cid = hashlib.md5(f"{fp}|{idx}|{ck[:200]}".encode()).hexdigest()[:16]
if cid in done:
continue
raw = call_llm(client, args.model, EXTRACT_SYS, f"文本：\\n{ck}\\n\\n抽取结果：")
data = parse_json(raw)
for _ in range(args.glean): # gleaning 追问
more = parse_json(call_llm(
client, args.model, GLEAN_SYS,
f"原文：\\n{ck}\\n\\n已抽取：\\n{raw}\\n\\n补充（只给 JSON）："))
data["entities"] += more.get("entities", [])
data["relations"] += more.get("relations", [])
kept_e = [e for e in data.get("entities", []) if e.get("confidence", 1) >= args.min_conf]
kept_r = [r for r in data.get("relations", []) if r.get("confidence", 1) >= args.min_conf]
fw_e.write(json.dumps({"cid": cid, "file": str(fp), "text": ck,
"entities": kept_e, "relations": kept_r},
ensure_ascii=False) + "\\n")
fw_e.flush()
with cache_f.open("a", encoding="utf-8") as f2:
f2.write(json.dumps({"cid": cid}, ensure_ascii=False) + "\\n")
nchunk += 1
if nchunk % 20 == 0:
print(f" …{nchunk} chunks")
if args.sleep:
time.sleep(args.sleep)
fw_e.close()

# 汇总：同名实体合并描述，边按 (src,dst) 聚合权重
for line in graph_f.open(encoding="utf-8"):
d = json.loads(line)
for e in d["entities"]:
k = norm_name(e["name"])
if not k:
continue
n = ents.setdefault(k, {"name": e["name"], "type": e.get("type", "Concept"),
"descriptions": [], "chunks": set()})
n["descriptions"].append(e.get("description", ""))
n["chunks"].add(d["cid"])
for r in d["relations"]:
s, t = norm_name(r.get("src", "")), norm_name(r.get("dst", ""))
if not s or not t or s == t:
continue
key = (s, t)
e = rels.setdefault(key, {"src": s, "dst": t, "descriptions": [],
"weight": 0, "chunks": set()})
e["descriptions"].append(r.get("description", ""))
e["weight"] += float(r.get("weight", 1))
e["chunks"].add(d["cid"])

graph = {
"nodes": [{"id": k, "name": v["name"], "type": v["type"],
"description": " / ".join(dict.fromkeys([d for d in v["descriptions"] if d]))[:800],
"chunks": sorted(v["chunks"])}
for k, v in ents.items()],
"edges": [{"src": v["src"], "dst": v["dst"], "weight": v["weight"],
"description": " / ".join(dict.fromkeys([d for d in v["descriptions"] if d]))[:400],
"chunks": sorted(v["chunks"])}
for v in rels.values()],
}
(outdir / "graph.json").write_text(json.dumps(graph, ensure_ascii=False, indent=2),
encoding="utf-8")
print(f"\\n完成：{nchunk} 个新 chunk，图 = {len(graph['nodes'])} 节点 / "
f"{len(graph['edges'])} 边 -> {outdir/'graph.json'}")

if __name__ == "__main__":
main()

5.3 双层检索的查询脚本

这份脚本实现了 LightRAG 的五种模式，图用 NetworkX 加载，向量库用一个极简的 numpy 实现（真实项目里换成你自己的向量库即可）。重点看 dual_level_retrieve 里 low-level 与 high-level 两套锚定是怎么合并的。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""kg_query.py —— LightRAG 风格的双层检索 + 五种查询模式。
依赖上一节产出的 graph.json 与 chunks.jsonl。

用法:
python kg_query.py –store ./kg_store –mode hybrid –q "张三负责的项目用了哪些自研中间件"
"""
import argparse, json, math
from pathlib import Path
import numpy as np
import networkx as nx
from openai import OpenAI

KW_SYS = """从用户问题里抽两类关键词，只输出 JSON：
{"low": ["具体实体名1","具体实体名2"], "high": ["抽象主题1","抽象主题2"]}
low 是可以精确匹配到图上节点的专有名词；high 是无法精确匹配的概念或主题。"""

ANSWER_SYS = """你是一个严谨的问答助手。只能使用下面提供的上下文回答。
如果上下文里没有答案，明确说"资料里没有提到"，不要猜。
每个关键结论后面用 [chunk_id] 标注来源。"""

def build_nx(graph):
G = nx.Graph()
for n in graph["nodes"]:
G.add_node(n["id"], name=n["name"], type=n["type"],
desc=n["description"], chunks=n.get("chunks", []))
for e in graph["edges"]:
if e["src"] in G and e["dst"] in G:
G.add_edge(e["src"], e["dst"], weight=e["weight"],
desc=e["description"], chunks=e.get("chunks", []))
return G

def embed(client, texts, model="text-embedding-3-small"):
r = client.embeddings.create(model=model, input=texts)
return np.array([d.embedding for d in r.data], dtype=np.float32)

class VectorIndex:
"""极简 numpy 向量库，真实项目请换成自己的向量库。"""

def __init__(self, ids, mat):
self.ids = ids
self.mat = mat / (np.linalg.norm(mat, axis=1, keepdims=True) + 1e-9)

def search(self, q, topk=5):
qv = q / (np.linalg.norm(q) + 1e-9)
sim = self.mat @ qv
idx = np.argsort(–sim)[:topk]
return [(self.ids[i], float(sim[i])) for i in idx]

def extract_keywords(client, model, q):
r = client.chat.completions.create(
model=model, temperature=0.0,
messages=[{"role": "system", "content": KW_SYS},
{"role": "user", "content": q}],
response_format={"type": "json_object"})
return json.loads(r.choices[0].message.content)

def match_entity(G, vi_nodes, name):
"""先从图里精确/子串匹配，失败再退到向量相似度。"""
key = name.strip().lower()
for nid in G.nodes:
if key == nid or key in str(G.nodes[nid].get("name", "")).lower():
return nid, 1.0
hit = vi_nodes.search(embed(client, [name])[0], topk=1) if vi_nodes else None
return (hit[0][0], hit[0][1]) if hit and hit[0][1] > 0.75 else (None, 0.0)

def dual_level_retrieve(client, model, G, vi_nodes, vi_chunks, kw, mode,
hops=1, topk_chunk=6, max_ctx=6000):
"""双层检索：low-level 锚定实体邻域，high-level 锚定主题边。"""
ctx, budget = [], max_ctx

def take(text, tag):
nonlocal budget
if budget <= 0 or not text:
return
t = text[:1500]
ctx.append(f"[{tag}] {t}")
budget -= len(t)

anchors = []
if mode in ("local", "hybrid", "mix"):
for name in kw.get("low", []):
nid, sc = match_entity(G, vi_nodes, name)
if nid:
anchors.append(nid)
nb = nx.single_source_shortest_path_length(G, nid, cutoff=hops)
for m in nb:
d = G.nodes[m]
take(f"{d.get('name')} ({d.get('type')}): {d.get('desc')}", f"entity:{m}")
for u, v, ed in G.edges(nid, data=True):
take(f"{G.nodes[u].get('name')} –[{ed.get('desc')}]–> {G.nodes[v].get('name')}",
f"rel:{u}->{v}")
if mode in ("global", "hybrid", "mix"):
for topic in kw.get("high", []):
qv = embed(client, [topic])[0]
scored = []
for u, v, ed in G.edges(data=True):
if ed.get("desc"):
scored.append(((u, v, ed), 0.0))
# 主题关键词落不到边上时，退到"高频边的描述向量"近似
if scored and vi_chunks is not None:
descs = [e[2].get("desc", "") for e, _ in scored[:400]]
sims = embed(client, descs) @ (qv / (np.linalg.norm(qv) + 1e-9))
order = np.argsort(–sims)[:8]
for i in order:
u, v, ed = scored[i][0]
take(f"{G.nodes[u].get('name')} –[{ed.get('desc')}]–> {G.nodes[v].get('name')}",
f"theme:{u}->{v}")
if mode in ("naive", "mix") and vi_chunks is not None:
qv = embed(client, [kw.get("raw", "") or " ".join(kw.get("low", []))])[0]
for cid, sc in vi_chunks.search(qv, topk=topk_chunk):
take(f"{cid}", f"chunk:{cid}")
# 兜底：锚定实体对应的原始 chunk
seen = set()
for a in anchors:
for cid in G.nodes[a].get("chunks", [])[:3]:
if cid not in seen:
seen.add(cid)
take(cid, f"chunk:{cid}")
return "\\n".join(ctx), anchors

def main():
ap = argparse.ArgumentParser()
ap.add_argument("–store", required=True)
ap.add_argument("–q", required=True)
ap.add_argument("–mode", default="hybrid",
choices=["naive", "local", "global", "hybrid", "mix"])
ap.add_argument("–model", default="gpt-4o-mini")
ap.add_argument("–embed-model", default="text-embedding-3-small")
args = ap.parse_args()

store = Path(args.store)
client = OpenAI(base_url=None)
graph = json.loads((store / "graph.json").read_text(encoding="utf-8"))
G = build_nx(graph)
print(f"图：{G.number_of_nodes()} 节点 / {G.number_of_edges()} 边")

texts, ids = [], []
for line in (store / "chunks.jsonl").open(encoding="utf-8"):
d = json.loads(line)
ids.append(d["cid"]); texts.append(d["text"])
vi_chunks = VectorIndex(ids, embed(client, texts, args.embed_model)) if texts else None

node_ids = list(G.nodes)
node_txt = [f"{G.nodes[n].get('name')} {G.nodes[n].get('desc','')}" for n in node_ids]
vi_nodes = VectorIndex(node_ids, embed(client, node_txt, args.embed_model)) if node_ids else None

kw = extract_keywords(client, args.model, args.q)
kw["raw"] = args.q
print("关键词:", kw)

ctx, anchors = dual_level_retrieve(client, args.embed_model, G, vi_nodes,
vi_chunks, kw, args.mode)
print(f"\\n上下文 {len(ctx)} 字符，锚定实体 {anchors}\\n" + "-" * 60)
print(ctx[:2000])

r = client.chat.completions.create(
model=args.model, temperature=0.2,
messages=[{"role": "system", "content": ANSWER_SYS},
{"role": "user", "content": f"问题：{args.q}\\n\\n上下文：\\n{ctx}"}])
print("\\n" + "=" * 60 + "\\n回答：\\n" + r.choices[0].message.content)

if __name__ == "__main__":
main()

5.4 增量并集合并的最小实现

这是 LightRAG 相对 GraphRAG 的核心优势所在，也是全文最该自己写一遍的部分。算法本身非常朴素：

G = (V, E) 已有图
G' = (V', E') 新文档抽出的图
1. 对 V' 每个实体 v'，在 V 中解析出 canonical id（命中则合并，否则新增）
2. 合并节点：描述去重拼接、chunks 并集、出现次数累加
3. 对 E' 每条边：两端解析到 canonical id，边已存在则权重累加 + 描述去重，否则新增
4. 新 chunk 追加进向量库
5. 不做任何全局重算

复杂度是 O(|V’| + |E’|)，与已有语料规模无关。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""kg_incremental.py —— 最小增量并集合并：新图并入旧图，不触发任何全局重算。
含实体解析（归一化 -> 字符串相似 -> embedding 相似 -> 可选 LLM 判定）与图退化检测。

用法:
python kg_incremental.py –base ./kg_store/graph.json \\
–delta ./kg_store/graph_new.json –out ./kg_store/graph_merged.json –check
"""
import argparse, json, math, re
from pathlib import Path
from collections import defaultdict
from difflib import SequenceMatcher

import networkx as nx

CJK = re.compile(r"[\\u4e00-\\u9fff]")

def norm_name(s):
s = str(s).strip().lower()
s = re.sub(r"[（）()\\[\\]【】\\"'“”‘’。，,、；;：:]", " ", s)
s = re.sub(r"\\b(the|a|an|inc|ltd|llc|corp|co|有限公司|股份有限公司|集团)\\b", " ", s)
s = re.sub(r"\\s+", " ", s).strip()
# 中文没有词边界，实体内部的空格几乎全是噪声（"消息队列 MQ" 与 "消息队列MQ"）
if CJK.search(s):
s = re.sub(r"\\s+", "", s)
return s

def str_sim(a, b):
return SequenceMatcher(None, a, b).ratio()

def bigrams(s, n=2):
return {s[i:i + n] for i in range(len(s) – n + 1)} or {s}

def jaccard(a, b):
"""字符 bigram Jaccard：中英文都适用，词级 Jaccard 在中文上会直接归零。"""
A, B = bigrams(a), bigrams(b)
return len(A & B) / max(len(A | B), 1)

class Resolver:
"""四级实体解析：精确 -> 字符串 -> 类型一致 + 上下文 -> 阈值区间外的拒绝。"""

def __init__(self, high=0.90, low=0.72):
self.high, self.low = high, low
self.canon = {} # normalized -> canonical id

def add(self, nid, display=None):
"""id 与显示名都注册：边里可能引用 id，节点表里存的是显示名。"""
self.canon.setdefault(norm_name(nid), nid)
if display:
self.canon.setdefault(norm_name(display), nid)

def resolve(self, name, ntype=None, ctx_words=None):
key = norm_name(name)
if not key:
return None, 0.0
if key in self.canon:
return self.canon[key], 1.0
best, bs = None, 0.0
for k, cid in self.canon.items():
s = 0.6 * str_sim(key, k) + 0.4 * jaccard(key, k)
if s > bs:
best, bs = cid, s
if best is not None and bs >= self.high:
return best, bs # 高分：自动合并
if best is not None and bs >= self.low:
return best, bs # 中间区：交给人工或 LLM 判定
return None, bs # 低分：当作新实体

def merge_descriptions(old, new, limit=800):
parts = [p.strip() for p in (old or "").split("/") if p.strip()]
parts += [p.strip() for p in (new or "").split("/") if p.strip()]
seen, out = set(), []
for p in parts:
if p not in seen:
seen.add(p); out.append(p)
return " / ".join(out)[:limit]

def union_merge(base, delta, resolver, ambiguous="new"):
"""并集合并。ambiguous: 'new'（中间区当新实体）或 'merge'（合并）。"""
stats = {"new_nodes": 0, "merged_nodes": 0, "new_edges": 0,
"merged_edges": 0, "ambiguous": 0, "dropped_edges": 0}
nodes = {n["id"]: dict(n) for n in base["nodes"]}
for n in nodes.values():
resolver.add(n["id"], n.get("name"))

for n in delta["nodes"]:
cid, sc = resolver.resolve(n["name"], n.get("type"))
if cid is None:
cid = norm_name(n["name"]) or n["id"]
nodes[cid] = {"id": cid, "name": n["name"], "type": n.get("type", "Concept"),
"description": n.get("description", ""),
"chunks": list(n.get("chunks", []))}
resolver.add(cid, n["name"])
stats["new_nodes"] += 1
else:
if sc < resolver.high:
stats["ambiguous"] += 1
if ambiguous == "new":
cid = cid + "__dup" + str(stats["ambiguous"])
nodes[cid] = {"id": cid, "name": n["name"], "type": n.get("type", "Concept"),
"description": n.get("description", ""),
"chunks": list(n.get("chunks", []))}
resolver.add(cid, n["name"])
stats["new_nodes"] += 1
continue
cur = nodes[cid]
cur["description"] = merge_descriptions(cur.get("description"), n.get("description"))
cs = set(cur.get("chunks", [])) | set(n.get("chunks", []))
cur["chunks"] = sorted(cs)
stats["merged_nodes"] += 1

edges = {}
def ekey(u, v):
return tuple(sorted((u, v)))
for e in base["edges"]:
edges[ekey(e["src"], e["dst"])] = dict(e)
for e in delta["edges"]:
su, _ = resolver.resolve(e["src"])
du, _ = resolver.resolve(e["dst"])
su, du = su or norm_name(e["src"]), du or norm_name(e["dst"])
# 端点不在节点表里 -> 悬空边，丢弃而不是造一个空节点（空节点会污染孤立率指标）
if not su or not du or su == du or su not in nodes or du not in nodes:
stats["dropped_edges"] += 1
continue
k = ekey(su, du)
if k in edges:
edges[k]["weight"] = float(edges[k].get("weight", 1)) + float(e.get("weight", 1))
edges[k]["description"] = merge_descriptions(edges[k].get("description"),
e.get("description"), limit=400)
cs = set(edges[k].get("chunks", [])) | set(e.get("chunks", []))
edges[k]["chunks"] = sorted(cs)
stats["merged_edges"] += 1
else:
edges[k] = {"src": k[0], "dst": k[1], "weight": float(e.get("weight", 1)),
"description": e.get("description", ""),
"chunks": list(e.get("chunks", []))}
stats["new_edges"] += 1
return {"nodes": list(nodes.values()), "edges": list(edges.values())}, stats

def health_check(graph):
"""图退化检测：巨型节点、碎片化、稀疏度。"""
G = nx.Graph()
for n in graph["nodes"]:
G.add_node(n["id"])
for e in graph["edges"]:
if e["src"] in G and e["dst"] in G:
G.add_edge(e["src"], e["dst"], weight=e.get("weight", 1))
if G.number_of_nodes() == 0:
return {}
degs = [d for _, d in G.degree()]
comps = sorted((len(c) for c in nx.connected_components(G)), reverse=True)
iso = sum(1 for _, d in G.degree() if d == 0)
return {
"nodes": G.number_of_nodes(),
"edges": G.number_of_edges(),
"avg_degree": round(sum(degs) / len(degs), 3),
"max_degree": max(degs),
"hub_ratio": round(max(degs) / max(sum(degs) / len(degs), 1e-9), 1),
"isolated_nodes": iso,
"isolated_ratio": round(iso / len(degs), 4),
"largest_component_ratio": round(comps[0] / len(degs), 4) if comps else 0,
"num_components": len(comps),
"relation_per_node": round(G.number_of_edges() / G.number_of_nodes(), 3),
}

def main():
ap = argparse.ArgumentParser()
ap.add_argument("–base", required=True)
ap.add_argument("–delta", required=True)
ap.add_argument("–out", required=True)
ap.add_argument("–high", type=float, default=0.90)
ap.add_argument("–low", type=float, default=0.72)
ap.add_argument("–ambiguous", default="new", choices=["new", "merge"])
ap.add_argument("–check", action="store_true")
args = ap.parse_args()

base = json.loads(Path(args.base).read_text(encoding="utf-8"))
delta = json.loads(Path(args.delta).read_text(encoding="utf-8"))
merged, stats = union_merge(base, delta, Resolver(args.high, args.low), args.ambiguous)
Path(args.out).write_text(json.dumps(merged, ensure_ascii=False, indent=2), encoding="utf-8")
print("合并统计:", json.dumps(stats, ensure_ascii=False))
print(f"合并后: {len(merged['nodes'])} 节点 / {len(merged['edges'])} 边 -> {args.out}")

if args.check:
for tag, g in (("base", base), ("delta", delta), ("merged", merged)):
h = health_check(g)
print(f"\\n[{tag}] " + json.dumps(h, ensure_ascii=False))
h = health_check(merged)
warns = []
if h.get("hub_ratio", 0) > 100:
warns.append("存在巨型节点（hub），检索时会撑爆上下文，需要在查询侧做邻域裁剪")
if h.get("isolated_ratio", 0) > 0.2:
warns.append("孤立节点超过 20%，大概率是实体消歧失败或抽取噪声")
if h.get("largest_component_ratio", 1) < 0.5:
warns.append("图碎片化严重，多跳路径会大量断裂")
if h.get("relation_per_node", 0) < 1.5:
warns.append("边节点比过低，图太稀疏，多跳检索没有可走的路径")
print("\\n".join("[WARN] " + w for w in warns) or "\\n图健康度检查通过")

if __name__ == "__main__":
main()

这份实现刻意保留了两个"不做"：不做任何全局重算，不修订任何已有边。 前一个是优点（O(新文档)），后一个是硬限制——如果一个事实会过期（“现任 CEO 是谁”），并集合并会让新旧两条边同时存在，图里出现自相矛盾。这种场景要换 Graphiti 那一类带失效区间的 mutate 模式。

实现里有三处是踩过坑才加上的，别删：

• norm_name 里对含中文的字符串去掉所有空格。 中文没有词边界，“消息队列 MQ” 和 “消息队列MQ” 是同一个实体，但按英文规则归一化会得到两个 key。这一条能让中文语料的重复率直接降几个点。
• 相似度用字符 bigram 的 Jaccard，不用词级 Jaccard。 词级 Jaccard 在两个中文串上几乎必然返回 0（因为整串是一个"词"），会让你以为相似度算法坏了。
• 悬空边直接丢弃并计数，不补空节点。 端点不在节点表里时补一个空节点会让孤立率指标失真，反而掩盖问题。dropped_edges 这个计数要盯着看，它突然变大通常意味着消歧阈值被改坏了。

5.5 实体消歧：真正的质量瓶颈

抽取模型的选择只贡献十几个点的差距，而消歧做砸了能让整个图退化。有一组公开的消融数据很能说明问题（CORE-KG，法律文书领域，LLaMA 3.3 70B）：

共指消解主要压重复，结构化提示主要压噪声，两者正交，缺一个就废一半。 而 R/N 比从 1.67 提到 2.88，意味着每个节点能走的路更多，多跳检索才真的有路可走。

我的消歧流水线是四级，越往后越贵：

两个关键工程点：

• 必须做 blocking（分块）。 朴素两两比较是 O(n²)，5 万节点就是 12.5 亿次比较，跑不完。按实体类型 + 名称首字母 + embedding 聚类先分块，把候选降到几十个。
• 阈值要按业务定，不是拍脑袋。 合并错（false positive）的代价和漏合并（false negative）的代价不一样：金融风控场景误合并两个公司是灾难，应该把阈值调到 0.95；客服知识库场景漏合并只是召回差一点，0.80 就够。Adobe 的 KG-RAG 那篇就是用 confidence ≥ 0.6 过滤三元组，把不相关回答降了 51.9%、完全相关回答提了 88.2%。

失败会怎样： 同一实体裂成多个节点 → 图变稀疏 → 三跳路径在第二跳断掉 → 检索召回不到 → 表现为"图检索还不如向量"。同时过度合并会产生巨型节点（hub），一个节点挂几千条边，查询时 1-hop 邻域就撑爆上下文。这两个失败方向是相反的，所以阈值不能两头都想要。

六、结果对比与数据分析

把我这边 500 页内部语料（工程文档 + 会议纪要混合，中英混杂）的实测贴出来。抽取用中端 API 模型，回答用同一个模型，embedding 用 bge-m3 级别的模型。

（延迟是我这边单机实测，不同硬件差异很大；准确率是内部 300 题评测集，多跳题 90 道、全局题 60 道、单事实题 150 道。）

四条结论：

还有一组我特意做的对比：语料每周新增 5% 的情况下，半年（26 周，语料从 500 页涨到约 1150 页）、日均 200 次查询、按混合价 $0.30/1M token 计的累计成本：

查询成本那一列的差距才是决定性的：global 与 LightRAG 差 $787，而索引只差 $51。 日均 200 次这个量级下，索引成本根本不是决策因素。

七、踩坑记录与排查手册

坑 1：新文档持续进来之后，全局问题的答案质量悄悄变差

现象：上线第一个月全局归纳题准确率 74%，第三个月掉到 51%，模型没换、提示没换。 定位：统计社区数，从 75 涨到 130，但每份社区报告的 token 数从 900 掉到 520。 原因：我们当初为了省事，新文档进来后只做图的并集，没有重跑 Leiden。社区边界早就不对了，而社区报告是基于旧划分生成的——新实体被塞进了语义不匹配的旧社区，报告被稀释。 解决：要么接受"全局问题会退化"（改用 LightRAG hybrid，全局题从 74% 掉到 68%，但稳定）；要么定一个重建周期（我们后来改成月度全量重建 + 周度增量补丁）。没有第三种：Leiden 是全局算法，局部更新在数学上不成立。

坑 2：同一个公司裂成了 40 多个节点

现象：图里有 3 万个节点，但人工看明显重复极多，“腾讯云"出现了"腾讯云”“Tencent Cloud”“腾讯云 Tencent Cloud”"腾讯云计算"等一堆。 定位：跑 health_check，孤立节点比例 34%，R/N 比只有 0.9。 原因：只做了大小写归一化（L1），中文别名、中英混写、组织后缀完全没处理。抽取模型每次写的规范形式都不一样。 解决：补 L2/L3 两级消歧 + 一张人工维护的同义词表。R/N 从 0.9 提到 2.4，多跳准确率涨 5.4 个点。教训：消歧不是可选优化，是能不能用的分水岭。

坑 3：查询时上下文被一个巨型节点撑爆

现象：某次查询返回了 12 万字符的上下文，模型直接报超长。 定位：health_check 报 hub_ratio = 430，最大度数是平均度数的 430 倍。 原因：“公司”（一个泛概念实体）被抽成了节点，几乎所有实体都和它有关系，1-hop 邻域就是整个图。 解决：两条一起上。一是抽取时把过于泛化的概念（公司、系统、方案、问题）加进停用词表；二是查询侧做邻域裁剪——按边权重排序只取 top-20 条边，且对 hub 节点单独设更小的截断。工程上必须假设图中一定会有 hub，不能指望抽取侧永远干净。

坑 4：图碎片化严重，多跳根本走不通

现象：多跳问题召回率极低，甚至不如向量。 定位：largest_component_ratio 只有 0.23，图被切成了几千个小连通块。 原因：两个叠加。一是 chunk 边界把跨段的关系切断了（实体 A 在 chunk 3、实体 B 在 chunk 4，抽取时看不到彼此）；二是过度严格的消歧阈值让本该合并的实体没合并。 解决：把 chunk overlap 从 100 提到 300，并在抽取 prompt 里显式要求"如果提到某个实体但描述不完整，仍然抽出实体，描述留空"——空描述的实体靠共指合并补齐。overlap 提到 25% 之后，最大连通块占比从 0.23 回到 0.61。

坑 5：关系谓词碎片化，图看起来很密但没法查

现象：边很多，但查"雇佣关系"要写十几个 OR 条件。 定位：统计边的 description 里出现的谓词，IS_EMPLOYED_BY / WORKS_AT / STAFF_OF / EMPLOYEE_OF 有 40 多种写法。 原因：开放关系抽取没有约束谓词表，LLM 每次自由发挥。 解决：在抽取 prompt 里给一个封闭的谓词白名单（我们用了 22 个），要求模型只能从里面选，再单独留一个 other 字段放描述。这个改动让边的可用率从约 60% 提到 90% 以上，而且 Cypher 查询从十几个 OR 变成一次索引查找。

坑 6：增量合并之后，旧描述永远是旧的

现象：文档更新了"X 项目已下线"，但图里还同时存在"X 项目进行中"和"X 项目已下线"两条描述。 定位：翻代码，union_merge 里只做描述拼接，从不删除。 原因：这是并集合并的定义——它只增不改。我们当初选 LightRAG 就是冲着增量来的，但需求里其实包含"事实会变"。 解决：对确实需要"事实过期"的那一小部分实体（项目状态、负责人、版本号），单独建一张关系表用 SQL 存，图里只存稳定的结构关系。图和会变的事实要分开存，硬塞进图里一定会出问题。

坑 7：NetworkX 的 GraphML 在 10 万节点时加载要 3 分钟

现象：服务重启一次要等 3 分钟，健康检查一直失败。 定位：profile 下来时间全在 nx.read_graphml，这是纯 Python 的 XML 解析。 原因：GraphML 是 XML，NetworkX 的实现是逐节点解析，10 万节点就是 10 万次 XML 元素构造。 解决：开发期继续用 GraphML（人能看、Gephi 能开），生产改成启动时从 JSON/Parquet 构建 DiGraph，或者直接用 Neo4j。加载时间从 180 秒降到 4 秒。GraphML 是交换格式不是存储格式，别拿它当生产存储。

坑 8：图检索和向量检索的结果不一致，用户觉得系统在自相矛盾

现象：同一个问题，走图检索给的答案是 A，走向量给的答案是 B，而且都有引用来源。 定位：两个索引的语料版本不一致——图是上周建的，向量库昨天更新过。 原因：我们把图和向量当成两个独立的索引在维护，更新流程没有事务。 解决：所有 chunk 用同一个内容哈希做主键，图和向量库在同一次 ingestion 事务里更新，落一份 manifest 记录两者的版本。** indexing 的原子性比算法重要。**

八、优化方案与进阶打法

8.1 成本侧的四个杠杆

按杠杆从大到小排：

8.2 质量侧的三个杠杆

8.3 图存储选型

我的默认建议：如果你已经在用 PostgreSQL，直接上 AGE（配 pgvector），不要为一个图再引入一套数据库。 图查询只是你查询形态里的一部分，大部分时候还是要和关系表 JOIN。AGE 支持到 PostgreSQL 18，openCypher 的常用语法都覆盖。

顺便提一句有争议的观点：图灵奖得主 Stonebraker 在 2026 年的一次访谈里说，图完全可以建模成"一张边表加一张节点表"，基于关系表执行查询的性能往往优于原生图数据库，他没看到图数据库真正不可替代的场景。这个观点偏激，但有一半是对的：如果你的图查询主要是 1-hop 邻域和 2-hop 路径，用 SQL 递归 CTE 完全够，而且能白拿事务和 JOIN。真正需要原生图引擎的场景是可变长度路径遍历和图算法（PageRank、社区检测）跑在全图上——这恰好是 GraphRAG 索引期才做的事，查询期反而很少。

8.4 图退化检测要常态化

把 health_check 接进 CI，每次 ingestion 之后跑一次，指标超过阈值就告警。我用的阈值：

最后两项是过程指标，比结构指标更早发现问题——节点数突然暴涨基本都是消歧阈值被改坏了。

九、适用场景与选型建议

9.1 选型决策树

语料会不会变？
├─ 基本不变（归档、法规、上季度工单）
│ └─ 需要全局归纳吗？
│ ├─ 需要 → GraphRAG（global + local 双路径）
│ └─ 不需要 → 向量 + BM25 + rerank，别上图
└─ 持续在变（产品目录、PR、工单）
├─ 事实会过期吗？
│ ├─ 会 → Graphiti 那一类 mutate 模式（带失效区间）
│ └─ 不会 → LightRAG（hybrid 默认）
└─ 但如果你要亚秒级 → 回到向量 + BM25，图只做离线分析

9.2 什么情况下不该用本方案

这一节比前面所有内容都重要。以下几种情况我明确不推荐上图：

9.3 渐进式路线

别跳步。生产上有效的顺序是：

1. 基线：语义切分 + 像样的 embedding 模型
2. 混合搜索：BM25 与向量并行召回再融合 → +10%~30%，延迟几乎不变
3. 重排：对前 50 个候选做交叉编码取前 5 → +，多花 100~200 ms
4. 如果多跳/全局问题仍然是主要痛点，再上图

前三步做完，大部分业务的多跳问题会得到明显改善（因为很多所谓的"多跳"其实是单跳召回不准）。只有确认剩下的失败案例确实是 I_cross 缺失（答案分散在多个 chunk、需要沿关系走）时，图的投入才有回报。

十、总结与展望

这篇想留下三件事。

第一，向量 RAG 在多跳和全局归纳上的失败是索引阶段的信息丢失，不是检索阶段的调参问题。 chunk 独立编码这个动作把跨块的互信息 I_cross 直接置零，后面加多少 reranker 都补不回来。判断要不要上图的标准很简单：你的问题形态里 I_cross 到底占多少。

第二，成本账要算在查询侧，不是索引侧。 2026 年的模型定价下，500 页语料的 GraphRAG 索引不到 1 美元，社区流传的 50~200 美元是 2024 年 GPT-4 定价的历史包袱。但查询侧"遍历全量社区摘要 vs 取有界邻域"的差距是架构性的，日均 200 次查询半年能差出一万多美元。而"语料会不会变"决定了你要付几次索引的钱——Leiden 是全局算法，模块化度这个目标函数的定义决定了新文档进来只能重建，这不是实现偷懒。

第三，实体消歧才是图质量的瓶颈，而且它是工程问题不是模型问题。 我这次项目里，40 条同义词规则加阈值调优换来了 5.4 个点，比换方案（4.7 个点）和换模型都划算。图退化检测要常态化接进 CI，hub_ratio 和 R/N 比这两个指标能提前发现大部分问题。

往后看，我判断有两个方向值得盯：一个是"图作为持久上下文层"这个形态正在成为共识（OpenAI 2026-09 的 V7 案例里，agent 先查预建的 Context Graph，查不到才回落到 RAG），图从"检索增强"变成"知识层"；另一个是成本结构的进一步分化——抽取会越来越便宜（本地小模型足够），而"事实会过期"这个需求会推着 mutate 模式的框架往前走。并集合并解决不了事实过期，这是 LightRAG 这类方案的硬边界。

参考素材与延伸阅读

• GraphRAG: The End-to-End Guide to Reducing Hallucination — WildGraphBench 全套对比数字与 500 页语料成本
• LightRAG vs Microsoft GraphRAG — 2026-08-26 核验，Leiden 全局性与增量并集的对比
• GraphRAG vs LightRAG vs Graphiti vs Cognee: choose by write pattern — 按 build / append / mutate 三种写入模式分类
• Agent Retrieval: Knowledge Graphs — 单次查询 610K token vs 数百 token 的差距来源，以及不该上图的场景
• 企业知识库问答，别急着上 GraphRAG：三条路线的成本账 — 延迟 20~24 s vs 20~30 ms，渐进式路线
• LightRAG 解剖：向量检索丢掉的上下文关系，图结构怎么捡回来 — 论文 4.5 节成本表与双层检索细节
• Graph RAG / Agentic RAG 选型 — CSDN，2026 年图检索落地选型
• Inside CORE-KG: Evaluating Structured Prompting and Coreference Resolution — 共指消解与结构化提示的消融数据
• From Documents to Dialogue: Building KG-RAG Enhanced AI Assistants — Adobe 与 CMU，置信度过滤与种子概念消歧
• GraphRAG Meets TypeSafe Jev — 实体消歧、谓词标准化、边权重、子图裁剪四类模式
• Apache AGE 仓库 — PostgreSQL 图扩展，2026-08-26，支持 PG 11~18
• 别太高估 AI 了：图灵奖得主 Stonebraker 谈 LLM、智能体与数据库的未来 — 图数据库争议观点
• OpenAI Demotes RAG: Context Graphs Become Primary — 2026-09，HERB benchmark 与 Context Graph 架构

关键词：GraphRAG、LightRAG、知识图谱、Leiden 社区检测、实体消歧、增量更新、RAG 成本模型、双层检索、Neo4j、Apache AGE