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 核心组件拆解智能体Agent这是系统的大脑。它通常基于一个大语言模型LLM负责理解用户意图、制定计划、决定下一步行动。在Python生态中LangChain的AgentExecutor、LlamaIndex的AgentRunner或是基于OpenAI Function Calling自定义的Agent都是常见实现。它的核心能力是“决策”根据当前上下文决定是调用检索工具、进行计算还是直接给出最终答案。工具Tools这是智能体的手脚。一个“带动态检索的RAG”系统其核心工具就是可多次、条件性调用的检索器。除此之外还可能包括计算器、代码执行器、网络搜索API等。工具的设计要点在于给智能体提供清晰、可靠的函数接口和描述让LLM知道在什么情况下该用什么工具。检索器Retriever这是传统RAG的核心在Agentic架构中它降级为“工具之一”但重要性丝毫未减。它需要支持按需调用并能处理智能体发出的、可能更精确或更抽象的查询。例如智能体在分析过程中可能会生成这样的查询“查找与‘Q3 revenue decline’和‘supply chain disruption’同时相关的内部会议纪要”。记忆与状态管理Memory State这是实现“动态”和“多轮”的关键。智能体在与工具交互、生成中间结果的过程中需要有一个地方来存储对话历史、已检索到的文档片段、以及任务执行的中间状态。这可以是简单的列表也可以是更复杂的图结构用于记录推理路径。规划与反思模块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_clsTextLoader) documents loader.load() # 分块策略是关键对于Agentic RAG块可以稍大一些因为智能体会进行多轮精炼检索。 # 但重叠度overlap建议设置以保持上下文连贯。 text_splitter RecursiveCharacterTextSplitter( chunk_size1000, # 块大小 chunk_overlap200, # 重叠字符 length_functionlen, 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(modeltext-embedding-3-small) # 持久化存储到本地目录 ./chroma_db vectorstore Chroma.from_documents( documentschunks, embeddingembeddings, persist_directory./chroma_db ) vectorstore.persist() # 持久化保存 # 3. 创建检索器 retriever vectorstore.as_retriever( search_typesimilarity, # 相似度检索 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( nameKnowledge_Base_Search, funcrag_retriever, description用于从内部知识库中搜索产品文档、技术指南和解决方案。当你需要基于公司内部文档来回答问题时优先使用此工具。 ), # 你可以在这里添加更多工具例如 # Tool(nameCalculator, funccalculator, description用于执行数学计算。), # Tool(nameWeb_Search, funcduckduckgo_search, description用于搜索最新的公开网络信息。), ] # 2. 初始化LLM智能体的大脑 llm ChatOpenAI(modelgpt-4-turbo-preview, temperature0) # 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 # 创建执行器设置verboseTrue以查看智能体的“思考”过程 agent_executor AgentExecutor( agentagent, toolstools, verboseTrue, # 关键这将打印出智能体的推理链 handle_parsing_errorsTrue, # 优雅地处理解析错误 max_iterations5, # 防止无限循环设置最大迭代次数 early_stopping_methodgenerate # 当智能体输出最终答案时停止 ) # 一个需要多步检索的复杂问题 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])当verboseTrue时你会在控制台看到类似以下的输出这清晰地展示了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。陷入循环智能体在两个不充分的检索结果间来回切换。解决方案强化提示词与后处理定制化系统提示词不要完全依赖LangChain Hub的通用模板。根据你的领域定制提示词强制要求智能体必须使用工具。custom_prompt 你是一个严谨的产品技术支持助手必须严格依据公司内部知识库回答问题。 你拥有一个名为‘Knowledge_Base_Search’的工具可以检索最新的产品文档、案例和解决方案。 回答用户问题时请遵循以下步骤 1. 分析用户问题确定是否需要从知识库查找信息。几乎所有具体技术问题都需要。 2. 如果需要构思一个简洁、精准的搜索查询词然后调用‘Knowledge_Base_Search’工具。 3. 仔细阅读工具返回的文档内容。 4. 如果返回的文档完全回答了问题请基于这些文档组织答案。 5. 如果返回的文档只回答了部分问题请分析还缺什么信息再次构思新的查询词进行检索。 6. 严禁在未使用工具检索的情况下编造任何关于产品特性、版本、配置的具体信息。 7. 如果多次检索后仍找不到相关信息请如实告知用户“在现有知识库中未找到相关记录”并建议其联系人工支持。 当前对话 {chat_history} 用户问题{input} 开始思考实现检索结果重排序Re-ranking简单的向量相似度检索可能会把一些相关但排名靠后的文档漏掉。可以在检索器返回初步结果后加入一个重排序模型如BAAI/bge-reranker-large对Top K个结果进行精排将最相关的文档提到最前面提高输入给LLM的上下文质量。设置迭代上限与超时务必在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(nameInternal_KB_Search, funcrag_retriever, description搜索公司内部产品知识库和文档。), Tool(nameWeb_Search, funcsearch.run, description使用搜索引擎查找最新的公共信息、新闻或通用知识。), Tool(nameArxiv_Search, funcarxiv.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(agentagent, toolstools, verboseTrue, max_iterations6)更成熟的框架如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_errorsTrue并可以定义一个错误处理函数尝试修复或提示用户重新提问。上下文长度管理多轮对话和多次检索结果会迅速撑爆LLM的上下文窗口。需要实现一个“摘要式记忆”或“关键信息提取”机制只把最相关的历史信息保留在上下文中而不是全部堆进去。5.3 评估与监控如何知道你的Agentic RAG系统工作得好不好定义评估指标除了传统的检索精度PrecisionK、答案准确性还需要评估工具调用准确率智能体在应该调用工具时是否调用了调用的工具是否正确查询词质量智能体生成的搜索查询是否有效可以通过人工评估或与理想查询的相似度来衡量。任务完成度对于多步任务智能体是否完成了所有必要的步骤建立监控看板记录每次会话的日志包括用户问题、智能体的思考链、每次工具调用的输入输出、最终答案、总耗时、Token消耗等。这有助于分析失败案例和优化系统。A/B测试对比静态RAG和Agentic RAG在复杂问题上的回答质量。质量评估可以结合自动化评分如使用GPT-4作为裁判和人工评分。构建一个健壮的、生产可用的Agentic RAG系统是一个持续迭代的过程。从最简单的动态检索开始逐步引入反思、规划、多工具协作等能力同时牢牢把控性能、成本和稳定性这条生命线。这个过程充满挑战但当你看到系统能够自主拆解并解决一个复杂问题时所带来的价值提升是巨大的。