开发者日记

外观

一文搞懂 GraphRAG：实体抽取到社区摘要 ​

为什么普通向量 RAG 答不好跨文档多跳问题 ​

典型企业知识库往往由会议纪要、合同、工单、PRD、组织架构和客户访谈组成。用户问：“A 客户投诉的模块由哪个团队负责？该团队最近一次架构调整影响了哪些服务？”这不是在找一段相似文本，而是要沿着“客户—问题—模块—团队—架构调整—服务”的关系链跨越多篇文档完成推理。

普通向量 RAG 会把问题编码成向量，再取 Top-K 相似块。它有三个结构性短板：第一，实体之间的显式关系被切块切碎；第二，“客户 A”和“客户简称 ACorp”可能无法被识别为同一节点；第三，Top-K 偏向局部语义相似，缺少全局主题视角。HyDE、查询改写和重排序能缓解召回问题，但没有真正补上关系结构。

GraphRAG 的核心思路是：离线阶段从文档中抽取实体与关系，构建知识图谱；再通过社区检测把关联紧密的节点聚成社区，生成局部摘要和全局摘要；在线问答时根据问题选择实体子图或社区摘要，让 LLM 基于结构化上下文回答。

GraphRAG 与普通 RAG 的选型对比 ​

方案检索对象适合问题主要成本主要风险
向量 RAG文本块单文档事实查询、定义解释低，建库快多跳链路断、全局问题弱
HyDE + Rerank假设答案与重排文本块提问表述和文档表述差异大中等，增加 LLM 与重排延迟假设答案错误会带偏召回
GraphRAG实体、关系、社区摘要跨文档归因、关系推理、全局总结高，需要抽取与摘要图谱噪声、实体歧义、更新复杂
GraphRAG + 向量图谱子图与原文块混合既要关系推理又要引用证据较高排序与上下文拼装复杂

我的建议是：如果 80% 问题都能在一两个段落中找到答案，不要为了“图谱”而上图谱；如果评测集中大量问题需要 3 跳以上连接，或经常要求“总结某类客户、部门、项目之间的共同模式”，GraphRAG 的收益会明显。

GraphRAG 离线构建流程 ​

下面这张图概括了从原始文档到多跳答案的主链路。

文档切分保留标题与来源实体关系抽取LLM 输出 JSON图谱构建实体消歧与边合并社区检测Leiden/连通分量社区摘要局部到全局层级混合检索子图、摘要、原文块多跳问答引用证据生成

第一步是切分。GraphRAG 不建议按固定 token 粗暴切块，因为一个句子可能同时包含主体、动作和客体。更稳妥的方式是按标题、段落和事件边界切分，并给每个 chunk 附加文档 ID、章节路径、时间、来源类型等元数据。

第二步是实体关系抽取。抽取提示词必须约束类型、输出格式和证据字段，否则图谱会充满“系统”“问题”“能力”这类无意义大节点。下面是可复用提示词模板。

text
你是严格的知识图谱抽取器。请从给定文本中抽取实体和关系。

实体类型只能从以下集合选择：
Person, Team, Product, Module, Service, Customer, Incident, Project, Document, Risk

关系类型只能从以下集合选择：
owns, maintains, depends_on, caused_by, reported_by,
belongs_to, participates_in, affects, mentions, related_to

输出 JSON：
{
  "entities": [{"name":"", "type":"", "aliases":[], "description":""}],
  "relations": [{
    "subject":"", "subject_type":"",
    "relation":"",
    "object":"", "object_type":"",
    "evidence":"原文片段，不得改写"
  }]
}

要求：
1. 只输出 JSON，不要解释。
2. 证据不足时不要猜测。
3. 名称使用规范全称，别名放入 aliases。
4. 同一关系在文本中重复出现，只输出一次。

第三步是实体消歧和边合并。比如“支付组”“支付团队”“Payment Team”应合并为同一团队；但“苹果公司”和“苹果水果”不能只因字面相似而合并。实践中可采用“名称归一化 + 向量相似度 + 类型一致 + 别名表”的组合策略，低置信度合并进入人工审核。

第四步是社区检测。小规模图谱可直接用 NetworkX 的连通分量或 Louvain；中大型图谱更适合 Leiden，因为它能减少断开的社区。社区不是简单聚类，而是为后续摘要提供主题边界。

第五步是社区摘要。每个社区摘要应回答：这个群体包含哪些关键实体、它们之间的核心关系、有哪些冲突或风险、证据来自哪些文档。对大社区还要递归生成父级摘要，使系统既能回答局部问题，也能回答“全局有哪些共性风险”。

可运行 Python：最小 GraphRAG 原型 ​

下面示例不依赖商业框架，使用 NetworkX 模拟图谱构建、连通社区和摘要输入生成。真实项目中可将 llm_json() 替换为任意支持结构化输出的模型接口，并为 chunk 和摘要建立向量索引。

python
# 安装依赖：pip install networkx
import json
import re
from collections import defaultdict
import networkx as nx


def llm_json(prompt: str) -> dict:
    """模拟 LLM 的结构化 JSON 返回。
    生产环境可替换为：
    1. 调用支持 response_format=json_object 的聊天模型；
    2. 用 jsonschema 校验；
    3. 解析失败时带错误信息重试一次。
    """
    fake_text = """
    {
      "entities": [
        {"name": "ACorp", "type": "Customer", "aliases": ["A客户"], "description": "企业客户"},
        {"name": "支付网关", "type": "Module", "aliases": ["Payment Gateway"], "description": "处理支付请求"},
        {"name": "交易团队", "type": "Team", "aliases": ["支付组"], "description": "维护交易链路"},
        {"name": "INC-204", "type": "Incident", "aliases": [], "description": "支付超时故障"}
      ],
      "relations": [
        {"subject": "ACorp", "subject_type": "Customer",
         "relation": "reported_by", "object": "INC-204", "object_type": "Incident",
         "evidence": "A客户反馈支付请求频繁超时，工单编号 INC-204。"},
        {"subject": "INC-204", "subject_type": "Incident",
         "relation": "affects", "object": "支付网关", "object_type": "Module",
         "evidence": "故障影响支付网关的下单回调。"},
        {"subject": "交易团队", "subject_type": "Team",
         "relation": "maintains", "object": "支付网关", "object_type": "Module",
         "evidence": "支付网关由交易团队维护。"}
      ]
    }
    """
    return json.loads(re.search(r"\{.*\}", fake_text, re.S).group(0))


def normalize_name(name: str) -> str:
    """名称归一化：去空格、统一英文大小写。"""
    return re.sub(r"\s+", "", name.strip()).lower()


# 实际项目中应从数据库或别名服务加载
ALIAS_TO_CANONICAL = {
    normalize_name("A客户"): "ACorp",
    normalize_name("支付组"): "交易团队",
    normalize_name("Payment Gateway"): "支付网关",
}


def canonical(entity_name: str) -> str:
    key = normalize_name(entity_name)
    return ALIAS_TO_CANONICAL.get(key, entity_name.strip())


def build_graph(extractions: list[dict]) -> nx.MultiDiGraph:
    G = nx.MultiDiGraph()
    for item in extractions:
        for ent in item.get("entities", []):
            name = canonical(ent["name"])
            if not G.has_node(name):
                G.add_node(
                    name,
                    type=ent["type"],
                    aliases=set(ent.get("aliases", [])),
                    descriptions=[ent.get("description", "")],
                )
        for rel in item.get("relations", []):
            s = canonical(rel["subject"])
            o = canonical(rel["object"])
            edge_key = (s, rel["relation"], o)
            if not G.has_edge(s, o, key=edge_key):
                G.add_edge(
                    s, o, key=edge_key,
                    relation=rel["relation"],
                    evidence=[rel["evidence"]],
                )
            else:
                G.edges[s, o, edge_key]["evidence"].append(rel["evidence"])
    return G


def detect_communities(G: nx.Graph) -> list[list[str]]:
    """最小原型使用弱连通分量；生产环境可替换为 Leiden 算法。"""
    undirected = G.to_undirected()
    return [list(nodes) for nodes in nx.connected_components(undirected)]


def community_summary_prompt(G: nx.MultiDiGraph, nodes: list[str]) -> str:
    lines = ["请基于以下图谱证据生成社区摘要，包含关键实体、关系、风险和证据来源。", ""]
    for n in nodes:
        node = G.nodes[n]
        lines.append(f"实体：{n}（类型：{node.get('type')}）")
    lines.append("")
    for s, o, data in G.edges(data=True):
        if s in nodes and o in nodes:
            lines.append(f"关系：{s} -[{data['relation']}]-> {o}")
            for ev in data["evidence"][:2]:
                lines.append(f"证据：{ev}")
    return "\n".join(lines)


def retrieve_subgraph(G: nx.MultiDiGraph, seeds: list[str], hops: int = 2) -> nx.MultiDiGraph:
    """从问题命中的实体出发，扩展固定跳数形成推理子图。"""
    selected = set(canonical(s) for s in seeds if canonical(s) in G)
    frontier = set(selected)
    for _ in range(hops):
        added = set()
        for node in frontier:
            added.update(G.successors(node))
            added.update(G.predecessors(node))
        frontier = added - selected
        selected.update(added)
    return G.subgraph(selected).copy()


# 1. 离线抽取并建图
extraction = llm_json("抽取实体关系")
G = build_graph([extraction])

# 2. 社区检测并生成摘要输入
communities = detect_communities(G)
for idx, nodes in enumerate(communities, 1):
    print(f"--- 社区 {idx}: {nodes} ---")
    print(community_summary_prompt(G, nodes))

# 3. 在线多跳检索：从 ACorp 出发扩展两跳
answer_graph = retrieve_subgraph(G, seeds=["A客户"], hops=2)
print("\n多跳子图节点：", list(answer_graph.nodes))
print("多跳子图边：")
for s, o, data in answer_graph.edges(data=True):
    print(s, data["relation"], o)

运行后可以看到，问题只从“A客户”出发，也能沿 reported_by → affects 找到“支付网关”，再通过反向 maintains 找到“交易团队”。这正是跨文档多跳问答需要的关系路径。

在线问答：子图与社区摘要如何协同 ​

在线阶段不建议把整张图谱塞进提示词。更合理的策略是按问题类型分流：

  1. 实体关系型问题：先用别名表和向量检索识别种子实体，再扩展 1 到 3 跳，最后附带边的原文证据。
  2. 全局总结型问题：优先检索相关社区摘要，必要时读取父级摘要，而不是盲目取大量文本块。
  3. 对比分析型问题：分别抽取多个实体的邻域子图，再让模型比较共同邻居、路径和时间线。
  4. 证据引用型问题：图谱只负责找路径，最终答案必须引用 chunk 原文，避免摘要层层传播造成幻觉。

生成答案时可使用如下结构：先列出检索到的路径，再给出结论，最后给出证据编号。若某条边没有原文证据，应明确标记“关系缺失”，而不是让模型补全。

避坑总结：GraphRAG 落地最容易踩的 4 个坑 ​

1. 实体抽取过宽，导致超级节点吞掉检索结果 ​

常见错误现象是图谱中出现大量“系统”“数据”“客户”“问题”节点，一个节点连接上千条边，扩展两跳后上下文爆炸。修正方法是在提示词中限制实体类型，把抽象名词改为具体对象；同时设置节点度数阈值，对超级节点按业务域拆分或降权，并在评测中统计平均子图节点数。

2. 只做字符串匹配，未处理别名和同名歧义 ​

错误现象是查询“支付组”找不到“交易团队”，或者把不同文档中的“蓝鲸”同时识别为项目、系统和客户。修正方法是维护别名表、要求模型输出 aliases，并将实体类型纳入合并条件。对高风险实体不要自动合并，应生成人工审核队列。

3. 社区摘要写成泛泛总结，丢失时间、方向和证据 ​

不少初版摘要只写“多个团队协作处理客户问题”，无法支持推理。正确做法是摘要中保留关系方向、关键时间、因果链和证据 ID，例如“ACorp 报告 INC-204；INC-204 影响支付网关；交易团队维护支付网关”。社区过大时要递归切分，避免摘要把冲突信息平均掉。

4. 只看回答主观效果，不评测图谱链路质量 ​

GraphRAG 的失败可能发生在抽取、消歧、社区、检索、生成任一环节。建议同时评测四类指标：实体准确率、关系准确率、路径命中率和最终答案正确率。构造黄金集时为每个多跳问题标注种子实体、期望路径和证据片段。否则看起来“答案像那么回事”，实际上模型可能绕过图谱凭常识编造。

最后，GraphRAG 不是替换向量库的银弹，而是把显式关系和全局主题引入 RAG 的工程化方案。最稳妥的落地路径是先选一批跨文档多跳问题做黄金评测，再用最小图谱验证路径命中率；确认收益后，再投入增量更新、权限过滤、社区摘要层级和混合检索排序，避免一开始就建设复杂而难以维护的大而全知识图谱。

免责声明

本工具与文章内容仅用于学习测试场景，优化建议仅供参考，请勿直接用于生产环境，由此产生的风险自行承担。

本文来自开发者日记（devdiary.cn）· 记录开发者实战笔记与免费在线工具箱

免责声明

本工具与文章内容仅用于学习测试场景，优化建议仅供参考，请勿直接用于生产环境，由此产生的风险自行承担。