构建基础RAG管道：文档加载 -> 分割 -> 存储 -> 检索

核心知识点讲解

什么是RAG？

RAG（Retrieval-Augmented Generation）是一种结合了信息检索和生成模型的技术，它的主要特点包括：

1. 外部知识增强：通过检索外部知识库的相关信息来增强生成模型的回答
2. 减少幻觉：基于检索到的事实性信息生成回答，减少模型的幻觉
3. 知识更新：无需重新训练模型，通过更新知识库即可更新模型的知识
4. 领域适应：可以快速适应特定领域的知识
5. 可解释性：可以提供回答的知识来源，增强可解释性

在AI智能体系统中，RAG主要用于：

• 知识库问答：基于企业或个人知识库回答问题
• 信息整合：整合多个来源的信息生成综合回答
• 实时信息获取：结合实时检索的信息生成回答
• 个性化服务：基于用户的历史数据和偏好提供个性化回答

RAG管道的核心组件

一个完整的RAG管道包含以下核心组件：

1. 文档加载器（Document Loader）

文档加载器负责从不同来源加载文档，支持多种文件格式：

• 文本文件：TXT、MD、JSON等
• 办公文档：PDF、DOCX、PPTX、XLSX等
• 网页内容：HTML、URL等
• 数据库：关系型数据库、NoSQL数据库等
• 其他来源：API、邮件、代码仓库等

2. 文本分割器（Text Splitter）

文本分割器负责将长文档分割成短文本块，以便于：

• 适应上下文窗口：符合语言模型的上下文窗口大小
• 提高检索精度：更精确地匹配相关内容
• 减少噪声：减少无关信息的干扰
• 提高处理速度：加快嵌入和检索速度

3. 嵌入模型（Embedding Model）

嵌入模型负责将文本转换为向量表示：

• 语义捕捉：捕捉文本的语义信息
• 向量生成：生成低维稠密的向量
• 相似度计算：支持基于向量的相似度计算

4. 向量数据库（Vector Database）

向量数据库负责存储和检索向量：

• 向量存储：存储文本块的向量表示
• 索引构建：构建高效的索引加速检索
• 相似度搜索：根据查询向量检索最相似的文本块

5. 检索器（Retriever）

检索器负责根据用户查询检索相关信息：

• 查询处理：处理用户的自然语言查询
• 向量转换：将查询转换为向量
• 相似度匹配：匹配最相关的文本块
• 结果排序：根据相似度排序返回结果

6. 生成器（Generator）

生成器负责基于检索结果生成回答：

• 上下文构建：构建包含检索结果的上下文
• 提示词设计：设计有效的提示词引导生成
• 回答生成：生成准确、连贯的回答
• 引用标注：标注回答的知识来源

实用案例分析

案例1：构建基础RAG管道

# 安装必要的库
# pip install langchain chromadb pypdf sentence-transformers

from langchain.document_loaders import PyPDFLoader, TextLoader, WebBaseLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI
import os

# 设置环境变量
os.environ["OPENAI_API_KEY"] = "your-openai-api-key"

# 1. 文档加载
print("=== 文档加载 ===")

# 加载PDF文件
pdf_loader = PyPDFLoader("example.pdf")
pdf_docs = pdf_loader.load()
print(f"PDF文档数量: {len(pdf_docs)}")

# 加载文本文件
text_loader = TextLoader("example.txt", encoding="utf-8")
text_docs = text_loader.load()
print(f"文本文档数量: {len(text_docs)}")

# 加载网页内容
web_loader = WebBaseLoader("https://example.com")
web_docs = web_loader.load()
print(f"网页文档数量: {len(web_docs)}")

# 合并所有文档
all_docs = pdf_docs + text_docs + web_docs
print(f"总文档数量: {len(all_docs)}")

# 2. 文本分割
print("\n=== 文本分割 ===")

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,      # 每个块的大小
    chunk_overlap=200,     # 块之间的重叠
    length_function=len,   # 长度计算函数
    separators=["\n\n", "\n", " ", ""]  # 分割符
)

# 分割文档
split_docs = text_splitter.split_documents(all_docs)
print(f"分割后的文档数量: {len(split_docs)}")
print(f"第一个文档块: {split_docs[0].page_content[:200]}...")

# 3. 嵌入模型
print("\n=== 嵌入模型 ===")

embeddings = HuggingFaceEmbeddings(
    model_name="sentence-transformers/all-MiniLM-L6-v2",  # 轻量级模型
    model_kwargs={"device": "cpu"},  # 运行设备
    encode_kwargs={"normalize_embeddings": True}  # 归一化嵌入
)

# 测试嵌入
query = "测试嵌入模型"
query_embedding = embeddings.embed_query(query)
print(f"查询嵌入维度: {len(query_embedding)}")
print(f"嵌入示例: {query_embedding[:3]}...")

# 4. 向量存储
print("\n=== 向量存储 ===")

# 创建Chroma向量库
vectorstore = Chroma.from_documents(
    documents=split_docs,
    embedding=embeddings,
    persist_directory="./chroma_db"
)
vectorstore.persist()
print("向量库创建成功并持久化")

# 测试相似度搜索
search_results = vectorstore.similarity_search(
    query="测试相似度搜索",
    k=3
)
print(f"相似度搜索结果数量: {len(search_results)}")
for i, result in enumerate(search_results):
    print(f"结果 {i+1}: {result.page_content[:150]}...")

# 5. 构建RAG链
print("\n=== 构建RAG链 ===")

# 初始化语言模型
llm = OpenAI(temperature=0.7)

# 构建检索器
retriever = vectorstore.as_retriever(
    search_type="similarity",  # 搜索类型
    search_kwargs={"k": 3}  # 搜索参数
)

# 构建RAG链
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",  # 链类型
    retriever=retriever,
    return_source_documents=True,  # 返回源文档
    verbose=True  # 启用详细输出
)

# 6. 测试RAG链
print("\n=== 测试RAG链 ===")

# 测试问题
questions = [
    "文档的主要内容是什么？",
    "文档中提到了哪些关键概念？",
    "文档的作者是谁？"
]

for question in questions:
    print(f"\n问题: {question}")
    result = qa_chain(question)
    print(f"回答: {result['result']}")
    print(f"源文档数量: {len(result['source_documents'])}")
    for i, doc in enumerate(result['source_documents']):
        print(f"源文档 {i+1}: {doc.page_content[:100]}...")

print("\n基础RAG管道构建完成！")

案例2：高级RAG配置

from langchain.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter, TokenTextSplitter
from langchain.embeddings import OpenAIEmbeddings, HuggingFaceEmbeddings
from langchain.vectorstores import FAISS
from langchain.chains import RetrievalQA, ConversationalRetrievalChain
from langchain.chat_models import ChatOpenAI
from langchain.memory import ConversationBufferMemory
import os

# 设置环境变量
os.environ["OPENAI_API_KEY"] = "your-openai-api-key"

# 1. 文档加载与处理
print("=== 文档加载与处理 ===")

# 加载PDF文档
loader = PyPDFLoader("technical_document.pdf")
documents = loader.load()
print(f"原始文档数量: {len(documents)}")

# 2. 高级文本分割
print("\n=== 高级文本分割 ===")

# 选项1：递归字符分割器（推荐）
recursive_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", " ", ""]
)

# 选项2：Token分割器
 token_splitter = TokenTextSplitter(
    chunk_size=512,
    chunk_overlap=100
)

# 使用递归字符分割器
split_docs = recursive_splitter.split_documents(documents)
print(f"分割后的文档数量: {len(split_docs)}")

# 3. 嵌入模型选择
print("\n=== 嵌入模型选择 ===")

# 选项1：OpenAI嵌入模型
openai_embeddings = OpenAIEmbeddings(
    model="text-embedding-ada-002"
)

# 选项2：HuggingFace嵌入模型
hf_embeddings = HuggingFaceEmbeddings(
    model_name="sentence-transformers/all-mpnet-base-v2"
)

# 使用OpenAI嵌入模型
embeddings = openai_embeddings

# 4. 向量存储配置
print("\n=== 向量存储配置 ===")

# 创建FAISS向量库
vectorstore = FAISS.from_documents(
    documents=split_docs,
    embedding=embeddings
)

# 保存向量库
vectorstore.save_local("./faiss_db")
print("FAISS向量库创建并保存成功")

# 5. 高级检索配置
print("\n=== 高级检索配置 ===")

# 选项1：相似度搜索
retriever_similarity = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 3}
)

# 选项2：最大边际相关性搜索（MMR）
retriever_mmr = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 3, "lambda_mult": 0.5}
)

# 选项3：相似度分数阈值搜索
retriever_threshold = vectorstore.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={"score_threshold": 0.7, "k": 3}
)

# 使用MMR检索器
retriever = retriever_mmr

# 6. 不同链类型测试
print("\n=== 不同链类型测试 ===")

# 初始化语言模型
llm = ChatOpenAI(temperature=0.7, model="gpt-3.5-turbo")

# 选项1：stuff链
chain_stuff = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever,
    return_source_documents=True,
    verbose=True
)

# 选项2：map_reduce链
chain_map_reduce = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="map_reduce",
    retriever=retriever,
    return_source_documents=True,
    verbose=True
)

# 选项3：refine链
chain_refine = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="refine",
    retriever=retriever,
    return_source_documents=True,
    verbose=True
)

# 7. 对话式RAG
print("\n=== 对话式RAG ===")

# 初始化记忆系统
memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True,
    output_key="answer"
)

# 构建对话式RAG链
conversational_chain = ConversationalRetrievalChain.from_llm(
    llm=llm,
    retriever=retriever,
    memory=memory,
    return_source_documents=True,
    verbose=True
)

# 测试对话式RAG
print("\n测试对话式RAG:")
print("你可以连续提问，测试对话上下文理解能力")
print("输入'退出'结束对话")

while True:
    question = input("\n问题: ")
    if question.lower() == "退出":
        print("对话结束！")
        break
    
    result = conversational_chain({
        "question": question
    })
    
    print(f"回答: {result['answer']}")
    print(f"源文档数量: {len(result['source_documents'])}")
    for i, doc in enumerate(result['source_documents']):
        print(f"源文档 {i+1}: {doc.page_content[:100]}...")

print("\n高级RAG配置测试完成！")

案例3：RAG与智能体集成

from langchain.agents import initialize_agent, AgentType
from langchain.chat_models import ChatOpenAI
from langchain.tools import Tool
from langchain.utilities import SerpAPIWrapper
from langchain.chains import RetrievalQA
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain.memory import ConversationBufferMemory
import os

# 设置环境变量
os.environ["OPENAI_API_KEY"] = "your-openai-api-key"
os.environ["SERPAPI_API_KEY"] = "your-serpapi-api-key"

# 1. 加载向量库
print("=== 加载向量库 ===")

embeddings = OpenAIEmbeddings()
vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings
)

print("向量库加载成功")

# 2. 构建RAG工具
print("\n=== 构建RAG工具 ===")

# 初始化语言模型
llm = ChatOpenAI(temperature=0.7)

# 构建检索器
retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 3, "lambda_mult": 0.5}
)

# 构建RAG链
rag_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever,
    return_source_documents=True
)

# 创建RAG工具
def rag_tool(query):
    """使用RAG回答关于文档的问题"""
    result = rag_chain(query)
    answer = result["result"]
    sources = result["source_documents"]
    
    # 格式化结果
    formatted_sources = "\n".join([f"- {source.page_content[:100]}..." for source in sources[:2]])
    return f"回答: {answer}\n\n参考来源:\n{formatted_sources}"

# 3. 初始化其他工具
print("\n=== 初始化其他工具 ===")

search = SerpAPIWrapper()

# 4. 定义工具列表
tools = [
    Tool(
        name="Search",
        func=search.run,
        description="用于搜索网络上的实时信息"
    ),
    Tool(
        name="DocumentQA",
        func=rag_tool,
        description="用于回答关于文档内容的问题"
    )
]

# 5. 初始化记忆系统
print("\n=== 初始化记忆系统 ===")

memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

# 6. 初始化智能体
print("\n=== 初始化智能体 ===")

agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent=AgentType.CHAT_ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True,
    memory=memory
)

print("\n=== 测试智能体 ===")
print("你可以询问关于文档的问题或其他问题，例如：")
print("1. 文档的主要内容是什么？")
print("2. 文档中提到了哪些技术？")
print("3. 今天北京的天气怎么样？")
print("4. 文档的作者是谁？")
print("输入'退出'结束对话")
print("===================")

# 7. 测试智能体
while True:
    user_input = input("\n用户: ")
    if user_input.lower() == "退出":
        print("智能体: 再见！如果有任何问题，随时再来咨询我。")
        break
    
    try:
        response = agent.run(user_input)
        print(f"智能体: {response}")
    except Exception as e:
        print(f"智能体: 抱歉，我在处理您的问题时遇到了一些困难。错误信息: {str(e)}")

print("\n智能体测试完成！")

代码解析

基础RAG管道

1. 文档加载：

   • 使用不同的加载器加载不同类型的文档
   • 支持PDF、文本文件、网页等多种格式
   • 合并多个来源的文档

2. 文本分割：

   • 使用RecursiveCharacterTextSplitter进行递归分割
   • 配置合适的chunk_size和chunk_overlap
   • 确保分割后的文本块大小适中

3. 嵌入与存储：

   • 使用HuggingFaceEmbeddings生成文本嵌入
   • 创建Chroma向量库并持久化
   • 测试相似度搜索功能

4. RAG链构建：

   • 初始化语言模型
   • 构建检索器
   • 使用RetrievalQA构建RAG链
   • 测试RAG链的回答能力

高级RAG配置

1. 高级文本分割：

   • 比较不同的文本分割器
   • 根据文档类型选择合适的分割策略

2. 嵌入模型选择：

   • 比较OpenAI和HuggingFace嵌入模型
   • 根据性能和成本选择合适的模型

3. 向量存储优化：

   • 使用FAISS提高检索性能
   • 保存和加载向量库

4. 高级检索策略：

   • 实现相似度搜索、MMR搜索和阈值搜索
   • 根据场景选择合适的检索策略

5. 不同链类型：

   • 测试stuff、map_reduce和refine链类型
   • 分析不同链类型的优缺点

6. 对话式RAG：

   • 集成记忆系统
   • 实现多轮对话能力

RAG与智能体集成

1. 向量库加载：

   • 加载已有的向量库
   • 准备RAG功能

2. RAG工具创建：

   • 构建RAG链
   • 封装为工具函数
   • 定义清晰的工具描述

3. 智能体配置：

   • 集成RAG工具和搜索工具
   • 配置记忆系统
   • 初始化智能体

4. 交互测试：

   • 测试智能体对不同问题的响应
   • 观察智能体选择工具的策略
   • 验证RAG功能的有效性

高级技巧

1. 文档加载优化

针对不同类型的文档，使用合适的加载策略：

from langchain.document_loaders import (
    PyPDFLoader, PDFPlumberLoader, UnstructuredPDFLoader,
    Docx2txtLoader, UnstructuredWordDocumentLoader,
    TextLoader, CSVLoader, JSONLoader,
    WebBaseLoader, SeleniumURLLoader,
    GitHubRepoLoader, NotionDirectoryLoader
)
import os

# 1. PDF文档加载
print("=== PDF文档加载 ===")

# 选项1：PyPDFLoader（基础）
pdf_loader = PyPDFLoader("document.pdf")

# 选项2：PDFPlumberLoader（高级）
# pdf_loader = PDFPlumberLoader("document.pdf")

# 选项3：UnstructuredPDFLoader（支持OCR）
# pdf_loader = UnstructuredPDFLoader("document.pdf", mode="elements")

pdf_docs = pdf_loader.load()
print(f"PDF文档页数: {len(pdf_docs)}")

# 2. Word文档加载
print("\n=== Word文档加载 ===")
docx_loader = Docx2txtLoader("document.docx")
docx_docs = docx_loader.load()
print(f"Word文档数量: {len(docx_docs)}")

# 3. CSV文档加载
print("\n=== CSV文档加载 ===")
csv_loader = CSVLoader(
    file_path="data.csv",
    csv_args={
        "delimiter": ",",
        "quotechar": '"',
        "fieldnames": ["column1", "column2", "column3"]
    }
)
csv_docs = csv_loader.load()
print(f"CSV文档数量: {len(csv_docs)}")

# 4. 网页加载
print("\n=== 网页加载 ===")

# 选项1：WebBaseLoader（基础）
web_loader = WebBaseLoader("https://example.com")

# 选项2：SeleniumURLLoader（支持JavaScript渲染）
# web_loader = SeleniumURLLoader(urls=["https://example.com"])

web_docs = web_loader.load()
print(f"网页文档数量: {len(web_docs)}")

# 5. GitHub仓库加载
print("\n=== GitHub仓库加载 ===")
if os.environ.get("GITHUB_TOKEN"):
    github_loader = GitHubRepoLoader(
        repo="username/repository",
        access_token=os.environ.get("GITHUB_TOKEN"),
        recursive=True,
        ignore_directories=[".git", "node_modules", "venv"]
    )
    github_docs = github_loader.load()
    print(f"GitHub文档数量: {len(github_docs)}")

# 6. 批量加载
print("\n=== 批量加载 ===")
document_loaders = [
    PyPDFLoader("doc1.pdf"),
    TextLoader("doc2.txt"),
    WebBaseLoader("https://example.com")
]

all_docs = []
for loader in document_loaders:
    try:
        docs = loader.load()
        all_docs.extend(docs)
        print(f"加载成功: {type(loader).__name__}")
    except Exception as e:
        print(f"加载失败 {type(loader).__name__}: {str(e)}")

print(f"\n总加载文档数量: {len(all_docs)}")

2. 文本分割策略

根据文档类型选择合适的分割策略：

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

# 1. 递归字符分割器（通用推荐）
print("=== 递归字符分割器 ===")
recursive_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", " ", ""]
)

# 2. 字符分割器（简单分割）
print("\n=== 字符分割器 ===")
character_splitter = CharacterTextSplitter(
    separator="\n",
    chunk_size=1000,
    chunk_overlap=200,
    length_function=len
)

# 3. Token分割器（基于Token计数）
print("\n=== Token分割器 ===")
token_splitter = TokenTextSplitter(
    chunk_size=512,
    chunk_overlap=100
)

# 4. Markdown分割器（保留Markdown结构）
print("\n=== Markdown分割器 ===")
markdown_splitter = MarkdownTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)

# 5. Python代码分割器（保留代码结构）
print("\n=== Python代码分割器 ===")
python_splitter = PythonCodeTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)

# 测试不同分割器
text = """# 标题

这是一段正文内容，包含多个句子。

## 子标题

这是子标题下的内容，也包含多个句子。

```python
def hello():
    print("Hello world!")

这是代码后的内容。"""

print("\n=== 分割结果对比 ===")

递归字符分割

recursive_chunks = recursive_splitter.split_text(text)
print(f"递归字符分割: {len(recursive_chunks)} 块")
for i, chunk in enumerate(recursive_chunks):
print(f"块 {i+1}: {chunk[:100]}...")

Markdown分割

markdown_chunks = markdown_splitter.split_text(text)
print(f"\nMarkdown分割: {len(markdown_chunks)} 块")
for i, chunk in enumerate(markdown_chunks):
print(f"块 {i+1}: {chunk[:100]}...")

Python代码分割

python_chunks = python_splitter.split_text(text)
print(f"\nPython代码分割: {len(python_chunks)} 块")
for i, chunk in enumerate(python_chunks):
print(f"块 {i+1}: {chunk[:100]}...")

选择合适的分割器

print("\n=== 分割器选择建议 ===")
print("- 通用文档: 递归字符分割器")
print("- Markdown文档: Markdown分割器")
print("- 代码文件: Python代码分割器（或相应语言的分割器）")
print("- 对Token敏感的模型: Token分割器")

### 3. 检索优化

优化检索策略，提高检索精度：

```python
from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings
from langchain.docstore.document import Document
import numpy as np

# 1. 创建测试文档
documents = [
    Document(
        page_content="人工智能是研究、开发用于模拟、延伸和扩展人的智能的理论、方法、技术及应用系统的一门新的技术科学。",
        metadata={"category": "ai", "type": "definition"}
    ),
    Document(
        page_content="机器学习是人工智能的一个分支，它使计算机系统能够从数据中学习并改进，而无需明确编程。",
        metadata={"category": "ai", "type": "branch"}
    ),
    Document(
        page_content="深度学习是机器学习的一个分支，它使用多层神经网络来模拟人脑的学习过程。",
        metadata={"category": "ai", "type": "branch"}
    ),
    Document(
        page_content="自然语言处理是人工智能的一个领域，它使计算机能够理解、解释和生成人类语言。",
        metadata={"category": "ai", "type": "field"}
    ),
    Document(
        page_content="计算机视觉是人工智能的一个领域，它使计算机能够理解和解释图像和视频。",
        metadata={"category": "ai", "type": "field"}
    )
]

# 2. 创建向量库
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_documents(
    documents=documents,
    embedding=embeddings
)

# 3. 不同检索策略对比
print("=== 不同检索策略对比 ===")
query = "什么是机器学习和深度学习？"

# 策略1：相似度搜索
print("\n策略1：相似度搜索")
similarity_results = vectorstore.similarity_search(
    query=query,
    k=3
)
for i, result in enumerate(similarity_results):
    print(f"{i+1}. {result.page_content}")

# 策略2：带分数的相似度搜索
print("\n策略2：带分数的相似度搜索")
similarity_score_results = vectorstore.similarity_search_with_score(
    query=query,
    k=3
)
for i, (result, score) in enumerate(similarity_score_results):
    print(f"{i+1}. {result.page_content} (分数: {score:.4f})")

# 策略3：MMR搜索
print("\n策略3：MMR搜索")
mmr_results = vectorstore.max_marginal_relevance_search(
    query=query,
    k=3,
    lambda_mult=0.5
)
for i, result in enumerate(mmr_results):
    print(f"{i+1}. {result.page_content}")

# 策略4：基于向量的搜索
print("\n策略4：基于向量的搜索")
query_vector = embeddings.embed_query(query)
vector_results = vectorstore.similarity_search_by_vector(
    embedding=query_vector,
    k=3
)
for i, result in enumerate(vector_results):
    print(f"{i+1}. {result.page_content}")

# 4. 元数据过滤
print("\n=== 元数据过滤 ===")

# 过滤特定类型的文档
filtered_results = vectorstore.similarity_search(
    query="人工智能领域",
    k=3
)

# 手动过滤元数据
filtered_by_metadata = [
    doc for doc in filtered_results 
    if doc.metadata.get("type") == "field"
]

print("过滤后的结果（只显示field类型）:")
for i, result in enumerate(filtered_by_metadata):
    print(f"{i+1}. {result.page_content} (类型: {result.metadata.get('type')})")

# 5. 组合检索策略
print("\n=== 组合检索策略 ===")

def hybrid_search(query, k=3):
    """混合检索策略"""
    # 1. 首先使用MMR获取多样化结果
    mmr_results = vectorstore.max_marginal_relevance_search(
        query=query,
        k=k*2,  # 获取更多结果
        lambda_mult=0.3
    )
    
    # 2. 然后对结果进行相似度排序
    query_vector = embeddings.embed_query(query)
    
    # 计算每个结果与查询的相似度
    results_with_similarity = []
    for doc in mmr_results:
        doc_vector = embeddings.embed_query(doc.page_content)
        similarity = np.dot(query_vector, doc_vector) / (
            np.linalg.norm(query_vector) * np.linalg.norm(doc_vector)
        )
        results_with_similarity.append((doc, similarity))
    
    # 按相似度排序并返回前k个
    results_with_similarity.sort(key=lambda x: x[1], reverse=True)
    top_results = [result[0] for result in results_with_similarity[:k]]
    
    return top_results

# 测试混合检索
hybrid_results = hybrid_search(query, k=3)
print("混合检索结果:")
for i, result in enumerate(hybrid_results):
    print(f"{i+1}. {result.page_content}")

print("\n检索优化完成！")

4. RAG评估

评估RAG系统的性能：

from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI
import json
import os

# 1. 加载向量库
embeddings = OpenAIEmbeddings()
vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings
)

# 2. 构建RAG链
llm = OpenAI(temperature=0.7)
retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 3}
)

rag_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever,
    return_source_documents=True
)

# 3. 准备评估数据集
evaluation_data = [
    {
        "question": "文档的主要内容是什么？",
        "ground_truth": "文档主要介绍了人工智能的基本概念、发展历史、核心技术和应用领域。"
    },
    {
        "question": "文档中提到了哪些人工智能技术？",
        "ground_truth": "文档中提到了机器学习、深度学习、自然语言处理、计算机视觉等人工智能技术。"
    },
    {
        "question": "人工智能有哪些应用领域？",
        "ground_truth": "人工智能的应用领域包括医疗、金融、教育、交通、制造、娱乐等。"
    }
]

# 4. 评估指标
from sklearn.metrics import f1_score, accuracy_score
import re

def normalize_text(text):
    """标准化文本"""
    text = text.lower()
    text = re.sub(r'\s+', ' ', text)
    text = re.sub(r'[\n\r]', ' ', text)
    return text.strip()

def compute_metrics(predictions, ground_truths):
    """计算评估指标"""
    # 简单准确率（基于关键词匹配）
    correct = 0
    for pred, gt in zip(predictions, ground_truths):
        pred_norm = normalize_text(pred)
        gt_norm = normalize_text(gt)
        # 检查预测是否包含 ground truth 的主要内容
        if all(keyword in pred_norm for keyword in gt_norm.split()[:5]):
            correct += 1
    accuracy = correct / len(predictions)
    
    return {
        "accuracy": accuracy
    }

# 5. 执行评估
print("=== RAG系统评估 ===")

predictions = []
ground_truths = []
evaluation_results = []

for item in evaluation_data:
    question = item["question"]
    ground_truth = item["ground_truth"]
    
    # 获取预测结果
    result = rag_chain(question)
    prediction = result["result"]
    sources = result["source_documents"]
    
    # 保存结果
    predictions.append(prediction)
    ground_truths.append(ground_truth)
    
    # 记录详细结果
    evaluation_results.append({
        "question": question,
        "prediction": prediction,
        "ground_truth": ground_truth,
        "sources": [source.page_content[:200] for source in sources]
    })
    
    print(f"\n问题: {question}")
    print(f"预测: {prediction}")
    print(f"真实答案: {ground_truth}")

# 6. 计算指标
metrics = compute_metrics(predictions, ground_truths)
print("\n=== 评估指标 ===")
for metric_name, metric_value in metrics.items():
    print(f"{metric_name}: {metric_value:.4f}")

# 7. 保存评估结果
with open("./rag_evaluation_results.json", "w", encoding="utf-8") as f:
    json.dump({
        "metrics": metrics,
        "results": evaluation_results
    }, f, ensure_ascii=False, indent=2)

print("\n评估完成，结果已保存到 rag_evaluation_results.json")

最佳实践

1. RAG管道优化

2. 性能优化

• 批处理：批量处理文档加载和嵌入
• 缓存机制：缓存频繁查询的结果和嵌入
• 并行处理：并行执行文档分割和嵌入
• 索引优化：为向量数据库创建合适的索引
• 硬件加速：使用GPU加速嵌入生成和检索

3. 常见问题与解决方案

4. 部署建议

• 开发环境：

  • 使用Chroma内存模式快速测试
  • 启用详细日志便于调试
  • 使用轻量级嵌入模型

• 生产环境：

  • 使用FAISS或专业向量数据库
  • 实现向量库的备份和恢复
  • 监控系统性能和响应时间
  • 考虑使用容器化部署
  • 实现自动扩展机制

总结与展望

本集要点总结

1. RAG基础：

   • RAG的概念和优势
   • RAG在智能体系统中的应用
   • RAG管道的核心组件

2. RAG管道构建：

   • 文档加载：支持多种文件格式
   • 文本分割：适应语言模型的上下文窗口
   • 向量存储：使用Chroma和FAISS
   • 相似度检索：实现准确的信息检索
   • 生成回答：基于检索结果生成准确回答

3. 高级配置：

   • 不同分割策略的选择
   • 多种检索策略的对比
   • 不同链类型的应用
   • 对话式RAG的实现

4. 智能体集成：

   • 将RAG封装为工具
   • 集成到智能体系统
   • 实现多工具协同

5. 高级技巧：

   • 文档加载优化
   • 文本分割策略
   • 检索优化
   • RAG评估

未来发展方向

1. RAG技术演进：

   • 更先进的检索策略
   • 多模态RAG（结合文本、图像、音频等）
   • 自适应RAG（根据查询自动调整策略）

2. 智能体集成：

   • 更深层次的RAG与智能体集成
   • 多智能体RAG系统
   • 个性化RAG（基于用户偏好）

3. 性能优化：

   • 更高效的嵌入和检索
   • 分布式RAG系统
   • 边缘设备上的RAG

4. 应用扩展：

   • 企业级知识库系统
   • 实时信息获取与整合
   • 跨语言RAG系统

5. 研究方向：

   • RAG的理论基础
   • 最优检索策略的研究
   • RAG与微调的结合

通过本集的学习，我们了解了如何构建基础的RAG管道，掌握了文档加载、文本分割、向量存储和相似度检索等核心技术，以及如何将RAG集成到智能体系统中。RAG作为一种强大的知识增强技术，将在构建更智能、更可靠的AI系统中发挥越来越重要的作用。

在后续的课程中，我们将深入探讨RAG的高级应用，以及如何构建更复杂、更强大的智能体记忆系统。