从概念到生产：工程化构建Agentic RAG智能问答系统

30款热门AI模型一站整合DeepSeek/GLM/Claude 随心用限时 5 折。 点击领海量免费额度在构建企业级智能问答系统时你是否遇到过这样的困境用户一个看似简单的业务问题比如“上个月华东区A产品的销售额是多少”系统却因为数据分散在订单、产品和区域等多个数据库中只能返回“未找到相关信息”或一个基于片面数据的错误答案。这正是传统检索增强生成RAG架构的典型痛点——单次、静态的检索难以应对复杂、多跳的查询场景。近期Google推出的Agentic RAG框架为解决这一难题提供了新思路。它不再将RAG视为一次性的“检索-生成”管道而是引入了一个由多个智能体Agent协作的、具备自我检查和迭代能力的动态系统。其核心的“质检Agent”Sufficient Context Agent能够评估检索到的信息是否足以回答问题并在发现缺口时主动引导补充检索从而显著提升答案的完整性和准确性。据公开测试其在多跳问答任务中的准确率比传统RAG提升了34%。然而从阅读一篇技术新闻到真正构建一个稳定、高效、可投入生产的Agentic RAG系统中间隔着巨大的工程化鸿沟。本文将带你跨越这道鸿沟从核心概念拆解开始逐步构建一个从模拟Google Search到具备生产级可信度的AI Agent的完整工程化方案。我们将聚焦于架构设计、关键组件实现、性能优化与可信保障并提供可直接复用的代码示例。1. Agentic RAG 核心概念与工程化挑战在深入代码之前我们必须厘清Agentic RAG与传统RAG的本质区别并明确将其工程化所面临的核心挑战。1.1 传统RAG vs. Agentic RAG从管道到智能体工作流传统RAG是一个线性管道Pipeline检索Retrieval将用户问题转换为向量在向量数据库中执行一次相似性搜索返回Top-K个相关文档片段。生成Generation将问题和检索到的片段一起输入大语言模型LLM生成最终答案。这个模型的缺陷显而易见检索是静态的、一次性的。如果第一次检索没有找到全部必要信息例如问题需要关联多个数据源或首次检索的关键词不够精确系统就会失败。它缺乏“反思”和“补救”的能力。Agentic RAG则将这个过程重构为一个由智能体驱动的工作流Workflow任务规划与分解智能体理解复杂问题并将其拆解为一系列可执行的子任务或查询。动态迭代检索智能体可以发起多次、多路径的检索。每次检索后会有一个“评估”环节判断信息是否充足。判断与决策核心的“质检Agent”评估当前收集的上下文是否足以回答问题。如果不足它会分析缺失什么并生成新的、更精确的检索指令。综合与生成当信息被判定为充足或达到迭代上限时另一个智能体负责整合所有检索结果生成连贯、准确的最终答案。简单来说Agentic RAG为系统装上了“大脑”和“质量控制环节”使其具备了感知-思考-行动-再评估的循环能力。1.2 工程化核心挑战将上述概念落地为生产系统我们需要解决以下挑战复杂性管理多智能体协作带来了复杂的控制流和状态管理。如何清晰定义各Agent的职责、通信协议和协作流程延迟与成本多次调用LLM用于规划、重写、评估、生成和多次向量检索必然增加延迟和API成本。如何优化可靠性与稳定性智能体的决策可能出错如陷入死循环、生成无效查询。如何设计熔断、超时和回退机制可信度与可解释性生产环境要求系统不仅是准确的还是可信的。如何追溯答案的来源多跳检索路径如何评估答案的置信度与现有基础设施集成如何将Agentic RAG工作流嵌入到现有的微服务、数据库和监控体系中本文的后续章节将围绕这些挑战提供一个模块化、可观测、可运维的工程化实现方案。2. 环境准备与项目架构我们选择Python作为实现语言因为它拥有最丰富的AI开发生态。本项目将基于LangChain和LangGraph来构建智能体工作流因为它们提供了强大的工具编排和图工作流定义能力。2.1 技术栈与版本说明Python: 3.9核心框架:langchain0.2.0用于智能体、工具链和基础组件构建。langgraph0.0.30用于构建有状态、多智能体的工作流图。langchain-openai0.1.0用于接入OpenAI的LLM也可替换为其他模型。向量数据库与检索:chromadb0.4.22轻量级向量数据库用于演示。langchain-chroma0.0.4集成包。大语言模型:OpenAI GPT-4系列或GPT-3.5-Turbo需自备API Key。生产环境可考虑本地部署模型如Qwen、Llama等以控制成本和延迟。辅助工具:pydantic2.5.0用于数据验证和状态定义。tenacity8.2.0用于重试逻辑。重要提示AI框架迭代迅速以下代码基于上述版本编写。实际部署时请根据官方文档调整可能发生变化的API。2.2 项目结构与模块职责一个清晰的工程化结构是成功的第一步。我们建议如下目录结构agentic_rag_system/ ├── core/ │ ├── __init__.py │ ├── state.py # 定义工作流全局状态Pydantic模型 │ ├── agents.py # 各个智能体的定义规划、重写、评估、生成 │ └── tools.py # 智能体可调用的工具如检索工具 ├── workflows/ │ ├── __init__.py │ └── agentic_rag_graph.py # 使用LangGraph定义的工作流图 ├── config/ │ └── settings.py # 配置管理API Keys 模型参数等 ├── services/ │ ├── __init__.py │ ├── retriever_service.py # 封装检索逻辑可接入多种数据源 │ └── evaluation_service.py # 可信度评估服务可选 ├── app.py # 主应用入口FastAPI/CLI └── requirements.txt3. 核心组件实现构建智能体与工具我们将实现一个简化但功能完整的Agentic RAG系统包含以下关键角色规划者Planner、查询重写器Query Rewriter、检索执行器Retriever、充足性评估者Sufficient Context Agent和生成器Synthesizer。3.1 定义工作流状态所有智能体共享和修改一个全局状态。这是LangGraph工作流的核心。# core/state.py from typing import List, Optional, Dict, Any from pydantic import BaseModel, Field class AgenticRAGState(BaseModel): Agentic RAG 工作流的全局状态 # 用户原始输入 original_query: str Field(description用户的原始问题) # 当前迭代的查询可能被重写 current_query: str Field(description当前用于检索的查询) # 检索到的文档片段列表 retrieved_documents: List[str] Field(default_factorylist, description检索到的文档列表) # 所有历史检索到的文档用于最终合成 all_documents: List[str] Field(default_factorylist, description所有检索到的文档) # 评估结果当前上下文是否充足 is_context_sufficient: bool Field(defaultFalse, description上下文是否足以回答问题) # 评估反馈如果不足缺失什么 feedback_for_next_query: Optional[str] Field(defaultNone, description用于指导下一轮检索的反馈) # 最终答案 final_answer: Optional[str] Field(defaultNone, description最终生成的答案) # 迭代次数控制防止死循环 iteration_count: int Field(default0, description当前迭代次数) # 最大迭代次数 max_iterations: int Field(default5, description最大允许迭代次数) # 可选的元数据如检索来源、置信度等 metadata: Dict[str, Any] Field(default_factorydict, description工作流元数据)3.2 实现检索工具这是智能体与知识库交互的“手”。我们模拟一个多源检索场景。# core/tools.py from langchain.tools import tool from langchain_chroma import Chroma from langchain_openai import OpenAIEmbeddings from typing import List import os # 假设我们已初始化了两个向量库代表不同数据源 # 生产环境中这里可能是不同的数据库连接 embeddings OpenAIEmbeddings(modeltext-embedding-3-small) # 模拟初始化两个向量库实际应从持久化存储加载 vectorstore_a Chroma(persist_directory./data/chroma_db_a, embedding_functionembeddings) vectorstore_b Chroma(persist_directory./data/chroma_db_b, embedding_functionembeddings) tool def retrieve_from_knowledge_base(query: str, source: str all) - List[str]: 从指定的知识库源检索与查询相关的文档。 Args: query: 检索查询字符串。 source: 数据源可选 source_a, source_b, 或 all。 Returns: 检索到的相关文档文本列表。 docs [] try: if source in [source_a, all]: retriever_a vectorstore_a.as_retriever(search_kwargs{k: 3}) docs_a retriever_a.invoke(query) docs.extend([doc.page_content for doc in docs_a]) if source in [source_b, all]: retriever_b vectorstore_b.as_retriever(search_kwargs{k: 3}) docs_b retriever_b.invoke(query) docs.extend([doc.page_content for doc in docs_b]) except Exception as e: # 生产环境应有更细致的错误处理和日志 print(f检索时发生错误: {e}) return [f检索失败: {str(e)}] # 去重简单的基于文本的去重 seen set() unique_docs [] for doc in docs: if doc not in seen: seen.add(doc) unique_docs.append(doc) return unique_docs[:5] # 返回最多5个片段3.3 构建核心智能体我们使用LangChain的create_react_agent模式来创建具备推理和工具调用能力的智能体。# core/agents.py from langchain import hub from langchain.agents import create_react_agent, AgentExecutor from langchain_openai import ChatOpenAI from core.tools import retrieve_from_knowledge_base from core.state import AgenticRAGState import os # 从LangChain Hub拉取一个优化的ReAct提示词模板 # 你也可以自定义这个模板以获得更好的控制 react_prompt hub.pull(hwchase17/react) # 初始化LLM llm ChatOpenAI(modelgpt-4-turbo-preview, temperature0, api_keyos.getenv(OPENAI_API_KEY)) # 1. 规划与查询重写智能体 # 这个智能体负责分析原始问题并可能将其分解或重写为更适合检索的查询。 planner_tools [] # 规划阶段可能不需要工具或需要不同的工具集 planner_agent create_react_agent(llm, planner_tools, react_prompt) planner_executor AgentExecutor(agentplanner_agent, toolsplanner_tools, handle_parsing_errorsTrue, verboseTrue) def run_planner(state: AgenticRAGState) - dict: 规划智能体分析问题决定检索策略和初始查询。 system_message 你是一个查询分析专家。你的任务是根据用户问题生成一个或多个最有效的检索查询。 如果问题复杂可能需要拆解。如果问题模糊可能需要澄清或重写。 直接输出优化后的查询语句。如果问题很简单可以直接使用原问题。 human_message f用户问题: {state.original_query}\n请生成用于知识库检索的查询语句。 # 这里简化处理实际规划智能体可以输出更复杂的结构如子问题列表 response llm.invoke([(system, system_message), (human, human_message)]) new_query response.content.strip() # 更新状态 return {current_query: new_query, iteration_count: state.iteration_count 1} # 2. 检索执行器这更像一个工具调用节点而非复杂智能体 def run_retrieval(state: AgenticRAGState) - dict: 执行检索将结果添加到状态中。 # 这里可以加入更复杂的逻辑比如根据规划智能体的输出选择数据源 retrieved_docs retrieve_from_knowledge_base.invoke({query: state.current_query, source: all}) # 更新本次检索结果和总文档库 new_all_docs state.all_documents retrieved_docs # 简单的去重 unique_all_docs list(dict.fromkeys(new_all_docs)) return { retrieved_documents: retrieved_docs, all_documents: unique_all_docs } # 3. 充足性评估智能体 (Sufficient Context Agent / 质检Agent) # 这是Agentic RAG的核心判断当前信息是否足够回答。 sufficiency_tools [] # 评估本身通常不需要调用外部工具 sufficiency_agent create_react_agent(llm, sufficiency_tools, react_prompt) sufficiency_executor AgentExecutor(agentsufficiency_agent, toolssufficiency_tools, handle_parsing_errorsTrue) def run_sufficiency_check(state: AgenticRAGState) - dict: 评估当前检索到的上下文是否足以回答问题。 if not state.retrieved_documents: # 如果什么都没检索到直接判定不足 return { is_context_sufficient: False, feedback_for_next_query: 未检索到任何相关文档。请尝试使用更通用或不同的关键词。 } context_str \n---\n.join(state.retrieved_documents[-3:]) # 评估最近几次检索的结果 system_message 你是一个严格的上下文质量评估员。你的任务是判断给定的上下文信息是否足以准确、完整地回答用户的问题。 你必须非常严格。如果上下文缺失关键事实、数据、定义或逻辑步骤就必须判定为不足。 你的输出必须是严格的JSON格式 { is_sufficient: true/false, reason: 解释为什么充足或不足。如果不足明确指出缺失的信息类型。, suggested_next_query: 如果不足建议一个用于补充检索的新查询。否则为null。 } human_message f用户问题: {state.original_query} 当前检索到的上下文: {context_str} 请根据上述上下文进行评估。 response llm.invoke([(system, system_message), (human, human_message)]) try: # 这里需要解析LLM返回的JSON简化处理假设返回是纯文本 # 生产环境应使用LLM的JSON模式或更稳健的解析 import json # 尝试提取JSON部分 content response.content # 简单查找第一个{和最后一个} start content.find({) end content.rfind(}) 1 if start ! -1 and end ! -1: json_str content[start:end] result json.loads(json_str) is_sufficient result.get(is_sufficient, False) reason result.get(reason, ) next_query_suggestion result.get(suggested_next_query) else: # 回退逻辑如果没找到JSON根据关键词判断 if 不足 in content or 不充分 in content or 缺失 in content or false in content.lower(): is_sufficient False reason LLM判定上下文信息不足。 next_query_suggestion 请根据原始问题和我之前检索的内容生成一个更具体或不同角度的查询。 else: is_sufficient True reason LLM判定上下文信息充足。 next_query_suggestion None except Exception as e: print(f解析评估结果失败: {e}, 内容: {response.content}) # 出错时保守处理继续迭代 is_sufficient False reason f评估解析错误: {e} next_query_suggestion state.current_query (请提供更详细的信息) return { is_context_sufficient: is_sufficient, feedback_for_next_query: next_query_suggestion, metadata: {**state.metadata, sufficiency_reason: reason} } # 4. 答案生成智能体 def run_synthesis(state: AgenticRAGState) - dict: 当上下文被判定充足时综合所有信息生成最终答案。 if not state.is_context_sufficient: # 如果状态显示不足不应进入此节点。这里作为安全保护。 return {final_answer: 错误在上下文不足的情况下尝试生成答案。} context_str \n---\n.join(state.all_documents) system_message 你是一个专业的问答助手。请严格基于提供的上下文信息生成一个准确、完整、简洁的答案。 如果上下文明确包含答案请直接给出。 如果上下文信息不足以完全回答问题请在答案中说明哪些部分是基于已知信息的哪些部分是未知的。 绝对不要编造上下文之外的信息。 human_message f基于以下上下文回答用户的问题。 上下文: {context_str} 用户问题: {state.original_query} 请生成最终答案: response llm.invoke([(system, system_message), (human, human_message)]) return {final_answer: response.content}4. 组装工作流使用LangGraph构建智能体协作图现在我们将上述分散的智能体函数组装成一个有向图定义它们之间的执行流和条件逻辑。# workflows/agentic_rag_graph.py from langgraph.graph import StateGraph, END from core.state import AgenticRAGState from core.agents import run_planner, run_retrieval, run_sufficiency_check, run_synthesis def create_agentic_rag_workflow() - StateGraph: 创建并返回Agentic RAG的工作流图 workflow StateGraph(AgenticRAGState) # 1. 添加节点 workflow.add_node(planner, run_planner) # 规划/重写查询 workflow.add_node(retriever, run_retrieval) # 执行检索 workflow.add_node(sufficiency_checker, run_sufficiency_check) # 评估充足性 workflow.add_node(synthesizer, run_synthesis) # 生成最终答案 # 2. 设置入口点 workflow.set_entry_point(planner) # 3. 定义边执行流 # 规划后总是进行检索 workflow.add_edge(planner, retriever) # 检索后总是进行充足性检查 workflow.add_edge(retriever, sufficiency_checker) # 4. 定义条件边基于检查结果路由 def should_continue(state: AgenticRAGState) - str: 决定是继续检索还是生成答案。 # 条件1: 上下文已充足 - 去生成答案 if state.is_context_sufficient: return proceed_to_answer # 条件2: 达到最大迭代次数 - 强制结束即使不足 if state.iteration_count state.max_iterations: print(f达到最大迭代次数 {state.max_iterations}将使用现有上下文生成答案。) # 这里可以更新状态标记为“尽力而为”模式 return proceed_to_answer_forced # 条件3: 上下文不足且可继续 - 返回规划器进行下一轮 return continue_retrieval # 5. 添加条件路由 workflow.add_conditional_edges( sufficiency_checker, should_continue, { proceed_to_answer: synthesizer, # 充足去生成 proceed_to_answer_forced: synthesizer, # 强制生成 continue_retrieval: planner, # 不足继续规划下一轮检索 } ) # 6. 从生成器到结束 workflow.add_edge(synthesizer, END) return workflow # 编译工作流 agentic_rag_graph create_agentic_rag_workflow().compile() # 可视化工作流可选需要graphviz # try: # from IPython.display import Image, display # display(Image(agentic_rag_graph.get_graph().draw_mermaid_png())) # except: # pass5. 运行与验证从模拟Google Search到生产级思考现在我们可以运行这个工作流来体验Agentic RAG的动态检索过程。5.1 启动工作流# app.py (简化示例) from workflows.agentic_rag_graph import agentic_rag_graph from core.state import AgenticRAGState import asyncio async def main(): # 模拟一个多跳查询 # 假设知识库中源A有“产品A的负责人是张三”源B有“张三的联系电话是13800138000” user_query 告诉我产品A的负责人的联系电话是多少 # 初始化工作流状态 initial_state AgenticRAGState( original_queryuser_query, current_queryuser_query, # 初始查询 max_iterations3 # 设置较小的迭代次数用于演示 ) print(f开始处理查询: {user_query}) print(- * 50) # 运行工作流 final_state await agentic_rag_graph.ainvoke(initial_state) print(\n *50) print(工作流执行完成) print(f最终答案: {final_state[final_answer]}) print(f总迭代次数: {final_state[iteration_count]}) print(f使用的总文档数: {len(final_state[all_documents])}) print(*50) # 打印详细的执行路径和中间结果生产环境应记录到日志 print(\n--- 执行追踪 ---) print(f初始查询: {initial_state.current_query}) for i, doc in enumerate(final_state[all_documents]): print(f检索到的文档[{i1}]: {doc[:100]}...) # 打印前100字符 if __name__ __main__: asyncio.run(main())5.2 预期执行流程与输出分析对于一个设计良好的多跳查询工作流可能会按以下步骤执行第一轮:Planner接收原始问题“产品A的负责人的联系电话”可能直接将其作为初始查询current_query。Retriever用该查询检索。假设在source_a中匹配到“产品A的负责人是张三”。Sufficiency Checker评估上下文提到了负责人是张三但没有电话号码。判定为不足。反馈“缺失张三的联系电话信息”建议下一轮查询“张三 联系电话”。第二轮:Planner根据反馈将current_query更新为“张三 联系电话”。Retriever用新查询检索。假设在source_b中匹配到“张三的联系电话是13800138000”。Sufficiency Checker再次评估现在上下文同时包含“产品A负责人是张三”和“张三电话是13800138000”。判定为充足。第三轮:路由到Synthesizer综合两段文档生成最终答案“产品A的负责人是张三其联系电话是13800138000。”输出示例:开始处理查询: 告诉我产品A的负责人的联系电话是多少 -------------------------------------------------- [Planner Agent Log]... 生成查询“产品A的负责人联系电话” [Retriever Log]... 从 source_a 检索到1个文档。 [Sufficiency Checker Log]... 评估不足。缺失具体电话号码。 [Planner Agent Log]... 根据反馈生成新查询“张三 电话” [Retriever Log]... 从 source_b 检索到1个文档。 [Sufficiency Checker Log]... 评估充足。 [Synthesizer Log]... 生成最终答案。 工作流执行完成 最终答案: 根据已有信息产品A的负责人是张三其联系电话是13800138000。 总迭代次数: 2 使用的总文档数: 2 这个流程清晰地展示了Agentic RAG如何通过智能体间的协作自动完成信息的“拼图”解决了传统RAG在信息分散时的短板。6. 生产级工程化性能、可靠性与可信度一个演示原型与生产级系统之间的差距正是工程化的价值所在。以下是关键考量点。6.1 性能优化策略异步与并行Search Fanout并行检索多源是Agentic RAG的优势。在run_retrieval节点可以对不同数据源的检索调用改为异步并行如使用asyncio.gather大幅降低I/O等待时间。缓存层引入缓存如Redis存储频繁查询的中间结果或最终答案。对于相同或相似的查询可以直接返回缓存结果避免重复的LLM调用和检索。LLM调用优化流式输出对于生成环节使用流式响应改善用户体验。上下文窗口管理在run_synthesis时如果all_documents很大需要做智能截断或摘要防止超出模型上下文限制并增加成本。模型分级对“规划”、“评估”等对创造力要求较低的任务使用更小、更快的模型如GPT-3.5-Turbo仅在最终“生成”环节使用大模型如GPT-4。向量检索优化索引优化使用HNSW等高性能索引算法。混合搜索结合向量相似性搜索和关键词BM25搜索提高召回率。元数据过滤在检索时加入来源、时间等元数据过滤提升精度。6.2 可靠性保障模式超时与熔断为每个LLM调用和工具调用设置超时。使用类似tenacity的库实现重试机制针对瞬时故障。为整个工作流设置总超时。防死循环我们已经通过max_iterations进行了限制。还可以增加更复杂的策略如检测重复查询、评估进展停滞等。优雅降级当Agentic工作流失败或超时时应能回退到传统的、一次性的RAG管道作为保底方案。结构化输出与验证强制关键智能体如评估器输出结构化数据JSON并使用Pydantic进行验证避免解析失败导致流程崩溃。状态持久化对于长时间运行的工作流需要将AgenticRAGState持久化到数据库支持断点续跑。6.3 构建可信AI Agent可信度是生产应用的基石尤其在医疗、法律等领域。可追溯性在AgenticRAGState.metadata中详细记录每一步每次检索的查询、数据源、返回的文档ID、评估结果和理由。最终答案应附带完整的“溯源链”。置信度评分在run_sufficiency_check和run_synthesis节点让LLM输出一个置信度分数例如0-1或对答案中不同部分标注置信度。低置信度答案需要特殊标记或人工审核。事实一致性检查在生成最终答案后可以增加一个“事实核查”智能体将答案与检索到的源文档进行交叉验证标记可能存在的矛盾或幻觉。不确定性表达训练或提示生成器在信息不足时明确说出“根据现有资料无法确定...”而不是猜测。人工审核闭环设计接口允许用户对答案进行“点赞”、“点踩”或“标记不正确”。这些反馈可以用于持续优化检索和评估模型。6.4 监控、日志与可观测性生产系统必须有完善的可观测性。结构化日志使用structlog或jsonlogger记录每个节点的输入、输出、耗时和错误。关键指标包括各阶段延迟、迭代次数、检索文档数、LLM Token消耗、最终置信度等。分布式追踪为每个用户请求生成唯一的trace_id贯穿整个工作流方便在像Jaeger这样的系统中进行端到端问题排查。指标与告警暴露Prometheus指标如请求量、成功率、各阶段P99延迟、迭代次数分布等。设置告警规则如错误率升高、平均迭代次数异常增加。成本监控密切监控LLM API的调用费用按业务线或团队进行分摊。7. 常见问题与排查思路在开发和运维Agentic RAG系统时你可能会遇到以下典型问题。问题现象可能原因排查与解决思路工作流陷入死循环不断检索1.Sufficiency Checker智能体逻辑有误始终判定为“不足”。2. 检索工具始终返回无关内容无法满足评估条件。3.max_iterations设置过高或未生效。1. 检查评估智能体的提示词Prompt确保其判断标准清晰。可以加入少量示例Few-shot进行引导。2. 检查检索相关性。优化嵌入模型或引入重排序Re-ranker。3. 在should_continue函数中添加日志确认迭代计数逻辑。设置一个合理的默认值如3-5次。答案质量不稳定时好时坏1. LLM生成具有随机性即使temperature0。2. 检索结果排序波动大。3. 多跳查询中某一跳的检索失败导致连锁反应。1. 对于生成环节可以尝试使用“自我一致性”Self-Consistency策略生成多个答案并投票选择最佳。2. 为检索工具设置固定的随机种子如果支持或使用更确定的检索算法如精确匹配向量搜索。3. 增强工作流的鲁棒性例如当某一轮检索结果为空时让规划器尝试更宽泛或更具体的替代查询而不是直接失败。系统延迟过高用户体验差1. 串行调用LLM次数过多。2. 向量检索慢或网络延迟高。3. 未使用缓存。1. 分析关键路径将可并行的步骤并行化如多源检索。考虑将某些智能体合并如将规划和首次查询重写合并。2. 对向量数据库进行性能调优索引、分片、硬件。考虑将检索服务部署在靠近LLM服务或应用服务的区域。3. 为常见查询和中间结果引入多级缓存内存、Redis。LLM API调用成本失控1. 工作流迭代次数过多。2. 每次调用传入的上下文检索到的文档过长。3. 使用了过于昂贵的大模型处理简单任务。1. 优化评估逻辑避免不必要的迭代。设置成本预算当单次请求的预估Token消耗超过阈值时提前终止或降级。2. 在合成前对检索文档进行摘要或精炼只保留最相关的部分。3. 实施模型路由策略根据问题复杂度选择不同规格的模型。答案缺乏可解释性用户不信任1. 最终答案未提供引用来源。2. 内部决策过程如为何判定充足对用户不可见。1. 修改run_synthesis要求LLM在答案中以内联形式引用文档编号如[1]并在最终输出后附上“参考来源”列表。2. 将关键中间状态如每次检索的查询、评估结论作为元数据随答案一同返回给前端以“思考链”或“诊断报告”的形式展示给高级用户。8. 总结与进阶方向通过本文的拆解与实践我们完成了一个工程化Agentic RAG系统的核心构建。我们从理解其与传统RAG的本质区别出发设计了由多智能体协作的动态工作流并使用LangGraph将其实现为一个可控、可观测的图。更重要的是我们深入探讨了将其推向生产环境必须考虑的性能、可靠性、可信度和可运维性等工程问题。下一步你可以从以下几个方向深化工具扩展让智能体不仅能检索向量库还能调用API查询数据库、搜索互联网注意合规、执行计算等打造功能更强大的AI Agent。学习与优化引入强化学习RL或基于人类反馈的强化学习RLHF让评估智能体和规划智能体根据最终答案的质量自动优化其策略。复杂工作流处理需要多个工具顺序调用、且有严格依赖关系的复杂任务例如“查询天气 - 如果下雨则推荐室内活动 - 并预订附近餐厅”。与MCP集成探索与Model Context ProtocolMCP服务器集成以标准化的方式连接更丰富的外部工具和数据源。本地化部署为了满足数据隐私和成本要求将整个系统包括LLM、嵌入模型、向量数据库部署在私有基础设施中。Agentic RAG代表了RAG技术从“检索增强”向“认知增强”演进的重要一步。它不再满足于做一个被动的信息提取器而是试图成为一个主动的问题解决者。虽然其架构更为复杂但通过精心的工程化设计我们完全能够驾驭这种复杂性构建出真正智能、可靠、值得信赖的新一代企业级问答系统。 30款热门AI模型一站整合DeepSeek/GLM/Claude 随心用限时 5 折。 点击领海量免费额度