RAG 系统完全指南：检索增强生成的一切

GPT-4 或 Claude 这样的 LLM 具备惊人的能力，但也存在本质上的局限。它们不了解训练数据截止之后的信息，缺乏特定领域的专业知识，有时还会以斩钉截铁的口吻编造出错误信息(幻觉，Hallucination)。RAG(Retrieval-Augmented Generation)正是解决这些问题最实用的架构。

本指南从基础 RAG 出发，一路讲到 Self-RAG、Corrective RAG、GraphRAG 等最新架构，并配有可直接用于实战的完整代码。

1. 什么是 RAG？

1.1 LLM 的知识局限

LLM 虽然基于海量文本进行预训练，但存在两个根本性的局限。

知识截止(Knowledge Cutoff)：无法得知训练完成时点之后的信息。GPT-4 的训练数据只包含到某个特定日期为止。

幻觉(Hallucination)：LLM 是概率性的语言模型。面对自己不知道的内容，它往往不会说"不知道"，而是倾向于生成听起来言之凿凿的内容。这种情况尤其常见于具体的事实、日期、引文和数字。

缺乏领域专业性：企业内部文档、最新技术规格，以及医疗/法律等专业领域知识，很难被充分包含进通用 LLM 中。

1.2 RAG 的核心思路

RAG 的核心很简单：在 LLM 生成答案之前，先检索相关信息，并将其作为上下文提供给它。

用户提问 → 检索相关文档 → 将[文档 + 问题]提供给 LLM → 生成答案

这就是全部内容。但"如何检索""如何准备文档""如何传递给 LLM"这些细节，才是决定系统质量的关键。

1.3 RAG 与微调(Fine-tuning)对比

在许多情况下，RAG 更为实用。但如果需要特定的输出格式、风格，或专门领域的推理能力，Fine-tuning 可以作为补充手段使用。

1.4 RAG 系统架构概览

整个 RAG 流水线分为两个阶段。

离线(索引)阶段：

1. 文档收集(PDF、HTML、数据库等)
2. 文本分块(切分)
3. 生成嵌入
4. 存入向量数据库

在线(查询)阶段：

1. 对用户查询做嵌入
2. 在向量数据库中检索相似分块
3. 组装上下文
4. 用 LLM 生成答案

2. 文本嵌入(Text Embeddings)

2.1 嵌入的概念

嵌入是将文本转换为高维实数向量的过程。其核心在于：语义相近的文本，在向量空间中的位置也会彼此靠近。

举例来说：

• "小狗正在跑来跑去地玩耍"
• "狗正在奔跑"

这两句话的嵌入向量之间，余弦相似度非常高。

2.2 主要嵌入模型

OpenAI 嵌入

• text-embedding-3-small：1536 维，速度快、成本低
• text-embedding-3-large：3072 维，质量更高
• 基于 API，使用简便，付费

Sentence-Transformers

• all-MiniLM-L6-v2：384 维，速度快、通用性强
• BAAI/bge-large-en-v1.5：1024 维，高性能
• 可本地运行，免费

韩语嵌入模型

• jhgan/ko-sroberta-multitask：韩语专用 Sentence-BERT
• snunlp/KR-ELECTRA-discriminator：基于韩语 ELECTRA
• BAAI/bge-m3：支持多语言，韩语表现也很出色

2.3 用余弦相似度检索

基于嵌入的检索，是计算查询嵌入与已存储的文档嵌入之间的余弦相似度。

$\text{similarity}(q, d) = \frac{q \cdot d}{\|q\| \|d\|}$

from sentence_transformers import SentenceTransformer
import numpy as np

# 加载嵌入模型
model = SentenceTransformer('all-MiniLM-L6-v2')

# 文档嵌入
documents = [
    "Python 是广泛用于数据科学的编程语言。",
    "机器学习是从数据中学习模式的人工智能领域。",
    "巴黎是法国的首都。",
    "深度学习是使用神经网络的机器学习方法。",
]

doc_embeddings = model.encode(documents)
print(f"嵌入形状: {doc_embeddings.shape}")  # (4, 384)

# 查询嵌入
query = "人工智能与数据分析"
query_embedding = model.encode([query])[0]

# 计算余弦相似度
def cosine_similarity(a, b):
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

similarities = [cosine_similarity(query_embedding, doc_emb)
                for doc_emb in doc_embeddings]

# 按结果排序
ranked = sorted(zip(similarities, documents), reverse=True)
print("\n按相似度排序:")
for score, doc in ranked:
    print(f"  {score:.4f}: {doc}")

2.4 评估嵌入质量(MTEB)

MTEB(Massive Text Embedding Benchmark)是一个系统性评估嵌入模型的基准测试，涵盖检索(Retrieval)、分类(Classification)、聚类(Clustering)等多种任务。

在实际 RAG 系统中选择嵌入模型的标准：

1. 目标语言下的表现(如果是韩语 RAG，则参考韩语基准分数)
2. 相对于嵌入维度的性能(维度越大，存储成本越高)
3. 推理速度(对实时系统而言尤为重要)
4. 许可协议(是否可商用)

3. 文档分块策略(Chunking Strategies)

分块是决定 RAG 性能的最重要设计决策之一。分块过大会引入过多噪声，过小又会导致上下文不足。

3.1 固定大小分块(Fixed-Size Chunking)

这是最简单的方法，按指定的字符/token 数量均匀切分。

from langchain.text_splitter import CharacterTextSplitter

text = """
机器学习是人工智能的一个分支，是让计算机无需
显式编程即可从数据中学习的技术。它包括监督学习、
无监督学习、强化学习等方法论，各自解决不同类型的问题。
"""

splitter = CharacterTextSplitter(
    chunk_size=100,
    chunk_overlap=20,
    separator="\n"
)
chunks = splitter.split_text(text)
for i, chunk in enumerate(chunks):
    print(f"分块 {i}: {chunk[:50]}...")

3.2 递归分块(Recursive Chunking)

LangChain 的 RecursiveCharacterTextSplitter 按段落、句子、单词的顺序递归切分，尽最大可能保留语义边界。

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separators=["\n\n", "\n", ".", "!", "?", ",", " ", ""]
)

with open("document.txt", "r", encoding="utf-8") as f:
    text = f.read()

chunks = splitter.split_text(text)
print(f"总分块数: {len(chunks)}")
print(f"平均分块长度: {sum(len(c) for c in chunks) / len(chunks):.0f} 字符")

3.3 语义分块(Semantic Chunking)

利用嵌入相似度在语义边界处切分。相邻句子之间的嵌入相似度骤然下降的地方，就是切分点。

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

# 语义分块器(基于嵌入相似度切分)
semantic_splitter = SemanticChunker(
    embeddings=OpenAIEmbeddings(),
    breakpoint_threshold_type="percentile",  # 或 "standard_deviation"
    breakpoint_threshold_amount=95           # 在相似度变化最大的前 5% 处切分
)

chunks = semantic_splitter.create_documents([text])
print(f"语义分块结果: {len(chunks)} 个分块")

3.4 父子分块(Parent-Child Chunking)

检索用小分块(子)，传递上下文用大分块(父)，分开处理的策略。

• 子分块：小而精准(提升检索准确度)
• 父分块：大而全面(为 LLM 提供充分的上下文)

from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore
from langchain_community.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter

# 子分块器：小分块(用于检索)
child_splitter = RecursiveCharacterTextSplitter(chunk_size=200)

# 父分块器：大分块(用于上下文)
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=1000)

# 向量存储(存储子分块嵌入)
vectorstore = Chroma(
    collection_name="full_documents",
    embedding_function=OpenAIEmbeddings()
)

# 父文档存储(存储完整的父分块)
store = InMemoryStore()

retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=store,
    child_splitter=child_splitter,
    parent_splitter=parent_splitter,
)

3.5 分块大小决策指南

4. 向量数据库(Vector Databases)

向量数据库是专门用于存储高维向量、并快速查找相似向量的数据库。

4.1 主要向量数据库对比

FAISS(Facebook AI Similarity Search)

• Meta 开发的库
• 内存内处理，速度非常快
• 无需生产服务器(是一个库)
• 最适合大规模批处理

Chroma

• 开源，内置嵌入
• Python 原生 API
• 适合开发/原型阶段
• 支持持久化(基于 SQLite)

Pinecone

• 完全托管的云服务
• 企业级扩展能力
• 付费服务，运维简单

Weaviate

• 开源 + 云端选项
• 内置混合检索
• GraphQL API

Milvus

• 高性能开源
• 分布式架构
• 数十亿级向量规模

pgvector

• PostgreSQL 扩展
• 利用现有 PostgreSQL 基础设施
• 用 SQL 做向量检索

4.2 ANN 算法

精确最近邻(KNN)需要 $O(n)$ 时间。面对大规模数据时，会使用近似算法(ANN)。

HNSW(Hierarchical Navigable Small World)

通过分层图结构支持快速检索。

• 插入：$O(\log n)$
• 检索：$O(\log n)$
• 高召回率，查询速度快
• Chroma、Weaviate 的默认算法

IVF(Inverted File Index)

把数据划分为多个簇，只在相关的簇内检索。

• 内存效率高
• 通过 nprobe 参数调节精度与速度之间的权衡
• 在 FAISS 中被广泛使用

4.3 FAISS 与 Chroma 实现对比

import numpy as np
import faiss
from langchain_community.vectorstores import FAISS, Chroma
from langchain_openai import OpenAIEmbeddings

# ===== 直接使用 FAISS =====
# 随机向量数据(实际使用中应为嵌入)
d = 384          # 向量维度
n = 10000        # 文档数量
vectors = np.random.randn(n, d).astype('float32')

# 创建索引(L2 距离)
index_flat = faiss.IndexFlatL2(d)
index_flat.add(vectors)
print(f"FAISS 索引大小: {index_flat.ntotal}")

# 检索相似向量
query = np.random.randn(1, d).astype('float32')
k = 5
distances, indices = index_flat.search(query, k)
print(f"前 {k} 个结果: {indices[0]}")

# HNSW 索引(快速近似检索)
index_hnsw = faiss.IndexHNSWFlat(d, 32)  # M=32 条连接
index_hnsw.add(vectors)
distances_hnsw, indices_hnsw = index_hnsw.search(query, k)
print(f"HNSW 结果: {indices_hnsw[0]}")

# ===== LangChain + Chroma =====
from langchain.schema import Document

documents = [
    Document(page_content="Python 被用于数据科学。", metadata={"source": "doc1"}),
    Document(page_content="机器学习从数据中学习模式。", metadata={"source": "doc2"}),
    Document(page_content="深度学习是基于神经网络的机器学习。", metadata={"source": "doc3"}),
    Document(page_content="自然语言处理用于分析文本。", metadata={"source": "doc4"}),
]

# 创建 Chroma 向量存储
embeddings = OpenAIEmbeddings()
chroma_db = Chroma.from_documents(
    documents,
    embeddings,
    persist_directory="./chroma_db"
)

# 相似度检索
results = chroma_db.similarity_search("AI 与机器学习", k=2)
for doc in results:
    print(f"来源: {doc.metadata['source']}, 内容: {doc.page_content}")

# 附带分数一起检索
results_with_score = chroma_db.similarity_search_with_score("深度学习", k=2)
for doc, score in results_with_score:
    print(f"分数: {score:.4f}, 内容: {doc.page_content}")

5. 实现基础 RAG 流水线

5.1 完整的 RAG 流水线(LangChain)

from langchain_community.document_loaders import PyPDFLoader, DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate
import os

# ===== 1. 加载文档 =====
# PDF 加载器
loader = PyPDFLoader("company_handbook.pdf")
# 或者加载整个目录
# loader = DirectoryLoader("./docs/", glob="**/*.pdf", loader_cls=PyPDFLoader)
pages = loader.load()
print(f"已加载页数: {len(pages)}")

# ===== 2. 切分文本 =====
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    length_function=len,
    separators=["\n\n", "\n", ".", "!", "?", ",", " "]
)
chunks = text_splitter.split_documents(pages)
print(f"生成的分块数: {len(chunks)}")

# ===== 3. 生成嵌入并存入向量数据库 =====
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(
    chunks,
    embeddings,
    persist_directory="./rag_db"
)
print("向量数据库存储完成")

# ===== 4. 构建 RAG 链 =====
# 自定义提示词模板
prompt_template = """你是一个乐于助人的 AI 助手。
请仅使用给定的上下文来回答问题。
如果上下文中没有相关信息，请回答"在提供的文档中找不到相关信息"。

上下文:
{context}

问题: {question}

回答:"""

PROMPT = PromptTemplate(
    template=prompt_template,
    input_variables=["context", "question"]
)

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 4}
)

# RetrievalQA 链
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever,
    chain_type_kwargs={"prompt": PROMPT},
    return_source_documents=True
)

# ===== 5. 问答 =====
query = "公司的休假政策是怎样的？"
result = qa_chain.invoke({"query": query})

print(f"\n问题: {query}")
print(f"回答: {result['result']}")
print(f"\n参考文档:")
for doc in result['source_documents']:
    print(f"  - 第 {doc.metadata.get('page', '?')} 页: {doc.page_content[:100]}...")

5.2 用 LlamaIndex 实现 RAG

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings
from llama_index.core.node_parser import SentenceSplitter
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding

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

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

# 创建索引
index = VectorStoreIndex.from_documents(documents)

# 查询引擎
query_engine = index.as_query_engine(
    similarity_top_k=4,
    response_mode="compact"  # "tree_summarize"、"refine" 等
)

# 查询
response = query_engine.query("What is the main topic of the documents?")
print(f"回答: {response}")
print(f"\n来源节点:")
for node in response.source_nodes:
    print(f"  - Score: {node.score:.4f}")
    print(f"    Text: {node.text[:100]}...")

6. 高级检索技术

6.1 混合检索(Hybrid Search)

结合向量检索(语义)与 BM25 检索(基于关键词)，兼具两种方式的优点。

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

# BM25 检索(基于关键词)
bm25_retriever = BM25Retriever.from_documents(chunks)
bm25_retriever.k = 4

# 向量检索(基于语义)
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

# 集成(混合)
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.5, 0.5]  # 各方式的权重
)

results = ensemble_retriever.invoke("Python 编程教程")
print(f"混合检索结果数: {len(results)}")

6.2 多查询检索(Multi-Query Retrieval)

把一个问题改写成多种不同的表达方式，从而扩大检索范围。

from langchain.retrievers.multi_query import MultiQueryRetriever
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(temperature=0)
multi_query_retriever = MultiQueryRetriever.from_llm(
    retriever=vectorstore.as_retriever(),
    llm=llm
)

# 内部由 LLM 将问题改写为多个版本
# 例如: "RAG 的优点是什么?"
# → "RAG 技术的优势是什么?"
# → "检索增强生成相比普通 LLM 好在哪里?"
# → "为什么应该使用 RAG?"

results = multi_query_retriever.invoke("RAG 的优点是什么?")
print(f"多查询检索结果: {len(results)} 个")

6.3 MMR(Maximal Marginal Relevance)

如果只考虑相似度，可能会选到重复的分块。MMR 会同时考虑相似度与多样性。

$\text{MMR} = \arg\max_{d_i \in D \setminus R} [\lambda \cdot \text{sim}(d_i, q) - (1-\lambda) \cdot \max_{d_j \in R} \text{sim}(d_i, d_j)]$

# MMR 检索
mmr_retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={
        "k": 4,           # 最终返回数量
        "fetch_k": 20,    # 初始候选数量
        "lambda_mult": 0.5  # 相似度 vs 多样性的平衡(0=多样性, 1=相似度)
    }
)

results = mmr_retriever.invoke("机器学习算法")
print(f"MMR 结果: {len(results)} 个")

6.4 元数据过滤

检索时附加元数据条件，以缩小范围。

from langchain.schema import Document

# 带元数据的文档
docs_with_metadata = [
    Document(
        page_content="2024年第一季度营收为100亿韩元。",
        metadata={"year": 2024, "quarter": "Q1", "category": "financial"}
    ),
    Document(
        page_content="2024年第二季度营收为120亿韩元。",
        metadata={"year": 2024, "quarter": "Q2", "category": "financial"}
    ),
    Document(
        page_content="技术路线图：计划强化 AI 功能。",
        metadata={"year": 2024, "quarter": "Q1", "category": "strategy"}
    ),
]

# 用元数据过滤缩小检索范围
filtered_results = vectorstore.similarity_search(
    "营收表现",
    k=2,
    filter={"category": "financial", "year": 2024}
)

7. 重排序(Reranking)

为了改进检索结果的排序，用更精细的模型重新排列。这是一种两阶段策略：先靠检索提高 recall，再靠重排序提高 precision。

7.1 Cross-Encoder 重排序器

相比双编码器(bi-encoder，分别编码两段文本后再计算相似度)，使用更精确的交叉编码器(cross-encoder，将两段文本一起编码)。

from sentence_transformers import CrossEncoder
import numpy as np

# 加载 Cross-Encoder 模型
cross_encoder = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')

# 初始检索结果(基于双编码器)
query = "机器学习算法的种类"
initial_results = vectorstore.similarity_search(query, k=20)  # 多检索一些

# 用交叉编码器重新排序
pairs = [[query, doc.page_content] for doc in initial_results]
scores = cross_encoder.predict(pairs)

# 按分数排序
ranked = sorted(zip(scores, initial_results), reverse=True)
top_k = [doc for _, doc in ranked[:5]]  # 只使用前 5 个

print("重排序后的头部结果:")
for score, doc in ranked[:3]:
    print(f"  分数 {score:.4f}: {doc.page_content[:80]}...")

7.2 Cohere Rerank API

import cohere
from langchain.retrievers.document_compressors import CohereRerank
from langchain.retrievers import ContextualCompressionRetriever

co = cohere.Client("your-api-key")

# Cohere 重排序压缩器
compressor = CohereRerank(
    cohere_api_key="your-api-key",
    top_n=3,
    model="rerank-multilingual-v3.0"  # 支持韩语
)

# 压缩检索器(检索 + 重排序)
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vectorstore.as_retriever(search_kwargs={"k": 20})
)

results = compression_retriever.invoke("请告诉我公司的政策")
print(f"重排序后的文档数: {len(results)}")

7.3 BGE Reranker(开源)

from FlagEmbedding import FlagReranker

# BGE 重排序器(开源，本地运行)
reranker = FlagReranker('BAAI/bge-reranker-large', use_fp16=True)

query = "What is RAG?"
passages = [
    "RAG stands for Retrieval-Augmented Generation.",
    "A rag is a piece of cloth used for cleaning.",
    "RAG systems combine retrieval with generation for better LLM responses.",
]

# 逐对计算分数
scores = reranker.compute_score([[query, p] for p in passages])
ranked = sorted(zip(scores, passages), reverse=True)

for score, passage in ranked:
    print(f"  {score:.4f}: {passage}")

8. HyDE(Hypothetical Document Embeddings)

8.1 HyDE 的思路

普通 RAG 是直接比较查询嵌入与文档嵌入。但短查询的嵌入，在语义空间中可能与长文档的嵌入相距甚远。

HyDE 的解决办法：让 LLM 生成一份假想的答案文档，然后用这份假想文档的嵌入来检索。

问题 → LLM 生成假想答案 → 对假想答案做嵌入 → 检索真实文档

8.2 实现 HyDE

from langchain.chains import HypotheticalDocumentEmbedder
from langchain_openai import OpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

# HyDE 提示词
hyde_prompt = """你是一位专家。请针对以下问题撰写一份详细的答案文档。
请让这份文档写得像是真实存在的一样。

问题: {question}

假想答案文档:"""

# HyDE 嵌入生成器
llm = OpenAI()
embeddings = OpenAIEmbeddings()

hyde_embeddings = HypotheticalDocumentEmbedder.from_llm(
    llm=llm,
    embeddings=embeddings,
    prompt_key="web_search"  # 或使用自定义提示词
)

# 使用 HyDE 的检索
query = "量子计算对加密的影响"
# HyDE 内部会先生成假想文档，再做嵌入
hyde_retriever = vectorstore.as_retriever(embedding_function=hyde_embeddings)
results = hyde_retriever.invoke(query)

# 手动实现 HyDE
def manual_hyde(query, llm, embeddings, vectorstore, k=4):
    # 1. 生成假想文档
    hypothetical_doc = llm.invoke(
        f"请针对以下问题撰写详细答案: {query}"
    )

    # 2. 对假想文档做嵌入
    hyp_embedding = embeddings.embed_query(hypothetical_doc.content)

    # 3. 用假想文档的嵌入进行检索
    results = vectorstore.similarity_search_by_vector(hyp_embedding, k=k)

    return results, hypothetical_doc.content

from langchain_openai import ChatOpenAI
chat_llm = ChatOpenAI(temperature=0.7)
results, hyp_doc = manual_hyde(
    "深度学习的历史", chat_llm, embeddings, vectorstore
)
print(f"生成的假想文档: {hyp_doc[:200]}...")
print(f"检索到的实际文档: {len(results)} 个")

9. 高级 RAG 架构

9.1 Self-RAG

Self-RAG(2023, Asai et al.)让 LLM 自行判断是否需要检索，并对检索到的文档的相关性、以及应答质量进行批判性评估。

它使用 4 种特殊 token：

• [Retrieve]：是否需要检索？(Yes/No)
• [IsRel]：检索到的文档是否相关？(Relevant/Irrelevant)
• [IsSup]：应答是否得到文档支持？(Supported/Partially/Not)
• [IsUse]：应答是否有用？(1-5 分)

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# Self-RAG 模拟(真正的 Self-RAG 需要专门训练的模型)
class SelfRAGSimulator:
    def __init__(self, retriever, llm):
        self.retriever = retriever
        self.llm = llm

    def should_retrieve(self, query: str) -> bool:
        """判断是否需要检索"""
        prompt = ChatPromptTemplate.from_template("""
请判断回答以下问题是否需要检索外部文档。
如果仅凭常识或推理即可回答，请只回答 'NO'；
如果需要特定事实或专业知识，请只回答 'YES'。

问题: {query}
判断 (YES/NO):""")
        response = self.llm.invoke(prompt.format_messages(query=query))
        return "YES" in response.content.upper()

    def is_relevant(self, query: str, doc_content: str) -> bool:
        """判断文档是否与问题相关"""
        prompt = ChatPromptTemplate.from_template("""
请判断以下文档是否与问题相关。
只回答 'RELEVANT' 或 'IRRELEVANT'。

问题: {query}
文档: {doc}
判断:""")
        response = self.llm.invoke(
            prompt.format_messages(query=query, doc=doc_content[:500])
        )
        return "RELEVANT" in response.content.upper()

    def generate_with_reflection(self, query: str) -> str:
        """以 Self-RAG 方式生成应答"""
        # 1. 判断是否需要检索
        need_retrieve = self.should_retrieve(query)
        print(f"是否需要检索: {need_retrieve}")

        if not need_retrieve:
            # 不检索，直接回答
            response = self.llm.invoke(query)
            return response.content

        # 2. 检索文档
        docs = self.retriever.invoke(query)

        # 3. 按相关性过滤
        relevant_docs = [d for d in docs if self.is_relevant(query, d.page_content)]
        print(f"相关文档: {len(relevant_docs)}/{len(docs)} 个")

        if not relevant_docs:
            return "未能找到相关文档，将以常识回答: " + \
                   self.llm.invoke(query).content

        # 4. 基于上下文生成答案
        context = "\n\n".join([d.page_content for d in relevant_docs[:3]])
        prompt = f"""请使用上下文回答问题。
上下文: {context}
问题: {query}
回答:"""
        return self.llm.invoke(prompt).content

9.2 Corrective RAG(CRAG)

CRAG 会评估检索到文档的质量，若质量不足，则用网络搜索来补充。

from langchain_community.tools.tavily_search import TavilySearchResults
from typing import List, Tuple

class CorrectiveRAG:
    def __init__(self, retriever, llm):
        self.retriever = retriever
        self.llm = llm
        self.web_search = TavilySearchResults(max_results=3)

    def evaluate_documents(self, query: str, docs: list) -> Tuple[str, List]:
        """
        评估文档相关性
        返回: ("CORRECT"|"INCORRECT"|"AMBIGUOUS", 过滤后的文档)
        """
        evaluation_prompt = """请评估检索到的文档与问题的相关性。
- CORRECT: 文档可以直接回答问题
- INCORRECT: 文档与问题无关
- AMBIGUOUS: 部分相关但不完整

问题: {query}
文档:
{docs}

评估 (CORRECT/INCORRECT/AMBIGUOUS):"""

        docs_text = "\n---\n".join([d.page_content[:300] for d in docs[:4]])
        response = self.llm.invoke(
            evaluation_prompt.format(query=query, docs=docs_text)
        )

        evaluation = response.content.strip().upper()
        if "CORRECT" in evaluation:
            return "CORRECT", docs
        elif "INCORRECT" in evaluation:
            return "INCORRECT", []
        else:
            return "AMBIGUOUS", docs

    def run(self, query: str) -> str:
        # 1. 初始检索
        docs = self.retriever.invoke(query)

        # 2. 评估文档质量
        status, filtered_docs = self.evaluate_documents(query, docs)
        print(f"文档评估结果: {status}")

        # 3. 按情况处理
        if status == "INCORRECT":
            # 用网络搜索补充
            print("正在用网络搜索补充...")
            web_results = self.web_search.invoke(query)
            context = "\n".join([r['content'] for r in web_results])
        elif status == "AMBIGUOUS":
            # 知识精炼 + 网络搜索相结合
            web_results = self.web_search.invoke(query)
            web_context = "\n".join([r['content'] for r in web_results])
            doc_context = "\n".join([d.page_content for d in filtered_docs[:2]])
            context = doc_context + "\n\n[网络搜索补充]\n" + web_context
        else:
            # 直接使用检索到的文档
            context = "\n\n".join([d.page_content for d in filtered_docs[:4]])

        # 4. 生成最终应答
        response = self.llm.invoke(
            f"上下文:\n{context}\n\n问题: {query}\n回答:"
        )
        return response.content

9.3 Adaptive RAG

根据查询的复杂度，动态选择检索策略。

class AdaptiveRAG:
    def __init__(self, simple_retriever, advanced_retriever, llm):
        self.simple_retriever = simple_retriever   # 简单向量检索
        self.advanced_retriever = advanced_retriever  # 混合检索 + 重排序
        self.llm = llm

    def classify_query(self, query: str) -> str:
        """对查询复杂度分类"""
        prompt = f"""请对以下问题的复杂度进行分类:
- simple: 单纯的事实确认，或可直接回答
- complex: 需要综合多个来源、多步推理

问题: {query}
分类 (simple/complex):"""

        response = self.llm.invoke(prompt)
        return "complex" if "complex" in response.content.lower() else "simple"

    def run(self, query: str) -> str:
        query_type = self.classify_query(query)
        print(f"查询类型: {query_type}")

        if query_type == "simple":
            docs = self.simple_retriever.invoke(query)
        else:
            docs = self.advanced_retriever.invoke(query)

        context = "\n\n".join([d.page_content for d in docs])
        return self.llm.invoke(
            f"上下文:\n{context}\n\n问题: {query}\n回答:"
        ).content

9.4 GraphRAG(Microsoft)

Microsoft 的 GraphRAG 会从文档中构建知识图谱，并利用图结构来执行检索。

核心思路：

1. 从文档中抽取实体(人物、地点、概念)与关系
2. 用社区检测算法将相关实体分组
3. 为每个社区生成摘要
4. 全局查询使用社区摘要，局部查询使用图遍历

# 安装并初始化 GraphRAG
pip install graphrag

# 初始化项目
python -m graphrag.index --init --root ./ragtest

# 修改配置文件后建立索引
python -m graphrag.index --root ./ragtest

# 全局检索(需要理解整体文档)
python -m graphrag.query --root ./ragtest --method global "What are the main themes?"

# 局部检索(围绕特定实体)
python -m graphrag.query --root ./ragtest --method local "Tell me about company X"

10. RAG 评估指标

客观衡量 RAG 系统的质量，对于持续改进是必不可少的。

10.1 RAGAS(RAG Assessment)

RAGAS 是一个自动评估 RAG 流水线的框架。

主要指标：

• Faithfulness：答案对上下文的忠实程度(衡量幻觉)
• Answer Relevancy：答案与问题的相关程度
• Context Recall：相关上下文的检索完整程度
• Context Precision：检索到的上下文中实际有用的比例

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_recall,
    context_precision
)
from datasets import Dataset

# 准备评估数据
evaluation_data = {
    "question": [
        "公司的年假制度是怎样的？",
        "远程办公政策是什么？",
    ],
    "answer": [
        "入职一年后可获得15天年假，此后每年增加1天。",
        "每周可远程办公3天。",
    ],
    "contexts": [
        ["员工在入职一年后获得15天年假。此后每年增加1天。"],
        ["员工每周可远程办公2天。经特殊批准可增加。"],
    ],
    "ground_truth": [
        "入职一年后15天，每年增加1天",
        "每周远程办公2天为基本，经批准可增加",
    ]
}

dataset = Dataset.from_dict(evaluation_data)

result = evaluate(
    dataset,
    metrics=[
        faithfulness,
        answer_relevancy,
        context_recall,
        context_precision
    ]
)

print(result)
# faithfulness: 0.75 (检测到答案与上下文不一致的部分)
# answer_relevancy: 0.92
# context_recall: 0.85
# context_precision: 0.78

10.2 实战评估流水线

import json
from typing import List, Dict

class RAGEvaluator:
    def __init__(self, rag_chain, llm):
        self.rag_chain = rag_chain
        self.llm = llm

    def evaluate_faithfulness(self, answer: str, context: str) -> float:
        """评估答案是否忠实于上下文 (0-1)"""
        prompt = f"""请评估以下答案是否仅基于给定上下文中的信息写成。
0.0 (完全不是) ~ 1.0 (完全忠实)

上下文: {context}
答案: {answer}

忠实度分数 (仅数字):"""
        response = self.llm.invoke(prompt)
        try:
            return float(response.content.strip())
        except:
            return 0.5

    def evaluate_answer_relevancy(self, question: str, answer: str) -> float:
        """评估答案与问题的相关性"""
        prompt = f"""请以 0.0~1.0 评估以下答案与问题的相关程度。

问题: {question}
答案: {answer}

相关度分数 (仅数字):"""
        response = self.llm.invoke(prompt)
        try:
            return float(response.content.strip())
        except:
            return 0.5

    def run_evaluation(self, test_cases: List[Dict]) -> Dict:
        results = []
        for case in test_cases:
            question = case["question"]
            expected = case.get("expected_answer", "")

            # 生成 RAG 应答
            result = self.rag_chain.invoke({"query": question})
            answer = result["result"]
            context = "\n".join([d.page_content for d in result["source_documents"]])

            # 评估
            faithfulness_score = self.evaluate_faithfulness(answer, context)
            relevancy_score = self.evaluate_answer_relevancy(question, answer)

            results.append({
                "question": question,
                "answer": answer,
                "faithfulness": faithfulness_score,
                "relevancy": relevancy_score,
            })

        # 汇总
        avg_faithfulness = sum(r["faithfulness"] for r in results) / len(results)
        avg_relevancy = sum(r["relevancy"] for r in results) / len(results)

        return {
            "results": results,
            "avg_faithfulness": avg_faithfulness,
            "avg_relevancy": avg_relevancy,
            "overall_score": (avg_faithfulness + avg_relevancy) / 2
        }

11. 生产环境 RAG 系统

11.1 缓存策略

import hashlib
import json
from functools import lru_cache
import redis

class CachedRAGSystem:
    def __init__(self, rag_chain, redis_client=None, ttl=3600):
        self.rag_chain = rag_chain
        self.redis = redis_client
        self.ttl = ttl

    def _get_cache_key(self, query: str) -> str:
        return f"rag:{hashlib.md5(query.encode()).hexdigest()}"

    def query(self, query: str) -> dict:
        cache_key = self._get_cache_key(query)

        # 检查缓存
        if self.redis:
            cached = self.redis.get(cache_key)
            if cached:
                print("缓存命中！")
                return json.loads(cached)

        # 执行 RAG
        result = self.rag_chain.invoke({"query": query})
        response = {
            "answer": result["result"],
            "sources": [d.metadata for d in result["source_documents"]]
        }

        # 存储缓存
        if self.redis:
            self.redis.setex(cache_key, self.ttl, json.dumps(response))

        return response

# 嵌入缓存
from langchain.globals import set_llm_cache
from langchain_community.cache import InMemoryCache, SQLiteCache

# 内存缓存(用于开发)
set_llm_cache(InMemoryCache())

# SQLite 缓存(用于生产)
set_llm_cache(SQLiteCache(database_path=".langchain.db"))

11.2 流式响应

from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
from langchain_openai import ChatOpenAI

# 流式 LLM
streaming_llm = ChatOpenAI(
    model="gpt-4o-mini",
    streaming=True,
    callbacks=[StreamingStdOutCallbackHandler()]
)

# 通过 FastAPI 实现的流式端点
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio

app = FastAPI()

async def generate_rag_stream(query: str):
    docs = retriever.invoke(query)
    context = "\n\n".join([d.page_content for d in docs])

    async for chunk in streaming_llm.astream(
        f"上下文:\n{context}\n\n问题: {query}\n回答:"
    ):
        if chunk.content:
            yield f"data: {chunk.content}\n\n"

@app.get("/rag/stream")
async def rag_stream_endpoint(query: str):
    return StreamingResponse(
        generate_rag_stream(query),
        media_type="text/event-stream"
    )

11.3 成本优化

# 追踪 token 用量
from langchain.callbacks import get_openai_callback

with get_openai_callback() as cb:
    result = qa_chain.invoke({"query": "问题文本"})
    print(f"总 token 数: {cb.total_tokens}")
    print(f"提示词 token 数: {cb.prompt_tokens}")
    print(f"补全 token 数: {cb.completion_tokens}")
    print(f"成本: ${cb.total_cost:.6f}")

# 通过压缩分块节省上下文 token
from langchain.retrievers.document_compressors import LLMChainExtractor
from langchain.retrievers import ContextualCompressionRetriever

# 只抽取相关部分以压缩上下文
compressor = LLMChainExtractor.from_llm(llm)
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vectorstore.as_retriever(search_kwargs={"k": 8})
)

# 只把压缩后的文档传给 LLM(节省 token)
compressed_docs = compression_retriever.invoke("问题")
total_tokens = sum(len(d.page_content.split()) for d in compressed_docs)
print(f"压缩后的上下文 token 数(估算): {total_tokens}")

11.4 监控

import time
import logging
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class RAGMetrics:
    query: str
    retrieval_time: float = 0.0
    generation_time: float = 0.0
    num_docs_retrieved: int = 0
    answer_length: int = 0
    error: Optional[str] = None

class MonitoredRAGSystem:
    def __init__(self, rag_chain, logger=None):
        self.rag_chain = rag_chain
        self.logger = logger or logging.getLogger(__name__)
        self.metrics_history = []

    def query(self, query: str) -> dict:
        metrics = RAGMetrics(query=query)
        start_total = time.time()

        try:
            # 检索计时
            retrieval_start = time.time()
            docs = retriever.invoke(query)
            metrics.retrieval_time = time.time() - retrieval_start
            metrics.num_docs_retrieved = len(docs)

            # 生成计时
            gen_start = time.time()
            result = self.rag_chain.invoke({"query": query})
            metrics.generation_time = time.time() - gen_start
            metrics.answer_length = len(result["result"])

        except Exception as e:
            metrics.error = str(e)
            self.logger.error(f"RAG 错误: {e}")
            raise

        finally:
            total_time = time.time() - start_total
            self.metrics_history.append(metrics)
            self.logger.info(
                f"查询处理完成 | "
                f"检索: {metrics.retrieval_time:.2f}s | "
                f"生成: {metrics.generation_time:.2f}s | "
                f"总计: {total_time:.2f}s | "
                f"文档: {metrics.num_docs_retrieved} 个"
            )

        return result

    def get_stats(self) -> dict:
        if not self.metrics_history:
            return {}
        retrieval_times = [m.retrieval_time for m in self.metrics_history if not m.error]
        gen_times = [m.generation_time for m in self.metrics_history if not m.error]
        return {
            "total_queries": len(self.metrics_history),
            "error_rate": sum(1 for m in self.metrics_history if m.error) / len(self.metrics_history),
            "avg_retrieval_time": sum(retrieval_times) / len(retrieval_times) if retrieval_times else 0,
            "avg_generation_time": sum(gen_times) / len(gen_times) if gen_times else 0,
        }

12. RAG 实现清单

以下是构建生产级 RAG 系统时需要考虑的事项。

文档处理

• 支持多种文件格式(PDF、Word、HTML、Markdown)
• 保留元数据(来源、日期、作者)
• 图片、表格处理策略
• 支持增量(incremental)更新

检索质量

• 选择适合领域的嵌入模型
• 考虑混合检索(关键词 + 语义)
• 设置合适的分块大小与重叠
• 用重排序提升 precision

LLM 集成

• 清晰的系统提示词(强调只使用上下文)
• 要求引用来源
• 允许表达不确定性

运维

• 通过响应缓存降低成本
• 监控 token 用量
• 通过 A/B 测试优化分块/检索参数
• RAGAS 之类的自动化评估流水线

结语

RAG 是克服 LLM 局限性最实用的方法。总结本指南所涉及的内容：

1. 基础 RAG：文档分块 → 嵌入 → 向量数据库 → 检索 → 生成
2. 提升检索质量：混合检索、MMR、重排序
3. 高级架构：Self-RAG、CRAG、HyDE、GraphRAG
4. 评估：用 RAGAS 衡量 faithfulness、relevancy
5. 生产环境：缓存、监控、成本优化

RAG 系统的性能，并不取决于单一组件，而是取决于整条流水线的协调配合。尤其是分块策略与嵌入模型的选择，决定了检索质量的 80%，因此专注于这两个要素，能带来最高的投资回报率。

参考资料

• Lewis et al. (2020), "Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks" — RAG 原始论文(arXiv:2005.11401)
• Asai et al. (2023), "Self-RAG: Learning to Retrieve, Generate, and Critique" (arXiv:2310.11511)
• Gao et al. (2023), "Precise Zero-Shot Dense Retrieval without Relevance Labels" — HyDE(arXiv:2305.14283)
• Microsoft GraphRAG: https://microsoft.github.io/graphrag/
• LangChain 官方文档: https://python.langchain.com/docs/
• LlamaIndex 官方文档: https://docs.llamaindex.ai/
• RAGAS 官方文档: https://docs.ragas.io/
• FAISS: https://faiss.ai/
• Chroma: https://www.trychroma.com/