Resource / Python
RAG
Cheatsheet
从文档切分、Embedding、Vector Store 和 Retriever 开始,一直到混合检索、Rerank、Context 组装、引用、Agentic RAG 和评估。代码以 LangChain Python 当前接口为主,检索方法本身不绑定具体向量数据库。
- 资料核验
- 2026-09-09
- 语言范围
- Python · LangChain
- 适用范围
- RAG · Retrieval
documentschunksembeddingretrievecontextRAG 不是“把所有资料塞进 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 示例。
pip install -U \
langchain \
langchain-core \
langchain-text-splitters \
langchain-openai
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 最基本的工作就是:
- 保存可检索的 Document。
- 根据问题检索相关 Document。
- 把 Document 内容组成 Context。
- 让模型基于 Context 回答。
03 / DOCUMENT
Document 与 Metadata
LangChain 的检索接口最终通常围绕 Document 工作。正文放在 page_content,来源、标题、章节和业务字段放在 metadata。
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 至少保留什么
原始文件、页面、数据库记录或其他来源标识。用于引用和问题排查。
标题、章节或段落位置。比只保存文件名更容易定位原文。
稳定的原始文档 ID。用于更新、删除和增量索引。
多用户或企业系统应保存 tenant、user、department 等检索权限字段。
对经常更新的资料,保留版本、更新时间或内容哈希。
04 / CHUNK
文档切分
Chunk 的目标不是得到大小完全一致的文本块,而是让每个块能够独立被检索,同时尽量保留完整语义。
对普通文本,优先从 RecursiveCharacterTextSplitter 开始。
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 大小。
Chunk 太小
容易把定义、条件、结论拆开。虽然命中更精确,但内容本身可能不足以回答问题。
大小合适
一个 Chunk 能表达相对完整的信息,同时不会包含大量与查询无关的文本。
Chunk 太大
Embedding 表达会变得模糊,检索回来后还会占用大量模型 Context。
Markdown 文档
对技术文档和博客,不建议完全忽略标题结构直接按字符切。可以先按标题拆章节,再对过长章节做第二次切分。
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。
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。
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 也提供统一初始化接口:
from langchain.embeddings import init_embeddings
embeddings = init_embeddings(
"openai:text-embedding-3-small"
)
选择 Embedding 模型时看什么
中文或中英混合知识库必须实际测试中文检索,不要只看英文榜单。
维度影响索引空间和检索成本。更高维度不等于一定更准确。
Chunk 大小不能超过 Embedding 模型支持的输入限制。
部分模型会区分检索查询和索引文档的任务类型,应按模型说明使用。
Embedding 模型发生变化后,旧向量通常不能直接与新向量混用。
06 / INDEX
Vector Store
Vector Store 负责保存向量以及对应的文档信息,并执行相似度搜索。LangChain 为不同实现提供了一套相对统一的接口。
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])
查看分数
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
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 = 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
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"
)
08 / SEARCH
检索策略
Similarity Search
最常见的 Dense Retrieval。适合语义相近但用词不同的查询。
MMR
Maximum Marginal Relevance 会在相关性和结果多样性之间做平衡,适合相似 Chunk 很多、结果容易重复的知识库。
retriever = vector_store.as_retriever(
search_type="mmr",
search_kwargs={
"k": 5,
"fetch_k": 20,
"lambda_mult": 0.5,
},
)
docs = retriever.invoke(
"LangGraph persistence"
)
Keyword / BM25
对错误码、API 名、类名、产品型号、法规编号等精确关键词,纯向量检索未必稳定。BM25 或其他关键词搜索往往更合适。
Hybrid Search
Hybrid Search 把 Dense Retrieval 和 Sparse / Keyword Retrieval 结合起来。技术知识库通常很适合这种方式,因为查询里经常同时存在自然语言和精确标识符。
query
│
├─ Dense Search ── top 20
│
└─ BM25 / Sparse Search ── top 20
│
▼
Merge / RRF
│
▼
top 20
│
▼
Rerank
│
▼
top 5
DenseBM25HybridMMR09 / RERANK
Rerank
初步检索的目标偏向“不要漏掉相关资料”,Rerank 再对候选结果做更精细的排序。
常见流程是:
- 先召回 10~50 个候选 Chunk。
- 使用 Reranker 对 Query + Document 重新打分。
- 只保留前几条进入模型 Context。
# 常见两阶段检索思路:
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 拼起来。模型还需要知道每段资料来自哪里。
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 最重要的不是写得复杂,而是明确资料使用规则和资料不足时的行为。
SYSTEM_PROMPT = """
你根据提供的参考资料回答问题。
要求:
1. 优先使用参考资料中的事实。
2. 不要把模型已有知识冒充成参考资料内容。
3. 资料不足时直接说明无法从当前资料确定。
4. 涉及关键事实时标明资料编号。
5. 不要编造引用、文件名、页码或链接。
"""
不要让模型自己编来源
更可靠的方式是应用提前给每个 Document 分配来源编号,让模型只返回使用过的编号。
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。
from langchain.tools import tool
@tool
def search_knowledge_base(
query: str,
) -> str:
"""搜索内部知识库。"""
docs = retriever.invoke(query)
return format_context(docs)
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 如何恢复中断的执行?",
}
]
})
固定检索
每次请求都检索一次。流程简单、延迟稳定、容易测试,普通知识问答优先使用。
Agent 决定检索
问题类型复杂,需要多步搜索、Query Rewrite 或多个数据源时再使用。
显式工作流
需要检索评分、重写、重试、人工确认等确定流程时,可以使用 LangGraph。
13 / EVAL
RAG 评估
RAG 不能只评估最终 Answer。至少要把 Retrieval 和 Generation 分开看。
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 等传统信息检索指标。
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。
纯向量搜索对精确标识符未必稳定,可以增加 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 后重新跑评估。
增量索引
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 速查
langchain_core.documentslangchain_text_splitterslangchain_text_splitterslangchain.embeddingslangchain_openailangchain_core.vectorstoreslangchain_core.retrieversVectorStoreVectorStoreRetriever17 / 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