RAG（检索增强生成）完整教程

基于 LangChain 官方文档整理
文档来源：https://docs.langchain.com/oss/python/langchain/rag

---

📚 目录

1. 什么是RAG
2. 核心概念
3. 环境准备
4. 第一部分：索引（Indexing）
5. 第二部分：检索与生成（Retrieval & Generation）
6. 两种实现方式对比
7. 完整代码示例
8. 最佳实践
9. 下一步学习

---

什么是RAG？

定义

RAG（Retrieval-Augmented Generation，检索增强生成） 是一种结合了检索系统和生成模型的技术，用于构建能够回答特定领域问题的智能问答系统。

RAG解决什么问题？

传统LLM的局限性：

• ❌ 只能回答训练数据中的知识
• ❌ 无法访问最新信息
• ❌ 容易产生"幻觉"（编造不存在的信息）
• ❌ 无法处理私有/专有数据

RAG的解决方案：

• ✅ 从外部知识库检索相关信息
• ✅ 将检索到的信息作为上下文提供给LLM
• ✅ LLM基于真实数据生成答案
• ✅ 减少幻觉，提高准确性

RAG的典型应用场景

1. 企业知识库问答 - 回答公司文档、政策相关问题
2. 客户支持 - 基于产品文档提供技术支持
3. 学术研究 - 在大量论文中查找信息
4. 法律咨询 - 基于法律文档提供咨询
5. 医疗诊断 - 基于医学文献辅助诊断

---

核心概念

RAG应用包含两个主要阶段：

1. 索引（Indexing）

这是一个离线过程，通常在应用启动前完成。

原始数据 → 加载 → 分割 → 向量化 → 存储到向量数据库

流程图：

┌─────────────┐      ┌──────────────┐      ┌─────────────┐
│  数据源     │      │  文档加载器  │      │  文本分割器 │
│ (PDF/Web)   │ ───> │ (Loaders)    │ ───> │ (Splitters) │
└─────────────┘      └──────────────┘      └─────────────┘
                                                    │
                                                    ▼
┌─────────────┐      ┌──────────────┐      ┌─────────────┐
│  向量数据库 │ <─── │  嵌入模型    │ <─── │  文档块     │
│ (Vector DB) │      │ (Embeddings) │      │ (Chunks)    │
└─────────────┘      └──────────────┘      └─────────────┘

2. 检索与生成（Retrieval and Generation）

这是在线过程，在用户查询时实时执行。

用户问题 → 向量化 → 相似度搜索 → 检索相关文档 → LLM生成答案

流程图：

┌─────────────┐      ┌──────────────┐      ┌─────────────┐
│  用户问题   │      │  嵌入模型    │      │  向量数据库 │
│  (Query)    │ ───> │ (Embeddings) │ ───> │  相似度搜索 │
└─────────────┘      └──────────────┘      └─────────────┘
                                                    │
                                                    ▼
┌─────────────┐      ┌──────────────┐      ┌─────────────┐
│  最终答案   │ <─── │     LLM      │ <─── │  相关文档   │
│  (Answer)   │      │              │      │ + 用户问题  │
└─────────────┘      └──────────────┘      └─────────────┘

---

环境准备

1. 安装依赖

# 核心依赖
pip install langchain langchain-text-splitters langchain-community bs4

# 根据你选择的模型和向量数据库安装对应的包
pip install langchain-openai          # 如果使用OpenAI
pip install langchain-chroma           # 如果使用Chroma

2. 配置 LangSmith（可选，用于调试）

LangSmith 是 LangChain 的调试和监控平台，可以帮助你：

• 查看每一步的执行过程
• 追踪token使用量
• 调试问题

# 设置环境变量
export LANGSMITH_TRACING="true"
export LANGSMITH_API_KEY="your-api-key"

或在Python代码中：

import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "your-api-key"

3. 选择三个核心组件

RAG应用需要三个核心组件：

① 聊天模型（Chat Model）

用于生成最终答案。

from langchain.chat_models import init_chat_model

# OpenAI
model = init_chat_model("gpt-4o-mini")

# Anthropic
model = init_chat_model("claude-3-5-sonnet-20241022")

② 嵌入模型（Embeddings）

用于将文本转换为向量。

from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-large")

③ 向量数据库（Vector Store）

用于存储和检索文档向量。

# 内存向量存储（适合测试）
from langchain_core.vectorstores import InMemoryVectorStore
vector_store = InMemoryVectorStore(embeddings)

# Chroma（适合生产，支持持久化）
from langchain_chroma import Chroma
vector_store = Chroma(
    collection_name="my_docs",
    embedding_function=embeddings,
    persist_directory="./chroma_db"
)

---

第一部分：索引（Indexing）

这部分内容详细说明如何将数据加载、分割并存储到向量数据库。

步骤1：加载文档（Loading）

目标： 从数据源加载文档内容。

示例：从网页加载内容

import bs4
from langchain_community.document_loaders import WebBaseLoader

# 只保留特定CSS类的内容
bs4_strainer = bs4.SoupStrainer(class_=("post-title", "post-header", "post-content"))

# 创建加载器
loader = WebBaseLoader(
    web_paths=("https://lilianweng.github.io/posts/2023-06-23-agent/",),
    bs_kwargs={"parse_only": bs4_strainer},
)

# 加载文档
docs = loader.load()

print(f"加载了 {len(docs)} 个文档")
print(f"总字符数: {len(docs[0].page_content)}")
# 输出: 总字符数: 43131

Document对象结构：

Document(
    page_content="文档的文本内容...",
    metadata={
        "source": "https://...",
        "title": "文档标题"
    }
)

其他常用加载器

# PDF加载器
from langchain_community.document_loaders import PyPDFLoader
loader = PyPDFLoader("path/to/file.pdf")

# 文本文件加载器
from langchain_community.document_loaders import TextLoader
loader = TextLoader("path/to/file.txt")

# CSV加载器
from langchain_community.document_loaders import CSVLoader
loader = CSVLoader("path/to/file.csv")

💡 提示： LangChain提供160+种文档加载器，支持PDF、Word、网页、数据库等各种数据源。

---

步骤2：分割文档（Splitting）

为什么要分割文档？

1. 模型上下文限制 - 大多数LLM有token限制（如4K、8K、128K）
2. 检索精确度 - 较小的文本块更容易匹配查询
3. 提高相关性 - 避免检索到不相关的长文档

使用 RecursiveCharacterTextSplitter

这是推荐的通用文本分割器，它会：

1. 尝试按段落分割
2. 如果段落太长，按句子分割
3. 如果句子太长，按单词分割
4. 最后按字符分割

from langchain_text_splitters import RecursiveCharacterTextSplitter

# 创建分割器
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,        # 每个块的最大字符数
    chunk_overlap=200,      # 块之间的重叠字符数
    add_start_index=True,   # 记录每个块在原文档中的位置
)

# 执行分割
all_splits = text_splitter.split_documents(docs)

print(f"文档被分割为 {len(all_splits)} 个块")
# 输出: 文档被分割为 66 个块

参数说明：

• chunk_size=1000：每个文本块最多1000个字符
• chunk_overlap=200：相邻块之间重叠200字符
  • 为什么需要重叠？ 保持上下文连贯性，避免语义被截断
• add_start_index=True：在metadata中记录原始位置

分割后的Document示例：

Document(
    page_content="这是第一个文本块的内容...",
    metadata={
        "source": "https://...",
        "start_index": 0        # 在原文档中的起始位置
    }
)

其他分割器

# 按字符数分割
from langchain_text_splitters import CharacterTextSplitter

# 按token分割（更精确）
from langchain_text_splitters import TokenTextSplitter

# 按代码结构分割（适合代码文档）
from langchain_text_splitters import PythonCodeTextSplitter

---

步骤3：存储文档（Storing）

目标： 将文档块向量化并存储到向量数据库。

一步完成向量化和存储

# 将所有文档块添加到向量存储
document_ids = vector_store.add_documents(documents=all_splits)

print(f"存储了 {len(document_ids)} 个文档块")
print(f"前3个ID: {document_ids[:3]}")

底层发生了什么？

# LangChain自动执行以下步骤：
for doc in all_splits:
    # 1. 使用嵌入模型将文本转换为向量
    vector = embeddings.embed_query(doc.page_content)

    # 2. 存储向量、原文和元数据到向量数据库
    vector_store.add(
        vector=vector,
        text=doc.page_content,
        metadata=doc.metadata
    )

手动向量化（了解底层原理）

# 手动生成向量
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings()

# 为单个文本生成向量
text = "这是一段示例文本"
vector = embeddings.embed_query(text)

print(f"向量维度: {len(vector)}")  # 输出: 1536 (OpenAI text-embedding-3-large)
print(f"向量前5个值: {vector[:5]}")

✅ 索引完成！ 现在你有了一个可查询的向量数据库，包含了博客文章的66个文本块。

---

第二部分：检索与生成（Retrieval & Generation）

现在数据已经索引好了，我们可以构建实际的RAG应用来回答用户问题。

LangChain提供两种主要实现方式：

1. RAG Agent（代理） - 灵活，LLM自主决定是否搜索
2. RAG Chain（链） - 快速，总是执行搜索

---

方式一：RAG Agent（推荐）

什么是RAG Agent？

Agent是一个能够自主决定是否需要使用工具的智能体。

工作流程：

用户问题 → Agent分析 → 决定是否需要搜索
                    ↓ 是
              调用检索工具 → 获取上下文 → 生成答案
                    ↓ 否
              直接回答（如问候语）

实现步骤

步骤1：创建检索工具

from langchain.tools import tool

@tool(response_format="content_and_artifact")
def retrieve_context(query: str):
    """检索信息以帮助回答查询。"""

    # 执行相似度搜索，返回最相关的2个文档
    retrieved_docs = vector_store.similarity_search(query, k=2)

    # 将文档序列化为字符串（发送给LLM）
    serialized = "\n\n".join(
        (f"来源: {doc.metadata}\n内容: {doc.page_content}")
        for doc in retrieved_docs
    )

    # 返回: (字符串表示, 原始文档对象)
    # 字符串 → LLM
    # 原始对象 → 应用逻辑（如引用来源）
    return serialized, retrieved_docs

关键点：

• @tool 装饰器将函数转换为LangChain工具
• response_format="content_and_artifact" 允许返回两个值：
  • content：文本格式，发送给LLM
  • artifact：原始对象，供应用使用

步骤2：创建Agent

from langchain.agents import create_agent

# 工具列表
tools = [retrieve_context]

# 自定义系统提示（可选）
system_prompt = (
    "你有一个工具可以从博客文章中检索上下文信息。"
    "使用该工具来帮助回答用户查询。"
)

# 创建Agent
agent = create_agent(model, tools, system_prompt=system_prompt)

步骤3：使用Agent

# 提出一个需要多次检索的复杂问题
query = (
    "什么是任务分解的标准方法？\n\n"
    "获得答案后，查找该方法的常见扩展。"
)

# 流式输出Agent的执行过程
for event in agent.stream(
    {"messages": [{"role": "user", "content": query}]},
    stream_mode="values",
):
    event["messages"][-1].pretty_print()

执行过程（输出）：

================================ Human Message =================================
什么是任务分解的标准方法？
获得答案后，查找该方法的常见扩展。

================================== Ai Message ==================================
Tool Calls:
  retrieve_context (call_xxx)
  Args:
    query: 任务分解的标准方法

================================= Tool Message =================================
Name: retrieve_context

来源: {'source': 'https://...'}
内容: 任务分解可以通过...

================================== Ai Message ==================================
Tool Calls:
  retrieve_context (call_yyy)
  Args:
    query: 任务分解方法的常见扩展

================================= Tool Message =================================
Name: retrieve_context

来源: {'source': 'https://...'}
内容: 常见扩展包括...

================================== Ai Message ==================================
任务分解的标准方法通常是思维链（Chain of Thought）...

Agent的智能行为：

1. ✅ 识别需要检索信息
2. ✅ 生成第一个检索查询：“任务分解的标准方法”
3. ✅ 获得答案后，生成第二个查询：“常见扩展”
4. ✅ 综合所有信息，生成最终答案

Agent的优势

优势	说明
智能判断	能识别何时需要搜索，何时不需要
上下文查询	基于对话历史生成更准确的查询
多次检索	可以执行多次搜索来回答复杂问题
处理问候语	无需搜索即可回答简单问候

高级：添加查询参数

你可以强制LLM指定额外的搜索参数：

from typing import Literal

@tool(response_format="content_and_artifact")
def retrieve_context(
    query: str,
    section: Literal["beginning", "middle", "end"]
):
    """
    检索上下文信息。

    参数:
        query: 搜索查询
        section: 文档的哪个部分（开头/中间/结尾）
    """
    # 实现按section过滤的逻辑
    ...

---

方式二：RAG Chain（快速方案）

什么是RAG Chain？

Chain是一个固定流程的处理链，总是执行相同的步骤。

工作流程：

用户问题 → 自动执行搜索 → 检索上下文 → LLM生成答案（单次调用）

Agent vs Chain 对比

特性	RAG Agent	RAG Chain
调用次数	2次（查询生成 + 答案生成）	1次（仅答案生成）
灵活性	高（LLM自主决定）	低（固定流程）
延迟	较高	较低
适用场景	复杂查询、需要多次检索	简单查询、固定场景
控制力	低（LLM可能跳过搜索）	高（总是搜索）

实现步骤

from langchain.agents.middleware import dynamic_prompt, ModelRequest

@dynamic_prompt
def prompt_with_context(request: ModelRequest) -> str:
    """将上下文注入到系统消息中。"""

    # 获取最后一条用户消息
    last_query = request.state["messages"][-1].text

    # 执行搜索（使用原始查询）
    retrieved_docs = vector_store.similarity_search(last_query)

    # 合并文档内容
    docs_content = "\n\n".join(doc.page_content for doc in retrieved_docs)

    # 构建包含上下文的系统消息
    system_message = (
        "你是一个有帮助的助手。在回答时使用以下上下文信息："
        f"\n\n{docs_content}"
    )

    return system_message


# 创建不带工具的Agent（实际上是Chain）
agent = create_agent(model, tools=[], middleware=[prompt_with_context])

使用Chain

query = "什么是任务分解？"

for step in agent.stream(
    {"messages": [{"role": "user", "content": query}]},
    stream_mode="values",
):
    step["messages"][-1].pretty_print()

输出：

================================ Human Message =================================
什么是任务分解？

================================== Ai Message ==================================
任务分解是将复杂任务分解为更小、更易管理的子任务的过程...

特点：

• ✅ 只需要1次LLM调用
• ✅ 延迟更低
• ❌ 无法处理需要多次检索的复杂问题
• ❌ 对所有查询都执行搜索（即使不需要）

---

两种实现方式对比

详细对比表

对比维度	RAG Agent	RAG Chain
LLM调用次数	2次或更多	1次
是否总是搜索	❌ 由LLM决定	✅ 总是搜索
延迟	较高（多次调用）	较低（单次调用）
灵活性	高（自适应）	低（固定流程）
多次检索	✅ 支持	❌ 不支持
处理问候语	✅ 无需搜索	❌ 仍会搜索
复杂查询	✅ 适合	❌ 不适合
简单查询	⚠️ 可能过度	✅ 高效
控制力	低（LLM决策）	高（完全控制）
推荐场景	通用问答、复杂场景	简单查询、受限场景

使用建议

选择 RAG Agent 当：

• ✅ 需要处理各种类型的查询（问候、简单问题、复杂问题）
• ✅ 查询可能需要多次检索
• ✅ 需要LLM根据上下文生成更好的搜索查询
• ✅ 用户体验优先于响应速度

选择 RAG Chain 当：

• ✅ 查询类型固定且简单
• ✅ 总是需要执行搜索
• ✅ 响应速度是关键
• ✅ 需要严格控制流程
• ✅ 成本敏感（减少LLM调用）

---

完整代码示例

示例1：完整的RAG Agent实现

"""
完整的RAG Agent示例
使用场景：通用问答系统
"""

import os
from langchain.chat_models import init_chat_model
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain_community.document_loaders import WebBaseLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain.tools import tool
from langchain.agents import create_agent
import bs4

# ==================== 配置 ====================
os.environ["OPENAI_API_KEY"] = "your-api-key"

# 初始化组件
model = init_chat_model("gpt-4o-mini")
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
vector_store = Chroma(
    collection_name="blog_posts",
    embedding_function=embeddings,
    persist_directory="./chroma_db"
)

# ==================== 索引（只需运行一次）====================
def index_documents():
    """加载、分割并索引文档"""

    # 1. 加载文档
    bs4_strainer = bs4.SoupStrainer(class_=("post-title", "post-header", "post-content"))
    loader = WebBaseLoader(
        web_paths=("https://lilianweng.github.io/posts/2023-06-23-agent/",),
        bs_kwargs={"parse_only": bs4_strainer},
    )
    docs = loader.load()
    print(f"✓ 加载了 {len(docs)} 个文档")

    # 2. 分割文档
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,
        chunk_overlap=200,
        add_start_index=True,
    )
    all_splits = text_splitter.split_documents(docs)
    print(f"✓ 分割为 {len(all_splits)} 个块")

    # 3. 存储到向量数据库
    document_ids = vector_store.add_documents(documents=all_splits)
    print(f"✓ 存储了 {len(document_ids)} 个文档块")

# 首次运行时执行索引
# index_documents()

# ==================== 检索与生成 ====================

# 创建检索工具
@tool(response_format="content_and_artifact")
def retrieve_context(query: str):
    """从博客文章中检索相关上下文。"""
    retrieved_docs = vector_store.similarity_search(query, k=3)
    serialized = "\n\n".join(
        (f"来源: {doc.metadata}\n内容: {doc.page_content}")
        for doc in retrieved_docs
    )
    return serialized, retrieved_docs

# 创建Agent
tools = [retrieve_context]
system_prompt = """
你是一个专业的AI助手，可以回答关于LLM代理的问题。
你有一个工具可以检索相关信息。

使用指南：
- 对于问候或简单问题，直接回答
- 对于需要具体信息的问题，使用检索工具
- 可以多次使用工具来获取完整信息
- 基于检索到的上下文生成准确答案
"""

agent = create_agent(model, tools, system_prompt=system_prompt)

# ==================== 使用Agent ====================

def ask_question(query: str):
    """提问并获取答案"""
    print(f"\n问题: {query}")
    print("=" * 70)

    for event in agent.stream(
        {"messages": [{"role": "user", "content": query}]},
        stream_mode="values",
    ):
        event["messages"][-1].pretty_print()

# 测试不同类型的查询
if __name__ == "__main__":
    # 简单问候（不需要搜索）
    ask_question("你好！")

    # 具体问题（需要搜索）
    ask_question("什么是任务分解？")

    # 复杂问题（需要多次搜索）
    ask_question(
        "什么是Chain of Thought？"
        "它有哪些常见的扩展方法？"
    )

示例2：RAG Chain实现

"""
快速RAG Chain示例
使用场景：固定查询场景，追求低延迟
"""

from langchain.agents.middleware import dynamic_prompt, ModelRequest
from langchain.agents import create_agent

@dynamic_prompt
def prompt_with_context(request: ModelRequest) -> str:
    """自动注入检索上下文"""
    last_query = request.state["messages"][-1].text

    # 自动执行搜索
    retrieved_docs = vector_store.similarity_search(last_query, k=3)
    docs_content = "\n\n".join(doc.page_content for doc in retrieved_docs)

    # 构建系统提示
    system_message = f"""
你是一个AI助手。请基于以下上下文回答用户问题。

【上下文】
{docs_content}

【重要提示】
- 只基于上下文回答
- 如果上下文中没有相关信息，明确说明
- 保持回答简洁准确
"""
    return system_message

# 创建Chain
chain = create_agent(model, tools=[], middleware=[prompt_with_context])

# 使用
def ask_quick(query: str):
    """快速问答"""
    for step in chain.stream(
        {"messages": [{"role": "user", "content": query}]},
        stream_mode="values",
    ):
        step["messages"][-1].pretty_print()

# 测试
ask_quick("什么是任务分解？")

---

最佳实践

1. 文档分割策略

✅ 推荐做法

# 根据内容类型选择合适的chunk_size
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,      # 通用文本：1000-1500
    chunk_overlap=200,    # 重叠：chunk_size的10-20%
    add_start_index=True,
)

不同内容类型的建议：

内容类型	chunk_size	chunk_overlap	说明
技术文档	800-1200	150-200	保持完整的代码块和说明
新闻文章	1000-1500	200-300	保持段落完整性
法律文档	1500-2000	300-400	保持条款完整性
对话记录	500-800	100-150	保持对话连贯性

❌ 避免的做法

# 太小 - 上下文不足
chunk_size=200, chunk_overlap=0

# 太大 - 检索不精确
chunk_size=5000, chunk_overlap=1000

# 无重叠 - 可能截断语义
chunk_overlap=0

2. 检索优化

选择合适的 k 值

# k = 返回的文档数量

# 太少 - 可能遗漏重要信息
retrieved_docs = vector_store.similarity_search(query, k=1)  # ❌

# 推荐 - 平衡相关性和上下文
retrieved_docs = vector_store.similarity_search(query, k=3)  # ✅

# 太多 - 引入噪音，增加成本
retrieved_docs = vector_store.similarity_search(query, k=20) # ❌

建议：

• 简单问答：k=2-3
• 复杂分析：k=4-6
• 总结任务：k=5-10

使用混合检索

# 方案1：相似度搜索（默认）
docs = vector_store.similarity_search(query, k=3)

# 方案2：MMR（最大边际相关性）- 提高多样性
docs = vector_store.max_marginal_relevance_search(
    query, k=3, fetch_k=10
)

# 方案3：带分数的搜索 - 过滤低相关度结果
docs_and_scores = vector_store.similarity_search_with_score(query, k=5)
relevant_docs = [doc for doc, score in docs_and_scores if score > 0.7]

3. Prompt工程

✅ 好的系统提示

system_prompt = """
你是一个专业的AI助手，专门回答关于[领域]的问题。

【工作流程】
1. 仔细阅读用户问题
2. 使用检索工具查找相关信息
3. 基于检索到的上下文生成答案

【回答要求】
- 只基于检索到的上下文回答
- 如果信息不足，明确说明需要更多上下文
- 引用具体的来源
- 保持回答简洁专业

【特别注意】
- 不要编造信息
- 不确定时要说"我不知道"
"""

❌ 避免的做法

# 太简单 - 缺乏指导
system_prompt = "你是一个助手。"

# 太复杂 - 消耗token
system_prompt = """
[5000字的详细说明...]
"""

4. 元数据利用

添加有用的元数据

from langchain_core.documents import Document

# 创建带丰富元数据的文档
doc = Document(
    page_content="文档内容...",
    metadata={
        "source": "https://example.com/doc.pdf",
        "title": "文档标题",
        "author": "作者",
        "date": "2024-01-01",
        "category": "技术文档",
        "page": 1,
    }
)

基于元数据过滤

# 只检索特定类别的文档
from langchain_core.vectorstores import VectorStoreRetriever

retriever = vector_store.as_retriever(
    search_kwargs={
        "k": 3,
        "filter": {"category": "技术文档"}
    }
)

5. 成本优化

选择合适的模型

# 嵌入模型选择
# 小型项目 - 使用免费/便宜的模型
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")  # 更便宜

# 大型项目 - 使用高质量模型
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")  # 更准确

# LLM选择
# 简单任务
model = init_chat_model("gpt-4o-mini")  # 便宜

# 复杂任务
model = init_chat_model("gpt-4o")       # 强大但贵

缓存策略

from langchain.cache import InMemoryCache
from langchain.globals import set_llm_cache

# 启用缓存 - 避免重复查询
set_llm_cache(InMemoryCache())

6. 错误处理

def safe_rag_query(query: str, max_retries=3):
    """带错误处理的RAG查询"""

    for attempt in range(max_retries):
        try:
            # 执行查询
            result = agent.invoke({"messages": [{"role": "user", "content": query}]})
            return result

        except Exception as e:
            print(f"尝试 {attempt + 1} 失败: {e}")

            if attempt == max_retries - 1:
                return {
                    "error": "查询失败",
                    "message": "请稍后重试或联系管理员"
                }

            # 指数退避
            time.sleep(2 ** attempt)

7. 监控和调试

# 使用LangSmith追踪
import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "your-key"

# 添加日志
import logging
logging.basicConfig(level=logging.INFO)

def retrieve_with_logging(query: str):
    logging.info(f"查询: {query}")

    docs = vector_store.similarity_search(query, k=3)

    logging.info(f"检索到 {len(docs)} 个文档")
    for i, doc in enumerate(docs):
        logging.info(f"文档 {i+1} 来源: {doc.metadata.get('source', 'unknown')}")

    return docs

---

下一步学习

1. 高级RAG技术

Multi-Query RAG

# 为单个用户查询生成多个搜索查询
queries = [
    "什么是任务分解？",
    "任务分解有哪些方法？",
    "如何实现任务分解？"
]
all_docs = []
for q in queries:
    docs = vector_store.similarity_search(q, k=2)
    all_docs.extend(docs)

重排序（Re-ranking）

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor

# 使用LLM重新排序和过滤文档
compressor = LLMChainExtractor.from_llm(model)
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vector_store.as_retriever()
)

HyDE (Hypothetical Document Embeddings)

# 先让LLM生成假设答案，然后用假设答案搜索
hypothetical_answer = model.invoke(f"回答这个问题：{query}")
docs = vector_store.similarity_search(hypothetical_answer, k=3)

2. 添加对话记忆

from langchain.memory import ConversationBufferMemory

# 添加记忆以支持多轮对话
memory = ConversationBufferMemory(return_messages=True)

# 在Agent中使用记忆
agent = create_agent(
    model,
    tools,
    memory=memory  # 记住对话历史
)

3. 流式输出

# 实时显示生成的答案
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": query}]},
    stream_mode="messages",
):
    if chunk.content:
        print(chunk.content, end="", flush=True)

4. 结构化输出

from pydantic import BaseModel, Field

class Answer(BaseModel):
    """结构化答案"""
    answer: str = Field(description="问题的答案")
    confidence: float = Field(description="置信度 (0-1)")
    sources: list[str] = Field(description="信息来源")

# 强制LLM返回结构化数据
structured_model = model.with_structured_output(Answer)

5. 部署

使用LangServe部署API

from langserve import add_routes
from fastapi import FastAPI

app = FastAPI(title="RAG API")

# 添加RAG端点
add_routes(app, agent, path="/rag")

# 运行: uvicorn main:app --reload

使用Streamlit创建UI

import streamlit as st

st.title("RAG问答系统")

query = st.text_input("输入你的问题：")

if st.button("提问"):
    with st.spinner("思考中..."):
        response = agent.invoke({"messages": [{"role": "user", "content": query}]})
        st.write(response)

---

参考资源

官方文档

• LangChain RAG Tutorial
• LangChain Documentation
• LangSmith

相关教程

• Semantic Search Tutorial
• LangGraph Agentic RAG

扩展阅读

• Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks - RAG原始论文
• LangChain Cookbook

---

总结

RAG的核心要点

1. 两个阶段

   • 索引：加载 → 分割 → 向量化 → 存储
   • 检索生成：查询 → 搜索 → 生成答案

2. 两种实现方式

   • Agent：灵活，适合复杂场景
   • Chain：快速，适合简单场景

3. 三个核心组件

   • 聊天模型（生成答案）
   • 嵌入模型（文本向量化）
   • 向量数据库（存储和检索）

4. 关键最佳实践

   • 合理的chunk_size和overlap
   • 适当的k值（3-5）
   • 清晰的系统提示
   • 丰富的元数据
   • 错误处理和监控

下一步行动

1. ✅ 选择一个小项目实践（如个人知识库）
2. ✅ 实现基础的RAG Agent
3. ✅ 添加对话记忆
4. ✅ 优化检索质量
5. ✅ 部署为API或Web应用

---

祝学习愉快！ 🎉

如有问题，欢迎查阅官方文档或社区讨论。