GraphRAG：基于知识图谱的 RAG 增强

[进阶选读] 本篇适合已完成必读五篇（01-05）、希望解决多跳推理问题（如"A 公司的创始人和 B 公司的 CEO 有什么共同背景"）的读者。如果你的 RAG 系统主要用于单文档事实查询，可暂时跳过。

1.1 标准 RAG 的根本局限：碎片化的向量空间

GraphRAG 通过知识图谱+向量双路检索，解决传统 RAG 丢失实体关系的问题

标准 RAG 有一个优雅的特性：它不依赖任何预先定义的知识结构，把任何文档都切成 chunk、向量化、存起来，查询时语义相似的 chunk 就能被检索出来。

但这个优雅背后隐藏着一个致命的架构问题：向量化抛弃了实体之间的关系信息。

当文档被切成 chunk 的那一刻，跨 chunk 的实体关系就断了。chunk A 说"张伟创立了公司 X"，chunk B 说"公司 X 和公司 Y 是竞争对手"，chunk C 说"公司 Y 的投资方是机构 Z"。这三个 chunk 存进向量数据库后，变成了三个独立的浮点数向量——关系信息消失了。

向量相似度检索只能回答"找和问题语义相似的 chunk"，但无法回答"从公司 X 出发，沿关系链找到机构 Z"这种需要图遍历的问题。

什么场景会遭遇这个限制

单文档内的事实查询，标准 RAG 没有问题。问题出现在需要跨文档多跳推理的场景：

"这家公司的创始人，和另一家公司的 CEO 有什么共同背景？"
"这篇研究报告里引用的三篇论文，它们的核心观点是否一致？"
"分散在五份合同中的甲方乙方关系，整体呈现什么格局？"

回答这类问题，需要先找到相关实体，再沿着实体之间的关系边遍历——这是图问题，不是相似度搜索问题。

知识图谱是什么，它为什么能解决这个问题

知识图谱（Knowledge Graph）的核心思想是：把知识表示为"实体 + 关系"的有向图，而不是文本片段的集合。节点是实体（人、公司、地点、概念），边是关系（创立、属于、竞争、投资）。

这种表示方式天然支持关系查询：从任意节点出发，沿着特定类型的边遍历，可以精确找到所有满足关系条件的实体，而不需要猜测哪个 chunk 可能相关。

GraphRAG 的核心思路就是：先用 LLM 从文档中自动抽取实体和关系构建知识图谱，再用图遍历替代向量相似度检索来回答跨文档关系问题。

标准 RAG 能回答"某个文档里说了什么"，但对于跨文档的复杂推理——"这三份报告里提到的公司，它们之间有什么关联？"——标准 RAG 几乎无能为力。这不是检索质量的问题，而是架构层面的根本限制。

GraphRAG 通过在标准 RAG 之前加入知识图谱（Knowledge Graph，一种把实体和实体之间的关系结构化存储成图的数据库，节点是实体，边是关系）构建步骤，来解决这个限制。

1.2 标准 RAG 的架构局限

标准 RAG 把文档切成 chunk，每个 chunk 独立向量化存储。检索时找最相似的 chunk，送给 LLM 回答。

这个架构有一个根本限制：chunk 之间的关联关系在向量化时就丢失了。

举一个具体的例子。三份文档：

文档 A：介绍某公司的创始人张伟
文档 B：介绍张伟创立的另一家公司
文档 C：介绍该公司的竞争对手

用户问："张伟创立的公司和其主要竞争对手，有哪些共同的投资人？"

这个问题需要：

从文档 A 找到张伟的信息
从文档 B 找到张伟的公司
从文档 C 找到竞争对手
推断出三者之间的关系
在这个关系网络上查找共同投资人

标准 RAG 最多能帮你找到相关 chunk，但它不知道这些 chunk 里的实体是怎么关联的。LLM 要做这种跨文档推理，需要把所有相关 chunk 都塞进 context，而且还需要 LLM 自己做关系推断——这是对 LLM 推理能力的重大考验，效果不稳定。

知识图谱把这些关系显式地结构化表示出来，查询时可以直接沿着关系边遍历，而不是靠语义相似度猜。

1.3 知识图谱的基本概念

知识图谱由三种基本要素构成：

实体（Entity）：现实世界中的"物"。人名、公司名、地名、产品名、概念名。例：张伟、OpenAI、GPT-4。

关系（Relation）：实体之间的连接。例：创立了、竞争对手、投资于、属于。

三元组（Triple）：一个完整的知识单元，格式为 (主体, 关系, 客体)，是知识图谱中存储信息的基本单位。例：(张伟, 创立了, OpenAI分部X)、(GPT-4, 属于, OpenAI)。

知识图谱就是这些三元组的集合，构成一个有向图，节点是实体，边是关系。

用上面的关系图，"张伟的公司和竞争对手有共同投资人吗"这个问题就变成了一个图遍历问题：从 张伟 出发，找到 公司A，再找到 公司C，再看两者是否共享同一个 投资人X 节点。这是确定性的图查询，不依赖向量相似度。

1.4 Microsoft GraphRAG 框架

2024 年，Microsoft Research 发布了 GraphRAG 论文和开源实现，系统性地解决了基于知识图谱的 RAG 问题。

Microsoft GraphRAG 的核心贡献：

自动化知识图谱构建：用 LLM 自动从文档中抽取实体和关系，不需要手动标注
社区检测：对图中的实体做聚类，找出"社区"（在图结构中联系紧密的实体集合，类似社交网络中的圈子）
层级摘要：为每个社区生成摘要，形成层级化的知识表示
两种查询模式：Local Search（精确实体查询）和 Global Search（全局主题查询）

1.4.1 Local Search vs Global Search

查询模式	原理	适用场景	代价
Local Search	从特定实体出发，沿关系边检索相关信息	针对具体实体的问题："某公司的业务范围？"	低，精确检索
Global Search	扫描所有社区摘要，做跨社区综合	全局主题问题："整个行业的趋势？"	高，需要处理大量摘要

Local Search 适合"关于 X 的问题"，Global Search 适合"关于整个语料的问题"。两种模式在同一个图上运行，由查询类型自动选择。

1.5 GraphRAG vs 标准 RAG 架构对比

1.6 用 LangChain + Neo4j 实现 Graph RAG

下面是一个完整可运行的示例，演示用 Neo4j 存储知识图谱，用 LangChain（一个构建 AI 应用的 Python 框架，提供图数据库连接、LLM 调用等现成组件）实现 Graph RAG 查询。

验证目的：用内存版知识图谱演示从文档抽取三元组、构建图结构、执行图遍历查询的完整流程，验证"跨文档关系问题可以通过图查询而非向量相似度检索来准确回答"。

python

# graph_rag_demo.py
# 演示基于知识图谱的 RAG：实体抽取 → 图存储 → 图检索
# 依赖：pip install langchain langchain-openai langchain-community neo4j

from typing import List, Dict, Tuple, Optional
import json


# ============================================================
# 第一步：模拟文档集合
# ============================================================

SAMPLE_DOCUMENTS = [
    {
        "id": "doc1",
        "content": "OpenAI 由 Sam Altman 担任 CEO，公司总部位于旧金山。OpenAI 开发了 GPT-4 和 ChatGPT 等产品。微软向 OpenAI 投资了超过 100 亿美元。"
    },
    {
        "id": "doc2",
        "content": "Anthropic 由前 OpenAI 员工 Dario Amodei 创立。Anthropic 开发了 Claude 系列 AI 助手。亚马逊向 Anthropic 投资了 40 亿美元。"
    },
    {
        "id": "doc3",
        "content": "Google DeepMind 是 Alphabet 的子公司，开发了 Gemini 模型。Google 同时也是 Anthropic 的投资方，投资金额约 3 亿美元。"
    },
]


# ============================================================
# 第二步：用 LLM 从文档中抽取三元组
# 实际使用时接入真实 LLM，这里用模拟数据演示结构
# ============================================================

def extract_triples_with_llm(text: str) -> List[Tuple[str, str, str]]:
    """
    从文本中抽取知识三元组 (主体, 关系, 客体)。
    实际场景调用 LLM：

    from langchain_openai import ChatOpenAI
    from langchain.prompts import ChatPromptTemplate

    llm = ChatOpenAI(model="gpt-4o")
    prompt = ChatPromptTemplate.from_template(
        "从以下文本中抽取所有实体关系三元组，格式为 JSON 数组：\n"
        "[{'subject': '主体', 'relation': '关系', 'object': '客体'}]\n"
        "文本：{text}"
    )
    result = (prompt | llm).invoke({"text": text})
    triples = json.loads(result.content)
    return [(t['subject'], t['relation'], t['object']) for t in triples]
    """
    # 模拟抽取结果（结构与真实 LLM 输出一致）
    if "OpenAI" in text and "Sam Altman" in text:
        return [
            ("Sam Altman", "担任CEO", "OpenAI"),
            ("OpenAI", "总部位于", "旧金山"),
            ("OpenAI", "开发了", "GPT-4"),
            ("OpenAI", "开发了", "ChatGPT"),
            ("微软", "投资了", "OpenAI"),
        ]
    elif "Anthropic" in text and "Dario Amodei" in text:
        return [
            ("Dario Amodei", "创立了", "Anthropic"),
            ("Anthropic", "开发了", "Claude"),
            ("亚马逊", "投资了", "Anthropic"),
        ]
    elif "DeepMind" in text:
        return [
            ("Google DeepMind", "是子公司", "Alphabet"),
            ("Google DeepMind", "开发了", "Gemini"),
            ("Google", "投资了", "Anthropic"),
        ]
    return []


# ============================================================
# 第三步：内存知识图谱（生产环境替换为 Neo4j）
# ============================================================

class InMemoryKnowledgeGraph:
    """
    简单的内存知识图谱实现，用于演示。
    生产环境使用 Neo4j：

    from langchain_community.graphs import Neo4jGraph
    graph = Neo4jGraph(url="bolt://localhost:7687", username="neo4j", password="password")
    """

    def __init__(self):
        # 邻接表：entity -> [(relation, target_entity), ...]
        self.edges: Dict[str, List[Tuple[str, str]]] = {}
        # 反向索引：entity -> [(source_entity, relation), ...]
        self.reverse_edges: Dict[str, List[Tuple[str, str]]] = {}
        # 所有三元组
        self.triples: List[Tuple[str, str, str]] = []

    def add_triple(self, subject: str, relation: str, obj: str):
        """添加一个三元组到图中。"""
        self.triples.append((subject, relation, obj))

        # 正向边
        if subject not in self.edges:
            self.edges[subject] = []
        self.edges[subject].append((relation, obj))

        # 反向边（方便从客体查主体）
        if obj not in self.reverse_edges:
            self.reverse_edges[obj] = []
        self.reverse_edges[obj].append((subject, relation))

    def get_neighbors(self, entity: str, max_hops: int = 2) -> List[str]:
        """
        获取实体的邻居节点（多跳）。
        max_hops 控制图遍历深度，深度越大获取的上下文越多，但也越慢。
        """
        visited = {entity}
        current_level = {entity}
        context_triples = []

        for _ in range(max_hops):
            next_level = set()
            for node in current_level:
                # 获取以当前节点为主体的边
                for relation, target in self.edges.get(node, []):
                    context_triples.append(f"({node}) -[{relation}]-> ({target})")
                    if target not in visited:
                        next_level.add(target)
                        visited.add(target)

                # 获取以当前节点为客体的边
                for source, relation in self.reverse_edges.get(node, []):
                    context_triples.append(f"({source}) -[{relation}]-> ({node})")
                    if source not in visited:
                        next_level.add(source)
                        visited.add(source)

            current_level = next_level

        return context_triples

    def search_by_relation(self, relation_keyword: str) -> List[str]:
        """
        按关系类型搜索三元组。
        例如搜索 "投资" 找出所有投资关系。
        """
        return [
            f"({s}) -[{r}]-> ({o})"
            for s, r, o in self.triples
            if relation_keyword.lower() in r.lower()
        ]

    def find_common_neighbors(self, entity1: str, entity2: str) -> List[str]:
        """
        找出两个实体的共同邻居。
        这是 Graph RAG 解决"共同投资人"类问题的核心操作。
        """
        def get_all_neighbors(entity):
            neighbors = set()
            for _, target in self.edges.get(entity, []):
                neighbors.add(target)
            for source, _ in self.reverse_edges.get(entity, []):
                neighbors.add(source)
            return neighbors

        neighbors1 = get_all_neighbors(entity1)
        neighbors2 = get_all_neighbors(entity2)
        return list(neighbors1 & neighbors2)


# ============================================================
# 第四步：Graph RAG 查询管道
# ============================================================

class GraphRAGPipeline:
    """
    Graph RAG 完整查询管道：
    1. 从问题中识别关键实体
    2. 在知识图谱中检索相关三元组（图遍历）
    3. 把三元组转为自然语言上下文
    4. 送给 LLM 生成答案
    """

    def __init__(self, kg: InMemoryKnowledgeGraph):
        self.kg = kg

    def extract_entities_from_query(self, query: str) -> List[str]:
        """
        从查询中识别关键实体。
        实际场景用 NER 模型或 LLM 来做，这里用简单规则演示。
        """
        # 简单规则：检查查询中是否包含图中已知的实体
        known_entities = set(self.kg.edges.keys()) | set(self.kg.reverse_edges.keys())
        found = [e for e in known_entities if e in query]

        # 如果没找到已知实体，返回空（后续使用全局搜索）
        return found

    def retrieve_graph_context(self, query: str) -> str:
        """
        核心检索函数：从知识图谱中获取与查询相关的上下文。
        根据查询类型选择 Local Search 或关系搜索。
        """
        entities = self.extract_entities_from_query(query)

        context_parts = []

        if entities:
            # Local Search：从识别出的实体出发做图遍历
            for entity in entities:
                neighbors = self.kg.get_neighbors(entity, max_hops=2)
                if neighbors:
                    context_parts.append(f"关于「{entity}」的关系：")
                    context_parts.extend(neighbors[:10])  # 限制数量，避免 context 太长

        # 如果问题涉及"共同"关系，做共同邻居查询
        if "共同" in query or "都" in query:
            if len(entities) >= 2:
                common = self.kg.find_common_neighbors(entities[0], entities[1])
                if common:
                    context_parts.append(f"\n{entities[0]} 和 {entities[1]} 的共同关联实体：{common}")

        # 如果没有找到具体实体，做关系类型搜索
        if not context_parts:
            for keyword in ["投资", "创立", "开发", "竞争"]:
                if keyword in query:
                    triples = self.kg.search_by_relation(keyword)
                    if triples:
                        context_parts.append(f"与「{keyword}」相关的关系：")
                        context_parts.extend(triples[:10])

        return "\n".join(context_parts) if context_parts else "未找到相关知识图谱信息"

    def answer_with_graph_context(self, query: str) -> Dict:
        """
        完整的 Graph RAG 问答：检索图上下文 → 生成答案。
        """
        graph_context = self.retrieve_graph_context(query)

        # 实际场景调用 LLM：
        # from langchain_openai import ChatOpenAI
        # llm = ChatOpenAI(model="gpt-4o")
        # prompt = f"基于以下知识图谱信息回答问题：\n{graph_context}\n\n问题：{query}"
        # answer = llm.invoke(prompt).content

        # 模拟 LLM 回答
        answer = f"基于知识图谱分析，{query.rstrip('？?')}的情况如下：\n从图谱中提取的相关信息显示了实体间的关联关系。"

        return {
            "query": query,
            "graph_context": graph_context,
            "answer": answer,
        }


# ============================================================
# 第五步：使用 Neo4j 的生产级实现（代码框架）
# ============================================================

def build_neo4j_graph_rag():
    """
    生产级 Graph RAG，使用 Neo4j 作为图数据库。
    需要安装：pip install langchain-community neo4j
    需要运行 Neo4j 实例：docker run -p 7474:7474 -p 7687:7687 neo4j
    """
    try:
        from langchain_community.graphs import Neo4jGraph
        from langchain_community.chains.graph_qa.cypher import GraphCypherQAChain
        from langchain_openai import ChatOpenAI

        # 连接 Neo4j
        graph = Neo4jGraph(
            url="bolt://localhost:7687",
            username="neo4j",
            password="your_password"
        )

        # 插入知识三元组（Cypher 是 Neo4j 的查询语言，类似 SQL 但专为图数据库设计）
        # Neo4j（一种专门存储图结构数据的数据库）使用 Cypher 语言，MERGE 表示"不存在则创建"
        graph.query("""
            MERGE (sam:Person {name: 'Sam Altman'})
            MERGE (openai:Company {name: 'OpenAI'})
            MERGE (sam)-[:CEO_OF]->(openai)
            MERGE (microsoft:Company {name: '微软'})
            MERGE (microsoft)-[:INVESTED_IN {amount: '100亿美元'}]->(openai)
        """)

        # 用 GraphCypherQAChain 实现自然语言转 Cypher 查询
        # 这个 Chain 会把用户的自然语言问题转成 Cypher 查询，在 Neo4j 上执行，再用 LLM 总结答案
        llm = ChatOpenAI(model="gpt-4o", temperature=0)
        chain = GraphCypherQAChain.from_llm(
            llm=llm,
            graph=graph,
            verbose=True,   # 显示生成的 Cypher 查询，方便调试
        )

        result = chain.invoke({"query": "谁投资了 OpenAI？"})
        return result

    except ImportError:
        return "需要安装：pip install langchain-community neo4j langchain-openai"


# ============================================================
# 演示运行
# ============================================================

def run_demo():
    print("=" * 60)
    print("GraphRAG 演示：知识图谱构建与查询")
    print("=" * 60)

    # 构建知识图谱
    kg = InMemoryKnowledgeGraph()

    print("\n[第一步] 从文档中抽取三元组...")
    for doc in SAMPLE_DOCUMENTS:
        triples = extract_triples_with_llm(doc["content"])
        for s, r, o in triples:
            kg.add_triple(s, r, o)
            print(f"  + ({s}) -[{r}]-> ({o})")

    print(f"\n知识图谱构建完成：{len(kg.triples)} 个三元组")

    # 初始化查询管道
    pipeline = GraphRAGPipeline(kg)

    # 测试不同类型的查询
    test_queries = [
        "OpenAI 开发了哪些产品？",
        "谁投资了 AI 公司？",
        "Anthropic 和 OpenAI 有哪些共同关联？",
    ]

    print("\n[第二步] 执行 Graph RAG 查询...")
    for query in test_queries:
        print(f"\n查询：{query}")
        result = pipeline.answer_with_graph_context(query)
        print(f"图谱上下文：\n{result['graph_context']}")
        print(f"回答：{result['answer']}")
        print("-" * 40)


if __name__ == "__main__":
    run_demo()

1.7 GraphRAG 的代价

GraphRAG 的效果好，但代价也是真实的：

构建时间：对一个中等规模的文档集（100 份文档），知识图谱构建需要大量 LLM 调用来做实体抽取和关系识别。每份文档调用一次 LLM，100 份就是 100 次调用，加上社区检测和摘要生成，总调用次数可能达到几百次。

Token 成本：Microsoft 的基准测试显示，对一个百万 token 的语料库，构建 GraphRAG 的成本约为 $3-5（使用 GPT-4o-mini），但使用 GPT-4o 会高出 10 倍以上。

维护成本：文档更新时，知识图谱需要对应更新——不只是添加新实体，还需要处理已有实体的关系变化。标准 RAG 只需重新 embedding，GraphRAG 还需要重新跑实体抽取和图更新。

适用判断

场景	推荐方案	原因
问题只涉及单个文档内的信息	标准 RAG	GraphRAG 的开销不值得
需要跨文档推理和关系查询	GraphRAG	标准 RAG 做不好
文档频繁更新（每天）	标准 RAG	图维护成本高
文档相对稳定、关系密集	GraphRAG	一次构建，长期使用
语料规模小（< 50 文档）	标准 RAG + 全文塞 context	直接全塞更简单

GraphRAG 解决的是标准 RAG 的架构缺陷：chunk 切碎了，实体间的关系就断了。知识图谱把这些关系显式存下来，查询时直接沿边遍历，而不是靠语义相似度猜。

代价是真实的：构建成本高、图的维护比向量索引复杂。不要为了用 GraphRAG 而用 GraphRAG——先确认你的问题确实需要跨文档多跳推理，再考虑引入它。