Python实战：构建具备动态检索能力的Agentic RAG系统

1. 从静态到动态：为什么我们需要Agentic RAG？

如果你在过去一年里折腾过RAG（检索增强生成），大概率经历过这样的场景：你精心准备了向量数据库，写好了检索逻辑，满心欢喜地输入一个问题，结果模型给出的答案要么是“根据现有资料，无法回答”，要么就是一本正经地胡说八道，引用了完全不相关的文档片段。你检查了分块策略、Embedding模型、甚至重写了提示词，但效果提升有限。问题出在哪？很多时候，问题不在于检索本身，而在于我们把它当作了一个“一锤子买卖”——用户提问，系统检索一次，然后生成答案。这种静态的、一次性的检索，在面对复杂、多跳或需要上下文推理的问题时，显得力不从心。

这就是“Agentic RAG”要解决的核心痛点。它不是一个全新的技术，而是一种架构思想的演进。传统的RAG像一个“图书管理员”，你问一个问题，他根据关键词去书架上找一本最相关的书给你。而Agentic RAG则更像一个“研究助理”。你给他一个复杂任务，比如“帮我分析一下公司上个季度在亚太地区销售下滑的原因，并对比一下北美市场的策略”，这个“研究助理”会自己拆解任务：他可能先检索“上季度亚太销售报告”，发现提到了“供应链延迟”；接着他会自主发起新的检索，去查“亚太地区供应链情况”；然后他可能意识到需要对比，再去检索“北美市场供应链策略”；最后，他综合所有信息，给你一份结构化的分析报告。整个过程是动态的、多轮的、有“思考”的。

所以，“基于Python的RAG开发手册——带动态检索的Agentic RAG”这个标题，指向的正是当前RAG落地中最具挑战性也最有趣的前沿：如何让RAG系统具备自主规划、决策和迭代检索的能力。这不仅仅是加个“Agent”的标签，而是涉及到任务分解、工具调用、状态管理、自我反思等一系列设计模式的深度融合。接下来，我将结合具体的Python实现，拆解如何从零构建这样一个系统，并分享我在实际项目中趟过的坑和积累的经验。

2. 架构基石：理解Agentic RAG的核心组件与工作流

在动手写代码之前，我们必须先厘清Agentic RAG系统由哪些核心部分组成，以及数据和控制流是如何在这些组件间运转的。一个典型的Agentic RAG架构可以抽象为以下几个层次，这远比简单的“检索-生成”链条要复杂。

2.1 核心组件拆解

1. 智能体（Agent）：这是系统的大脑。它通常基于一个大语言模型（LLM），负责理解用户意图、制定计划、决定下一步行动。在Python生态中，LangChain的AgentExecutor、LlamaIndex的AgentRunner或是基于OpenAI Function Calling自定义的Agent都是常见实现。它的核心能力是“决策”：根据当前上下文，决定是调用检索工具、进行计算，还是直接给出最终答案。

2. 工具（Tools）：这是智能体的手脚。一个“带动态检索的RAG”系统，其核心工具就是可多次、条件性调用的检索器。除此之外，还可能包括计算器、代码执行器、网络搜索API等。工具的设计要点在于：给智能体提供清晰、可靠的函数接口和描述，让LLM知道在什么情况下该用什么工具。

3. 检索器（Retriever）：这是传统RAG的核心，在Agentic架构中它降级为“工具之一”，但重要性丝毫未减。它需要支持按需调用，并能处理智能体发出的、可能更精确或更抽象的查询。例如，智能体在分析过程中可能会生成这样的查询：“查找与‘Q3 revenue decline’和‘supply chain disruption’同时相关的内部会议纪要”。

4. 记忆与状态管理（Memory & State）：这是实现“动态”和“多轮”的关键。智能体在与工具交互、生成中间结果的过程中，需要有一个地方来存储对话历史、已检索到的文档片段、以及任务执行的中间状态。这可以是简单的列表，也可以是更复杂的图结构，用于记录推理路径。

5. 规划与反思模块（Planning & Reflection）：高级的Agentic RAG会引入更复杂的机制。规划是指智能体在开始前就拆解任务步骤（如ReAct范式中的“Thought”）。反思是指在得到初步答案或遇到矛盾时，智能体能评估结果质量，并决定是否需要修正查询或进行更深度的检索。例如，如果首次检索到的文档相关性都很低，反思模块会触发智能体重新表述问题。

2.2 动态工作流示意

一个简化的工作流循环如下：

• 输入：用户复杂查询。
• 步骤1 - 规划/思考：智能体分析查询，决定第一步行动（例如：“我需要先了解事件背景，应该调用检索工具查询关键词A和B”）。
• 步骤2 - 执行：智能体调用检索工具，传入生成的查询语句。
• 步骤3 - 观察：检索工具返回相关文档片段。这些片段被添加到上下文中。
• 步骤4 - 反思/迭代：智能体评估现有信息是否足够回答用户问题。如果不够，回到步骤1，生成下一步的“思考”和新的检索查询（例如：“已有背景信息，现在需要查找事件的具体影响，应检索关键词C”）。
• 步骤5 - 终结：当智能体判断信息已充分，或达到最大迭代次数时，它综合所有检索到的上下文，生成最终答案并输出。

这个循环的核心是“检索”动作被智能体所掌控，可以根据中间结果动态调整，而不再是一次性的固定操作。

3. 实战构建：用LangChain搭建一个基础Agentic RAG系统

理论讲完了，我们上代码。这里我选择用LangChain来构建，因为它提供了丰富的Agent和Tool抽象，能让我们快速搭建原型。假设我们的知识库是一组关于产品技术的Markdown文档。

3.1 环境准备与知识库加载

首先，安装核心库并准备文档。这里我强烈建议使用Chroma作为向量数据库，它轻量且与LangChain集成良好。

pip install langchain langchain-community langchain-chroma langchain-openai tiktoken pypdf

# 1. 文档加载与分割 from langchain_community.document_loaders import DirectoryLoader, TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter # 假设文档在 ./docs 目录下 loader = DirectoryLoader('./docs', glob="**/*.md", loader_cls=TextLoader) documents = loader.load() # 分块策略是关键：对于Agentic RAG，块可以稍大一些，因为智能体会进行多轮精炼检索。 # 但重叠度（overlap）建议设置，以保持上下文连贯。 text_splitter = RecursiveCharacterTextSplitter( chunk_size=1000, # 块大小 chunk_overlap=200, # 重叠字符 length_function=len, separators=["\n\n", "\n", "。", "！", "？", "；", "，", " ", ""] ) chunks = text_splitter.split_documents(documents) print(f"共切分为 {len(chunks)} 个文本块。") # 2. 向量化与存储 from langchain_openai import OpenAIEmbeddings from langchain_chroma import Chroma # 使用OpenAI的Embedding模型，注意环境变量 OPENAI_API_KEY 需已设置 embeddings = OpenAIEmbeddings(model="text-embedding-3-small") # 持久化存储到本地目录 './chroma_db' vectorstore = Chroma.from_documents( documents=chunks, embedding=embeddings, persist_directory="./chroma_db" ) vectorstore.persist() # 持久化保存 # 3. 创建检索器 retriever = vectorstore.as_retriever( search_type="similarity", # 相似度检索 search_kwargs={"k": 4} # 每次检索返回4个最相关的块 )

注意：Embedding模型的选择极大影响检索质量。text-embedding-3-small在成本和性能间取得了很好平衡。对于中文场景，可能需要考虑text-embedding-3-small对中文的优化程度，或使用本地模型如bge-large-zh-v1.5。

3.2 创建智能体工具与系统提示词

接下来，我们将检索器包装成一个智能体可以调用的工具，并设计驱动智能体行为的系统提示词。

from langchain.agents import Tool, create_react_agent from langchain_openai import ChatOpenAI from langchain import hub # 1. 定义检索工具 # 工具的描述至关重要！它告诉LLM这个工具能做什么、什么时候用。 def rag_retriever(query: str) -> str: """一个用于从产品技术知识库中检索相关信息的工具。当你需要查找具体的产品特性、技术规格、故障解决方法或历史文档时，就使用这个工具。输入应该是一个清晰的搜索查询语句。""" docs = retriever.invoke(query) # 调用检索器 # 将检索结果格式化为字符串 content = "\n\n".join([doc.page_content for doc in docs]) return f"以下是从知识库中检索到的相关信息：\n\n{content}" # 将函数封装成Tool对象 tools = [ Tool( name="Knowledge_Base_Search", func=rag_retriever, description="用于从内部知识库中搜索产品文档、技术指南和解决方案。当你需要基于公司内部文档来回答问题时，优先使用此工具。" ), # 你可以在这里添加更多工具，例如： # Tool(name="Calculator", func=calculator, description="用于执行数学计算。"), # Tool(name="Web_Search", func=duckduckgo_search, description="用于搜索最新的公开网络信息。"), ] # 2. 初始化LLM（智能体的大脑） llm = ChatOpenAI(model="gpt-4-turbo-preview", temperature=0) # temperature设为0使输出更稳定 # 3. 拉取一个ReAct风格的智能体提示词模板 # LangChain Hub上有很多预设模板，`react-chat`适合对话场景。 prompt = hub.pull("hwchase17/react-chat") # 4. 创建智能体 agent = create_react_agent(llm, tools, prompt)

实操心得：工具描述（description）是智能体能否正确调用工具的灵魂。描述要具体、明确地指出使用场景和输入格式。模糊的描述会导致智能体滥用或不用工具。我习惯在描述中加上“当你需要...时，就使用这个工具”和“输入应该是...”这样的句式。

3.3 运行智能体并观察动态检索过程

现在，让我们用一个复杂的查询来测试这个系统，并观察其思考过程。

from langchain.agents import AgentExecutor # 创建执行器，设置verbose=True以查看智能体的“思考”过程 agent_executor = AgentExecutor( agent=agent, tools=tools, verbose=True, # 关键！这将打印出智能体的推理链 handle_parsing_errors=True, # 优雅地处理解析错误 max_iterations=5, # 防止无限循环，设置最大迭代次数 early_stopping_method="generate" # 当智能体输出最终答案时停止 ) # 一个需要多步检索的复杂问题 complex_question = “我们产品‘NexusHub’的‘智能路由’功能，在版本v2.1.0到v2.2.0的升级过程中，客户反馈的主要性能问题是什么？以及文档中提到的临时解决方案是什么？” print("用户问题：", complex_question) print("="*50) result = agent_executor.invoke({"input": complex_question, "chat_history": []}) print("="*50) print("\n最终答案：") print(result["output"])

当verbose=True时，你会在控制台看到类似以下的输出，这清晰地展示了Agentic RAG的动态过程：

> Entering new AgentExecutor chain... 思考：用户的问题涉及特定产品“NexusHub”的特定功能“智能路由”在两个特定版本间的升级问题。我需要找到客户反馈的性能问题和文档中的解决方案。我应该先使用知识库搜索工具查找关于“NexusHub 智能路由 v2.1.0 升级”的文档。 行动：Knowledge_Base_Search 行动输入：NexusHub 智能路由 版本 v2.1.0 升级 v2.2.0 客户反馈 性能问题 观察：以下是从知识库中检索到的相关信息： [文档片段1： v2.2.0发布公告，提到了智能路由模块的重构...] [文档片段2： 一份已知问题列表，其中有一条关于“智能路由在高峰时段延迟增加”...] 思考：我找到了一些关于性能问题的提及，但不够具体。我需要更详细的客户反馈报告。也许应该搜索“客户反馈”或“故障报告”相关的文档。 行动：Knowledge_Base_Search 行动输入：“智能路由” 性能 客户反馈 报告 v2.2.0 观察：以下是从知识库中检索到的相关信息： [文档片段3： 客户支持案例汇总，编号CS-2023-445，主题“升级后API延迟飙升”...] [文档片段4： 内部性能测试报告，对比v2.1.0和v2.2.0的延迟指标...] 思考：现在我对问题有了更具体的了解（案例CS-2023-445）。接下来，我需要查找文档中针对此问题的临时解决方案。 行动：Knowledge_Base_Search 行动输入：CS-2023-445 临时解决方案 智能路由 观察：以下是从知识库中检索到的相关信息： [文档片段5： 临时解决方案KB-00123： 通过配置参数‘routing_thread_pool_size’可以缓解...] 思考：我已经收集了必要的信息。现在可以综合回答了。 最终答案：根据知识库文档，在NexusHub从v2.1.0升级至v2.2.0后，关于“智能路由”功能，客户反馈的主要性能问题是升级后出现的API请求延迟显著增加，尤其是在业务高峰时段。这在一份客户支持案例（CS-2023-445）中有详细记录。文档中提供的临时解决方案（KB-00123）是：通过调整配置文件中的‘routing_thread_pool_size’参数值，将其从默认的10增加到20或30，以分配更多的处理线程来缓解队列拥堵，从而降低延迟。该方案注明为临时措施，建议在后续的v2.2.1热修复版本中彻底解决。

你可以看到，智能体自动执行了三次检索，每次的查询词都基于前一次的结果进行了优化和聚焦。这就是“动态检索”的威力：它模拟了人类研究员逐步深入、迭代搜索的思考过程。

4. 进阶优化：让Agentic RAG更可靠、更强大

基础版本跑通了，但离生产可用还有距离。下面分享几个关键的优化方向，这些都是我在实际项目中踩过坑后总结的经验。

4.1 处理检索失败与智能体“幻觉”

即使有工具，智能体也可能“偷懒”或不准确。常见问题包括：

• 工具调用不准确：智能体生成的查询词太模糊或语法奇怪，导致检索结果差。
• 逃避使用工具：智能体直接基于自身知识生成答案，造成“幻觉”（Hallucination）。
• 陷入循环：智能体在两个不充分的检索结果间来回切换。

解决方案：强化提示词与后处理

1. 定制化系统提示词：不要完全依赖LangChain Hub的通用模板。根据你的领域定制提示词，强制要求智能体必须使用工具。

   custom_prompt = """你是一个严谨的产品技术支持助手，必须严格依据公司内部知识库回答问题。 你拥有一个名为‘Knowledge_Base_Search’的工具，可以检索最新的产品文档、案例和解决方案。 回答用户问题时，请遵循以下步骤： 1. 分析用户问题，确定是否需要从知识库查找信息。几乎所有具体技术问题都需要。 2. 如果需要，构思一个简洁、精准的搜索查询词，然后调用‘Knowledge_Base_Search’工具。 3. 仔细阅读工具返回的文档内容。 4. 如果返回的文档完全回答了问题，请基于这些文档组织答案。 5. 如果返回的文档只回答了部分问题，请分析还缺什么信息，再次构思新的查询词进行检索。 6. 严禁在未使用工具检索的情况下，编造任何关于产品特性、版本、配置的具体信息。 7. 如果多次检索后仍找不到相关信息，请如实告知用户“在现有知识库中未找到相关记录”，并建议其联系人工支持。 当前对话： {chat_history} 用户问题：{input} 开始思考："""

2. 实现检索结果重排序（Re-ranking）：简单的向量相似度检索可能会把一些相关但排名靠后的文档漏掉。可以在检索器返回初步结果后，加入一个重排序模型（如BAAI/bge-reranker-large），对Top K个结果进行精排，将最相关的文档提到最前面，提高输入给LLM的上下文质量。

3. 设置迭代上限与超时：务必在AgentExecutor中设置max_iterations（如10次）和max_execution_time，防止智能体陷入死循环消耗资源。

4.2 集成更多工具与实现路由逻辑

一个强大的智能体不应该只有检索能力。我们可以为其增加更多工具，并设计路由逻辑。

from langchain_community.tools import DuckDuckGoSearchRun from langchain_community.utilities import ArxivAPIWrapper # 创建更多工具 search = DuckDuckGoSearchRun() arxiv = ArxivAPIWrapper() tools = [ Tool(name="Internal_KB_Search", func=rag_retriever, description="搜索公司内部产品知识库和文档。"), Tool(name="Web_Search", func=search.run, description="使用搜索引擎查找最新的公共信息、新闻或通用知识。"), Tool(name="Arxiv_Search", func=arxiv.run, description="搜索学术论文，获取前沿研究信息。"), ] # 更高级的做法：使用“工具检索器”（Tool Retriever）或“多智能体路由”。 # 例如，可以训练一个简单的分类器或使用一个小型LLM，根据用户问题判断应该将任务分配给“内部知识库智能体”还是“网络搜索智能体”。

经验技巧：工具不是越多越好。工具越多，智能体越容易困惑。建议从核心工具开始，逐步增加。并为每个工具提供极其清晰、互斥的描述，帮助智能体做选择。对于复杂路由，可以考虑使用LangChain的MultiRouteChain或自定义一个路由Chain。

4.3 为智能体注入“反思”能力

这是实现“智能”的关键一步。让智能体在行动后评估结果，决定下一步。

from langchain.agents import AgentExecutor from langchain_core.agents import AgentFinish class ReflexiveAgentExecutor(AgentExecutor): """一个带有简单反思机制的智能体执行器""" def _take_next_step(self, ...): # 继承并重写核心步骤函数 # 在智能体决定下一步行动前，插入反思逻辑 if intermediate_steps: # 如果已经有历史步骤 last_observation = intermediate_steps[-1][1] # 获取上一次工具返回的结果 # 简单的反思：如果上次检索结果为空或相关性低，则触发重新思考 if "未找到" in last_observation or len(last_observation) < 50: # 可以构造一个特殊的“反思提示”，让LLM分析失败原因并生成新的查询 reflection_prompt = f"上一次检索没有得到有用信息。原始问题是：{input}。上一次使用的查询词是：{last_query}。请分析为什么没找到，并生成一个更可能找到答案的新查询词。" # ... 调用LLM进行反思 ... new_query = llm.invoke(reflection_prompt) # 然后用新的查询词继续 # 调用父类的原始逻辑执行行动 return super()._take_next_step(...) # 使用自定义的执行器 agent_executor = ReflexiveAgentExecutor(agent=agent, tools=tools, verbose=True, max_iterations=6)

更成熟的框架如LangGraph，专门为构建有状态的、支持循环和分支的智能体工作流而设计，实现反思和规划更加自然。你可以将“检索”、“评估”、“生成”定义为不同的节点，通过条件边来控制流程的走向。

5. 生产环境部署的考量与避坑指南

将原型部署到生产环境，会面临一系列新的挑战。这里分享几个关键的考量点。

5.1 性能、成本与延迟优化

Agentic RAG的多轮LLM调用和检索意味着更高的成本和延迟。

• LLM调用成本：每次“思考”和“生成”都是一次API调用。优化策略：

  • 使用更小/更快的模型：对于工具调用的决策（即“思考”步骤），可以使用成本更低的模型（如gpt-3.5-turbo），仅在最终生成答案时使用大模型（如gpt-4）。这被称为“大小模型协同”。
  • 缓存：对频繁出现的、相似的智能体“思考”过程和检索结果进行缓存。可以使用langchain.cache配合SQLiteCache或RedisCache。
  • 设置超时和限流：对智能体的执行时间进行严格限制，避免复杂问题消耗过多资源。

• 检索延迟：

  • 优化向量数据库：确保向量索引构建正确，对于大规模数据，考虑使用FAISS、Weaviate或Pinecone等性能更高的专业向量数据库。
  • 并行检索：如果智能体需要多个不相关的信息，可以考虑在工具层面支持并行查询（但这需要智能体能提前规划好并行任务，实现较复杂）。

5.2 稳定性与错误处理

智能体在复杂环境中运行时，什么奇怪的事情都可能发生。

• 工具调用错误：检索器可能因为网络、数据库连接失败而抛出异常。必须在工具函数内部做好try-catch，并返回一个对智能体友好的错误信息，例如：“检索服务暂时不可用，请稍后再试或尝试简化您的问题。”
• LLM输出解析失败：LangChain的Agent依赖于LLM输出格式严格的Action和Action Input。有时LLM会输出不规范的内容。确保在AgentExecutor中设置handle_parsing_errors=True，并可以定义一个错误处理函数，尝试修复或提示用户重新提问。
• 上下文长度管理：多轮对话和多次检索结果会迅速撑爆LLM的上下文窗口。需要实现一个“摘要式记忆”或“关键信息提取”机制，只把最相关的历史信息保留在上下文中，而不是全部堆进去。

5.3 评估与监控

如何知道你的Agentic RAG系统工作得好不好？

• 定义评估指标：除了传统的检索精度（Precision@K）、答案准确性，还需要评估：
  • 工具调用准确率：智能体在应该调用工具时是否调用了？调用的工具是否正确？
  • 查询词质量：智能体生成的搜索查询是否有效？可以通过人工评估或与理想查询的相似度来衡量。
  • 任务完成度：对于多步任务，智能体是否完成了所有必要的步骤？
• 建立监控看板：记录每次会话的日志，包括：用户问题、智能体的思考链、每次工具调用的输入输出、最终答案、总耗时、Token消耗等。这有助于分析失败案例和优化系统。
• A/B测试：对比静态RAG和Agentic RAG在复杂问题上的回答质量。质量评估可以结合自动化评分（如使用GPT-4作为裁判）和人工评分。

构建一个健壮的、生产可用的Agentic RAG系统，是一个持续迭代的过程。从最简单的动态检索开始，逐步引入反思、规划、多工具协作等能力，同时牢牢把控性能、成本和稳定性这条生命线。这个过程充满挑战，但当你看到系统能够自主拆解并解决一个复杂问题时，所带来的价值提升是巨大的。