GraphRAG 是什么

GraphRAG 是微软研究院于 2024 年提出的 RAG 增强架构。传统 Vector RAG 基于向量相似度检索文档片段，难以回答需要跨文档推理、全局性总结的问题。GraphRAG 通过构建知识图谱，利用实体关系和社区结构来增强检索和生成能力。

核心问题：Vector RAG 擅长"找相关文档"，但不擅长"综合多文档的全局信息"。

问题示例：
  Vector RAG 擅长："公司2024年Q3营收是多少？"（单文档事实）
  Vector RAG 失败："总结整个行业的主要参与者及其关系"（全局综合）
  GraphRAG 擅长：以上两者都能处理

架构全貌

┌──────────────────────────────────────────────────────┐
│                   GraphRAG Pipeline                   │
│                                                       │
│  ┌─────────┐   ┌──────────┐   ┌──────────────────┐  │
│  │ 文档分块 │──▶│ 实体抽取  │──▶│ 关系图构建        │  │
│  └─────────┘   └──────────┘   └────────┬─────────┘  │
│                                        │              │
│                              ┌─────────▼──────────┐  │
│                              │ 社区检测 (Leiden)   │  │
│                              └─────────┬──────────┘  │
│                                        │              │
│                       ┌────────────────┼────────────┐ │
│                       ▼                ▼            │ │
│              ┌──────────────┐ ┌────────────────┐    │ │
│              │ 社区摘要生成  │ │ 向量索引        │    │ │
│              └──────┬───────┘ └───────┬────────┘    │ │
│                     │                 │              │
│                     ▼                 ▼              │
│              ┌──────────────┐ ┌────────────────┐    │ │
│              │ Global Search│ │ Local Search   │    │ │
│              └──────────────┘ └────────────────┘    │ │
└──────────────────────────────────────────────────────┘

核心步骤详解

1. 文档分块

与标准 RAG 一样，先对源文档分块：

from langchain.text_splitter import RecursiveCharacterTextSplitter

def chunk_documents(documents):
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=1200,
        chunk_overlap=100,
        separators=["\n\n", "\n", ". ", " ", ""]
    )
    chunks = []
    for doc in documents:
        chunks.extend(splitter.split_text(doc))
    return chunks

2. 实体与关系抽取

使用 LLM 从每个块中抽取实体和关系：

ENTITY_PROMPT = """从以下文本中抽取实体和关系，输出 JSON 格式。

文本：
{text}

输出格式：
{{
  "entities": [
    {{"name": "实体名", "type": "ORGANIZATION/PERSON/LOCATION/CONCEPT", "description": "描述"}}
  ],
  "relationships": [
    {{"source": "实体A", "target": "实体B", "description": "关系描述", "strength": 0.8}}
  ]
}}
"""

def extract_entities_relations(chunks, llm):
    all_entities = {}
    all_relations = []
    
    for chunk in chunks:
        result = llm.generate(ENTITY_PROMPT.format(text=chunk))
        data = json.loads(result)
        
        for entity in data["entities"]:
            key = entity["name"].lower()
            if key in all_entities:
                # 合并描述
                all_entities[key]["description"] += f"; {entity['description']}"
            else:
                all_entities[key] = entity
        
        all_relations.extend(data["relationships"])
    
    return list(all_entities.values()), all_relations

3. 知识图谱构建

import networkx as nx

def build_knowledge_graph(entities, relations):
    G = nx.Graph()
    
    # 添加实体节点
    for entity in entities:
        G.add_node(
            entity["name"],
            type=entity["type"],
            description=entity["description"]
        )
    
    # 添加关系边
    for rel in relations:
        if G.has_edge(rel["source"], rel["target"]):
            # 已有关系，合并描述
            existing = G[rel["source"]][rel["target"]]
            existing["description"] += f"; {rel['description']}"
            existing["weight"] = max(existing["weight"], rel.get("strength", 1.0))
        else:
            G.add_edge(
                rel["source"], rel["target"],
                description=rel["description"],
                weight=rel.get("strength", 1.0)
            )
    
    return G

4. 社区检测与层次聚类

使用 Leiden 算法检测社区结构，生成层次化的社区：

from community import community_louvain  # 或使用 graspologic

def detect_communities(G, resolution=1.0):
    # Leiden 算法进行社区检测
    partition = community_louvain.best_partition(G, resolution=resolution)
    
    # 构建层次结构
    communities = {}
    for node, community_id in partition.items():
        if community_id not in communities:
            communities[community_id] = []
        communities[community_id].append(node)
    
    return list(communities.values())

def hierarchical_communities(G, max_levels=5):
    """多层社区检测，从粗到细"""
    levels = []
    current_graph = G.copy()
    
    for level in range(max_levels):
        communities = detect_communities(current_graph, resolution=1.0 / (level + 1))
        levels.append(communities)
        
        if all(len(c) <= 1 for c in communities):
            break
    
    return levels

5. 社区摘要生成

为每个社区生成 LLM 摘要，这是 GraphRAG 的关键创新：

SUMMARY_PROMPT = """你是一个知识图谱分析专家。请为以下实体社区生成综合摘要。

社区成员：
{entities}

实体间关系：
{relationships}

请生成：
1. 社区主题概述
2. 主要实体及其角色
3. 关键关系和模式
4. 潜在重要发现
"""

def generate_community_summaries(communities, graph, llm):
    summaries = []
    for i, community in enumerate(communities):
        entities_info = "\n".join(
            f"- {node}: {graph.nodes[node].get('description', '')}"
            for node in community
        )
        relations_info = "\n".join(
            f"- {u} → {v}: {d.get('description', '')}"
            for u, v, d in graph.subgraph(community).edges(data=True)
        )
        
        summary = llm.generate(SUMMARY_PROMPT.format(
            entities=entities_info,
            relationships=relations_info
        ))
        summaries.append({
            "community_id": i,
            "members": community,
            "summary": summary
        })
    
    return summaries

两种搜索模式

Global Search：全局搜索

针对需要综合全局信息的问题，遍历社区摘要生成回答：

class GlobalSearch:
    def __init__(self, community_summaries, llm):
        self.summaries = community_summaries
        self.llm = llm
    
    def search(self, query, top_communities=10):
        # 1. 按相关性筛选社区
        scored = []
        for summary in self.summaries:
            score = self.llm.score_relevance(query, summary["summary"])
            scored.append((summary, score))
        
        scored.sort(key=lambda x: x[1], reverse=True)
        top = [s[0] for s in scored[:top_communities]]
        
        # 2. Map: 每个社区生成部分回答
        partial_answers = []
        for summary in top:
            prompt = f"""基于以下社区信息回答问题。
            问题：{query}
            社区摘要：{summary['summary']}
            回答（如不相关请说明）："""
            partial = self.llm.generate(prompt)
            partial_answers.append(partial)
        
        # 3. Reduce: 综合所有部分回答
        final_prompt = f"""综合以下多源信息给出最终回答。
        问题：{query}
        各来源回答：{json.dumps(partial_answers, ensure_ascii=False)}
        最终回答："""
        
        return self.llm.generate(final_prompt)

Local Search：局部搜索

针对具体实体的问题，从图谱中提取相关子图：

class LocalSearch:
    def __init__(self, graph, entity_index, llm):
        self.graph = graph
        self.entity_index = entity_index  # 向量索引
        self.llm = llm
    
    def search(self, query, k=5, hops=2):
        # 1. 找到最相关的实体
        entities = self.entity_index.search(query, k=k)
        
        # 2. 提取 k-hop 子图
        subgraph_nodes = set()
        for entity in entities:
            if entity in self.graph:
                subgraph_nodes.add(entity)
                for _ in range(hops):
                    neighbors = set()
                    for node in subgraph_nodes:
                        neighbors.update(self.graph.neighbors(node))
                    subgraph_nodes.update(neighbors)
        
        subgraph = self.graph.subgraph(subgraph_nodes)
        
        # 3. 构建上下文并生成
        context = self._format_subgraph(subgraph)
        return self.llm.generate(f"基于以下知识图谱信息回答：\n{context}\n\n问题：{query}")

GraphRAG vs Vector RAG

维度Vector RAGGraphRAG
检索方式向量相似度图遍历 + 社区摘要
全局综合弱强
单点事实强中
构建成本低（嵌入即可）高（LLM 抽取+图构建）
查询延迟低（向量搜索）高（Map-Reduce）
可解释性中（文档引用）高（关系链路）
更新成本重新嵌入重新抽取+图更新
Token 消耗低高（构建阶段大量 LLM 调用）

查询效果对比

Q1: "谁是公司最大的竞争对手？"
  Vector RAG: 检索到提到"竞争对手"的段落，可能遗漏
  GraphRAG: 从"公司"节点出发，遍历"竞争"关系，直接找到答案

Q2: "总结行业生态中的主要参与者和合作网络"
  Vector RAG: 几乎无法回答（需要综合全部文档）
  GraphRAG: Global Search 遍历社区摘要，自然回答全局问题

Q3: "产品 A 的保修期是多久？"
  Vector RAG: 直接检索到产品手册中的相关段落
  GraphRAG: 也能回答，但杀鸡用牛刀

适用场景

def should_use_graphrag(query_type, data_scale, budget):
    """GraphRAG 适用性判断"""
    if query_type == "global_summary":
        return True   # 全局综合问题
    if query_type == "multi_hop_reasoning":
        return True   # 多跳推理
    if data_scale == "large" and budget == "high":
        return True   # 大规模知识库 + 预算充足
    if query_type == "single_fact":
        return False  # 简单事实查询用 Vector RAG
    if budget == "low":
        return False  # 预算不足，GraphRAG 构建成本高
    return False

推荐场景：

  • 企业全量知识库的战略分析
  • 行业研究报告生成
  • 法律案例关系分析
  • 药物相互作用网络分析
  • 供应链关系追踪

不推荐场景：

  • 简单 FAQ 问答
  • 实时聊天客服
  • 文档频繁变更的系统
  • 预算有限的小团队

实践建议

  1. 混合使用：Vector RAG 处理事实查询，GraphRAG 处理全局分析
  2. 增量更新：设计增量图更新策略，避免全量重建
  3. 分层社区：多层社区结构，粗粒度用于全局搜索，细粒度用于细节查询
  4. 缓存社区摘要：摘要生成是最贵的步骤，缓存可大幅降低成本
  5. 先用小数据验证：GraphRAG 的 LLM 调用成本很高，先小规模验证效果

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。