Retrieval 概览

检索增强生成（Retrieval-Augmented Generation，RAG）是 LangChain 最核心的应用模式之一。它通过从外部知识源检索相关信息，增强语言模型的回答能力，解决模型知识有限和幻觉问题。

RAG 工作流程

典型 RAG 流程分为两个阶段：

阶段一：数据摄入（Ingestion）

原始文档 → 文档加载器 → 文本分割 → 向量嵌入 → 向量存储

阶段二：检索与生成（Retrieval + Generation）

用户问题 → 嵌入问题 → 向量检索 → 获取相关文档 → 模型生成答案

文档加载器（Document Loaders）

文档加载器用于从各种数据源加载文档内容：

python

from langchain.document_loaders import (
    TextLoader,
    PDFLoader,
    CSVLoader,
    WebBaseLoader,
    JSONLoader,
)

# 从文本文件加载
loader = TextLoader("data/notes.txt")
documents = loader.load()

# 从 PDF 加载
loader = PDFLoader("data/report.pdf")
documents = loader.load()

# 从网页加载
loader = WebBaseLoader("https://example.com/docs")
documents = loader.load()

print(f"加载了 {len(documents)} 个文档")
print(documents[0].page_content[:200])  # 文档内容
print(documents[0].metadata)            # 元数据（来源、页码等）

支持的加载器：

加载器	支持格式	用途
`TextLoader`	`.txt`	纯文本文件
`PDFLoader`	`.pdf`	PDF 文档
`CSVLoader`	`.csv`	表格数据
`JSONLoader`	`.json`	JSON 数据
`WebBaseLoader`	URL	网页内容
`MarkdownLoader`	`.md`	Markdown 文档
`S3Loader`	S3	AWS S3 文档
`GitHubLoader`	GitHub	GitHub 仓库文件

文本分割器（Text Splitters）

将长文档分割为合适大小的块（chunks），是 RAG 性能的关键：

python

from langchain.text_splitter import (
    RecursiveCharacterTextSplitter,
    CharacterTextSplitter,
    MarkdownTextSplitter,
    TokenTextSplitter,
)

# 递归字符分割器（推荐通用）
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,          # 每个块的目标大小（字符）
    chunk_overlap=200,        # 块之间的重叠字符数
    separators=["\n\n", "\n", "。", "，", " ", ""],  # 优先分割的字符
)

chunks = text_splitter.split_documents(documents)
print(f"分割为 {len(chunks)} 个块")

# 按 token 数分割
token_splitter = TokenTextSplitter(
    chunk_size=500,          # 每个块的目标 token 数
    chunk_overlap=50,
)
token_chunks = token_splitter.split_documents(documents)

分割策略选择

分割器	适用场景
`RecursiveCharacterTextSplitter`	通用文档，按层级分割
`MarkdownTextSplitter`	Markdown 文档，保留标题结构
`TokenTextSplitter`	需要精确控制 token 数
`PythonCodeTextSplitter`	Python 代码文件
`CharacterTextSplitter`	简单固定长度分割

嵌入向量（Embeddings）

将文本转换为向量表示，用于语义搜索：

python

from langchain.embeddings import init_embeddings

# 初始化嵌入模型
embeddings = init_embeddings("openai/text-embedding-3-small")

# 单个文本嵌入
vector = embeddings.embed_query("什么是人工智能？")
print(f"向量维度：{len(vector)}")

# 批量嵌入文档
vectors = embeddings.embed_documents([
    "人工智能是计算机科学的一个分支。",
    "机器学习是 AI 的子领域。",
    "深度学习使用多层神经网络。",
])
print(f"生成了 {len(vectors)} 个向量")

支持的嵌入模型：

python

# OpenAI
embeddings = init_embeddings("openai/text-embedding-3-small")  # 1536 维
embeddings = init_embeddings("openai/text-embedding-3-large")  # 3072 维

# Google
embeddings = init_embeddings("gemini/text-embedding-004")

# 本地模型
embeddings = init_embeddings("ollama/nomic-embed-text")
embeddings = init_embeddings("ollama/bge-m3")

向量存储（Vector Stores）

向量存储用于存储和检索嵌入向量：

python

from langchain.vectorstores import InMemoryVectorStore, ChromaVectorStore

# 内存向量存储（适合开发/测试）
vector_store = InMemoryVectorStore(embeddings)
vector_store.add_documents(chunks)

# ChromaDB（适合轻量生产）
vector_store = ChromaVectorStore(
    embeddings=embeddings,
    persist_directory="./chroma_db",  # 持久化目录
)
vector_store.add_documents(chunks)

# 其他向量数据库
# vector_store = PineconeVectorStore(embeddings, index_name="my-index")
# vector_store = QdrantVectorStore(embeddings, url="http://localhost:6333")
# vector_store = WeaviateVectorStore(embeddings)

支持的向量存储：

存储	部署方式	适用场景
`InMemoryVectorStore`	内存	开发测试
`ChromaVectorStore`	本地文件	小到中型项目
`PineconeVectorStore`	云服务	生产环境
`QdrantVectorStore`	本地/云	高性能需求
`WeaviateVectorStore`	本地/云	混合搜索
`PostgresVectorStore`	PostgreSQL	已有 PG 基础设施

检索器（Retrievers）

检索器封装了搜索逻辑，返回与查询最相关的文档：

python

from langchain.vectorstores import InMemoryVectorStore
from langchain.retrievers import VectorStoreRetriever

# 基础向量检索器
retriever = VectorStoreRetriever(
    vector_store=vector_store,
    search_type="similarity",    # 相似度搜索
    search_kwargs={"k": 4},      # 返回前 4 个最相关文档
)

# 检索
results = retriever.invoke("什么是机器学习？")
for doc in results:
    print(f"[相关度] {doc.metadata.get('score', 'N/A')}")
    print(f"[内容] {doc.page_content[:100]}\n")

检索策略

python

# 相似度搜索（默认）
retriever = VectorStoreRetriever(
    vector_store=vector_store,
    search_type="similarity",
    search_kwargs={"k": 5},
)

# MMR（最大边际相关性）
# 平衡相关性和多样性，避免结果过于相似
retriever = VectorStoreRetriever(
    vector_store=vector_store,
    search_type="mmr",
    search_kwargs={"k": 5, "fetch_k": 20, "lambda_mult": 0.5},
)

# 相似度阈值搜索
# 只返回超过相似度阈值的结果
retriever = VectorStoreRetriever(
    vector_store=vector_store,
    search_type="similarity_score_threshold",
    search_kwargs={"k": 10, "score_threshold": 0.6},
)

构建完整 RAG 系统

python

from langchain import create_agent
from langchain.chat_models import init_chat_model
from langchain.embeddings import init_embeddings
from langchain.vectorstores import InMemoryVectorStore
from langchain.retrievers import VectorStoreRetriever
from langchain.tools import tool

# 1. 摄入文档
from langchain.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter

loader = TextLoader("knowledge_base.txt")
documents = loader.load()

splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
chunks = splitter.split_documents(documents)

# 2. 构建向量存储
embeddings = init_embeddings("openai/text-embedding-3-small")
vector_store = InMemoryVectorStore(embeddings)
vector_store.add_documents(chunks)

# 3. 创建检索工具
@tool
def search_knowledge_base(query: str) -> list[dict]:
    """搜索知识库，返回相关文档。"""
    retriever = VectorStoreRetriever(
        vector_store=vector_store,
        search_type="similarity",
        search_kwargs={"k": 3},
    )
    results = retriever.invoke(query)
    return [
        {"content": doc.page_content, "source": doc.metadata.get("source", "")}
        for doc in results
    ]

# 4. 创建检索增强 Agent
agent = create_agent(
    model=init_chat_model("openai/gpt-4o"),
    tools=[search_knowledge_base],
    system_prompt="""你是一个知识助手。当回答问题时：
1. 首先搜索知识库获取相关信息
2. 基于搜索结果回答
3. 如果搜索不到，告诉用户你不知道
4. 引用信息来源""",
)

# 5. 查询
result = agent.invoke({
    "messages": [{"role": "user", "content": "什么是向量数据库？"}]
})
print(result)

高级检索技术

假设性文档嵌入（HyDE）

先让模型生成一个假设性答案，然后用这个答案去检索：

python

from langchain.chains import HyDEChain

hyde = HyDEChain(
    llm=init_chat_model("openai/gpt-4o-mini"),
    embeddings=embeddings,
)

# HyDE 提升检索准确性
query = "如何优化 Python 性能？"
results = hyde.retrieve(query, retriever=base_retriever)

多查询检索

生成多个不同角度的查询来检索：

python

from langchain.retrievers import MultiQueryRetriever

multi_query_retriever = MultiQueryRetriever(
    retriever=base_retriever,
    llm=init_chat_model("openai/gpt-4o-mini"),
    query_count=3,  # 生成 3 个不同查询
)

results = multi_query_retriever.invoke("Python 的性能特点")

重排序（Reranking）

检索更多文档后，用重排序模型重新排序：

python

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

# 先检索更多文档，再用重排序模型精排
retriever = VectorStoreRetriever(
    vector_store=vector_store,
    search_kwargs={"k": 20},  # 检索更多候选
)

compressor = CohereRerank(model="rerank-english-v3.0")
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=retriever,
)

results = compression_retriever.invoke("机器学习基础概念")

检索优化建议

问题	解决方案
检索结果不相关	改进块大小（500-1500字符）、使用重排序
遗漏关键信息	增加重叠（overlap）、使用多查询检索
回答冗长	限制检索数量（k=3-5）、优化提示词
知识更新慢	建立增量索引、定期重索引
处理多种语言	使用多语言嵌入模型（如 bge-m3）
性能慢	使用批量嵌入、向量索引加速（HNSW/IVF）

最佳实践

块大小选择：通用建议 500-1000 字符，根据文档类型调整
重叠设置：chunk_overlap 设为 10-20% 的块大小
检索数量：k=3-5 通常效果最好
嵌入模型选择：根据语言和领域选择
元数据保留：保留来源、页码便于溯源
定期更新索引：知识库变化时需要重建索引
混合检索：结合关键词搜索和向量搜索

下一步

了解项目结构建议中 RAG 项目的布局
学习模型接口不同模型的检索能力
查看提示词模板如何优化 RAG 提示词

Retrieval 概览 ​

RAG 工作流程 ​

阶段一：数据摄入（Ingestion） ​

阶段二：检索与生成（Retrieval + Generation） ​

文档加载器（Document Loaders） ​

文本分割器（Text Splitters） ​

分割策略选择 ​

嵌入向量（Embeddings） ​

向量存储（Vector Stores） ​

检索器（Retrievers） ​

检索策略 ​

构建完整 RAG 系统 ​

高级检索技术 ​

假设性文档嵌入（HyDE） ​

多查询检索 ​

重排序（Reranking） ​

检索优化建议 ​

最佳实践 ​

下一步 ​

Retrieval 概览

RAG 工作流程

阶段一：数据摄入（Ingestion）

阶段二：检索与生成（Retrieval + Generation）

文档加载器（Document Loaders）

文本分割器（Text Splitters）

分割策略选择

嵌入向量（Embeddings）

向量存储（Vector Stores）

检索器（Retrievers）

检索策略

构建完整 RAG 系统

高级检索技术

假设性文档嵌入（HyDE）

多查询检索

重排序（Reranking）

检索优化建议

最佳实践

下一步