知识图谱与LLM融合：GraphRAG实战

GraphRAG将知识图谱的结构化推理能力与LLM的自然语言生成能力深度融合，解决了传统RAG在多跳推理、全局摘要和事实一致性上的不足。本文基于Microsoft GraphRAG和LlamaIndex的实现，提供从原理到代码的完整实战指南。

一、为什么需要GraphRAG

1.1 传统RAG的局限

传统向量RAG流程：
  问题 → 向量化 → Top-K检索 → 拼接上下文 → LLM生成

局限：
├── 局部视角：只检索最相似的chunk，缺乏全局理解
├── 多跳推理困难：无法自然串联"A→B→C"的推理链
├── 上下文碎片化：检索到的chunk可能来自不相关的段落
├── 事实一致性差：LLM可能在不同chunk间产生矛盾
└── 全局摘要缺失：无法回答需要综合全文的问题

GraphRAG的解决思路：
├── 全局视角：知识图谱提供领域的结构化全景
├── 多跳推理：沿图谱路径自然串联信息
├── 上下文结构化：实体+关系提供有组织的上下文
├── 事实锚定：图谱三元组提供可验证的事实依据
└── 社区摘要：分层社区结构支持全局概括

1.2 GraphRAG核心架构

GraphRAG两阶段架构：

阶段1: 索引构建（离线）
  文档语料
      │
      ▼
  ┌─────────────────┐
  │ LLM实体/关系抽取 │
  └─────────────────┘
      │
      ▼
  ┌─────────────────┐
  │ 知识图谱构建      │
  │ (实体消歧+融合)  │
  └─────────────────┘
      │
      ▼
  ┌─────────────────┐      ┌──────────────┐
  │ 社区检测          │──→  │ 社区摘要生成   │
  │ (Leiden算法)      │      │ (LLM摘要)     │
  └─────────────────┘      └──────────────┘
      │
      ▼
  图谱 + 社区摘要 + 向量索引

阶段2: 查询（在线）
  用户问题
      │
      ├── 局部搜索(Local Search)
      │   └── 实体识别 → 子图检索 → 上下文组装 → LLM回答
      │
      └── 全局搜索(Global Search)
          └── 社区摘要检索 → Map-Reduce → LLM回答

二、Microsoft GraphRAG实现

2.1 环境搭建

# 安装GraphRAG
pip install graphrag

# 初始化项目
mkdir my-graphrag-project && cd my-graphrag-project
graphrag init --root .

# 项目结构
my-graphrag-project/
├── settings.yaml          # 配置文件
├── .env                   # API密钥
├── input/                 # 输入文档目录
│   └── *.txt              # 待索引的文档
├── output/                # 输出目录（索引后生成）
│   ├── entities.parquet
│   ├── relationships.parquet
│   ├── communities.parquet
│   └── community_reports.parquet
└── prompts/               # 自定义提示词
    ├── entity_extraction.txt
    └── community_report.txt

2.2 配置

# settings.yaml 关键配置
llm:
  api_key: ${GRAPHRAG_API_KEY}
  type: openai_chat
  model: gpt-4o-mini  # 索引构建用较便宜的模型
  max_tokens: 4000

embeddings:
  llm:
    api_key: ${GRAPHRAG_API_KEY}
    type: openai_embedding
    model: text-embedding-3-small

chunks:
  size: 1200
  overlap: 100

entity_extraction:
  max_gleanings: 1  # 多次抽取以提高召回
  entity_types:
    - organization
    - person
    - event
    - technology
    - location

community_reports:
  max_length: 2000

claim_extraction:
  enabled: true  # 提取声明/事实

2.3 索引构建

# 执行索引构建
graphrag index --root .

# 索引过程：
# 1. 文档分块 (Chunking)
# 2. 实体和关系抽取 (Entity/Relationship Extraction)
# 3. 实体消歧和图构建 (Graph Construction)
# 4. 社区检测 (Community Detection - Leiden Algorithm)
# 5. 社区摘要生成 (Community Summarization)
# 6. 向量嵌入生成 (Embedding Generation)

2.4 查询

# 局部搜索（适合具体问题）
graphrag query --root . --method local \
  --query "华为在AI芯片领域有哪些核心产品？"

# 全局搜索（适合宏观概括问题）
graphrag query --root . --method global \
  --query "这些文档的主要主题和趋势是什么？"

三、LlamaIndex PropertyGraphIndex实现

3.1 基于LlamaIndex的GraphRAG

from llama_index.core import (
    SimpleDirectoryReader,
    PropertyGraphIndex,
    Settings,
)
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.graph_stores.neo4j import Neo4jPropertyGraphStore

# 配置
Settings.llm = OpenAI(model="gpt-4o-mini", temperature=0)
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")

# 连接Neo4j图存储
graph_store = Neo4jPropertyGraphStore(
    username="neo4j",
    password="password",
    url="bolt://localhost:7687",
    database="neo4j"
)

# 加载文档
documents = SimpleDirectoryReader("./data").load_data()

# 构建属性图索引
index = PropertyGraphIndex.from_documents(
    documents,
    property_graph_store=graph_store,
    show_progress=True,
)

# 查询
query_engine = index.as_query_engine(
    include_text=True,
    response_mode="tree_summarize",
    similarity_top_k=5,
)

response = query_engine.query(
    "知识图谱在企业中的主要应用场景有哪些？"
)
print(response)

3.2 自定义实体抽取

from llama_index.core.indices.property_graph import (
    SchemaLLMPathExtractor,
    DynamicLLMPathExtractor,
)

# 方案A: Schema约束的抽取（更精确）
schema_extractor = SchemaLLMPathExtractor(
    llm=Settings.llm,
    possible_entities=[
        "Person", "Company", "Product", "Technology",
        "Location", "Event", "Policy"
    ],
    possible_relations=[
        "WORKS_AT", "FOUNDED", "PRODUCES", "USES",
        "LOCATED_IN", "PARTICIPATED_IN", "GOVERNS"
    ],
    strict=False,  # 允许schema外的实体/关系
)

# 方案B: 动态抽取（更灵活）
dynamic_extractor = DynamicLLMPathExtractor(
    llm=Settings.llm,
    max_triplets_per_chunk=20,
    num_workers=4,
)

# 使用自定义抽取器构建索引
index = PropertyGraphIndex.from_documents(
    documents,
    property_graph_store=graph_store,
    kg_extractors=[schema_extractor],
    show_progress=True,
)

3.3 混合检索

from llama_index.core.indices.property_graph import (
    LLMSynonymRetriever,
    VectorContextRetriever,
)

# 检索器1: LLM同义词扩展检索
synonym_retriever = LLMSynonymRetriever(
    index.property_graph_store,
    llm=Settings.llm,
    include_text=True,
    synonym_prompt=(
        "给定以下查询，生成相关的同义词和相关术语列表，"
        "用于在知识图谱中搜索相关实体。\n"
        "查询: {query_str}\n"
        "同义词: "
    ),
)

# 检索器2: 向量相似度检索
vector_retriever = VectorContextRetriever(
    index.property_graph_store,
    embed_model=Settings.embed_model,
    include_text=True,
    similarity_top_k=5,
)

# 组合检索器
query_engine = index.as_query_engine(
    sub_retrievers=[synonym_retriever, vector_retriever],
    response_mode="tree_summarize",
)

四、自定义GraphRAG实现

4.1 实体抽取模块

def extract_entities_and_relations(text, llm_client, schema=None):
    """从文本中抽取实体和关系"""

    schema_hint = ""
    if schema:
        schema_hint = f"""
可用的实体类型: {', '.join(schema.get('entity_types', []))}
可用的关系类型: {', '.join(schema.get('relation_types', []))}
"""

    prompt = f"""从以下文本中抽取所有实体和关系。
{schema_hint}

文本:
{text}

输出JSON格式:
{{
  "entities": [
    {{"id": "entity_唯一标识", "name": "实体名称", "type": "实体类型",
      "description": "一句话描述"}}
  ],
  "relations": [
    {{"source": "源实体id", "target": "目标实体id", "type": "关系类型",
      "description": "关系描述", "weight": 0.9}}
  ]
}}

要求:
1. 实体去重（相同实体只出现一次）
2. 关系必须有明确的文本依据
3. weight表示关系的置信度(0-1)
4. description用一句话描述实体或关系的含义"""

    response = llm_client.generate(prompt, temperature=0)
    return json.loads(response)

4.2 社区检测与摘要

import networkx as nx
from cdlib import algorithms

def detect_communities(graph_data):
    """使用Leiden算法检测社区"""
    G = nx.Graph()

    for entity in graph_data["entities"]:
        G.add_node(entity["id"], **entity)

    for rel in graph_data["relations"]:
        G.add_edge(
            rel["source"], rel["target"],
            weight=rel.get("weight", 1.0),
            type=rel["type"]
        )

    # Leiden社区检测
    communities = algorithms.leiden(G)

    return communities

def generate_community_summaries(communities, graph_data, llm_client):
    """为每个社区生成摘要报告"""
    summaries = []

    for i, community in enumerate(communities.communities):
        # 收集社区内的实体和关系
        community_entities = [
            e for e in graph_data["entities"]
            if e["id"] in community
        ]
        community_relations = [
            r for r in graph_data["relations"]
            if r["source"] in community and r["target"] in community
        ]

        # 用LLM生成摘要
        context = format_community_data(community_entities, community_relations)
        prompt = f"""基于以下知识图谱社区的实体和关系，生成一份结构化摘要报告。

社区数据:
{context}

报告格式:
1. 主题概述（2-3句话）
2. 核心实体列表
3. 关键关系和发现
4. 重要性评估（高/中/低）"""

        summary = llm_client.generate(prompt)
        summaries.append({
            "community_id": i,
            "members": list(community),
            "summary": summary,
            "size": len(community)
        })

    return summaries

4.3 查询引擎

class GraphRAGQueryEngine:
    def __init__(self, graph_store, community_summaries, llm_client, embed_model):
        self.graph_store = graph_store
        self.summaries = community_summaries
        self.llm = llm_client
        self.embed = embed_model

    def local_search(self, query, top_k=5):
        """局部搜索: 从相关实体出发遍历子图"""

        # 1. 识别查询中的实体
        entities = self._extract_query_entities(query)

        # 2. 在图谱中检索相关子图
        subgraph = self._retrieve_subgraph(entities, max_hops=2)

        # 3. 同时检索相关文本块
        text_chunks = self._vector_search(query, top_k=top_k)

        # 4. 组装上下文
        context = self._assemble_context(subgraph, text_chunks)

        # 5. LLM生成回答
        answer = self.llm.generate(f"""
基于以下知识图谱和文本信息回答问题。

知识图谱上下文:
{context['graph']}

相关文本:
{context['text']}

问题: {query}

要求：基于提供的信息回答，标注信息来源（图谱/文本），不确定时说明。
""")
        return answer

    def global_search(self, query):
        """全局搜索: Map-Reduce社区摘要"""

        # Map阶段: 每个社区独立回答
        community_answers = []
        for summary in self.summaries:
            partial = self.llm.generate(f"""
基于以下社区知识摘要，回答问题（如不相关回答N/A）。

社区摘要:
{summary['summary']}

问题: {query}
""")
            if "N/A" not in partial:
                community_answers.append(partial)

        # Reduce阶段: 汇总所有社区答案
        if not community_answers:
            return "无法从现有知识中找到相关信息。"

        final = self.llm.generate(f"""
以下是从不同知识社区中获得的部分答案，请综合它们生成最终回答。

部分答案:
{chr(10).join(f'[来源{i+1}] {a}' for i, a in enumerate(community_answers))}

问题: {query}

要求: 综合所有来源，给出完整、连贯的回答。
""")
        return final

五、效果评估

5.1 对比基准

维度	向量RAG	GraphRAG	提升
单跳事实问答	85%	88%	+3%
多跳推理问答	45%	72%	+27%
全局摘要	30%	78%	+48%
对比分析	50%	75%	+25%
事实一致性	70%	85%	+15%
延迟	1-2s	2-5s	-50-150%
成本	低	中-高(索引构建)	+200-500%

5.2 适用场景判断

使用GraphRAG的场景：
├── 需要跨文档多跳推理
├── 需要全局主题概括
├── 实体关系是核心信息
├── 数据量中等（<10万文档）
└── 对准确性要求高于延迟

使用向量RAG的场景：
├── 简单的信息检索
├── 对延迟敏感（<1s）
├── 数据量极大（>100万文档）
├── 成本敏感
└── 信息相对独立（不需要关联推理）

混合方案（推荐）：
├── 向量RAG作为基础层
├── GraphRAG作为增强层
├── 路由器根据查询类型分发
└── 简单问题走向量，复杂问题走图谱

六、总结

GraphRAG代表了RAG技术的重要演进方向。通过引入知识图谱的结构化表达和社区检测的层次化组织，它显著提升了LLM在多跳推理和全局理解上的能力。核心挑战在于索引构建的成本和延迟，但对于知识密集型场景，这些投入通常能够获得显著的质量回报。

---

Maurice | [email protected]