GraphRAG增强：图谱驱动RAG

可视化演示

图谱核心概念

想象你是一位刚上任的城市导航员，办公桌上堆满写着地标名字的卡片——“图书馆开放时间”“咖啡店招牌饮品”“学校历史沿革”。游客问：“离图书馆最近的咖啡店在哪？”你翻遍卡片，能背出咖啡店营业时间，却算不出它和图书馆的距离。

传统向量RAG就是这位只会翻卡片的导航员。它把文档切成块、用Embedding向量表示语义，检索时找“文字最相似的段落”，却看不见段落间“谁认识谁、谁属于谁”的关联。一旦问题涉及多跳关系——“A公司的供应商的竞争对手是谁”——它只能在卡片堆里大海捞针。

GraphRAG换了一种思路：在翻卡片前，先请测绘队把整座城市画成地图。这张地图就是知识图谱，数学形式为：

$$\displaystyle G = (V, E)$$

其中V是节点集合，代表语料中的实体；E是边集合，代表实体之间的关系。有了这张图，面对“离图书馆最近的咖啡店”时，只需找到“图书馆”节点，沿“相邻”“步行5分钟”的关系边走一两步，答案就浮现出来。

1. G 代表整个知识图谱：知识图谱是一个有向图，由节点和边共同构成。

$$G$$

2. V 是节点集合：每个节点对应语料中的一个实体，例如“北京大学”“腾讯公司”。

$$V = \{v_1, v_2, \dots, v_n\}$$

3. E 是边集合：每条边连接一个头实体和一个尾实体，并带上关系类型。

$$E \subseteq V \times V \times R$$

4. R 是关系类型集合：它限定了关系种类，防止抽取阶段出现关系爆炸。

$$R = \{\text{位于}, \text{合作}, \text{相邻}, \dots\}$$

和纸质地图一样，知识图谱也有“比例尺”：抽取多少实体、保留哪些关系，决定了地图的精细度。地图太粗，断头路多；地图太细，冗余小巷多。工程化第一步，就是找到适合业务的“比例尺”。

实体关系抽取

画地图的第一步是派测绘队。GraphRAG的“测绘队”是大语言模型，它接过一叠叠文档，用两种能力绘制草图：NER（命名实体识别）负责认出“哪里是建筑”——把人名、地名、机构名从句子中圈出来；RE（关系抽取）负责判断“两栋建筑之间是否通路”。

以句子“腾讯公司与北京大学合作建立实验室”为例，NER圈出“腾讯公司”“北京大学”“实验室”，RE判断出“腾讯公司与北京大学存在合作关系”，最终产出一条结构化三元组：（腾讯公司，合作，北京大学）。这个过程看似简单，却决定了整张地图的准确度。

工程落地时，LLM调用策略是成本与效果的分水岭。一种常见做法是：先按段落分块，设计统一的抽取提示词，要求LLM返回JSON三元组列表，解析失败自动重试；再用轻量规则或模型过滤明显错误，最后归一化实体名（“北大”→“北京大学”）。关系抽取的概率建模可写成：

$$\displaystyle P(r \mid s, o, c) = \frac{\exp(\text{score}(s, r, o, c))}{\sum_{r' \in \mathcal{R}} \exp(\text{score}(s, r', o, c))}$$

1. P(r|s,o,c) 是条件概率：给定头实体 s、尾实体 o 和上下文 c 时关系 r 成立的概率。

$$P(r \mid s, o, c)$$

2. score 是相关性打分：模型为候选关系 r 计算一个实数值分数，分数越高越可能成立。

$$\exp(\text{score}(s, r, o, c))$$

3. 分母是所有候选关系的指数和：对全部关系类型做归一化，使输出成为概率分布。

$$\sum_{r' \in \mathcal{R}} \exp(\text{score}(s, r', o, c))$$

4. 取概率最大的关系：argmax 选出最可能的关系作为抽取输出。

$$\hat{r} = \arg\max_{r \in R} P(r \mid s, o, c)$$

测绘队的工作质量直接决定地图可靠性。抽错实体（把“深圳大学”拆成“深圳”和“大学”）、判错关系（把“合作”误判为“收购”），都会让导航员把游客带到错误的地方。所以抽取后的清洗、消歧和置信度过滤，是GraphRAG工程质量的第一道防线。

图索引构建

地图上标好了建筑和道路，测绘队还要在每个地标门口挂一个“胶囊信箱”（Embedding向量），里面装着该地标的文字简介。导航员出发前，不用走到门口，只要扫一眼胶囊就能了解“这是什么地方”。给节点挂胶囊的过程，就是图索引构建。

图索引构建包含两层：一是图结构的存储，让节点和边能被快速检索和遍历；二是节点属性的向量化，让节点的语义可量化、可比较。前者回答“谁和谁连在一起”，后者回答“谁和谁本质相似”。

图结构可以用Neo4j、JanusGraph等图数据库存储，也可以直接用邻接表（Adjacency List）放在内存中——每行记录一个节点及其直接邻居。节点属性向量化则用预训练文本Embedding模型，将实体名与实体描述拼成文本后编码为稠密向量，连同向量一起存入向量数据库。节点之间的语义相似度用余弦相似度计算：

$$\displaystyle \text{sim}(n_i, n_j) = \cos(\mathbf{e}_i, \mathbf{e}_j) = \frac{\mathbf{e}_i \cdot \mathbf{e}_j}{\|\mathbf{e}_i\| \cdot \|\mathbf{e}_j\|}$$

1. e_i 是实体节点 n_i 的向量：用文本Embedding模型把实体名和描述编码成稠密向量。

$$\mathbf{e}_i = \text{Embedding}(\text{name}(n_i), \text{desc}(n_i))$$

2. 点积衡量方向一致性：两个向量对应维度相乘再求和，值越大方向越接近。

$$\mathbf{e}_i \cdot \mathbf{e}_j = \sum_k e_{ik} e_{jk}$$

3. 模长归一化：分母用两个向量的模长，把点积压缩到 [-1,1]。

$$\|\mathbf{e}_i\| = \sqrt{\sum_k e_{ik}^2}$$

4. 余弦相似度：结果接近 1 表示语义相近，接近 0 表示无关。

$$\text{sim}(n_i, n_j) = \frac{\mathbf{e}_i \cdot \mathbf{e}_j}{\|\mathbf{e}_i\| \|\mathbf{e}_j\|}$$

胶囊里装的简介越精确，导航员对地标的预判就越准。工程上，实体描述文本由多种来源拼接：实体名、抽取时的原文片段、以及一跳邻居的名称。如果某个实体的胶囊信息过于稀疏，它就会变成地图上的“无字标记”，既难被语义检索命中，也难参与相似度比较。

查询图遍历

游客终于问出具体问题。导航员打开地图，从入口实体出发，沿道路逐户走访相邻建筑——就像在街区里逐户敲门传话。这在图论中叫图遍历，是GraphRAG查询阶段的核心动作。

遍历策略主要有两种：BFS（广度优先搜索）像水面波纹一样逐层扩散，先访问全部一跳邻居，再访问二跳邻居；DFS（深度优先搜索）则像钻迷宫，沿一条路走到底再折返。GraphRAG中BFS更常用，因为它允许精确控制最大跳数（max_depth），避免被深层噪声关系带偏。

查询流程分为三步：第一步，通过实体链接或向量检索定位入口实体；若入口实体不存在，则触发回退策略——用向量检索召回语义相关文本块，再从文本块中提取候选实体作为入口。第二步，从入口出发做K跳BFS遍历，得到一组结构化候选路径。第三步，将图遍历得分与向量检索得分归一化后加权融合，得到最终排序：

$$\displaystyle \text{Score}(c) = \lambda \cdot \frac{\text{GraphScore}(c) - \mu_g}{\sigma_g} + (1-\lambda) \cdot \frac{\text{VecScore}(c) - \mu_v}{\sigma_v}$$

1. GraphScore(c) 是图遍历得分：候选上下文与入口实体之间路径越短，得分越高。

$$\text{GraphScore}(c) = \frac{1}{\text{depth}(c) + 1}$$

2. VecScore(c) 是向量检索得分：查询向量与候选向量之间的余弦相似度。

$$\text{VecScore}(c) = \cos(\mathbf{e}_q, \mathbf{e}_c)$$

3. Z-score 归一化：减去均值再除以标准差，让两类得分尺度对齐。

$$\frac{x - \mu}{\sigma}$$

4. λ 是融合权重：λ 大时图结构证据占主导，1-λ 大时语义相似证据占主导。

$$\lambda \in [0, 1]$$

导航员的高明不在于某一种能力特别强，而在于把“看地图找到的路”和“听路人描述的印象”结合起来。只靠地图可能漏掉新开的小店，只听路人描述可能迷路。图遍历与向量检索的融合，正是这种“地图+口碑”的决策机制。

全局局部摘要

出发前，资深导航员会做两件事：先看一眼整座城市的规划图，掌握东西南北的大格局；再放大目的地的街区图，看清每一条小巷。对应到GraphRAG，就是全局摘要（Global Summary）与局部摘要（Local Summary）的协同。

全局摘要回答“这个语料库整体在讲什么”。实现上，先对图谱做社区检测，把关系紧密的实体聚成社区；对每个社区，用LLM将其中所有三元组概括成一段摘要；再对社区摘要做摘要，逐层向上合成为一颗摘要树。查询时，从摘要树中取主题最相关的若干社区摘要作为背景信息。

局部摘要针对性更强：从查询的入口实体出发图遍历K跳，取出一个局部子图，把子图内的实体、关系、属性汇总成紧凑上下文。全局摘要提供主题背景，局部摘要提供直接证据，两者按查询性质动态组合。答案生成的上下文可以抽象为：

$$\displaystyle \text{ctx}(q) = [\text{GlobalSummary}(q);\; \text{LocalSubgraph}(q)]$$

1. LocalSubgraph(q) 是局部子图：从查询入口实体出发做 K 跳遍历，收集所有相关三元组作为直接证据。

$$\text{LocalSubgraph}(q) = \{(e, r, e') \mid \text{dist}(e, e_q) \le K\}$$

2. GlobalSummary(q) 是全局摘要块：从社区的摘要树中取出与查询主题最相关的摘要，提供宏观背景。

$$\text{GlobalSummary}(q) = \bigcup_{c \in \text{community}(q)} \text{summary}(c)$$

3. [;] 拼接上下文：把宏观背景和局部证据放入同一个上下文窗口。

$$[\text{GlobalSummary}(q); \text{LocalSubgraph}(q)]$$

4. LLM 生成最终答案：语言模型以拼接上下文为条件，输出概率最高的回答。

$$\hat{a} = \arg\max_a P(a \mid q, \text{ctx}(q))$$

全局摘要像城市的骨架印象——知道这座城建在山边还是海边；局部摘要像导航的最后500米——精确到哪个路口左转。两图叠加，导航才能一路顺畅。只用全局摘要，答案会泛泛而谈；只用局部摘要，答案会缺少大局观。

工程优化实践

再完美的城市地图，若测绘成本高到城市负担不起、导航慢到游客失去耐心，也只是一张废纸。GraphRAG工程化落地，本质上是构建成本、检索延迟、效果指标三者的平衡木。

成本大头在LLM抽取调用。若对全部文档做全量抽取，API费用会迅速失控。常用优化手段包括：只抽取与业务实体词典匹配的句子；按文档分块批量调用，减少往返次数；先让轻量模型初筛，再让大模型精抽。

延迟层面，图的邻接表常驻内存或使用图数据库缓存，热点实体的子图可预计算；向量检索用HNSW等近似最近邻索引加速。效果评估上，用精确率、召回率、F1等指标衡量检索质量，同时监控构建耗时、查询延迟与单次查询成本三项指标：

$$\displaystyle F1 = \frac{2 \times P \times R}{P + R}, \quad P = \frac{TP}{TP+FP}, \quad R = \frac{TP}{TP+FN}$$

1. 精确率 P：在模型返回的所有结果中，真正相关的比例，衡量“有没有检索到噪声”。

$$P = \frac{TP}{TP + FP}$$

2. 召回率 R：在全部正确答案中，模型找回的比例，衡量“有没有漏检”。

$$R = \frac{TP}{TP + FN}$$

3. F1 是调和平均：当精确率与召回率严重失衡时，F1 会给出较低分数，倒逼两者均衡。

$$F1 = \frac{2 \times P \times R}{P + R}$$

好的地图系统，既要在测绘时精打细算，又要在导航时响应敏捷，最终还要让游客顺利抵达目的地。这三项指标的持续平衡与调优，贯穿GraphRAG从原型到上线的每一步。

AI 应用场景

Python 演示

# -*- coding: utf-8 -*-"""简化版 GraphRAG：演示知识图谱增强 RAG 的核心流程."""import refrom collections import defaultdict  class SimpleGraphRAG:    """教学用简化 GraphRAG 实现."""     def __init__(self):        self.node_labels = {}        self.adjacency = defaultdict(list)     def extract_entities(self, sentences):        """用预置词典在句子中识别实体."""        vocab = ["腾讯公司", "北京大学", "北京图书馆", "深圳大学", "实验室"]        entity_dict = {}        for sent_idx, sent in enumerate(sentences):            for token in sorted(vocab, key=len, reverse=True):                if token in sent:                    entity_dict.setdefault(token, []).append(sent_idx)        return entity_dict     def extract_relations(self, sentences, entities):        """通过模式匹配抽取实体之间的关系."""        patterns = [            ("位于", r"(.+?)位于(.+?)。"),            ("合作", r"(.+?)与(.+?)合作"),            ("相邻", r"(.+?)与(.+?)相邻"),            ("成立于", r"(.+?)成立于(.+?)。"),        ]        relations = set()        for sent in sentences:            for rel, pat in patterns:                match = re.search(pat, sent)                if match:                    a, b = match.group(1), match.group(2)                    if a in entities and b in entities:                        relations.add((a, b, rel))        return relations     def build(self, sentences):        """抽取实体与关系并建立邻接表索引."""        entities = self.extract_entities(sentences)        relations = self.extract_relations(sentences, entities)        for name in entities:            nid = len(self.node_labels)            self.node_labels[nid] = name        name_to_id = {name: nid for nid, name in self.node_labels.items()}        for a, b, rel in relations:            src, dst = name_to_id[a], name_to_id[b]            self.adjacency[src].append((dst, rel))            self.adjacency[dst].append((src, rel))        return len(self.node_labels), len(relations)     def traverse(self, start_id, max_depth):        """BFS 图遍历：从起点出发逐层扩展."""        visited = set([start_id])        queue = [(start_id, 0, [])]        results = []        while queue:            node, depth, path = queue.pop(0)            if depth >= max_depth:                continue            for nxt, rel in self.adjacency[node]:                if nxt in visited:                    continue                visited.add(nxt)                new_path = path + [(self.node_labels[node], rel, self.node_labels[nxt])]                results.append((nxt, depth + 1, new_path))                queue.append((nxt, depth + 1, new_path))        return results     def answer(self, question, max_depth=2):        """定位起点实体后做图遍历并返回路径."""        start_id = None        for nid, name in self.node_labels.items():            if name in question:                start_id = nid                break        if start_id is None:            return "问题中未找到图谱内的实体"        paths = self.traverse(start_id, max_depth)        return [(self.node_labels[nid], depth, " -> ".join(            f"{s}-{r}->{t}" for s, r, t in path))            for nid, depth, path in paths]  if __name__ == "__main__":    corpus = [        "腾讯公司位于深圳。",        "腾讯公司与北京大学合作建立实验室。",        "北京大学成立于1898年。",        "北京图书馆与北京大学相邻。",        "深圳大学与腾讯公司合作研发AI。",    ]    rag = SimpleGraphRAG()    node_count, edge_count = rag.build(corpus)    print(f"构建完成: {node_count} 个实体, {edge_count} 条关系")    question = "北京大学与哪些实体有关联?"    print(f"查询: {question}")    for name, depth, path in rag.answer(question):        print(f"深度 {depth}: {name}  路径: {path}")

1. self.adjacency = defaultdict(list)邻接表是图谱的核心存储结构，记录每个节点的直接邻居和关系名。
2. for token in sorted(vocab, key=len, reverse=True):按长度降序匹配实体名，避免短词先命中而遮蔽长词。
3. if a in entities and b in entities:关系两侧必须是已识别实体，否则丢弃这条关系，过滤噪声。
4. self.adjacency[src].append((dst, rel))无向图中双向写入邻接表，保证从任意一端都能遍历到。
5. if depth >= max_depth: continueBFS的深度闸门，防止遍历在深层噪声关系上无限扩散。
6. if nxt in visited: continue访问集合去重，避免在环路中死循环。
7. if name in question:通过文本包含关系定位入口实体，是最简单的实体链接形态。
8. paths = self.traverse(start_id, max_depth)图遍历是查询核心：从入口出发、按深度扩展、返回结构化路径。

常见误区

面试题