Agent 智能体开发实战指南（六）：RAG 与向量存储实战

系列导读：这是《Agent 智能体开发实战指南》系列的第六篇，将深入讲解 RAG（检索增强生成）技术与向量数据库的实战应用，包括 Chroma 集成、文本分片、MD5 去重、检索优化等核心内容。

---

一、为什么需要 RAG？

1.1 LLM 的知识困境

问题场景：

用户：你们公司的扫地机器人 X1 型号怎么清理滤网？

LLM（训练截止 2024 年）：
 	我不知道这款产品（2025 年发布）
 	即使知道，也可能给出错误信息
    无法访问公司内部知识库

传统解决方案的局限：

方案	问题
微调模型	成本高、更新慢、容易遗忘
长上下文	Token 贵、有上限、检索慢
直接问答	无法访问私有数据

1.2 RAG 的核心思想

RAG = Retrieval（检索） + Augmented（增强） + Generation（生成）

用户提问
   ↓
┌─────────────────┐
│  检索相关知识    │ ← 从向量数据库查找
└────────┬────────┘
         ↓
┌─────────────────┐
│  增强提示词      │ ← 将知识注入 prompt
└────────┬────────┘
         ↓
┌─────────────────┐
│  LLM 生成答案    │ ← 基于检索到的知识
└────────┬────────┘
         ↓
   返回答案

核心优势：

• ✅ 无需训练，即时更新知识库
• ✅ 答案有依据，可追溯来源
• ✅ 减少幻觉，提高准确性
• ✅ Token 效率高，只传相关知识

---

二、RAG 架构详解

2.1 完整架构图

┌─────────────────────────────────────────────────────────────┐
│                      RAG 系统架构                            │
│                                                             │
│  ┌──────────────┐      ┌──────────────┐      ┌───────────┐  │
│  │  文档加载器   │  →   │  文本分片器   │  →   │ 嵌入模型   │  │
│  │  (Loader)    │      │  (Splitter)  │      │(Embedding)│  │
│  └──────────────┘      └──────────────┘      └─────┬─────┘  │
│                                                     ↓       │
│  ┌──────────────┐      ┌──────────────┐      ┌───────────┐  │
│  │  用户查询     │  →   │   检索器     │  ←   │ 向量数据库 │  │
│  │   (Query)    │      │  (Retriever) │      │  (Chroma) │  │
│  └──────────────┘      └──────┬───────┘      └───────────┘  │
│                               ↓                             │
│                       ┌──────────────┐                      │
│                       │  Prompt 模板 │                      │
│                       │  + 检索结果   │                      │
│                       └──────┬───────┘                      │
│                               ↓                             │
│                       ┌──────────────┐                      │
│                       │  聊天模型     │                      │
│                       │   (LLM)      │                      │
│                       └──────┬───────┘                      │
│                               ↓                             │
│                       ┌──────────────┐                      │
│                       │   最终答案    │                      │
│                       └──────────────┘                      │
└─────────────────────────────────────────────────────────────┘

2.2 核心组件

组件	作用	LangChain 实现
文档加载器	读取 PDF、TXT 等文件	PyPDFLoader, TextLoader
文本分片器	将长文档切分为小块	RecursiveCharacterTextSplitter
嵌入模型	将文本转为向量	DashScopeEmbeddings
向量数据库	存储和检索向量	Chroma
检索器	查询最相关的文档	vector_store.as_retriever()
提示词模板	组织检索结果和查询	PromptTemplate
聊天模型	生成最终答案	ChatTongyi

---

三、向量存储服务实现

3.1 模型工厂模式

为什么用工厂？

• 模型创建有很多种（聊天、嵌入等）
• 统一创建入口，便于管理
• 符合开闭原则，易于扩展

# model/factory.py
from abc import ABC, abstractmethod
from typing import Optional

from langchain_community.embeddings import DashScopeEmbeddings
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.embeddings import Embeddings
from langchain_core.language_models.chat_models import BaseChatModel
from utils.config_handler import rag_conf

class BaseModelFactory(ABC):
    """模型工厂抽象基类"""
    
    @abstractmethod
    def generate(self) -> Optional[Embeddings | BaseChatModel]:
        pass

class ChatModelFactory(BaseModelFactory):
    """聊天模型工厂"""
    
    def generate(self) -> BaseChatModel:
        return ChatTongyi(model=rag_conf['chat_model_name'])

class EmbeddingsFactory(BaseModelFactory):
    """嵌入模型工厂"""
    
    def generate(self) -> Embeddings:
        return DashScopeEmbeddings(model=rag_conf['embedding_model_name'])

# 使用
chat_model = ChatModelFactory().generate()
embedding_model = EmbeddingsFactory().generate()

3.2 向量存储服务核心类

# rag/vector_store.py
from langchain_chroma import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter
from utils.config_handler import chroma_conf
from utils.path_tool import get_abs_path
from model.factory import embedding_model
from utils.file_handler import pdf_loader, txt_loader, listdir_with_allowed_type, get_file_md5_hex
from utils.logger_handler import logger
import os

class VectorStoreService:
    def __init__(self):
        # 初始化向量数据库
        self.vector_store = Chroma(
            collection_name=chroma_conf["collection_name"],
            embedding_function=embedding_model,
            persist_directory=get_abs_path(chroma_conf["persist_directory"]),
        )
        
        # 初始化文本分片器
        self.spliter = RecursiveCharacterTextSplitter(
            chunk_size=chroma_conf["chunk_size"],
            chunk_overlap=chroma_conf["chunk_overlap"],
            separators=chroma_conf["separators"],
            length_function=len,
        )
    
    def get_retriever(self):
        """获取检索器"""
        return self.vector_store.as_retriever(
            search_kwargs={"k": chroma_conf["k"]}
        )

3.3 配置文件示例

# config/chroma.yml
collection_name: "zhisao_knowledge"
persist_directory: "data/chroma_db"
chunk_size: 500          # 每块 500 字符
chunk_overlap: 50        # 重叠 50 字符
separators:              # 分片分隔符优先级
  - "\n\n"
  - "\n"
  - "。 "
  - "！"
  - "？"
  - "；"
  - "，"
  - " "
  - ""
k: 3                     # 检索返回最相关的 3 条
allow_knowledge_file_type:  # 允许的文件类型
  - ".pdf"
  - ".txt"
data_path: "data/knowledge"  # 知识库文件目录
md5_hex_store: "data/md5_hex.txt"  # MD5 去重记录文件

---

四、文档加载与去重

4.1 完整加载流程

def load_document(self):
    """
    从数据文件夹内读取数据文件，转为向量存入向量库
    使用 MD5 进行去重
    """
    
    # 1. 检查 MD5 是否已处理
    def check_md5_hex(md5_for_check: str) -> bool:
        if not os.path.exists(get_abs_path(chroma_conf["md5_hex_store"])):
            open(get_abs_path(chroma_conf["md5_hex_store"]), "w", encoding="utf-8").close()
            return False  # 没处理过
        
        with open(get_abs_path(chroma_conf["md5_hex_store"]), "r", encoding="utf-8") as f:
            for line in f.readlines():
                if line.strip() == md5_for_check:
                    return True  # 已处理
        return False
    
    # 2. 保存 MD5 记录
    def save_md5_hex(md5_for_check: str):
        with open(get_abs_path(chroma_conf["md5_hex_store"]), "a", encoding="utf-8") as f:
            f.write(md5_for_check + "\n")
    
    # 3. 获取文件文档
    def get_file_documents(read_path: str):
        if read_path.endswith("txt"):
            return txt_loader(read_path)
        if read_path.endswith("pdf"):
            return pdf_loader(read_path)
        return []
    
    # 4. 获取所有允许的文件
    allowed_files = listdir_with_allowed_type(
        get_abs_path(chroma_conf["data_path"]),
        tuple(chroma_conf["allow_knowledge_file_type"]),
    )
    
    # 5. 遍历处理
    for path in allowed_files:
        md5_hex = get_file_md5_hex(path)
        
        # 去重检查
        if check_md5_hex(md5_hex):
            logger.info(f"[加载知识库]{path} 内容已存在，跳过")
            continue
        
        try:
            # 加载文档
            documents = get_file_documents(path)
            if not documents:
                logger.warning(f"[加载知识库]{path} 内没有有效文本内容，跳过")
                continue
            
            # 文本分片
            split_docs = self.spliter.split_documents(documents)
            if not split_docs:
                logger.warning(f"[加载知识库]{path} 分片后没有有效文本内容，跳过")
                continue
            
            # 存入向量库
            self.vector_store.add_documents(split_docs)
            
            # 记录 MD5
            save_md5_hex(md5_hex)
            
            logger.info(f"[加载知识库]{path} 内容加载成功")
            
        except Exception as e:
            logger.error(f"[加载知识库]{path} 加载失败：{str(e)}", exc_info=True)
            continue

4.2 MD5 去重的价值

问题：

• 每次启动都重新加载所有文档？
• 如何知道哪些文档已经处理过？

解决方案：

文件 → 计算 MD5 → 检查是否已存在 → 不存在则加载 → 记录 MD5

优势：

• 避免重复处理，节省时间
• 增量更新，只处理新文件
• 简单可靠，MD5 碰撞概率极低

4.3 文件处理工具

# utils/file_handler.py
import hashlib
from langchain_core.documents import Document
from langchain_community.document_loaders import PyPDFLoader, TextLoader

def get_file_md5_hex(filepath: str) -> str:
    """获取文件的 MD5 十六进制字符串"""
    if not os.path.exists(filepath):
        return None
    
    md5_obj = hashlib.md5()
    chunk_size = 4096  # 4KB 分片读取，避免大文件爆内存
    
    with open(filepath, "rb") as f:
        while chunk := f.read(chunk_size):
            md5_obj.update(chunk)
    
    return md5_obj.hexdigest()

def listdir_with_allowed_type(path: str, allowed_types: tuple) -> list:
    """列出文件夹中指定后缀的文件"""
    files = []
    for f in os.listdir(path):
        if f.endswith(allowed_types):
            files.append(os.path.join(path, f))
    return files

def pdf_loader(filepath: str, passwd=None) -> list[Document]:
    """PDF 文件加载器"""
    return PyPDFLoader(filepath, passwd).load()

def txt_loader(filepath: str) -> list[Document]:
    """文本文件加载器"""
    return TextLoader(filepath, encoding="utf-8").load()

---

五、RAG 总结服务

5.1 服务实现

# rag/rag_service.py
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from rag.vector_store import VectorStoreService
from utils.prompt_loader import load_rag_prompts
from model.factory import chat_model

class RagSummarizeService:
    def __init__(self):
        self.vector_store = VectorStoreService()
        self.retriever = self.vector_store.get_retriever()
        self.prompt_text = load_rag_prompts()
        self.prompt_template = PromptTemplate.from_template(self.prompt_text)
        self.model = chat_model
        self.chain = self._init_chain()
    
    def _init_chain(self):
        """初始化处理链：提示词 → 模型 → 输出解析"""
        return self.prompt_template | self.model | StrOutputParser()
    
    def retriever_docs(self, query: str):
        """检索相关文档"""
        return self.retriever.invoke(query)
    
    def rag_summarize(self, query: str) -> str:
        """RAG 总结：检索 + 生成"""
        # 1. 检索相关文档
        context_docs = self.retriever_docs(query)
        
        # 2. 格式化上下文
        context = ""
        for i, doc in enumerate(context_docs, 1):
            context += f"[参考资料{i}]: {doc.page_content}\n"
            context += f"  元数据：{doc.metadata}\n"
        
        # 3. 调用链生成答案
        return self.chain.invoke({
            "input": query,
            "context": context
        })

5.2 RAG 提示词模板

# config/prompts.yml
rag_summarize_prompt_path: "prompts/rag_prompt.txt"

# prompts/rag_prompt.txt
你是一个专业的客服助手，请根据提供的参考资料回答用户问题。

要求：
1. 严格基于参考资料回答，不要编造信息
2. 如果参考资料不足以回答问题，请如实告知
3. 回答要清晰、简洁、有条理
4. 必要时可以引用参考资料的来源

参考资料：
{context}

用户问题：
{input}

你的回答：

5.3 使用示例

if __name__ == "__main__":
    rag = RagSummarizeService()
    
    # 测试查询
    result = rag.rag_summarize("小户型适合哪种扫地机器人")
    print(result)
    
    # 查看检索到的文档
    docs = rag.retriever_docs("迷路")
    for doc in docs:
        print(doc.page_content)
        print("-" * 50)

---

六、文本分片策略

6.1 分片参数详解

RecursiveCharacterTextSplitter(
    chunk_size=500,        # 每块最大 500 字符
    chunk_overlap=50,      # 块与块之间重叠 50 字符
    separators=[           # 分隔符优先级（从高到低）
        "\n\n",            # 段落
        "\n",              # 换行
        "。 ",             # 句号
        "！ ",             # 感叹号
        "？ ",             # 问号
        "；",              # 分号
        "，",              # 逗号
        " ",               # 空格
        ""                 # 字符级
    ],
    length_function=len,   # 长度计算函数
)

6.2 参数调优指南

参数	太小	太大	推荐值
chunk_size	语义不完整	包含过多噪声	300-800
chunk_overlap	丢失上下文	冗余过多	50-200
k(检索数)	信息不足	噪声过多	3-5

6.3 不同场景的分片策略

技术文档：

chunk_size=800, chunk_overlap=100  # 保持代码完整性

客服问答：

chunk_size=400, chunk_overlap=50   # 问题 - 答案对通常较短

法律合同：

chunk_size=600, chunk_overlap=150  # 条款之间有关联

---

七、检索优化技巧

7.1 混合检索

# 同时使用语义检索 + 关键词检索
retriever = vector_store.as_retriever(
    search_type="mmr",  # Maximal Marginal Relevance
    search_kwargs={
        "k": 5,          # 返回 5 条
        "lambda_mult": 0.5  # 多样性参数（0-1）
    }
)

7.2 元数据过滤

# 只检索特定类别的文档
retriever = vector_store.as_retriever(
    search_kwargs={
        "k": 3,
        "filter": {"category": "故障排除"}
    }
)

7.3 查询重写

# 将用户查询改写为更适合检索的形式
def rewrite_query(query: str) -> str:
    # 可以使用 LLM 进行查询重写
    return f"扫地机器人 {query} 问题解决方法"

# 使用重写后的查询检索
rewritten = rewrite_query(user_query)
docs = retriever.invoke(rewritten)

---

八、本章小结

核心要点

1. RAG 架构：检索 → 增强 → 生成
2. 向量数据库：Chroma 存储和检索向量
3. 文本分片：合理的 chunk_size 和 overlap
4. MD5 去重：避免重复处理文档
5. 检索优化：混合检索、元数据过滤、查询重写

下章预告

下一篇我们将讲解 项目架构设计，学习：

• 分层架构设计
• 配置文件管理
• 日志系统实现
• 路径工具封装

---

1. Agent 智能体开发实战指南（一）：从 LLM 到 Agent 的认知升级
2. Agent 智能体开发实战指南（二）：工具调用系统深度解析
3. Agent 智能体开发实战指南（三）：ReAct 框架深度解析
4. Agent 智能体开发实战指南（四）：流式输出与状态管理
5. Agent 智能体开发实战指南（五）：中间件系统与动态提示词
6. Agent 智能体开发实战指南（六）：RAG 与向量存储实战(本文)
7. Agent 智能体开发实战指南（七）：项目架构设计与工程化实践
8. Agent 智能体开发实战指南（八）：UI 集成与生产部署

---

本文是《Agent 智能体开发实战指南》系列的第六篇，下一篇将深入讲解项目架构设计。