5.7 Retrieval

本节介绍 LangChain 中的检索功能，实现 RAG（检索增强生成）模式。

什么是 Retrieval？

检索（Retrieval） 解决了 LLM 的两个根本限制：

限制	说明
有限上下文	模型的上下文窗口有限
静态知识	模型知识有截止日期

检索在查询时获取相关的外部信息，形成 RAG（Retrieval-Augmented Generation，检索增强生成） 的基础。

知识库

知识库（Knowledge Base） 是文档或结构化数据的存储库：

自建知识库：使用 LangChain 工具构建
连接现有系统：SQL 数据库、CRM、文档系统

检索管道组件

┌─────────────────────────────────────────────────────────┐
│                    Retrieval Pipeline                    │
├─────────────────────────────────────────────────────────┤
│                                                          │
│  ┌──────────┐    ┌──────────┐    ┌──────────┐          │
│  │ Document │───▶│   Text   │───▶│ Embedding│          │
│  │  Loader  │    │ Splitter │    │  Model   │          │
│  └──────────┘    └──────────┘    └──────────┘          │
│       │               │               │                 │
│       ▼               ▼               ▼                 │
│  ┌─────────────────────────────────────────────────┐   │
│  │                 Vector Store                     │   │
│  └─────────────────────────────────────────────────┘   │
│                          │                              │
│                          ▼                              │
│                   ┌──────────┐                         │
│                   │ Retriever│                         │
│                   └──────────┘                         │
│                                                          │
└─────────────────────────────────────────────────────────┘

组件	功能
Document Loaders	从外部源加载数据，返回 `Document` 对象
Text Splitters	将大文档分割成小块
Embedding Models	将文本转换为向量
Vector Stores	存储和搜索向量
Retrievers	基于查询返回相关文档

三种 RAG 实现方式

1. 两步 RAG（2-Step RAG）

检索总是在生成之前发生：

python

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_core.prompts import ChatPromptTemplate

# 1. 创建向量存储
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_texts(
    ["LangChain 是一个 AI 框架", "Python 是一种编程语言"],
    embeddings
)

# 2. 创建检索器
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# 3. 检索相关文档
docs = retriever.invoke("什么是 LangChain?")

# 4. 生成回答
model = ChatOpenAI(model="gpt-4o")
prompt = ChatPromptTemplate.from_template("""
基于以下上下文回答问题:

上下文: {context}

问题: {question}
""")

context = "\n".join([doc.page_content for doc in docs])
response = model.invoke(prompt.format(context=context, question="什么是 LangChain?"))

特点：

✅ 简单可预测
✅ 高控制性
✅ 延迟稳定
适用场景：FAQ、文档问答

2. Agentic RAG

Agent 决定何时以及如何检索：

python

from langchain_core.tools import tool
from langchain.agents import create_agent
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings

# 创建向量存储
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_texts([...], embeddings)

@tool
def search_knowledge_base(query: str) -> str:
    """搜索知识库获取相关信息"""
    docs = vectorstore.similarity_search(query, k=3)
    return "\n\n".join([doc.page_content for doc in docs])

@tool
def fetch_documentation(url: str) -> str:
    """从 URL 获取文档内容"""
    # 验证 URL 并获取内容
    import requests
    response = requests.get(url)
    return response.text[:5000]

agent = create_agent(
    model="gpt-4o",
    tools=[search_knowledge_base, fetch_documentation],
    system_prompt="你是一个研究助手，可以搜索知识库和获取在线文档。"
)

特点：

✅ 高灵活性
✅ 支持多数据源
❌ 延迟不稳定
适用场景：研究助手、多工具场景

3. 混合 RAG（Hybrid RAG）

结合两种方式，添加验证层：

python

from langchain_core.tools import tool
from langchain.agents import create_agent, before_model, after_model

# 查询增强
@before_model
def enhance_query(state, runtime):
    """增强用户查询"""
    last_message = state["messages"][-1].content

    # 添加上下文提示
    enhanced = f"用户问题: {last_message}\n请先搜索相关信息再回答。"
    state["messages"][-1].content = enhanced

    return None

# 答案验证
@after_model
def validate_answer(state, response, runtime):
    """验证答案质量"""
    answer = response.content

    # 检查是否引用了来源
    if "根据" not in answer and "来源" not in answer:
        # 提示模型添加引用
        pass

    return None

@tool
def search_with_validation(query: str) -> str:
    """搜索并验证结果"""
    # 1. 检索
    docs = retriever.invoke(query)

    # 2. 相关性验证
    relevant_docs = [d for d in docs if d.metadata.get("score", 0) > 0.7]

    if not relevant_docs:
        return "未找到高度相关的信息"

    return "\n\n".join([doc.page_content for doc in relevant_docs])

agent = create_agent(
    model="gpt-4o",
    tools=[search_with_validation],
    middleware=[enhance_query, validate_answer],
)

特点：

✅ 中等灵活性和控制性
✅ 质量检查
适用场景：领域特定问答、质量要求高的场景

构建块详解

Document Loaders

从各种来源加载文档：

python

# 从文本文件加载
from langchain_community.document_loaders import TextLoader
loader = TextLoader("./document.txt")
docs = loader.load()

# 从 PDF 加载
from langchain_community.document_loaders import PyPDFLoader
loader = PyPDFLoader("./document.pdf")
docs = loader.load()

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

# 从目录加载
from langchain_community.document_loaders import DirectoryLoader
loader = DirectoryLoader("./docs/", glob="**/*.md")
docs = loader.load()

Text Splitters

分割长文档：

python

from langchain_text_splitters import (
    RecursiveCharacterTextSplitter,
    CharacterTextSplitter,
)

# 递归字符分割（推荐）
splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", " ", ""]
)
chunks = splitter.split_documents(docs)

# 按字符分割
splitter = CharacterTextSplitter(
    separator="\n",
    chunk_size=1000,
    chunk_overlap=200
)
chunks = splitter.split_documents(docs)

Embedding Models

将文本转换为向量：

python

from langchain_openai import OpenAIEmbeddings
from langchain_community.embeddings import HuggingFaceEmbeddings

# OpenAI Embeddings
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# HuggingFace Embeddings（本地）
embeddings = HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2")

# 使用
vector = embeddings.embed_query("Hello, world!")
vectors = embeddings.embed_documents(["Hello", "World"])

Vector Stores

存储和搜索向量：

python

from langchain_community.vectorstores import FAISS, Chroma
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings()

# FAISS（内存）
vectorstore = FAISS.from_documents(docs, embeddings)

# Chroma（持久化）
vectorstore = Chroma.from_documents(
    docs,
    embeddings,
    persist_directory="./chroma_db"
)

# 相似度搜索
results = vectorstore.similarity_search("查询内容", k=3)

# 带分数的搜索
results = vectorstore.similarity_search_with_score("查询内容", k=3)

Retrievers

统一的检索接口：

python

# 从向量存储创建
retriever = vectorstore.as_retriever(
    search_type="similarity",  # "similarity" | "mmr"
    search_kwargs={"k": 5}
)

# 检索
docs = retriever.invoke("查询内容")

# 异步检索
docs = await retriever.ainvoke("查询内容")

完整 RAG 示例

python

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_community.document_loaders import DirectoryLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.tools import tool
from langchain.agents import create_agent

# 1. 加载文档
loader = DirectoryLoader("./knowledge_base/", glob="**/*.md")
docs = loader.load()

# 2. 分割文档
splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)
chunks = splitter.split_documents(docs)

# 3. 创建向量存储
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_documents(chunks, embeddings)

# 4. 创建检索工具
@tool
def search_docs(query: str) -> str:
    """搜索文档库获取相关信息"""
    docs = vectorstore.similarity_search(query, k=3)

    if not docs:
        return "未找到相关信息"

    results = []
    for i, doc in enumerate(docs, 1):
        source = doc.metadata.get("source", "未知来源")
        results.append(f"[{i}] {doc.page_content}\n来源: {source}")

    return "\n\n".join(results)

# 5. 创建 Agent
agent = create_agent(
    ChatOpenAI(model="gpt-4o"),
    tools=[search_docs],
    system_prompt="""你是一个知识助手，帮助用户查找和理解文档内容。

    使用 search_docs 工具搜索相关信息，然后基于搜索结果回答问题。
    务必注明信息来源。"""
)

# 6. 使用
result = agent.invoke({
    "messages": [{"role": "user", "content": "如何配置 API 密钥？"}]
})

print(result["messages"][-1].content)

RAG 方式对比

方式	控制性	灵活性	延迟	适用场景
2-Step RAG	高	低	快	FAQ、文档问答
Agentic RAG	低	高	不定	研究助手、多工具
Hybrid RAG	中	中	中	领域问答、质量优先

最佳实践

实践	说明
合适的块大小	通常 500-1500 字符
重叠设置	10-20% 的重叠保持上下文
元数据保留	保留来源信息便于引用
混合搜索	结合关键词和语义搜索
定期更新	保持知识库时效性

上一节：5.6 Multi-agent

下一节：5.8 Long-term Memory

5.7 Retrieval ​

什么是 Retrieval？ ​

知识库 ​

检索管道组件 ​

三种 RAG 实现方式 ​

1. 两步 RAG（2-Step RAG） ​

2. Agentic RAG ​

3. 混合 RAG（Hybrid RAG） ​

构建块详解 ​

Document Loaders ​

Text Splitters ​

Embedding Models ​

Vector Stores ​

Retrievers ​

完整 RAG 示例 ​

RAG 方式对比 ​

最佳实践 ​

5.7 Retrieval

什么是 Retrieval？

知识库

检索管道组件

三种 RAG 实现方式

1. 两步 RAG（2-Step RAG）

2. Agentic RAG

3. 混合 RAG（Hybrid RAG）

构建块详解

Document Loaders

Text Splitters

Embedding Models

Vector Stores

Retrievers

完整 RAG 示例

RAG 方式对比

最佳实践