Resource / Python

RAG
Cheatsheet

从文档切分、Embedding、Vector Store 和 Retriever 开始,一直到混合检索、Rerank、Context 组装、引用、Agentic RAG 和评估。代码以 LangChain Python 当前接口为主,检索方法本身不绑定具体向量数据库。

资料核验
2026-09-09
语言范围
Python · LangChain
适用范围
RAG · Retrieval
LOADdocuments
SPLITchunks
INDEXembedding
SEARCHretrieve
ANSWERcontext
RAG 的核心流程:先把资料整理成可检索的 Document,再根据问题召回相关内容,最后把有限且可追踪的 Context 交给模型。
一句话理解

RAG 不是“把所有资料塞进 Prompt”,而是在模型回答之前先检索与当前问题有关的内容,再把这些内容作为 Context 提供给模型。

01 / FLOW

RAG 基本流程

一个普通 RAG 系统可以拆成两条流程:离线的索引流程,以及请求到来后的检索与生成流程。

Indexing

加载文档、清理内容、切分 Chunk、计算 Embedding,并写入 Vector Store 或搜索索引。

Retrieval

根据用户问题查找相关 Document。可以是向量检索、BM25、Hybrid Search 或其他搜索系统。

Augmentation

对检索结果进行过滤、去重、Rerank 和格式化,控制最终进入模型的 Context。

Generation

将问题和检索到的 Context 一起交给模型生成答案,并保留来源信息。

02 / BOOT

安装与最小示例

RAG 需要的包取决于具体模型、文档加载器和向量数据库。下面只安装一个最小 LangChain + OpenAI 示例。

terminalBash
pip install -U \
  langchain \
  langchain-core \
  langchain-text-splitters \
  langchain-openai
minimal_rag.pyPython
from langchain.chat_models import init_chat_model
from langchain_core.documents import Document
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_openai import OpenAIEmbeddings
 
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
)
 
vector_store = InMemoryVectorStore(
    embedding=embeddings,
)
 
vector_store.add_documents([
    Document(
        page_content="LangGraph 使用 StateGraph 组织有状态工作流。",
        metadata={
            "source": "langgraph.md",
            "section": "overview",
        },
    ),
    Document(
        page_content="Checkpoint 用于保存 Graph 的运行状态。",
        metadata={
            "source": "langgraph.md",
            "section": "persistence",
        },
    ),
])
 
question = "LangGraph 怎么保存运行状态?"
 
docs = vector_store.similarity_search(
    question,
    k=3,
)
 
context = "\n\n".join(
    doc.page_content
    for doc in docs
)
 
model = init_chat_model("openai:gpt-5.4")
 
response = model.invoke([
    {
        "role": "system",
        "content": (
            "只根据提供的参考资料回答。"
            "如果资料不足,明确说明无法确定。"
        ),
    },
    {
        "role": "user",
        "content": f"""
问题:
{question}
 
参考资料:
{context}
""",
    },
])
 
print(response.text)

这个例子故意没有使用复杂 Chain。RAG 最基本的工作就是:

  1. 保存可检索的 Document。
  2. 根据问题检索相关 Document。
  3. 把 Document 内容组成 Context。
  4. 让模型基于 Context 回答。

03 / DOCUMENT

Document 与 Metadata

LangChain 的检索接口最终通常围绕 Document 工作。正文放在 page_content,来源、标题、章节和业务字段放在 metadata

document.pyPython
from langchain_core.documents import Document
 
doc = Document(
    page_content="Checkpoint 用于保存 Graph 的运行状态。",
    metadata={
        "source": "langgraph-persistence.md",
        "title": "LangGraph Persistence",
        "section": "Checkpoint",
        "url": "/docs/langgraph/persistence",
        "updated_at": "2026-09-01",
        "tenant_id": "tenant-1",
    },
)
 
print(doc.page_content)
print(doc.metadata["source"])

Metadata 至少保留什么

source

原始文件、页面、数据库记录或其他来源标识。用于引用和问题排查。

section

标题、章节或段落位置。比只保存文件名更容易定位原文。

document_id

稳定的原始文档 ID。用于更新、删除和增量索引。

权限字段

多用户或企业系统应保存 tenant、user、department 等检索权限字段。

版本信息

对经常更新的资料,保留版本、更新时间或内容哈希。

04 / CHUNK

文档切分

Chunk 的目标不是得到大小完全一致的文本块,而是让每个块能够独立被检索,同时尽量保留完整语义。

对普通文本,优先从 RecursiveCharacterTextSplitter 开始。

split.pyPython
from langchain_text_splitters import (
    RecursiveCharacterTextSplitter,
)
 
splitter = RecursiveCharacterTextSplitter(
    chunk_size=800,
    chunk_overlap=120,
)
 
chunks = splitter.split_documents(documents)
 
for chunk in chunks[:3]:
    print(len(chunk.page_content))
    print(chunk.metadata)

chunk_size 怎么定

没有一个适合所有知识库的固定值。需要结合:

  • 文档本身的结构。
  • Embedding 模型输入限制。
  • 用户问题的粒度。
  • 召回数量。
  • 最终允许使用的 Context 大小。
TOO SMALL

Chunk 太小

容易把定义、条件、结论拆开。虽然命中更精确,但内容本身可能不足以回答问题。

BALANCED

大小合适

一个 Chunk 能表达相对完整的信息,同时不会包含大量与查询无关的文本。

TOO LARGE

Chunk 太大

Embedding 表达会变得模糊,检索回来后还会占用大量模型 Context。

Markdown 文档

对技术文档和博客,不建议完全忽略标题结构直接按字符切。可以先按标题拆章节,再对过长章节做第二次切分。

markdown_split.pyPython
from langchain_text_splitters import (
    MarkdownHeaderTextSplitter,
    RecursiveCharacterTextSplitter,
)
 
headers = [
    ("#", "h1"),
    ("##", "h2"),
    ("###", "h3"),
]
 
markdown_splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=headers,
    strip_headers=False,
)
 
sections = markdown_splitter.split_text(markdown_text)
 
# 标题切完以后,单个章节仍可能很长。
# 再做一次长度切分。
chunk_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=150,
)
 
chunks = chunk_splitter.split_documents(sections)

按 Token 切分

如果更关心模型和 Embedding 的 Token 限制,可以按 Token 数量控制 Chunk。

token_split.pyPython
from langchain_text_splitters import (
    CharacterTextSplitter,
)
 
splitter = CharacterTextSplitter.from_tiktoken_encoder(
    encoding_name="cl100k_base",
    chunk_size=500,
    chunk_overlap=80,
)
 
chunks = splitter.split_documents(documents)

05 / EMBEDDING

Embedding

Embedding 把文本转换成向量。索引文档时计算文档向量,检索时计算查询向量,再根据相似度查找候选 Chunk。

embedding.pyPython
from langchain_openai import OpenAIEmbeddings
 
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
)
 
document_vectors = embeddings.embed_documents([
    "LangGraph 支持 Checkpoint。",
    "Retriever 负责根据查询返回 Document。",
])
 
query_vector = embeddings.embed_query(
    "LangGraph 如何保存状态?"
)
 
print(len(query_vector))

LangChain 也提供统一初始化接口:

init_embedding.pyPython
from langchain.embeddings import init_embeddings
 
embeddings = init_embeddings(
    "openai:text-embedding-3-small"
)

选择 Embedding 模型时看什么

语言效果

中文或中英混合知识库必须实际测试中文检索,不要只看英文榜单。

向量维度

维度影响索引空间和检索成本。更高维度不等于一定更准确。

最大输入

Chunk 大小不能超过 Embedding 模型支持的输入限制。

Query / Document 模式

部分模型会区分检索查询和索引文档的任务类型,应按模型说明使用。

版本稳定性

Embedding 模型发生变化后,旧向量通常不能直接与新向量混用。

06 / INDEX

Vector Store

Vector Store 负责保存向量以及对应的文档信息,并执行相似度搜索。LangChain 为不同实现提供了一套相对统一的接口。

vector_store.pyPython
from langchain_core.vectorstores import (
    InMemoryVectorStore,
)
 
vector_store = InMemoryVectorStore(
    embedding=embeddings,
)
 
ids = vector_store.add_documents(chunks)
 
docs = vector_store.similarity_search(
    "LangGraph 如何恢复执行?",
    k=5,
)
 
vector_store.delete(ids=ids[:1])

查看分数

search_score.pyPython
results = vector_store.similarity_search_with_score(
    "LangGraph 如何恢复执行?",
    k=5,
)
 
for doc, score in results:
    print(score)
    print(doc.metadata)
    print(doc.page_content[:200])

Metadata Filter

metadata_filter.pyPython
docs = vector_store.similarity_search(
    "Checkpoint",
    k=5,
    filter={
        "source": "langgraph.md",
    },
)

不同 Vector Store 支持的 Filter 语法并不完全相同。通用代码可以依赖 LangChain 抽象,但复杂 Filter 仍应以具体数据库文档为准。

07 / RETRIEVE

Retriever

Retriever 是比 Vector Store 更通用的检索接口。输入查询字符串,返回 Document 列表。

retriever.pyPython
retriever = vector_store.as_retriever(
    search_type="similarity",
    search_kwargs={
        "k": 5,
    },
)
 
docs = retriever.invoke(
    "LangGraph 如何保存状态?"
)

Vector Store 可以作为 Retriever 使用,但 Retriever 并不要求底层一定是向量数据库。

Vector Store

负责保存向量、添加和删除文档,并执行向量相似度检索。

Retriever

负责“根据 Query 返回 Document”。底层可以是向量库、搜索引擎、API 或自己的业务逻辑。

自定义 Retriever

custom_retriever.pyPython
from langchain_core.documents import Document
from langchain_core.retrievers import BaseRetriever
 
class DocsRetriever(BaseRetriever):
    def _get_relevant_documents(
        self,
        query: str,
    ) -> list[Document]:
        rows = search_service.search(
            query=query,
            limit=10,
        )
 
        return [
            Document(
                page_content=row.text,
                metadata={
                    "source": row.source,
                    "score": row.score,
                },
            )
            for row in rows
        ]
 
retriever = DocsRetriever()
 
docs = retriever.invoke(
    "LangGraph Checkpoint"
)

09 / RERANK

Rerank

初步检索的目标偏向“不要漏掉相关资料”,Rerank 再对候选结果做更精细的排序。

常见流程是:

  1. 先召回 10~50 个候选 Chunk。
  2. 使用 Reranker 对 Query + Document 重新打分。
  3. 只保留前几条进入模型 Context。
rerank.pyPython
# 常见两阶段检索思路:
 
candidates = retriever.invoke(query)
 
# 第一阶段:
# 向量 / BM25 / Hybrid Search
# 尽量保证召回率,候选数量可以多一些。
 
reranked = reranker.rerank(
    query=query,
    documents=candidates,
)
 
# 第二阶段:
# Cross Encoder / Rerank Model
# 只把排名靠前的少量文档交给 LLM。
 
final_docs = reranked[:5]

Cross Encoder 会直接对 Query 和 Document 一起打分,通常比单独比较向量更适合精排,但也会增加一次推理成本。

10 / CONTEXT

Context 组装

检索完成后,不要简单把所有 page_content 拼起来。模型还需要知道每段资料来自哪里。

context.pyPython
def format_context(docs):
    blocks = []
 
    for index, doc in enumerate(docs, start=1):
        source = doc.metadata.get(
            "source",
            "unknown",
        )
        section = doc.metadata.get(
            "section",
            "",
        )
 
        blocks.append(
            f"""[资料 {index}]
来源:{source}
章节:{section}
内容:
{doc.page_content}"""
        )
 
    return "\n\n".join(blocks)
 
context = format_context(docs)

进入模型前建议处理

  • 删除完全重复的 Chunk。
  • 合并同一文档连续且相邻的 Chunk。
  • 保留 source、section 等引用信息。
  • 按最终相关性排序。
  • 控制总 Token 数。
  • 不要把检索分数直接当作事实可信度。

11 / GENERATE

生成与引用

RAG Prompt 最重要的不是写得复杂,而是明确资料使用规则和资料不足时的行为。

rag_prompt.pyPython
SYSTEM_PROMPT = """
你根据提供的参考资料回答问题。
 
要求:
1. 优先使用参考资料中的事实。
2. 不要把模型已有知识冒充成参考资料内容。
3. 资料不足时直接说明无法从当前资料确定。
4. 涉及关键事实时标明资料编号。
5. 不要编造引用、文件名、页码或链接。
"""

不要让模型自己编来源

更可靠的方式是应用提前给每个 Document 分配来源编号,让模型只返回使用过的编号。

answer_with_sources.pyPython
from pydantic import BaseModel, Field
 
class RagAnswer(BaseModel):
    answer: str
    sources: list[int] = Field(
        description="实际支持回答的资料编号"
    )
 
structured_model = model.with_structured_output(
    RagAnswer
)
 
result = structured_model.invoke([
    {
        "role": "system",
        "content": SYSTEM_PROMPT,
    },
    {
        "role": "user",
        "content": f"""
问题:
{question}
 
参考资料:
{context}
""",
    },
])
 
print(result.answer)
print(result.sources)

12 / AGENT

Agentic RAG

普通 RAG 每次请求都执行固定的“检索 → 生成”。Agentic RAG 则让 Agent 判断:

  • 当前问题是否需要检索。
  • 应该搜索什么。
  • 是否需要改写 Query。
  • 一次检索是否足够。
  • 是否需要调用多个知识源。

最简单的做法是把 Retriever 包装成一个 Tool。

retrieval_tool.pyPython
from langchain.tools import tool
 
@tool
def search_knowledge_base(
    query: str,
) -> str:
    """搜索内部知识库。"""
 
    docs = retriever.invoke(query)
 
    return format_context(docs)
rag_agent.pyPython
from langchain.agents import create_agent
 
agent = create_agent(
    model="openai:gpt-5.4",
    tools=[search_knowledge_base],
    system_prompt="""
你负责回答内部技术资料相关问题。
 
需要知识库信息时调用 search_knowledge_base。
如果检索结果不足,不要猜测。
""",
)
 
result = agent.invoke({
    "messages": [
        {
            "role": "user",
            "content": "LangGraph 如何恢复中断的执行?",
        }
    ]
})
2-STEP RAG

固定检索

每次请求都检索一次。流程简单、延迟稳定、容易测试,普通知识问答优先使用。

AGENTIC

Agent 决定检索

问题类型复杂,需要多步搜索、Query Rewrite 或多个数据源时再使用。

GRAPH

显式工作流

需要检索评分、重写、重试、人工确认等确定流程时,可以使用 LangGraph。

13 / EVAL

RAG 评估

RAG 不能只评估最终 Answer。至少要把 Retrieval 和 Generation 分开看。

rag_result.pyPython
def rag(question: str) -> dict:
    docs = retriever.invoke(question)
 
    answer = generate_answer(
        question=question,
        documents=docs,
    )
 
    return {
        "answer": answer,
        "documents": docs,
    }
Retrieval Relevance

检索出的 Document 是否和用户问题相关。主要检查 Retriever。

Groundedness

最终回答是否能从检索到的资料中得到支持,是否出现脱离 Context 的内容。

Answer Relevance

回答是否真正解决用户问题,而不是只复述检索资料。

Correctness

有标准答案时,将最终回答与 Ground Truth 比较,检查事实是否正确。

检索指标

如果测试集能标注“哪些文档应该被检索到”,可以直接计算 Recall@K、Precision@K、MRR、nDCG 等传统信息检索指标。

retrieval_metrics.pyPython
def recall_at_k(
    retrieved_ids: list[str],
    relevant_ids: set[str],
    k: int,
) -> float:
    if not relevant_ids:
        return 0.0
 
    hits = set(retrieved_ids[:k]) & relevant_ids
 
    return len(hits) / len(relevant_ids)
 
 
def precision_at_k(
    retrieved_ids: list[str],
    relevant_ids: set[str],
    k: int,
) -> float:
    selected = retrieved_ids[:k]
 
    if not selected:
        return 0.0
 
    hits = set(selected) & relevant_ids
 
    return len(hits) / len(selected)

建立最小测试集

不必一开始就准备几千条数据。先覆盖实际业务里最容易出问题的查询:

  • 普通事实查询。
  • 精确 API / 类名查询。
  • 跨多个 Chunk 才能回答的问题。
  • 知识库里不存在答案的问题。
  • 存在相似但错误资料的问题。
  • 有权限过滤要求的问题。

14 / DEBUG

常见问题与排查

搜不到明显存在的内容

先检查 Chunk 是否把关键内容拆散,再检查 Embedding、Query 表达、Metadata Filter 和 top_k。

总是返回相似的几段

检查文档是否大量重复;尝试 MMR、去重或 Rerank。

API 名称搜不准

纯向量搜索对精确标识符未必稳定,可以增加 BM25 / Keyword Search 或 Hybrid Search。

资料正确但回答还是错

检查 Context 是否过长、资料是否互相矛盾、Prompt 是否允许模型使用外部知识。

引用和答案对不上

不要让模型自由生成引用。将引用绑定到检索结果的稳定 metadata。

更新文档后仍搜到旧内容

检查增量索引、Chunk ID 和删除策略,确认旧向量是否真正被清理。

成本越来越高

检查重复 Embedding、过大的 Chunk overlap、过高 top_k、过多 Rerank 候选和不必要的 Agent 检索循环。

15 / PRODUCTION

生产检查清单

  • 原始文档有稳定的 document_id 和 source。
  • Chunk ID 可以稳定复现,不依赖随机值。
  • 文档更新支持增量索引和旧 Chunk 删除。
  • Embedding 模型和索引版本可以追踪。
  • 不同租户或用户的数据在检索阶段完成权限隔离。
  • Chunking 参数经过实际查询集验证,而不是只看字符长度。
  • Retriever 的 top_k 和 Filter 有测试覆盖。
  • 需要精确关键词时,不只依赖 Dense Retrieval。
  • Rerank 只处理有限候选集。
  • 最终进入模型的 Context 有 Token 上限。
  • 引用来自 Document metadata,而不是模型生成。
  • 知识库没有答案时允许系统明确返回“不知道”。
  • 保存 Query、检索结果、分数和最终 Answer 用于调试。
  • 分别评估 Retrieval 和 Generation。
  • Embedding、检索、Rerank、模型调用分别记录延迟和成本。
  • 建立回归测试集,修改 Chunking、Embedding 或 Retriever 后重新跑评估。

增量索引

indexing.pyPython
def index_documents(documents):
    chunks = splitter.split_documents(documents)
 
    for chunk in chunks:
        # 建议使用稳定 ID。
        # 不要每次重建索引都生成随机 ID。
        chunk_id = build_chunk_id(
            source=chunk.metadata["source"],
            section=chunk.metadata.get("section"),
            content=chunk.page_content,
        )
 
        vector_store.add_documents(
            documents=[chunk],
            ids=[chunk_id],
        )

16 / API

API 速查

Document检索文档基本结构langchain_core.documents
RecursiveCharacterTextSplitter通用文本切分langchain_text_splitters
MarkdownHeaderTextSplitter按 Markdown 标题切分langchain_text_splitters
init_embeddings统一初始化 Embeddinglangchain.embeddings
OpenAIEmbeddingsOpenAI Embedding 集成langchain_openai
InMemoryVectorStore内存 Vector Storelangchain_core.vectorstores
BaseRetriever自定义 Retriever 基类langchain_core.retrievers
similarity_search相似度搜索VectorStore
as_retrieverVector Store 转 RetrieverVectorStore
invoke执行 Retriever 查询Retriever

17 / SOURCES

官方资料

本页主要依据 LangChain 当前 Python 官方文档整理,重点核对以下内容:

  • LangChain Text Splitters: 文本切分策略与 RecursiveCharacterTextSplitter
  • LangChain Semantic Search: Document、Embedding、Vector Store 和基础 RAG 流程
  • LangChain Embedding Integrations: Embedding 接口与模型集成
  • LangChain Vector Store Integrations: Vector Store 添加、删除、搜索和过滤接口
  • LangChain Retriever Integrations: Retriever 接口与各种检索实现
  • LangGraph Agentic RAG: Retriever Tool、Query 判断、文档评分和 Query Rewrite
  • LangSmith RAG Evaluation: Correctness、Relevance、Groundedness 和 Retrieval Relevance