1.3 GraphRAG 快速入门实战

1.3 GraphRAG 快速入门实战 — 知识图谱增强检索从零到一

本节导读：本节通过端到端实战项目，带你从零搭建基于知识图谱的检索增强生成系统。你将完成环境搭建、知识图谱构建、图语义检索和 LLM 问答四个核心环节。

学习目标

• 搭建 GraphRAG 开发环境并完成依赖安装
• 构建基于三元组的小型知识图谱
• 实现基于实体和关系的图语义检索
• 将检索结果注入 LLM 提示词生成高质量回答
• 理解 GraphRAG 的完整数据流：文本→图谱→检索→生成

核心概念

GraphRAG 实战闭环

GraphRAG 的核心工作流程概括为四步：抽取→构建→检索→生成。传统 RAG 将文本切成无语义关联的 chunk 后丢失了结构关系，而 GraphRAG 通过构建知识图谱保留了实体之间的关联信息，使检索结果具备关系推理能力。

前两节我们从理论和架构层面理解了 GraphRAG。本节通过可运行的代码帮你建立工程直觉。

环境准备 / 前置知识

系统要求

• Python 3.10+，8GB+ 内存，网络连接

依赖安装

python -m venv graphrag-env && source graphrag-env/bin/activate
pip install networkx numpy scikit-learn sentence-transformers openai
​

分步实战

步骤 1：从文本中抽取知识三元组

知识图谱的起点是从非结构化文本提取"实体-关系-实体"三元组。快速入门使用轻量规则方法，生产环境应替换为 BERT-based 模型。

import re
from typing import List, Tuple

class SimpleTripleExtractor:
    """轻量级三元组抽取器"""
    ENTITY_PATTERNS = {
        "ORG": r"(腾讯|阿里巴巴|字节跳动|百度|华为|微软|OpenAI|谷歌)",
        "TECH": r"(人工智能|机器学习|深度学习|大语言模型|LLM|知识图谱|RAG)",
        "PRODUCT": r"(ChatGPT|Claude|文心一言|通义千问)",
    }
    RELATION_TEMPLATES = [
        (r"(.{2,10})是(.{2,10})的(子公司|母公司)", lambda m: (m.group(1), "属于", m.group(2))),
        (r"(.{2,10})(开发|推出)了(.{2,10})", lambda m: (m.group(1), "开发", m.group(3))),
        (r"(.{2,10})使用(.{2,10})技术", lambda m: (m.group(1), "使用", m.group(2))),
        (r"(.{2,10})专注于(.{2,10})", lambda m: (m.group(1), "专注领域", m.group(2))),
    ]
    
    def extract(self, text: str) -> List[Tuple[str, str, str]]:
        triples = []
        for pattern, extractor in self.RELATION_TEMPLATES:
            for match in re.finditer(pattern, text):
                try: triples.append(extractor(match))
                except: continue
        entities_found = []
        for etype, pattern in self.ENTITY_PATTERNS.items():
            for match in re.finditer(pattern, text):
                entities_found.append(match.group(1))
        sentences = re.split(r'[。！？]', text)
        for sent in sentences:
            found = [e for e in entities_found if e in sent]
            for i in range(len(found)):
                for j in range(i+1, len(found)):
                    t = (found[i], "相关", found[j])
                    if t not in triples: triples.append(t)
        return triples

extractor = SimpleTripleExtractor()
text = "腾讯是中国领先的互联网科技公司，专注于人工智能和云计算。腾讯开发了知识图谱检索系统。OpenAI推出了ChatGPT。百度推出了文心一言大模型。阿里巴巴使用深度学习技术。"
triples = extractor.extract(text)
for h, r, t in triples:
    print(f"  ({h}) --[{r}]--> ({t})")
​

步骤 2：构建知识图谱

import networkx as nx
from collections import defaultdict

class KnowledgeGraphBuilder:
    def __init__(self):
        self.graph = nx.DiGraph()
    
    def add_triples(self, triples):
        weights = {"开发":1.0, "推出":0.9, "使用":0.7, "专注领域":0.8, "相关":0.5}
        freq = defaultdict(int)
        for h, r, t in triples: freq[h] += 1; freq[t] += 1
        for head, rel, tail in triples:
            if not self.graph.has_node(head):
                self.graph.add_node(head, frequency=freq[head])
            if not self.graph.has_node(tail):
                self.graph.add_node(tail, frequency=freq[tail])
            self.graph.add_edge(head, tail, relation=rel, weight=weights.get(rel, 0.6))

builder = KnowledgeGraphBuilder()
builder.add_triples(triples)
print(f"图谱：{builder.graph.number_of_nodes()} 实体, {builder.graph.number_of_edges()} 关系")
​

步骤 3：实现混合语义检索

GraphRAG 的核心优势——融合图结构检索和向量语义检索两路结果。

import numpy as np
from sentence_transformers import SentenceTransformer

class GraphSemanticRetriever:
    def __init__(self, graph):
        self.graph = graph
        self.encoder = SentenceTransformer("paraphrase-multilingual-MiniLM-L12-v2")
        self.entity_vecs = {}
        for node in graph.nodes():
            nb = list(graph.successors(node)) + list(graph.predecessors(node))
            self.entity_vecs[node] = self.encoder.encode(f"{node} " + " ".join(nb[:5]))
    
    def graph_search(self, entity, max_hops=2):
        if entity not in self.graph: return []
        results, visited, cur = [], {entity}, [entity]
        for hop in range(1, max_hops+1):
            nxt = []
            for n in cur:
                for nb in list(self.graph.successors(n)) + list(self.graph.predecessors(n)):
                    if nb not in visited:
                        visited.add(nb); nxt.append(nb)
                        e = self.graph[n][nb] if self.graph.has_edge(n,nb) else self.graph[nb][n]
                        results.append({"entity":nb, "relation":e.get("relation","?"), "hop":hop, "score":e.get("weight",0.5)/hop})
            cur = nxt
        results.sort(key=lambda x: x["score"], reverse=True)
        return results
    
    def vector_search(self, query, top_k=10):
        qv = self.encoder.encode(query)
        scores = [(e, float(np.dot(qv,v)/(np.linalg.norm(qv)*np.linalg.norm(v)+1e-8))) for e,v in self.entity_vecs.items()]
        return sorted(scores, key=lambda x: x[1], reverse=True)[:top_k]
    
    def hybrid_search(self, query, entity=None, gw=0.6, vw=0.4, top_k=10):
        scores = {}
        if entity:
            for rank, r in enumerate(self.graph_search(entity)):
                scores[r["entity"]] = scores.get(r["entity"],0) + gw/(60+rank+1)
        for rank, (e, s) in enumerate(self.vector_search(query)):
            scores[e] = scores.get(e, 0) + vw/(60+rank+1)
        return sorted(scores.items(), key=lambda x: x[1], reverse=True)[:top_k]

retriever = GraphSemanticRetriever(builder.graph)
for r in retriever.hybrid_search("哪些公司在做AI", entity="腾讯")[:5]:
    print(f"  {r[0]} (分数: {r[1]:.4f})")
​

步骤 4：组装上下文并调用 LLM

class GraphRAGPipeline:
    def __init__(self, graph, retriever):
        self.graph, self.retriever = graph, retriever
    
    def answer(self, question, entity=None):
        results = self.retriever.hybrid_search(question, entity)
        ctx_parts = ["知识图谱检索结果："]
        for i, (ent, score) in enumerate(results[:10], 1):
            rels = []
            for nb in self.graph.successors(ent):
                rels.append(f"  {ent} --[{self.graph[ent][nb].get('relation','?')}]--> {nb}")
            for nb in self.graph.predecessors(ent):
                rels.append(f"  {nb} --[{self.graph[nb][ent].get('relation','?')}]--> {ent}")
            ctx_parts.append(f"{i}. {ent}:\n" + "\n".join(rels) if rels else f"{i}. {ent}")
        context = "\n".join(ctx_parts)
        prompt = f"你是知识问答助手。根据以下知识图谱信息回答问题。信息不足请如实说明。\n\n【知识图谱】\n{context}\n\n【问题】{question}\n\n请给出准确回答。"
        # 实际使用: from openai import OpenAI; client.chat.completions.create(...)
        return {"question": question, "context": context, "answer": "[配置LLM后运行]"}

pipeline = GraphRAGPipeline(builder.graph, retriever)
result = pipeline.answer("腾讯在AI领域做了什么", entity="腾讯")
​

扩展对比：GraphRAG vs 传统 RAG

传统 RAG 在处理开放式问题（如"介绍一下深度学习"）时表现优秀，但在需要关系推理的场景（如"A公司的CTO之前在哪里工作"）中容易丢失关键信息。GraphRAG 通过图谱结构天然保留了实体间的关系链路，能回答需要多跳推理的复杂问题。两种方案并非替代关系，混合使用效果最佳。

常见问题 FAQ

Q1：GraphRAG 和传统 RAG 在代码层面的核心区别？

A：核心区别在检索策略。传统 RAG 调用向量数据库的 search 接口检索文本 chunk；GraphRAG 调用图遍历接口检索实体-关系对。后续 LLM 调用逻辑完全相同。

Q2：规则抽取方法能用于生产环境吗？

A：不能。仅适合演示。生产环境应使用 BERT-NER + BERT-RE 或大模型 few-shot 抽取。详见本教程 2.1 节。

Q3：如何处理实体名称不一致？

A：常见策略：同义词合并、实体对齐（字符串相似度+上下文）、实体规范化。推荐使用实体链接技术。

Q4：为什么混合检索优于单一检索？

A：图检索擅长结构化查询（如"A和B什么关系"），向量检索擅长开放式查询。混合检索通过 RRF 融合算法统一排序，实测准确率提升 15-30%。

Q5：GraphRAG 适合哪些场景？

A：特别适合知识密集型场景：企业知识库、技术文档检索、专业领域问答、多实体关系分析。数据量小且关系简单的场景用传统 RAG 即可。

最佳实践与避坑

• 从小数据集开始：先 10-50 条文本跑通全流程，再逐步扩大
• 中文场景选 BGE 或 M3E 模型：比 MiniLM 效果好很多
• 缓存实体向量：预计算避免每次查询重复编码
• 实体链接是关键瓶颈：查询实体识别不准会导致图检索失效
• 不要跳过评估：对比 GraphRAG 和纯 LLM 回答质量差异

本节小结

本节通过端到端实战走完了 GraphRAG 核心流程：抽取三元组→构建图谱→混合检索→组装上下文→LLM 生成。核心收获：GraphRAG 检索粒度是实体和关系，天然具备结构化语义；混合检索显著优于单一策略；工程难点在模块间数据流转的一致性保障。

下一节进入第 2 章知识图谱构建模块，深入学习实体识别与关系抽取的生产级方案。

延伸阅读

• 官方文档：NetworkX 图算法文档最新版本
• 相关章节：本教程 2.1 节实体识别与关系抽取
• 深入学习：SentenceTransformers 多语言模型选型指南

关键词：GraphRAG, 知识图谱, 实体关系抽取, 混合检索, NetworkX, 快速入门, 实战, RAG对比
难度：入门
预计阅读：30 分钟