从朴素检索到图RAG：三种实现范式的选型与代码实战

核心洞察

这篇长文最值得看的地方，是作者把自己从2023年开始搭RAG系统的实战笔记全摊开了，从最基础的五步流程到GraphRAG的图构建，中间还夹着能直接跑的代码。我比较欣赏的一点是，他没有把RAG吹成万能药，后面专门有一节讲局限和替代方案。如果你正在纠结要不要上RAG，或者上了之后效果不达预期，这份材料能帮你少走不少弯路。

核心结论

1. RAG（检索增强生成）结合信息检索和文本生成的技术，通过检索相关文档来增强大型语言模型的生成能力。由Lewis等人在2020年于Facebook AI Research正式提出，核心思路是让LLM在需要时从外部知识库检索信息，而非将所有知识塞入模型参数。

2. RAG主要解决LLM的三个根本问题：时效性（训练截止后的事件）、私有知识（内部wiki、病历、合同等不在训练数据中）和幻觉（通过"只根据检索到的文档回答"约束加引用机制缓解）。

3. RAG已演进为四代：Naive（简单向量检索）、Advanced（检索前中后全流程优化）、Modular（路由、循环、工具集成）和Graph（实体关系图表示），各代在准确率、灵活性与构建成本之间权衡。

4. 切块参数建议：chunk_size为256-1024 tokens，chunk_overlap为chunk_size的10-20%；法律文档按条款切、技术文档按章节切、FAQ按问答对切、代码按函数/类切。

5. 选择embedding模型的决策依据是MTEB排行榜分数、领域契合度和成本/延迟；多语言场景可选BAAI/bge-m3、intfloat/multilingual-e5-large，英文闭源可选OpenAI text-embedding-3-large等。

目录

第一部分：基础

1. 什么是RAG？
2. 为什么要用RAG？
3. RAG的三代演进：Naive → Advanced → Modular → Graph

• 开始之前：术语表

第二部分：实现

4. Naive RAG：基础五步
5. 示例一：Naive RAG（LangChain + Chroma）
6. Advanced RAG深入
7. 示例二：Advanced RAG（混合检索 + 重排序 + 查询变换 + 引用 + 自评估）
8. Graph RAG深入
9. 示例三：Graph RAG（实体关系抽取 + 图遍历）

第三部分：运维与决策

10. 评估方法与指标
11. 近期趋势（2024-2026）
12. 局限与替代方案
13. 三方对比与选型指南
14. 生产环境检查清单
15. 环境搭建

第四部分：超越RAG

16. LLM Wiki：一个积累知识而非检索知识的系统
17. 示例四：LLM Wiki（自维护的wiki代理）

第一部分：基础

1. 什么是RAG？

RAG（Retrieval-Augmented Generation，检索增强生成）把检索、增强、生成三件事串在一起。这个概念是2020年Lewis等人在Facebook AI Research正式提出的。

核心思想一句话就能说完：

别把所有知识都塞进LLM的参数里，让它需要的时候自己去外面拿。

LLM是推理器，外部知识库是参考书。拿考试打比方，RAG把闭卷考试变成了开卷考试。

[用户提问] → [检索] → top-K相关文档 → [问题+文档] → LLM → [有据可查的回答]

2. 为什么要用RAG？

单独用LLM会碰到三个根本性问题，RAG就是冲着它们来的。

2.1 时效性

LLM不知道训练截止之后发生的事。公司政策昨天刚改，模型不可能知道。用了RAG，刷新一下索引，改动立刻生效。

2.2 私有知识

内部wiki、客服工单、病历、法律合同，这些东西都不在模型的训练数据里。训进去成本高得离谱，安全上还一堆麻烦。RAG把数据留在模型外面，权重完全不碰。

2.3 幻觉

LLM不知道答案的时候会编一个听起来很像那么回事的。RAG用“只根据检索到的文档回答”这条约束加上引用机制来缓解，幻觉少了，答案也能验证。

2.4 附带好处

• 成本：小模型配上好的RAG，往往比单用大模型便宜。
• 权限：可以在检索环节做按用户的权限控制。
• 可审计：每一条回答用了哪些文档，都能记录下来。受监管行业刚需。

3. RAG的三代演进：Naive → Advanced → Modular → Graph

2020年以来RAG进化得很快。学术界和工业界比较公认的分类是这样的：

3.1 Naive RAG（第一代，2023年初）

“把问题原样拿去做向量检索，结果直接丢给LLM。”

最简单的形态。切块、嵌入、相似度检索、生成。大部分入门教程就是这个。

局限：问题有歧义、同义词缺失、多跳推理、关键词匹配更重要的场景，都表现不好。

3.2 Advanced RAG（第二代，2023-2024）

“检索的每个阶段都做得更聪明，检索前、检索中、检索后。”

检索前：语义切块、元数据增强、查询重写/扩展/分解、HyDE。检索中：混合检索（稠密+稀疏）、多向量、ColBERT。检索后：重排序、上下文压缩、MMR、强制引用。

核心思路是“更聪明的检索”。数据表示还是块加嵌入。

3.3 Modular RAG（第2.5代）

“每个阶段都模块化、可替换，让系统自由路由、循环、调工具。”

路由器按问题类型分发给不同的子RAG，结果不够就循环再来，外部工具（SQL/API/网页）也能接进来。Self-RAG、CRAG、Adaptive RAG、Agentic RAG都归在这一类。

3.4 Graph RAG（以关系为中心的演进）

“把文档表示成实体关系图，而不是块。”

LLM从文档里抽取（实体、关系、实体）三元组，存进图数据库。查询时遍历图来收集多跳信息。微软GraphRAG（2024）、LightRAG（2024）、Neo4j-LangChain集成都是典型代表。

强项：多跳推理、关系本身就是重点的领域。弱项：图构建成本、schema设计开销。

一览表

开始之前：术语表

进入实现之前，先把这份文档里会反复出现的术语和工具名列一遍。已经知道的跳过，后面碰到卡壳的地方再回来查。

A. 基础概念

• LLM（大语言模型）：GPT、Claude、Gemini这些。在这里扮演生成答案的推理器角色。
• token：LLM处理的最小单位。一个英文单词大约1到1.5个token，一个汉字大约1到3个token。模型的上下文上限用token表示（比如“Claude 200K tokens”）。
• 上下文窗口：模型单次输入能接收的最大token数。
• embedding（嵌入）：把文本转成数值向量（比如一个1024维的浮点数组）。检索依赖的性质是“意思相近，向量就相近”。
• 向量：在这里就是一组数字。把一句话嵌入之后就得到一个N维向量。
• 向量数据库：专门存嵌入向量并快速找到相似向量的数据库。比如Chroma、Pinecone。
• 相似度：两个向量有多接近。最常用的是余弦相似度，越接近1越相似。
• top-k：前k个结果。“top-5文档”就是最相关的5篇。
• 块（chunk）：长文档切出来的一片，作为检索的基本单位。
• chunk_size / chunk_overlap：单个块的大小 / 相邻块之间重叠多少。

B. HuggingFace模型路径：BAAI/bge-m3是什么？

HuggingFace是一个共享AI模型的平台，可以理解成机器学习界的GitHub。模型用组织名/模型名来标识。

BAAI/bge-m3 这串东西的意思是：一个叫 BAAI 的机构发布的、名字叫 bge-m3一种先进的嵌入模型，能够生成1024维的语义嵌入向量，用于将文本转换为机器可理解的数值表示。 的模型。

常见模型家族

• BGE（BAAI General Embedding），智源的 embedding 产品线。bge-m3（多语言）、bge-large-en（英文）、bge-reranker（重排）等等。
• E5，微软研究院的 embedding。multilingual-e5-large、e5-mistral-7b-instruct 等。
• GTE，阿里的 embedding 产品线。
• ColBERT，late-interaction 检索模型。

代码里写 HuggingFaceEmbeddings(model_name="BAAI/bge-m3") 的时候，模型只会从 HuggingFace 下载一次，缓存在 ~/.cache/huggingface，之后再跑就直接读缓存。

C. 检索算法与技术

• BM25一种基于概率信息检索理论的文本检索模型，通过词频和逆文档频率对文档进行排名，考虑词项的相关性和稀有性。，信息检索里的标准关键词打分函数，1990 年代定型。它算的是“这个词在这篇文档里出现得多不多、有没有辨识度”。对精确标识符（错误码、专有名词）特别管用。
• Dense / Sparse Retrieval，Dense 是向量（稠密）检索，Sparse 是 BM25 这类基于词的检索。后者叫“稀疏”是因为它的表示里大部分是零。
• ANN（近似最近邻），在几百万个向量里快速找到最近邻的算法，结果是近似的，但快。常见变体有 HNSWHierarchical Navigable Small World - a graph-based algorithm for approximate nearest neighbor search that provides efficient similarity search in high-dimensional spaces.、IVF-PQ。基本上每个向量数据库内部都在用。
• Bi-encoder vs Cross-encoder，Bi-encoder 把问题和文档分别编码再比较，快，用于第一轮召回。Cross-encoder 把两者一起喂进去算分，准，慢，用于重排。
• RRF（Reciprocal Rank Fusion），合并多个检索器结果的标准做法，把每个检索器排名的倒数加起来。详见 §6.4。
• MMR（Maximal Marginal Relevance），给 top-k 加多样性，防止几个几乎一样的 chunk 占满名额。
• HyDE（Hypothetical Document Embeddings），先让 LLM 编一个假答案，拿这个答案去做检索。利用的是“答案到答案”通常比“问题到答案”更近。

D. 库与框架

• LangChain，LLM 应用框架（Python/JS）。本文所有例子的骨架。
• LCEL（LangChain Expression Language），LangChain 的 | 管道语法。像 prompt | llm | parser 这样把组件竖着串起来。跟 Unix shell 里 cat file | grep ... | wc -l 一个思路。
• Runnable，LCEL 里能用 | 串起来的组件的统一接口。RunnablePassthrough() 把输入原样传到下一级。
• LlamaIndex，LangChain 的主要竞争对手，在索引和知识图谱上更专。
• sentence-transformers，最常用的 Python 库，embedding 和 cross-encoder 都支持。
• NetworkX，Python 的内存图库。例 3 里拿它顶替真正的图数据库。
• rank_bm25，实现 BM25 的小型 Python 包。

E. 向量数据库 / 图数据库

• Cypher，Neo4j 的查询语言。比如 MATCH (p:Person)-[:WORKS_AT]->(c:Company) RETURN p, c。可以理解成图版的 SQL。

F. 评估与基准

• MTEB（Massive Text Embedding Benchmark），HuggingFace 上的 embedding 模型综合排行榜。挑 embedding 先看这里。
• RAGAS，自动化 RAG 评估框架。测 Faithfulness、Answer Relevance、Context Precision 等指标，走 LLM-as-judge 的路子。
• TruLens / DeepEval / ARES，RAGAS 的替代或补充。
• LLM-as-judge，让另一个 LLM 来判断“这个答案好不好”，以此作为评估方法。
• Faithfulness / Hallucination，Faithfulness 指答案有没有紧扣检索到的上下文；Hallucination 指没有依据、编得像模像样的答案。

G. 工具与服务（第四部分会重点用到）

• Obsidian，基于 markdown 的个人知识管理应用。支持 wikilink [[页面名]] 和图视图。免费。
• Web Clipper，浏览器扩展，把网页转成 markdown 存进 Obsidian。
• Dataview，Obsidian 插件，用类 SQL 语法查询页面 YAML frontmatter，生成动态表格和列表。
• Marp，用 markdown 做幻灯片的工具，有 Obsidian 插件。
• qmd，针对 markdown 文件夹的本地搜索引擎（BM25 + 向量 + LLM 重排），提供 CLI 和 MCP server。
• Claude Code / Codex，直接操作文件系统和 shell 的 agentic 编程工具，跟 LLM Wiki 天然契合。
• CLAUDE.md / AGENTS.md，给上面这些 agentic 工具读的项目使用说明。一份自然语言写的元文档，描述“这个仓库是这么组织的，请按这个方式来干活”。详见 §16.2。
• Microsoft GraphRAG，微软研究院官方的 GraphRAG 实现。
• LightRAG，港大出的轻量版 GraphRAG。

H. 常见缩写

4. Naive RAG：最基础的五步

最简单的 RAG 流程：

1. 加载：从 PDF、网页、数据库、Notion 等处收集原始素材。
2. 切块：把长文档切成适合检索的小块。
3. 向量化：把每个 chunk 转成向量。
4. 检索：找出跟问题 embedding 最接近的 K 个 chunk。
5. 生成：把检索到的文本塞进 prompt，让 LLM 回答。

切块参数建议

选 embedding 模型（2026 年的情况）

注：如果 BAAI/bge-m3 看着眼生，先翻一下术语表 §B，那是 HuggingFace 的模型路径格式，形如 机构名/模型名。

• 多语言 / 韩语：BAAI/bge-m3、intfloat/multilingual-e5-large、nlpai-lab/KURE-v1
• 英文 / 闭源：OpenAI text-embedding-3-large、Cohere embed-v3、Voyage voyage-3
• 决策依据：MTEB 排行榜分数 + 领域契合度 + 成本/延迟。

例 1：Naive RAG（LangChain + Chroma）

场景是公司内部 HR wiki。五步流程的最简形态。环境配置见 §15。

需要 ANTHROPIC_API_KEY。

"""example_1_naive_rag.py，最小可用的五步 Naive RAG"""
from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# ── 1) 文档（生产环境里从 PDF/Notion/数据库加载）──────────
raw = [
    Document(page_content=(
        "ACME 年假政策：全职员工入职满一年享 15 天年假。满 3 年 16 天，满 5 年 18 天，"
        "满 10 年 20 天。未休完的年假可顺延至次年 6 月 30 日，逾期作废。"),
        metadata={"source": "HR/leave_policy_v3.md"}),
    Document(page_content=(
        "特殊假：本人结婚 5 天，子女结婚 1 天，配偶生育 10 天，"
        "本人或配偶父母去世 5 天，祖父母去世 3 天。父母 60 或 70 大寿等家庭事件"
        "不算特殊假，需走年假。"),
        metadata={"source": "HR/special_leave.md"}),
    Document(page_content=(
        "远程办公：全职员工每周可居家办公两次，需提前获得主管批准。"
        "周二、周四不建议远程（全公司例会）。新员工前 3 个月必须每天到岗。"),
        metadata={"source": "HR/remote_work_policy.md"}),
]

# ── 2) 切块 ─────────────────────────────────────────────
splitter = RecursiveCharacterTextSplitter(
    chunk_size=300, chunk_overlap=50,
    separators=["\n\n", "\n", "。", " ", ""])
chunks = splitter.split_documents(raw)

# ── 3) 向量化 + 向量库 ──────────────────────────────────
emb = HuggingFaceEmbeddings(
    model_name="BAAI/bge-m3",
    encode_kwargs={"normalize_embeddings": True})
vectordb = Chroma.from_documents(chunks, emb, collection_name="acme_hr")

# ── 4) 检索器 ───────────────────────────────────────────
retriever = vectordb.as_retriever(search_kwargs={"k": 3})

# ── 5) Prompt + LLM + 链 ────────────────────────────────
prompt = ChatPromptTemplate.from_messages([
    ("system",
     "你是 ACME 的 HR 助手。只根据下面的[参考文档]回答。"
     "如果找不到答案，回复“我在提供的文档中找不到答案。”"
     "每条结论末尾用 [文件名] 标注引用。"),
    ("human", "[参考文档]\n{context}\n\n[问题]\n{question}")])

llm = ChatAnthropic(model="claude-opus-4-7", temperature=0)

def fmt(docs):
    return "\n\n".join(f"[{d.metadata['source']}]\n{d.page_content}" for d in docs)

# LCEL：LangChain 表达式语言，用 `|` 把组件串起来。
# 和 Unix shell 里 `cat file | grep ... | wc -l` 一个思路。
chain = ({"context": retriever | fmt, "question": RunnablePassthrough()}
         | prompt | llm | StrOutputParser())

# ── 跑起来 ──────────────────────────────────────────────
for q in [
    "如果我父母 60 大寿正好在我入职第一年，我能休几天假？",
    "工龄 7 年的员工年假多少天？",
    "新员工可以居家办公吗？",
    "公司管午饭吗？",  # 文档里没有 → 应该拒答
]:
    print(f"\n━━━ 问：{q}\n> {chain.invoke(q)}")

这段代码说明了什么

• 五步流程一屏就能放下。
• 你能亲眼验证“文档里没有的问题”会被正确拒答。
• 多语言 embedding（bge-m3）能处理中英混排的内容。

局限（这个例子里就能感受到）

• 同义词、改写很弱，“薪资”和“薪酬”对不上。
• 关键词是精确标识符时会漏，比如“ERR_404”。
• 多跳问题，比如“最近请假最多的那个员工的经理是谁？”，单个 chunk 里根本答不出来。
• 检索错了，答案就跟着错。

→ 下一步，Advanced RAG，就是逐条解决这些局限的。

6. Advanced RAG 深入

Advanced RAG 在 Naive RAG 的检索管线上，把前、中、后三个阶段各自加强。数据表示还是 chunk + embedding，但每个阶段都换上了更锋利的技术，检索准确率和答案质量会有实打实的提升。

                          ┌─ 检索前 ────────────┐
原始文档 ─→ 语义切块 ────→│ 元数据增强           │
                          │ Contextual Embedding │
                          └──────────┬───────────┘
                                     ▼
用户问题 ─→ 查询改写 ─→ 混合检索 ──→ 重排序 ──→ 上下文压缩 ─→ prompt ─→ LLM ─→ 答案
            (检索前)     (检索中)      (检索后)

6.1 语义切块

固定长度切块不管语义边界。语义切块用 embedding 相似度的突变点当边界，切出来的单元更自然。

from langchain_experimental.text_splitter import SemanticChunker
from langchain_community.embeddings import HuggingFaceEmbeddings

splitter = SemanticChunker(
    HuggingFaceEmbeddings(model_name="BAAI/bge-m3"),
    breakpoint_threshold_type="percentile",  # 也可以 "standard_deviation"、"interquartile"
    breakpoint_threshold_amount=95)
chunks = splitter.create_documents([long_text])

代价更高，但在长报告或访谈记录这种语义单元不规则的材料上，效果值得。

6.2 Contextual Retrieval（Anthropic，2024）

在 embedding 每个 chunk 之前，先用 LLM 给它加上一段所在文档的简短摘要作为前缀。

原始 chunk：“营收同比增长 12%。”

检索失败率能降 35% 到 67%。索引阶段确实会多烧一些 token，但这是一次性开销，再配上 prompt caching，成本就压得很低了。

6.3 查询改写

用户问的原话往往不适合直接拿去检索，得先重塑一下。

HyDE 示例

from langchain_core.prompts import ChatPromptTemplate
from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-opus-4-7", temperature=0)
hyde_prompt = ChatPromptTemplate.from_messages([
    ("system", "写一段看似合理的回答，不用管事实是否准确。"),
    ("human", "{question}")
])

def hyde_search(question: str, retriever):
    hypothetical = (hyde_prompt | llm).invoke({"question": question}).content
    # 拿假想答案去做向量检索，通常比直接拿问题检索更准
    return retriever.invoke(hypothetical)

6.4 混合检索

BM25（关键词）+ 稠密向量，两个一起用。几乎总是比单用其中一个强。

from langchain_community.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever

bm25 = BM25Retriever.from_documents(chunks); bm25.k = 10
dense = vectordb.as_retriever(search_kwargs={"k": 10})
hybrid = EnsembleRetriever(
    retrievers=[bm25, dense],
    weights=[0.4, 0.6])  # 按领域调权重，代码、标识符多就加重 BM25

RRF（倒数排名融合）Reciprocal Rank Fusion，一种按排名位置而非原始分数进行融合的算法，避免强行比较BM25分数和向量余弦分数的量纲差异，调参负担更低。

合并多路检索结果的标准做法。$$\text{RRF}(d) = \sum_{i} \frac{1}{k + \text{rank}_i(d)}$$

k 一般取 60。EnsembleRetriever 内部干的就是类似的事。

6.5 重排序

第一轮检索先粗放地捞 50 到 100 个候选，重排序器再收紧到精确的 5 到 10 个。

实测效果：光加一个重排序，回答准确率通常能涨 10 到 20 个点。

6.6 上下文压缩

把检索到的文档里跟问题无关的部分剪掉。省 token，也缓解 Lost in the Middle。

from langchain.retrievers.document_compressors import LLMChainExtractor
from langchain.retrievers import ContextualCompressionRetriever

compressor = LLMChainExtractor.from_llm(llm)  # 只抽出回答问题需要的那部分
compressed = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=hybrid)

6.7 Self-RAG / CRAG（模块化 RAG 的起点）

让模型自己评估检索质量，再决定走哪条分支。

• Self-RAG（Asai 等，2023）：模型用 [Retrieve] token 决定要不要检索，用 [IsRel]、[IsSup]、[IsUse] 这些 token 自评检索结果和回答质量。
• CRAG（Yan 等，2024）：判断检索结果，正确 → 直接用，模棱两可 → 补充，错误 → 丢掉，去网上搜。

6.8 应对 Lost in the Middle

LLM 对上下文开头和结尾的信息用得好，中间那段容易漏。几个缓解办法：

• 把最重要的文档放在最前面或最后面。
• 别盲目加大 top-k，控制在 5 到 10。
• 用重排序把列表头部的顺序排准。
• 用上下文压缩把整体体积缩下来。

示例 2：Advanced RAG

混合检索 + 重排序 + 多查询改写 + 上下文压缩 + 强制引用 + 自评估，全塞进一条流水线。

"""example_2_advanced_rag.py，生产形态的 Advanced RAG"""
from __future__ import annotations
from typing import List
from dataclasses import dataclass

from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever
from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate
from sentence_transformers import CrossEncoder

# ════════════════════════════════════════════════════════════════
# 0. 数据：六份工程制度文档
# ════════════════════════════════════════════════════════════════
KB = [
    {"source": "ENG/Coding_Standards.md",
     "content": "Python 遵循 PEP 8 和 black 格式化。行宽 100。每个公开函数都要类型标注。"
                "函数名 snake_case，类名 PascalCase，常量 UPPER_SNAKE_CASE。"},
    {"source": "ENG/Code_Review_Policy.md",
     "content": "PR 合并至少需要 2 个批准，其中一个必须是资深工程师。安全相关改动还要"
                "安全团队额外批准。建议 PR 大小 400 行，超过 1000 行要求拆分。"
                "评审在两个工作日内完成。"},
    {"source": "ENG/Deploy_Process.md",
     "content": "生产发布安排在周二、周三、周四的 10:00 到 16:00。周五和节假日前一天"
                "禁止发布。发布前在 staging 验证 24 小时。紧急修复经 CTO 批准可绕过时间限制。"
                "发布后监控 30 分钟。"},
    {"source": "ENG/On_Call_Policy.md",
     "content": "值班按周轮换。响应目标：P1 十五分钟内，P2 一小时内。夜间（22:00 到 08:00）"
                "和周末值班有额外时薪补偿。休假需要提前安排换班。"},
    {"source": "ENG/Tech_Stack.md",
     "content": "后端标准是 Python 3.12 + FastAPI。数据库 PostgreSQL 16，缓存 Redis 7，"
                "消息队列 RabbitMQ。前端 TypeScript + React 18。云上用 AWS（ECS/RDS/S3）配 Terraform。"},
    {"source": "HR/Remote_Work.md",
     "content": "全职员工每周可以居家办公两天，需要主管批准。周二、周四不鼓励居家。"
                "新员工前三个月每天到岗。境外办公需要单独审批并做税务审查。"},
]

# ════════════════════════════════════════════════════════════════
# 1.

先把混合检索的两条路都搭起来。Dense 走语义，BM25 走关键词，两个检索器各取 8 条，再用权重 0.4 和 0.6 做融合。分块这块用 350 字符、70 重叠，分隔符从双换行一路降到空字符串，尽量让切出来的片段语义完整。

# 索引阶段：同时构建 Dense 和 BM25
def build_retrievers():
    docs = [Document(page_content=d["content"], metadata={"source": d["source"]}) for d in KB]
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=350, chunk_overlap=70,
        separators=["\n\n", "\n", ". ", " ", ""])
    chunks = splitter.split_documents(docs)

    emb = HuggingFaceEmbeddings(
        model_name="BAAI/bge-m3", encode_kwargs={"normalize_embeddings": True})
    vectordb = Chroma.from_documents(chunks, emb, collection_name="adv_rag")

    dense = vectordb.as_retriever(search_kwargs={"k": 8})
    bm25 = BM25Retriever.from_documents(chunks); bm25.k = 8

    hybrid = EnsembleRetriever(retrievers=[bm25, dense], weights=[0.4, 0.6])
    return hybrid

查询改写用 Multi-Query 的思路：让模型把原问题改写成 3 个意思不变但措辞和角度不同的检索 query，加上原问题一共 4 条，分别去检索。这样能覆盖用户表达和文档表述之间的措辞差异。

# 查询改写：用 Multi-Query 生成多个变体
multi_query_prompt = ChatPromptTemplate.from_messages([
    ("system", "把用户的问题改写成 3 条检索 query，保持语义不变，但换措辞、换角度。一行一条，不要编号。"),
    ("human", "{question}")
])

llm = ChatAnthropic(model="claude-opus-4-7", temperature=0)

def multi_queries(question: str) -> List[str]:
    raw = (multi_query_prompt | llm).invoke({"question": question}).content
    qs = [q.strip("-•123456789. ").strip() for q in raw.split("\n") if q.strip()]
    return [question] + qs[:3]  # 原问题 + 3 个变体

检索回来的候选文档可能几十条，直接塞给模型既浪费 token 又容易淹没关键信息。加一层 cross-encoder 重排，拿原问题逐条打分，按分数排序后去重，只留前 4 条。

# 重排：用 cross-encoder 对候选精排
class Reranker:
    def __init__(self, name="BAAI/bge-reranker-v2-m3"):
        self.m = CrossEncoder(name, max_length=512)

    def __call__(self, query: str, docs: List[Document], top_n=4) -> List[Document]:
        if not docs: return []
        scores = self.m.predict([(query, d.page_content) for d in docs])
        ranked = sorted(zip(scores, docs), key=lambda x: x[0], reverse=True)
        # 按 page_content 去重
        seen, out = set(), []
        for s, d in ranked:
            if d.page_content in seen: continue
            seen.add(d.page_content)
            d.metadata["rerank_score"] = float(s)
            out.append(d)
            if len(out) == top_n: break
        return out

生成环节强制引用。system prompt 里写死规则：每个事实后面必须跟 [编号] 引用，多个文档支持同一结论就都标上，文档里没写就直说没写。这样答案可追溯，也方便后面做忠实度检查。

# 答案生成：强制引用
answer_prompt = ChatPromptTemplate.from_messages([
    ("system",
     "你是 ACME 的工程助手。只依据下面的[参考文档]回答。\n"
     "规则：\n"
     "1) 每个事实结尾都要带 [编号] 引用。\n"
     "2) 多条文档支持同一结论时，全部标上，如 [1][3]。\n"
     "3) 文档里没有的，回答“文档中未提及”。\n"
     "4) 简洁，直击要点。"),
    ("human", "[参考文档]\n{context}\n\n[问题]\n{question}")
])

@dataclass
class Result:
    answer: str
    sources: List[Document]
    queries_used: List[str]

def make_ctx(docs: List[Document]) -> str:
    return "\n\n".join(
        f"[{i}] (来源: {d.metadata['source']})\n{d.page_content}"
        for i, d in enumerate(docs, 1))

def advanced_rag(question: str, hybrid, reranker) -> Result:
    # ① Multi-Query 改写
    queries = multi_queries(question)

    # ② 混合检索（每条变体各检一遍）
    candidates: List[Document] = []
    seen = set()
    for q in queries:
        for d in hybrid.invoke(q):
            key = d.page_content
            if key not in seen:
                seen.add(key); candidates.append(d)

    # ③ 重排（拿原问题精排）
    top = reranker(question, candidates, top_n=4)

    # ④ 生成答案（强制引用）
    msg = answer_prompt.invoke({"context": make_ctx(top), "question": question})
    ans = llm.invoke(msg).content
    return Result(answer=ans, sources=top, queries_used=queries)

最后加一道自检。用另一个 prompt 让模型判断答案里的每个事实是否都能在参考文档里找到支撑，全部有支撑返回 PASS，有一条找不到就 FAIL，第一行给结论，后面写理由。

# 自评估：自动检查忠实度
judge_prompt = ChatPromptTemplate.from_messages([
    ("system", "判断一个 RAG 答案的忠实度。如果[答案]里每个事实都能在[参考文档]中找到支撑，返回 PASS；"
               "有任何事实缺乏支撑，返回 FAIL。第一行写 PASS/FAIL，后面写理由。"),
    ("human", "[参考文档]\n{context}\n\n[答案]\n{answer}\n\n判定：")
])

def judge(res: Result) -> str:
    msg = judge_prompt.invoke({"context": make_ctx(res.sources), "answer": res.answer})
    return llm.invoke(msg).content

if __name__ == "__main__":
    hybrid = build_retrievers()
    rerank = Reranker()

    for q in [
        "Who needs to approve a security-related PR merge?",
        "Can I push a hotfix on Friday afternoon?",
        "Can a new hire apply for remote work?",
        "What do we use for DB and cache?",
        "What's on the company lunch menu?",  # not in docs
    ]:
        print(f"\n{'='*72}\nQ: {q}")
        r = advanced_rag(q, hybrid, rerank)
        print(f"\nQuery variants: {r.queries_used}")
        print(f"\nRetrieved + reranked top-{len(r.sources)}:")
        for i, d in enumerate(r.sources, 1):
            print(f"  [{i}] {d.metadata['source']:28s}  "
                  f"score={d.metadata.get('rerank_score',0):+.2f}")
        print(f"\n> Answer:\n{r.answer}")
        print(f"\nSelf-Eval:\n{judge(r)}")

跑完这几条 query，你大概就能感受到整套流程的差别了。最后那条午饭菜单根本不在文档里，看看系统怎么处理这种“查无此事”的情况，挺有意思的。

对比朴素方案，到底强在哪

这个阶段还搞不定的

• 多跳推理：“最近部署的那个团队，他们的代码评审政策是什么？”单个 chunk 里找不到答案。
• 关系型问题：“在 Project X 上干过的人里，谁和安全团队有协作？”需要遍历关系。

下一步的 Graph RAG，就是冲着这类问题来的。

8. Graph RAG 深入

Graph RAG 把文档表示成实体-关系图，在图上做检索，不再以 chunk 为单位。这个方向的关键节点是微软研究院 2024 年的论文《From Local to Global: A Graph RAG Approach》。

8.1 为什么要图：向量 RAG 的天花板

向量 RAG 只能找到“这个 chunk 和问题语义接近”。下面这几种它就吃力了：

• 多跳：“Project Alpha 的 PM 之前在哪儿工作？”需要（项目 → PM → 前雇主）链式遍历。
• 关系查询：“John 和 Jane 都参与过哪些项目？”需要两个人取交集。
• 全局理解：“公司里最有影响力的 5 个人是谁？”需要整张图的结构。
• 时间、因果链：“事件 A 怎么一步步影响到事件 C 的？”需要沿因果图遍历。

这些场景里，关系本身就是信息。答案可能压根没写在任何一份文档里，你得把多个来源的信息拼起来。

8.2 核心思路：索引阶段

从文档里抽取**（主语，关系，宾语）**三元组，存进图数据库（Neo4j、Memgraph、NetworkX 都行）。

文档："John is the CTO of ACME and leads Project Alpha. Jane is the security lead for Project Alpha."

抽取出的三元组：
  (John, IS_CTO_OF, ACME)
  (John, LEADS, Project Alpha)
  (Jane, IS_SECURITY_LEAD_OF, Project Alpha)

图中，“John”和“Jane”通过“Project Alpha”形成两跳连接。

基于LLM的抽取：给GPT/Claude一个提示词“从这份文档中抽取实体和关系”。这是最常用的做法。

NER + 关系抽取模型：用spaCy + REBEL这类专用模型。可以针对领域微调。

手工schema + 解析器：在医疗、法律这类高度结构化的领域里用。

实体消解

“John Smith”、“J. Smith”、“the CTO”可能指的是同一个人。把它们合并成一个节点，直接决定图谱质量的好坏。通常用基于embedding的聚类来处理。

光有一张图谱，面对“全局性”问题（理解整体）时是很弱的。微软GraphRAG的做法是：

1. 建完图之后，用Leiden算法检测社区（连接密集的节点群）。
2. 为每个社区预先生成LLM摘要。
3. 全局性问题 → 对社区摘要做map-reduce。
4. 局部性问题 → 从相关实体周围的子图来回答。

这就是普通图遍历RAG和GraphRAG之间的决定性差异。

8.4 查询阶段：局部 vs 全局

8.5 GraphRAG vs 向量RAG混合

实践中，混合图RAG将文档表示为实体-关系图，通过图遍历进行多跳推理的RAG变体，代表工作有微软GraphRAG、LightRAG。才是标准做法。

[查询]
    ├─→ 实体抽取 → 图遍历（基于关系的证据）
    └─→ 向量检索（语义证据）
         ↓
    [合并证据 + LLM回答]

• 图负责关系，向量负责内容。
• LangChain的Neo4jVector + GraphCypherQAChain组合是经典配置。
• LlamaIndex的KnowledgeGraphIndex + VectorStoreIndex组合也很流行。

8.6 主流实现对比

8.7 Graph RAG的真实成本

• 索引成本膨胀：每份文档都要过一遍LLM做三元组抽取，token账单很可观。
• Schema设计：决定“有哪些实体类型？有哪些关系类型？”本身就很难。
• 图运维：你需要为Neo4j这类独立数据库积累运维经验。
• 实体消解做不好，图谱就散架：同义词处理是成败关键。

所以实践中常见的顺序是：“先用高级RAG第二代RAG，在检索前、中、后各阶段引入优化技术（如查询重写、混合检索、重排序），提升检索准确性。，等关系型问题真的开始占主导了，再加Graph RAG。”

示例3：Graph RAG

不依赖Neo4j这类外部数据库，用NetworkX内存图演示Graph RAG的核心流程（实体抽取 → 图构建 → 图遍历 → 回答）。生产环境中换成Neo4j配合LangChain的LLMGraphTransformer即可。

"""example_3_graph_rag.py，基于NetworkX的迷你GraphRAG"""
from __future__ import annotations
import json
from typing import List, Tuple, Dict, Set
from dataclasses import dataclass, field

import networkx as nx
from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate

# ════════════════════════════════════════════════════════════════
# 0. 数据：虚构的公司wiki，人物/项目/组织关系丰富
# ════════════════════════════════════════════════════════════════
DOCS = [
    "John Kim是ACME的CTO，2019年加入。在此之前他是BlueTech的高级工程师。"
    "他目前领导Project Alpha，同时兼任机器学习基础设施团队的总监。",

    "Jane Park是ACME的安全负责人，担任Project Alpha的安全owner。"
    "她之前在SecureCorp工作了10年，在BlueTech时和John Kim是同事。",

    "Minsoo Lee是ACME数据平台团队的高级工程师。他负责Project Alpha的数据管道，"
    "直接向John Kim汇报。他也在和Jane Park合作做一次安全审计。",

    "Jihoon Choi是Project Beta的PM。Beta的目标是构建新的支付系统，"
    "Minsoo Lee也部分参与Beta，支持数据迁移。",

    "Project Alpha是ACME的下一代推荐系统，2024年1月启动。"
    "Project Beta是支付系统项目，2024年6月启动。"
    "两个项目都由机器学习基础设施团队支持。",
]

# ════════════════════════════════════════════════════════════════
# 1. 用LLM抽取实体/关系三元组
# ════════════════════════════════════════════════════════════════
llm = ChatAnthropic(model="claude-opus-4-7", temperature=0)

extract_prompt = ChatPromptTemplate.from_messages([
    ("system",
     "从以下文档中抽取实体和关系，输出为JSON三元组列表。\n"
     "每个三元组格式为 {\"s\": 主语, \"r\": 关系, \"o\": 宾语}。\n"
     "实体只保留明确具体的事物：人物、组织、项目、角色。\n"
     "关系用简短动词短语（如WORKS_AT、LEADS、IS_CTO_OF、REPORTS_TO、COLLABORATES_WITH）。\n"
     "同一个人的不同称呼归一化为一个名字。\n"
     "只输出纯JSON数组，不要其他文字。"),
    ("human", "{document}")
])

def _safe_json_parse(text: str, default):
    """从LLM响应中提取JSON部分（去掉markdown围栏等）"""
    import re
    m = re.search(r"(\[.*\]|\{.*\})", text, re.DOTALL)
    if not m:
        return default
    try:
        return json.loads(m.group(1))
    except json.JSONDecodeError:
        return default

def extract_triples(doc: str) -> List[Dict]:
    raw = (extract_prompt | llm).invoke({"document": doc}).content
    return _safe_json_parse(raw, default=[])

# ════════════════════════════════════════════════════════════════
# 2.

先看建图这块。核心思路很简单：把每篇文档过一遍抽取，拿到三元组就往 NetworkX 里塞。用 MultiDiGraph 是因为同一对实体之间可能存在多种关系，普通 DiGraph 会覆盖掉。

@dataclass
class KnowledgeGraph:
    G: nx.MultiDiGraph = field(default_factory=nx.MultiDiGraph)
    # 实体 → 它出现过的文档编号集合
    ent2docs: Dict[str, Set[int]] = field(default_factory=dict)
    docs: List[str] = field(default_factory=list)

def build_kg(docs: List[str]) -> KnowledgeGraph:
    kg = KnowledgeGraph(docs=docs)
    for i, d in enumerate(docs):
        triples = extract_triples(d)
        for t in triples:
            s, r, o = t.get("s"), t.get("r"), t.get("o")
            if not (s and r and o): continue
            kg.G.add_edge(s, o, relation=r, doc_idx=i)
            kg.ent2docs.setdefault(s, set()).add(i)
            kg.ent2docs.setdefault(o, set()).add(i)
    return kg

ent2docs 这个索引别省。后面做子图检索时，要靠它反查哪些原始文档跟当前子图相关，没有它就得全量扫描。

接下来从用户问题里抽实体。这一步单独用一个轻量 prompt，只让它吐 JSON 数组，别掺别的：

query_ent_prompt = ChatPromptTemplate.from_messages([
    ("system", "Extract only the entities (people, organizations, projects, roles) mentioned in the question, "
               "as a JSON array. e.g., [\"John Kim\", \"Project Alpha\"]. No other text."),
    ("human", "{question}")
])

def extract_query_entities(q: str) -> List[str]:
    raw = (query_ent_prompt | llm).invoke({"question": q}).content
    return _safe_json_parse(raw, default=[])

_safe_json_parse 就是个容错包装，模型偶尔会在 JSON 外面裹一层 markdown 代码块，解析前先剥掉。

然后是图遍历。这里有两个细节值得说。第一，实体匹配不能只做精确匹配，用户问“John”但图里存的是“John Kim”，得靠子串兜底。第二，BFS 要在无向图上跑，因为关系是有方向的，但“谁跟谁有关联”这件事本身不分方向。

def find_node(kg: KnowledgeGraph, name: str) -> str | None:
    """先精确匹配，不行再退到子串匹配"""
    if name in kg.G: return name
    for n in kg.G.nodes:
        if name in n or n in name:
            return n
    return None

def subgraph_around(kg: KnowledgeGraph, entities: List[str], hops: int = 2) -> Tuple[nx.MultiDiGraph, Set[int]]:
    """从种子实体的 N 跳邻域收集子图 + 相关文档编号"""
    seed_nodes = {n for e in entities if (n := find_node(kg, e))}
    if not seed_nodes:
        return nx.MultiDiGraph(), set()

    # 转无向图做双向 BFS
    undirected = kg.G.to_undirected()
    visited = set(seed_nodes)
    frontier = set(seed_nodes)
    for _ in range(hops):
        next_frontier = set()
        for n in frontier:
            if n not in undirected: continue
            next_frontier.update(undirected.neighbors(n))
        frontier = next_frontier - visited
        visited |= frontier

    sub = kg.G.subgraph(visited).copy()
    # 收集相关文档编号
    doc_ids = set()
    for n in visited:
        doc_ids.update(kg.ent2docs.get(n, set()))
    return sub, doc_ids

两跳是个经验值。一跳经常不够，因为问题里的实体和答案实体之间往往隔着一层中间节点；三跳以上噪音涨得很快，子图文本塞进 prompt 会挤占文档的空间。如果你们的图特别稠密，可以考虑按边类型做剪枝，但那是另一个话题了。

子图拿到后要序列化成文本。这里用 (A) -[relation]-> (B) 的格式，跟后面 prompt 里要求模型引用关系的方式保持一致，模型照着格式抄就行：

def serialize_subgraph(sub: nx.MultiDiGraph) -> str:
    """把子图转成文本喂给 LLM"""
    if sub.number_of_edges() == 0:
        return "(no related graph)"
    lines = []
    for u, v, data in sub.edges(data=True):
        lines.append(f"({u}) -[{data['relation']}]-> ({v})")
    return "\n".join(sorted(set(lines)))

去重加排序，避免同一对实体多条边导致文本膨胀，也让输出稳定一点，方便调试。

最后是答案生成。关键点在于图和原文要一起给，让模型自己决定什么时候走图、什么时候看原文。只给图，模型容易在关系链断裂的地方硬编；只给原文，多跳问题就退化成普通 RAG 了。

graph_answer_prompt = ChatPromptTemplate.from_messages([
    ("system",
     "You are an internal knowledge assistant. Answer based only on the [knowledge graph] and [source documents] below.\n"
     "- Walk the graph relationships to perform multi-hop reasoning.\n"
     "- Cite the relations you used in the form (A) -[relation]-> (B).\n"
     "- If evidence is insufficient, answer 'Cannot be answered with the provided information.'"),
    ("human",
     "[knowledge graph]\n{graph}\n\n[source documents]\n{docs}\n\n[question]\n{question}")
])

def graph_rag(question: str, kg: KnowledgeGraph) -> str:
    ents = extract_query_entities(question)
    sub, doc_ids = subgraph_around(kg, ents, hops=2)

    graph_text = serialize_subgraph(sub)
    doc_text = "\n\n".join(f"[doc{i}] {kg.docs[i]}" for i in sorted(doc_ids)) or "(no related documents)"

    print(f"  · Query entities: {ents}")
    print(f"  · Subgraph nodes {sub.number_of_nodes()}, edges {sub.number_of_edges()}")
    print(f"  · Related source docs: {sorted(doc_ids)}")

    msg = graph_answer_prompt.invoke({
        "graph": graph_text, "docs": doc_text, "question": question})
    return llm.invoke(msg).content

那几行 print 别删。调试的时候能一眼看出是实体没抽对、子图没捞到东西、还是文档索引没关联上。这三种失败模式长得完全不一样，但最终表现都是“答得不对”。

跑起来：

if __name__ == "__main__":
    print("[1/3] Building the graph...")
    kg = build_kg(DOCS)
    print(f"     done. nodes={kg.G.number_of_nodes()}, edges={kg.G.number_of_edges()}")

整套流程串下来，一次查询大概是这样：抽实体（一次 LLM 调用）→ 图遍历（纯内存操作，毫秒级）→ 生成答案（一次 LLM 调用）。两次模型调用，中间夹一段确定性的图计算。比纯向量检索多了一次调用，但换来的是多跳推理能力，值不值取决于你的问题里有多少是“A 的上级负责的项目里，谁跟 B 有交集”这种。

到这里这个系列就写完了。从文档切分、向量检索、混合检索，一路到图增强，八篇文章覆盖的是一条完整的 RAG 流水线。实际落地时不用全上，先跑通最朴素的向量检索，看 badcase 集中在哪，再决定往哪个方向加东西。图这一层适合关系密集的场景，如果你们的文档本身是散的，硬建图反而会引入噪音。

常见问题（FAQ）

RAG是什么？为什么需要它？

RAG（检索增强生成）把检索、增强、生成三件事串在一起，让LLM需要时从外部知识库获取信息。它主要解决LLM的时效性、私有知识和幻觉三大问题，还能降低成本、实现权限控制和审计。

RAG有哪几代演进？各自特点是什么？

RAG经历了Naive RAG（简单检索后生成）、Advanced RAG（更聪明的检索流水线）、Modular RAG（路由、循环、工具）和Graph RAG（关系图）四代演进。每代在检索准确性、灵活性和关系推理上逐步增强，但复杂度也相应提高。

Graph RAG适合什么场景？有什么优缺点？

Graph RAG将文档表示为实体关系图，适合多跳推理和关系为核心的领域。优点是擅长多跳和关系推理，缺点是图构建成本高、schema设计开销大。