1. 项目概述：为什么现在必须掌握 RAG 与 Agent 的融合？

最近和几个做 AI 应用落地的朋友聊天，大家普遍有个共识：单纯调 API 调用大模型的时代已经过去了。客户不再满足于一个会聊天的“鹦鹉”，他们需要一个能真正理解业务、能查询内部知识、能执行复杂任务的“智能员工”。这背后，正是RAG（检索增强生成）和Agent（智能体）这两项技术从幕后走向台前。这个项目，就是一次从零开始的实战，目标不是复现某个教程，而是搭建一个能跑起来、能解决实际问题的检索增强生成系统，并为其注入 Agent 的“行动力”。

简单来说，RAG 解决了大模型的“幻觉”和知识滞后问题。它让模型在回答前，先去你的专属知识库（比如公司文档、产品手册、技术资料）里检索相关信息，然后基于这些“证据”来生成答案，确保回答的准确性和时效性。而 Agent 则赋予了系统“思考”和“执行”的能力。它不再是被动的一问一答，而是可以理解复杂意图、拆解任务、调用工具（比如检索、计算、写代码）、并持续完成多轮目标。

所以，这个实战项目要解决的核心问题是：如何将一个静态的 RAG 知识库问答系统，升级为一个能主动规划、能利用工具、具备记忆能力的智能体？这不仅仅是技术堆砌，更是架构设计和工程思维的考验。无论你是想为自己的产品增加智能客服、知识助手功能，还是想深入理解当前 AI 应用开发的核心范式，这次从零搭建的过程都会给你带来一手经验。

2. 核心架构设计：从 RAG 管道到 Agent 大脑的演进

在动手写代码之前，我们必须把架构想清楚。一个健壮的 RAG-Agent 系统不是简单的 1+1，它需要清晰的层次和职责划分。

2.1 经典 RAG 管道的再审视

一个标准的 RAG 系统，通常包含以下几个核心环节，我习惯称之为“数据处理与响应流水线”：

1. 文档加载与解析：这是源头。你的知识可能来自 PDF、Word、网页、数据库甚至 API。关键是要选对解析器，比如用PyPDF2或pdfplumber处理复杂排版的 PDF，用BeautifulSoup处理网页，确保文本和格式被正确提取。
2. 文本分割（分块）：这是影响检索效果最关键的步骤之一。你不能把整本书扔给模型，也不能切得太碎丢失上下文。常见的策略有：
   • 固定大小重叠分块：比如每块 500 个字符，重叠 100 字符。简单，但对语义完整性不友好。
   • 基于分隔符分块：按段落、标题等自然分隔符切分。更符合阅读习惯。
   • 语义分块：使用嵌入模型计算句子相似度，在语义边界处切割。效果更好，但计算成本高。我的经验是，对于技术文档，优先按章节标题（如##）分割；对于普通文章，固定大小重叠分块是稳妥的起点。
3. 向量化与存储：将文本块转换为向量（嵌入），并存入向量数据库。这里有两个关键选择：
   • 嵌入模型：选择适合你语种和领域的模型。中文场景下，BGE、text2vec系列是不错的选择。OpenAI 的text-embedding-3系列效果稳定但需调用 API。
   • 向量数据库：Chroma轻量易用，适合原型和中小规模数据；Milvus或Qdrant适合生产环境，支持分布式和高级检索功能；PGVector如果你已经在用 PostgreSQL，集成起来会很方便。本项目为了展示完整性和性能，我们选择Qdrant，它在性能和易用性上取得了很好的平衡。
4. 检索器（Retriever）：负责根据用户问题，从向量库中找出最相关的文本块。最简单的就是相似性搜索（余弦相似度）。但工业级系统往往会用上多路召回，比如同时用关键词搜索（BM25）和向量搜索，再将结果融合，以提高召回率。
5. 重排序（Reranker）：检索器返回的 Top K 个结果（比如 20 个），在相关性上可能仍有噪音。用一个更精细但更慢的重排序模型（如BGE-Reranker）对这 20 个结果重新打分排序，只保留 Top N（如 5 个）最相关的送入大模型。这能显著提升最终答案的质量。
6. 提示工程与生成：将检索到的上下文（Context）和用户问题（Question）组装成提示（Prompt），喂给大语言模型（LLM），让它生成最终答案。提示模板的设计至关重要，要明确指令模型“基于以下上下文回答”，并处理“上下文不包含答案”的情况。

2.2 引入 Agent 框架：为系统装上“大脑”

当 RAG 管道就绪后，它还是一个“反射弧”系统：输入问题，输出答案。Agent 的引入，让它变成了一个具有“目标感”的系统。

1. Agent 核心循环：经典的 ReAct（Reasoning + Acting）框架是基础。Agent 接收目标，然后进入“思考 -> 行动 -> 观察”的循环。
   • 思考：分析当前状态，决定下一步该用什么工具，或者直接给出答案。
   • 行动：调用选定的工具（Tool），比如我们的 RAG 检索工具、计算器、代码执行器、搜索引擎 API 等。
   • 观察：获取工具执行的结果，作为下一轮思考的输入。
2. 工具（Tools）抽象：这是 Agent 与外界交互的手脚。我们将 RAG 检索功能封装成一个标准的工具，比如search_knowledge_base(query: str) -> str。Agent 在需要查询内部知识时，就会调用这个工具。同样，你可以封装计算、网络搜索等工具。
3. 记忆（Memory）管理：Agent 需要有记忆，才能进行多轮对话和复杂任务规划。记忆通常分为：
   • 短期记忆/对话历史：保存当前会话的上下文。
   • 长期记忆：可以是一个向量数据库，存储之前对话或学习的摘要，供未来检索。这其实就是将 RAG 技术用于 Agent 自身的经验存储。
4. 规划（Planning）与执行：对于复杂任务，Agent 需要先拆解（Plan）。比如用户问“对比我们产品 A 和竞品 B 在安全特性上的差异”，Agent 可能会规划成：1) 检索产品 A 的安全文档；2) 搜索竞品 B 的公开安全信息；3) 综合两者信息生成对比表格。

架构融合点：在这个项目中，RAG 系统将主要扮演两个角色：一是作为 Agent 的一个核心工具，供其查询结构化知识；二是作为 Agent长期记忆的存储后端（可选）。我们将使用LangChain或LlamaIndex这类框架来高效地搭建管道和定义 Agent，因为它们提供了良好的抽象和丰富的集成。

3. 实战搭建：一步步构建你的智能知识引擎

理论说再多不如一行代码。我们开始从零搭建。环境准备：Python 3.9+，一个干净的虚拟环境是必须的。

3.1 第一步：搭建基础 RAG 管道

我们先抛开 Agent，确保最核心的“检索-增强-生成”链路是通畅的。

# 安装核心依赖 pip install langchain langchain-community langchain-qdrant langchain-openai sentence-transformers pypdf

# 示例代码结构，关键步骤注释 import os from langchain_community.document_loaders import PyPDFLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_qdrant import QdrantVectorStore from langchain_openai import OpenAIEmbeddings, ChatOpenAI from langchain.chains import RetrievalQA from qdrant_client import QdrantClient # 1. 加载文档 loader = PyPDFLoader("path/to/your/product_manual.pdf") documents = loader.load() # 2. 分割文本 text_splitter = RecursiveCharacterTextSplitter( chunk_size=500, # 块大小 chunk_overlap=50, # 重叠大小，避免语义割裂 separators=["\n\n", "\n", "。", "，", " ", ""] # 中文友好的分隔符 ) chunks = text_splitter.split_documents(documents) print(f"将文档切分为 {len(chunks)} 个块") # 3. 向量化与存储 embeddings = OpenAIEmbeddings(model="text-embedding-3-small") # 或使用本地模型 HuggingFaceEmbeddings(model_name="BAAI/bge-small-zh-v1.5") # 初始化 Qdrant 客户端，本地用内存模式，生产用远程服务器 client = QdrantClient(location=":memory:") # 或 `host="localhost", port=6333` vector_store = QdrantVectorStore.from_documents( documents=chunks, embedding=embeddings, client=client, collection_name="product_knowledge", ) # 4. 创建检索器 retriever = vector_store.as_retriever( search_type="similarity", # 相似度搜索 search_kwargs={"k": 5} # 检索前5个相关块 ) # 5. 创建提示模板 from langchain.prompts import PromptTemplate prompt_template = """基于以下上下文信息，回答用户的问题。如果你不知道答案，就说你不知道，不要编造答案。 上下文： {context} 问题：{question} 请用中文给出有帮助的答案：""" PROMPT = PromptTemplate( template=prompt_template, input_variables=["context", "question"] ) # 6. 构建 QA 链 llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0) # temperature=0 使输出更确定 qa_chain = RetrievalQA.from_chain_type( llm=llm, chain_type="stuff", # 最简单的方式，将所有上下文塞入提示 retriever=retriever, chain_type_kwargs={"prompt": PROMPT}, return_source_documents=True # 返回源文档，便于调试 ) # 测试 result = qa_chain.invoke({"query": "产品支持哪些支付方式？"}) print("答案：", result["result"]) print("来源：", result["source_documents"])

注意：嵌入模型的选择是性能和成本的权衡。如果数据敏感或想离线，务必使用本地模型（如BGE）。text-embedding-3虽然效果好，但会产生 API 调用费用和网络延迟。chunk_size需要根据你的模型上下文长度和文档特点调整，不是越大越好。

3.2 第二步：升级 RAG —— 实现多路召回与重排序

基础版直接用向量相似度检索，在有些场景下可能不够准。我们来增强它。

# 安装额外依赖：用于关键词召回和重排序 # pip install rank_bm25 sentence-transformers from langchain.retrievers import BM25Retriever, EnsembleRetriever from langchain.retrievers.document_compressors import CrossEncoderReranker from langchain.retrievers import ContextualCompressionRetriever from langchain_community.cross_encoders import HuggingFaceCrossEncoder # 1. 创建关键词检索器 (BM25) # 需要先将文档转换为纯文本列表 texts = [chunk.page_content for chunk in chunks] bm25_retriever = BM25Retriever.from_texts(texts) bm25_retriever.k = 5 # 也取前5个 # 2. 创建混合检索器 vector_retriever = vector_store.as_retriever(search_kwargs={"k": 5}) ensemble_retriever = EnsembleRetriever( retrievers=[bm25_retriever, vector_retriever], weights=[0.4, 0.6] # 可以调整权重，给向量检索更高权重 ) # 3. 创建重排序模型 # 使用一个轻量级的交叉编码器模型，它比嵌入模型更擅长判断相关性，但更慢 cross_encoder = HuggingFaceCrossEncoder(model_name="BAAI/bge-reranker-base") compressor = CrossEncoderReranker(model=cross_encoder, top_n=3) # 从混合结果中重排，只留最好的3个 # 4. 创建压缩检索器（即带重排序的检索器） compression_retriever = ContextualCompressionRetriever( base_compressor=compressor, base_retriever=ensemble_retriever ) # 5. 将增强后的检索器接入 QA 链 enhanced_qa_chain = RetrievalQA.from_chain_type( llm=llm, chain_type="stuff", retriever=compression_retriever, # 使用增强检索器 chain_type_kwargs={"prompt": PROMPT}, return_source_documents=True ) # 测试复杂问题 result = enhanced_qa_chain.invoke({"query": "请总结一下第三章提到的安全协议的具体实施步骤？"}) print("增强检索后的答案：", result["result"])

实操心得：多路召回和重排序会显著增加响应时间，但对于复杂、关键的查询，准确率的提升是值得的。在线上系统，可以考虑对简单查询走快速通道（仅向量检索），对复杂查询走增强通道。权重的设置（如weights=[0.4, 0.6]）需要在自己的数据集上做 AB 测试来确定。

3.3 第三步：定义 Agent 的工具与技能

现在，我们把上面构建的增强版 RAG 系统封装成 Agent 可以调用的工具。

from langchain.agents import Tool, initialize_agent, AgentType from langchain.memory import ConversationBufferMemory # 1. 将 RAG QA 链封装成工具函数 def search_knowledge_base(query: str) -> str: """用于查询内部知识库的工具。输入是一个问题，输出是相关的答案。""" try: result = enhanced_qa_chain.invoke({"query": query}) answer = result["result"] # 可以在这里添加一些后处理，比如截断长度 return answer except Exception as e: return f"查询知识库时出错：{str(e)}" # 2. 创建 Tool 对象 tools = [ Tool( name="Product_Knowledge_Base", func=search_knowledge_base, description="当需要查询关于产品功能、规格、使用指南、故障排除等内部知识时，使用此工具。输入应该是一个明确的问题。" ), # 你可以继续添加其他工具，例如： # Tool(name="Calculator", func=calculator, description="用于执行数学计算。"), # Tool(name="Web_Search", func=duckduckgo_search, description="用于搜索最新的公开信息。"), ] # 3. 创建对话记忆 memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True) # 4. 初始化 Agent agent = initialize_agent( tools=tools, llm=llm, # 使用和 RAG 相同或更强的 LLM，如 gpt-4 agent=AgentType.CHAT_CONVERSATIONAL_REACT_DESCRIPTION, # 适合多轮对话的 Agent 类型 verbose=True, # 开启详细日志，可以看到 Agent 的“思考过程” memory=memory, handle_parsing_errors=True # 优雅处理解析错误 )

现在，这个agent就是一个具备了内部知识查询能力的智能体了。你可以像对话一样向它提问，它会自己判断是否需要调用Product_Knowledge_Base工具。

3.4 第四步：运行与测试你的 RAG-Agent

让我们进行端到端的测试，看看它如何工作。

# 测试 1：直接的知识查询（Agent 应调用工具） print("=== 测试 1：知识查询 ===") response = agent.invoke({"input": "我们产品的数据备份策略是什么？"}) print("Agent 回复：", response["output"]) # 在 verbose=True 模式下，你会看到类似以下的思考过程： # > Entering new AgentExecutor chain... # 思考：用户问的是产品具体功能，我需要使用 Product_Knowledge_Base 工具来查找。 # 行动：调用 Product_Knowledge_Base，参数为“数据备份策略是什么？” # 观察：工具返回了从知识库检索到的具体备份策略文本... # 思考：我得到了相关信息，现在可以组织语言回答用户。 # 最终回复：根据知识库，我们的产品提供每日全量备份和每小时增量备份... # 测试 2：混合任务（需要知识+推理） print("\n=== 测试 2：混合任务 ===") response = agent.invoke({"input": "基于我们的安全协议，如果用户密码输错三次，接下来应该建议他怎么做？"}) print("Agent 回复：", response["output"]) # Agent 会先调用知识库工具获取“安全协议中关于密码错误处理”的条款，然后结合常识（建议重置密码或联系管理员）生成回答。 # 测试 3：无需工具的一般对话 print("\n=== 测试 3：一般对话 ===") response = agent.invoke({"input": "你好，今天天气怎么样？"}) print("Agent 回复：", response["output"]) # 因为工具描述不匹配，Agent 会判断这是一个普通对话，直接使用 LLM 的能力回答，不会调用知识库。

通过以上测试，你可以清晰地看到 Agent 的决策过程。它像一个项目经理，根据任务类型（需求），决定是派专家（工具）去查资料，还是自己直接处理。

4. 工程化与优化：让系统健壮、可维护

一个能跑通的 Demo 和一个可上线的系统之间，隔着无数个工程细节。以下是几个必须考虑的优化方向。

4.1 知识库的持续更新与版本管理

文档不是一成不变的。你需要设计一个流程来处理知识库的更新。

• 增量更新：最简单的办法是定期（如每天）全量重建向量库。但对于大规模知识库，成本太高。理想方案是支持增量更新。Qdrant支持通过point_id更新或删除特定向量。你需要建立一套映射关系，记录文档块（Chunk）的源文件 ID 和版本。
• 版本控制：可以考虑将文档内容和其对应的向量嵌入存储在同一事务中，并带有版本号。当文档更新时，先标记旧向量为失效，再插入新向量。查询时只检索最新版本。
• 更新策略：可以设计一个监听文件夹或 API 的守护进程，当有新文档放入或旧文档修改时，自动触发解析、分块、向量化并更新数据库。

4.2 检索质量监控与评估

你怎么知道你的 RAG 系统工作得好不好？需要建立监控评估体系。

• 人工评估样本集：构建一个包含“问题-标准答案-相关文档”的测试集，定期运行，评估答案的准确性和相关性。
• 自动评估指标：
  • 检索阶段：计算Hit Rate（标准答案文档是否在检索结果中）和MRR（标准答案文档的排名倒数均值）。
  • 生成阶段：使用ROUGE、BLEU或基于 LLM 的评估器（如用 GPT-4 判断生成答案与标准答案的一致性）来评估生成质量。
• 日志与分析：记录每一次查询的检索结果（source_documents）、生成的答案、以及用户的反馈（如果有）。这些数据是迭代优化分块策略、检索模型和提示模板的黄金资料。

4.3 Agent 的稳定性与安全性增强

让 Agent 自主运行，必须考虑约束和安全。

• 工具调用限制：为工具调用设置超时和重试机制，避免因单个工具挂起导致整个 Agent 卡死。
• 输入输出过滤：对用户输入和工具输出进行必要的清洗和过滤，防止提示词注入攻击或暴露内部敏感信息。
• 验证与确认：对于高风险操作（如发送邮件、执行数据库写操作），可以让 Agent 生成计划后，先向用户确认，再执行。
• 迭代次数限制：在AgentExecutor中设置max_iterations参数，防止 Agent 陷入无限循环的“思考-行动”中。

4.4 性能优化与成本控制

随着用户量增长，性能和成本成为关键。

• 嵌入缓存：对相同的文本块进行重复向量化是浪费。可以建立本地嵌入缓存（如用Redis或SQLite存储文本 -> 向量的映射），在向量化前先查缓存。
• 检索优化：
  • 索引优化：向量数据库使用 HNSW 等近似最近邻算法索引，在精度和速度间取得平衡。
  • 分层检索：先使用简单的关键词匹配（如 Elasticsearch）快速缩小范围，再在候选集上做精细的向量检索。
• LLM 调用优化：
  • 提示压缩：在将检索到的上下文喂给 LLM 前，尝试用更小的模型进行摘要压缩，减少 token 消耗。
  • 流式响应：对于长答案，使用流式输出提升用户体验。
  • 模型路由：简单问题用便宜快速的模型（如gpt-3.5-turbo），复杂问题再用强大模型（如gpt-4）。

5. 常见问题与排查实录

在开发和调试过程中，我踩过不少坑。这里把一些典型问题和解决方法记录下来，希望能帮你节省时间。

5.1 检索效果不佳，总是找不到相关内容

这是 RAG 系统最常见的问题。可以按以下步骤排查：

我的心得：分块策略是 RAG 的“地基”。花 30% 的时间优化分块，能解决 50% 的检索问题。对于技术文档，我强烈推荐先尝试按 Markdown 标题（#,##）分块，效果往往比固定尺寸分块好得多。

5.2 Agent 不调用工具，或错误调用工具

这通常与工具的描述（description）和 Agent 的类型有关。

• Agent 完全不调用工具：
  • 检查工具描述：描述必须清晰、具体，说明在什么情况下使用此工具。LLM 主要靠描述做判断。例如，“用于计算数学表达式”就比“一个计算工具”好。
  • 检查 Agent 类型：AgentType.ZERO_SHOT_REACT_DESCRIPTION适合简单任务，AgentType.CHAT_CONVERSATIONAL_REACT_DESCRIPTION更适合多轮对话。复杂的规划任务可能需要OPENAI_FUNCTIONS或PLANNER类型。
  • 开启 verbose 模式：这是最重要的调试手段，直接看 Agent 的思考链（Chain of Thought），它为什么决定不调用工具？
• Agent 调用了错误的工具：
  • 工具描述区分度不够：确保每个工具的描述有明确的边界。如果两个工具都描述为“处理用户问题”，LLM 就会困惑。
  • 给 LLM 更强的指令：在初始化 Agent 的agent_kwargs中，可以提供更详细的系统提示（system_message），明确指导它如何选择工具。

5.3 响应速度慢，用户体验差

性能瓶颈可能出现在多个环节。

1. 嵌入模型速度慢：如果使用本地大模型，考虑换成更小的模型（如all-MiniLM-L6-v2），或使用 GPU 加速。对于生产环境，异步处理和缓存是关键。
2. 向量检索慢：检查向量数据库的索引类型。对于千万级以下数据，HNSW 索引通常性能很好。确保查询时使用了合适的搜索参数（如ef或search_k）。
3. LLM 生成慢：这是主要瓶颈。可以考虑：
   • 使用更快的模型（如从gpt-4降级到gpt-3.5-turbo）。
   • 优化提示词，减少不必要的上下文。
   • 实现流式输出，让用户先看到部分结果。
4. 网络延迟：如果使用云端 API，网络波动会影响速度。需要设置合理的超时和重试机制，并考虑使用后端轮询或 WebSocket 向前端推送结果。

5.4 如何处理“知识库中没有答案”的情况

这是评估 RAG 系统可靠性的重要一点。我们必须在提示模板中明确要求模型“不知道就说不知道”。

# 一个更健壮的提示模板示例 robust_prompt_template = """你是一个专业的客服助手，请严格根据提供的上下文信息来回答问题。 如果上下文中的信息不足以完全回答问题，请基于已知部分诚实回答，并明确说明哪些信息是缺失的。 如果上下文与问题完全无关，请直接说“根据现有资料，我无法回答这个问题”。 上下文信息如下： {context} 用户问题：{question} 请根据以上上下文，给出准确、有帮助的回答："""

同时，在检索阶段，可以设定一个相似度阈值。当最相关文档块的相似度分数低于某个值（如 0.7）时，直接判定为“未找到相关信息”，不将其送入 LLM，而是返回预设的“知识库暂无此信息”回复，这可以节省 LLM 调用成本并避免误导。

搭建这个系统的过程，就像在组装一个既有“海量记忆”（RAG）又有“聪明大脑”（Agent）的数字生命体。最初的版本可能笨拙，但通过持续的迭代——优化分块、调整检索策略、丰富工具集、完善提示——你会亲眼看到它的能力边界一步步拓宽。最让我有成就感的时刻，往往是看到它成功地将一个模糊的用户需求，通过几次工具调用和逻辑推理，转化为一个准确、 actionable 的结果。这不仅仅是技术的实现，更是对复杂问题解决流程的一种自动化建模。