从RAG到Agentic RAG：构建多智能体协作的生产级可信AI问答系统

30款热门AI模型一站整合DeepSeek/GLM/Claude 随心用限时 5 折。 点击领海量免费额度大家好我是专注于AI应用落地的技术博主。在构建企业级知识问答系统时你是否遇到过这样的困境传统RAG检索增强生成模型在面对复杂、多跳的查询时要么给出的答案信息不全要么干脆回答“未找到相关信息”尤其是在医疗、法律、金融等对准确性要求极高的领域这种“信息缺口”可能导致严重的后果。近期Google推出的Agentic RAG框架为解决这一痛点提供了全新的思路。它不再是一个简单的“检索-生成”管道而是引入了一个多智能体协作系统其中核心的“质检Agent”能够主动判断信息是否足够并引导补充检索在测试中准确率提升了34%。这标志着RAG技术正从“工具”向“智能体”演进。本文将深入探讨如何将一个前沿的Agentic RAG概念工程化为一个从Google Search接入实时信息到最终部署为生产级可信AI Agent的完整系统。无论你是想了解下一代RAG的架构师还是正在寻找可靠AI应用落地方案的工程师都能从本文中获得从理论到实战的完整路径。我们将涵盖核心概念、系统架构设计、关键模块实现、工程化考量以及避坑指南。1. 背景与核心概念从RAG到Agentic RAG在深入工程实践之前我们必须厘清几个核心概念理解技术演进的脉络。1.1 传统RAG的局限性检索增强生成RAG已成为连接大语言模型与私有知识库的标准范式。其基本流程是用户提问 - 将问题转换为向量进行语义检索 - 从向量数据库中获取相关文档片段 - 将片段与问题一同提交给LLM生成答案。然而这个流程存在几个固有缺陷单次检索的局限性一次检索可能无法覆盖所有相关信息特别是当答案需要综合多个分散文档的内容时即“多跳问答”。“检索即答”的僵化无论检索到的上下文是否足够回答用户问题系统都会尝试生成答案容易导致“幻觉”或信息不全。缺乏自我评估与迭代系统没有能力评估当前检索结果的“充分性”无法在发现信息不足时主动发起新一轮、更精准的检索。1.2 Agentic RAG引入智能体思维Agentic RAG的核心思想是将RAG流程中的关键决策点交给专门的AI智能体Agent来处理使整个系统具备感知、规划、执行和反思的能力。根据Google公开的资料一个典型的Agentic RAG框架可能包含以下角色智能体Orchestrator协调器接收用户原始查询负责任务的初步理解和拆分。例如将“比较A产品和B产品在能耗和价格上的优劣”拆解为“A产品能耗”、“B产品能耗”、“A产品价格”、“B产品价格”等多个子查询。Planner规划器为拆分后的子任务规划检索路径和策略。决定是并行检索所有子查询还是按特定顺序执行。Query Rewriter查询重写器优化每个子查询的表述使其更适合底层检索系统如向量数据库或搜索引擎。例如将口语化问题转化为包含关键术语的查询语句。Search Fanout搜索分发器执行并行或串行的多源检索。可以同时查询多个向量数据库、知识图谱或外部API如Google Search。Sufficient Context Agent上下文充分性质检Agent这是最核心的突破。它不直接生成答案而是评估当前检索到的所有上下文片段判断它们是否足以准确、完整地回答原始问题。如果判断为“不充分”它会分析缺失的信息类型并生成新的、更精确的检索指令反馈给Planner或Search Fanout触发新一轮的补充检索。这个“质检-反馈-再检索”的循环直到信息被判定为充分为止。Synthesis综合器当上下文被判定为充分后该智能体负责整合所有检索到的信息生成最终连贯、准确的答案。1.3 工程化与生产级可信AI Agent理解了Agentic RAG的概念后“工程化”意味着我们需要设计一个稳定、可维护、可监控、高性能的系统来实现这一架构。“生产级可信”则进一步要求这个系统高可用与可扩展能处理高并发请求组件可水平扩展。可观测性具备完善的日志、指标和追踪能清晰看到每个智能体的决策过程、检索轮次和耗时。安全性对输入输出进行过滤防止提示词注入保护私有数据。可控性与可解释性运营人员能够理解系统为何给出某个答案必要时进行干预或调整。成本可控智能体的多次调用和检索会增加LLM Token消耗和延迟需要有预算管理和优化策略。2. 系统架构设计与环境准备我们将设计一个简化但功能完整的Agentic RAG系统原型它能够处理复杂查询利用Google Search补充实时信息并具备基本的自检循环能力。2.1 整体架构图我们的系统将包含以下核心模块[用户请求] | v [API网关/入口] | v [Orchestrator Agent] (任务分析与拆解) | v [Planner Query Rewriter Agent] (规划与查询优化) | v |----------------- [向量数据库检索] (私有知识) | [Search Fanout] --|----------------- [Google Search API] (公共/实时知识) | |----------------- [其他数据源...] | v [Sufficient Context Agent] (上下文充分性质检) | | | (不充分) | (充分) v v [反馈新查询] [Synthesis Agent] (答案生成) | | --------------------- | v [最终答案] - [用户]2.2 环境与工具栈说明我们将使用Python作为主要开发语言因为它拥有最丰富的AI和LLM生态。以下是我们需要准备的核心组件LLM服务我们将使用OpenAI的GPT-4或GPT-3.5-Turbo作为核心的“大脑”。你也可以替换为 Claude、国产大模型API或本地部署的Llama 3等。向量数据库用于存储和检索私有知识。这里选用ChromaDB因为它轻量、易用适合原型和中小规模生产。生产环境也可考虑Qdrant、Weaviate或Pinecone。搜索引擎API用于获取实时公共信息。我们将使用Google Custom Search JSON API。你需要一个Google Cloud项目并启用该API。Agent开发框架为了高效构建和管理多个智能体我们使用LangChain和LangGraph。LangChain提供了丰富的Agent、Tool和Chain抽象LangGraph则擅长构建有状态、可循环的图工作流完美契合Agentic RAG的需求。开发环境Python 3.10pip 包管理器2.3 项目初始化与依赖安装首先创建一个新的项目目录并初始化虚拟环境。# 创建项目目录 mkdir agentic-rag-system cd agentic-rag-system # 创建虚拟环境 (可选但强烈推荐) python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Mac/Linux: source venv/bin/activate # 创建 requirements.txt 文件并安装核心依赖requirements.txt文件内容langchain0.1.0 langchain-openai0.0.5 langchain-community0.0.10 langgraph0.0.33 chromadb0.4.22 langchain-chroma0.1.0 openai1.12.0 google-api-python-client2.108.0 python-dotenv1.0.0 pydantic2.5.0 fastapi0.104.1 uvicorn[standard]0.24.0安装依赖pip install -r requirements.txt接下来创建环境变量配置文件.env用于安全地存储API密钥等敏感信息。# .env 文件 OPENAI_API_KEYyour_openai_api_key_here GOOGLE_API_KEYyour_google_custom_search_api_key_here GOOGLE_CSE_IDyour_google_custom_search_engine_id_here请替换your_*部分为你自己的密钥。获取GOOGLE_API_KEY和GOOGLE_CSE_ID需要到 Google Cloud Console 创建项目并启用Custom Search API。3. 核心模块实现构建智能体工作流我们将使用LangGraph来定义和编排整个Agentic RAG的工作流。LangGraph的“图”概念允许我们清晰地定义智能体之间的状态流转和循环。3.1 定义系统状态State首先我们需要定义一个共享的“状态”对象它会在各个智能体节点之间传递包含工作流所需的所有信息。# state.py from typing import TypedDict, List, Optional, Annotated from langchain_core.messages import BaseMessage import operator class AgenticRAGState(TypedDict): Agentic RAG 工作流的共享状态 # 输入与原始信息 original_query: str messages: Annotated[List[BaseMessage], operator.add] # 对话消息历史 # 任务分解 sub_queries: List[str] # 分解后的子查询列表 # 检索结果 retrieved_contexts: List[str] # 从所有来源检索到的上下文片段 # 质检与循环控制 is_context_sufficient: bool # 上下文是否已充分的标志 missing_info_hint: Optional[str] # 如果信息不足缺失信息的提示 # 最终输出 final_answer: Optional[str]3.2 实现Orchestrator Agent协调器这个智能体负责理解用户意图并进行任务分解。我们使用LangChain的create_react_agent模式并为其配备一个“任务分解工具”。# orchestrator_agent.py from langchain_openai import ChatOpenAI from langchain.agents import create_react_agent, AgentExecutor from langchain.tools import Tool from langchain.prompts import PromptTemplate from langchain_core.messages import SystemMessage, HumanMessage import os from dotenv import load_dotenv load_dotenv() llm ChatOpenAI(modelgpt-4-turbo-preview, temperature0, api_keyos.getenv(OPENAI_API_KEY)) def decompose_query(query: str) - str: 工具函数将复杂查询分解为多个子查询。 返回一个用分号分隔的子查询字符串。 decomposition_prompt PromptTemplate.from_template( 你是一个任务分解专家。请将以下复杂问题分解为几个独立的、可单独检索的子问题。 每个子问题应该聚焦于一个具体的方面或实体。 输出格式仅返回用中文分号‘’分隔的子问题不要编号不要额外解释。 原始问题{query} 分解后的子问题 ) chain decomposition_prompt | llm result chain.invoke({query: query}) return result.content.strip() # 将函数包装成Tool decompose_tool Tool( nameQueryDecomposer, funcdecompose_query, description将复杂的用户查询分解为多个更简单、更聚焦的子查询。输入是原始查询字符串输出是用分号分隔的子查询字符串。 ) # 创建Orchestrator Agent orchestrator_prompt PromptTemplate.from_template( 你是一个协调器Orchestrator。你的任务是根据用户问题判断是否需要以及如何分解任务。 如果问题是简单的、单一的例如公司的成立时间是什么则无需分解直接原样输出。 如果问题是复杂的、涉及多个方面或需要多步推理的例如比较产品A和产品B的优缺点并给出购买建议则使用QueryDecomposer工具将其分解。 你的输出应该是最终确定的需要检索的子查询列表。如果未分解列表中就只包含原始问题。 用户问题{input} 请开始你的工作 ) orchestrator_agent create_react_agent(llm, tools[decompose_tool], promptorchestrator_prompt) orchestrator_agent_executor AgentExecutor(agentorchestrator_agent, tools[decompose_tool], verboseTrue) def run_orchestrator(state: AgenticRAGState): Orchestrator节点函数 query state[original_query] # 调用Agent执行分解任务 result orchestrator_agent_executor.invoke({input: query}) agent_response result[output] # 解析输出得到子查询列表 # 简单逻辑如果输出包含分号按分号分割否则将整个输出作为一个子查询 if in agent_response: sub_qs [sq.strip() for sq in agent_response.split() if sq.strip()] else: sub_qs [agent_response.strip()] # 更新状态 new_state state.copy() new_state[sub_queries] sub_qs # 初始化其他字段 new_state[retrieved_contexts] [] new_state[is_context_sufficient] False new_state[missing_info_hint] None return new_state3.3 实现Search Fanout与工具检索分发器这个节点不直接是一个对话智能体而是一个执行组件。它并行或串行地执行所有子查询的检索。我们需要实现两个关键的“工具”向量数据库检索和Google搜索。# search_tools.py from langchain_chroma import Chroma from langchain_openai import OpenAIEmbeddings from langchain_community.utilities import GoogleSearchAPIWrapper from langchain.tools import Tool import os # 1. 初始化向量数据库检索工具假设已有存储好的ChromaDB embeddings OpenAIEmbeddings(api_keyos.getenv(OPENAI_API_KEY)) # 请确保 persist_directory 指向你已有的ChromaDB持久化目录 vectorstore Chroma( persist_directory./my_chroma_db, embedding_functionembeddings, collection_namemy_knowledge_base ) retriever vectorstore.as_retriever(search_kwargs{k: 3}) # 每个查询检索3个片段 def retrieve_from_vector_db(query: str) - str: 从私有向量数据库检索相关文档 docs retriever.invoke(query) context \n\n.join([doc.page_content for doc in docs]) return f[来自私有知识库的检索结果]\n{context} if context else [私有知识库未找到相关信息] # 2. 初始化Google搜索工具 search GoogleSearchAPIWrapper( google_api_keyos.getenv(GOOGLE_API_KEY), google_cse_idos.getenv(GOOGLE_CSE_ID) ) def search_google(query: str) - str: 使用Google搜索获取实时信息 try: results search.results(query, num_results3) # 每个查询取3条结果 context \n\n.join([f来源{r[link]}\n摘要{r[snippet]} for r in results]) return f[来自Google搜索的实时信息]\n{context} if context else [Google搜索未找到相关信息] except Exception as e: return f[Google搜索出错{str(e)}] # 将函数包装成Tool供Agent调用 vector_db_tool Tool( namePrivateKnowledgeSearch, funcretrieve_from_vector_db, description从公司内部私有知识库中检索相关信息。输入是一个具体的查询语句。 ) google_search_tool Tool( nameGoogleSearch, funcsearch_google, description使用Google搜索获取最新的公共信息、新闻或通用知识。输入是一个搜索查询语句。 ) # search_fanout_node.py def run_search_fanout(state: AgenticRAGState): Search Fanout节点函数对每个子查询并行调用所有检索工具 sub_queries state[sub_queries] all_contexts state.get(retrieved_contexts, []) for sub_q in sub_queries: # 并行检索在实际生产中这里可以使用 asyncio 或线程池 private_context retrieve_from_vector_db(sub_q) web_context search_google(sub_q) # 将检索到的上下文添加到总列表中 all_contexts.append(f## 针对子查询{sub_q}\n{private_context}\n{web_context}) new_state state.copy() new_state[retrieved_contexts] all_contexts return new_state3.4 实现Sufficient Context Agent质检Agent这是系统的“大脑”负责评估已检索的上下文是否足够。我们将其设计为一个独立的Agent它根据当前上下文和原始问题做出判断。# sufficient_context_agent.py from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_core.messages import SystemMessage, HumanMessage, AIMessage import json sufficient_context_prompt ChatPromptTemplate.from_messages([ SystemMessage(content你是一个严格的上下文质量检查员Sufficient Context Agent。你的任务是根据用户原始问题和当前检索到的所有上下文判断这些上下文是否足以生成一个准确、完整、可靠的答案。 你的输出必须是严格的JSON格式 {{ is_sufficient: true/false, reasoning: 你的推理过程解释为什么足够或为什么不足。, missing_hint: 如果不足请明确指出缺失了哪类信息或哪个方面的证据。如果足够此项为null。 }} 评估标准 1. **相关性**上下文是否直接针对用户问题的核心 2. **完整性**对于多子问题或复杂问题是否所有部分都有覆盖 3. **时效性**如果问题涉及时效信息是否足够新 4. **权威性**如果重要信息源是否可靠 只有在所有标准都高度满足时才判断为‘足够’。 ), MessagesPlaceholder(variable_namechat_history), HumanMessage(content原始用户问题{original_query}\n\n当前检索到的所有上下文\n{all_context}\n\n请进行评估。) ]) sufficient_context_chain sufficient_context_prompt | llm def run_sufficient_context_agent(state: AgenticRAGState): Sufficient Context Agent节点函数 original_query state[original_query] all_context \n---\n.join(state[retrieved_contexts]) # 准备对话历史简单起见这里只放当前交互 chat_history [] # 调用LLM进行评估 response sufficient_context_chain.invoke({ original_query: original_query, all_context: all_context, chat_history: chat_history }) # 解析LLM的JSON输出 try: evaluation json.loads(response.content) is_sufficient evaluation.get(is_sufficient, False) missing_hint evaluation.get(missing_hint) reasoning evaluation.get(reasoning, ) except json.JSONDecodeError: # 如果LLM没有返回标准JSON保守处理认为不足 is_sufficient False missing_hint 无法解析质检Agent的评估结果。 reasoning response.content print(f[质检Agent日志] 是否足够{is_sufficient}。 理由{reasoning}。 缺失提示{missing_hint}) new_state state.copy() new_state[is_context_sufficient] is_sufficient new_state[missing_info_hint] missing_hint # 将质检Agent的推理也作为消息历史的一部分有助于后续综合 new_state[messages].append(AIMessage(contentf质检评估{reasoning})) return new_state3.5 实现Synthesis Agent综合器与路由逻辑当信息足够时由Synthesis Agent生成最终答案。我们还需要一个“路由”函数根据质检结果决定工作流下一步走向。# synthesis_agent.py from langchain.prompts import ChatPromptTemplate synthesis_prompt ChatPromptTemplate.from_messages([ SystemMessage(content你是一个信息综合专家Synthesis Agent。你的任务是根据用户原始问题和所有经过验证的上下文信息生成一个最终答案。 要求 1. **准确无误**答案必须严格基于提供的上下文不得捏造信息。 2. **完整全面**覆盖用户问题的所有方面特别是那些在子查询中被分解的部分。 3. **结构清晰**答案应组织良好如果涉及比较、列表或步骤请使用合适的格式。 4. **注明来源**如果上下文提到了信息来源如链接在答案中可简要提及以增加可信度。 5. **诚实**如果某些方面在上下文中确实没有找到可以说明“根据现有信息关于XX方面暂无明确信息”。 请生成最终答案。 ), HumanMessage(content原始用户问题{original_query}\n\n所有相关上下文\n{all_context}) ]) synthesis_chain synthesis_prompt | llm def run_synthesis_agent(state: AgenticRAGState): Synthesis Agent节点函数生成最终答案 original_query state[original_query] all_context \n---\n.join(state[retrieved_contexts]) response synthesis_chain.invoke({ original_query: original_query, all_context: all_context }) new_state state.copy() new_state[final_answer] response.content return new_state # router.py def route_after_sufficiency_check(state: AgenticRAGState) - str: 路由函数根据质检结果决定下一步 if state[is_context_sufficient]: # 信息足够前往综合节点生成答案 return generate_answer else: # 信息不足需要规划新一轮检索 return plan_new_search3.6 实现Planner for New Search新检索规划器当质检Agent认为信息不足时这个智能体被触发它根据缺失提示生成新的、更精确的搜索查询。# new_search_planner.py from langchain.prompts import PromptTemplate new_search_planner_prompt PromptTemplate.from_template( 你是一个检索策略规划师Planner。上一轮检索后质检Agent认为信息不足以回答用户问题。 原始问题{original_query} 已检索的上下文摘要{retrieved_contexts_summary} 质检Agent指出的缺失信息{missing_hint} 你的任务是基于缺失的信息提示生成1-3个新的、更精确、更有可能找到缺失信息的搜索查询。 这些查询应该 1. 更具体包含更明确的关键词。 2. 可能从不同角度切入同一个缺失点。 3. 避免重复已检索过的宽泛查询。 输出格式仅返回用中文分号‘’分隔的新搜索查询语句。 新搜索查询 ) new_search_planner_chain new_search_planner_prompt | llm def run_new_search_planner(state: AgenticRAGState): 新检索规划器节点函数 original_query state[original_query] # 简单摘要已检索的上下文取前500字符 contexts_summary \n.join(state[retrieved_contexts])[:500] ... missing_hint state[missing_info_hint] response new_search_planner_chain.invoke({ original_query: original_query, retrieved_contexts_summary: contexts_summary, missing_hint: missing_hint }) # 解析新的子查询 new_sub_queries [q.strip() for q in response.content.split() if q.strip()] new_state state.copy() # 用新的子查询列表替换旧的准备进行新一轮检索 new_state[sub_queries] new_sub_queries # 清空之前的上下文或可以选择追加 # new_state[retrieved_contexts] [] # 选择清空 # 重置质检标志 new_state[is_context_sufficient] False new_state[missing_info_hint] None return new_state3.7 使用LangGraph组装完整工作流现在我们将所有节点和路由逻辑组装成一个完整的有向图工作流。# workflow.py from langgraph.graph import StateGraph, END from state import AgenticRAGState from orchestrator_agent import run_orchestrator from search_fanout_node import run_search_fanout from sufficient_context_agent import run_sufficient_context_agent from synthesis_agent import run_synthesis_agent from new_search_planner import run_new_search_planner from router import route_after_sufficiency_check # 1. 创建图 workflow StateGraph(AgenticRAGState) # 2. 添加节点 workflow.add_node(orchestrator, run_orchestrator) workflow.add_node(search_fanout, run_search_fanout) workflow.add_node(sufficiency_checker, run_sufficient_context_agent) workflow.add_node(answer_synthesizer, run_synthesis_agent) workflow.add_node(new_search_planner, run_new_search_planner) # 3. 设置入口点 workflow.set_entry_point(orchestrator) # 4. 定义边连接 # 从协调器到检索分发 workflow.add_edge(orchestrator, search_fanout) # 从检索分发到质检器 workflow.add_edge(search_fanout, sufficiency_checker) # 从质检器根据路由决定下一步 workflow.add_conditional_edges( sufficiency_checker, route_after_sufficiency_check, # 路由函数 { generate_answer: answer_synthesizer, plan_new_search: new_search_planner } ) # 从新检索规划器回到检索分发形成循环 workflow.add_edge(new_search_planner, search_fanout) # 从答案综合器到结束 workflow.add_edge(answer_synthesizer, END) # 5. 编译图 app workflow.compile() # 可视化图需要graphviz try: from IPython.display import Image, display display(Image(app.get_graph().draw_mermaid_png())) except: print(Graph compiled successfully. Set up graphviz for visualization.)4. 运行与测试完整系统现在我们可以初始化状态并运行这个Agentic RAG工作流了。# main.py from workflow import app from state import AgenticRAGState from langchain_core.messages import HumanMessage def run_agentic_rag(query: str, max_cycles3): 运行Agentic RAG系统并限制最大循环次数以防无限循环 # 初始化状态 initial_state: AgenticRAGState { original_query: query, messages: [HumanMessage(contentquery)], sub_queries: [], retrieved_contexts: [], is_context_sufficient: False, missing_info_hint: None, final_answer: None } current_state initial_state cycle_count 0 print(f开始处理查询: {query}) print(- * 50) # 手动控制执行流程以便跟踪循环 # 我们从 orchestrator 节点开始 next_node orchestrator while next_node ! END and cycle_count max_cycles: cycle_count 1 print(f\n[循环 #{cycle_count}] 执行节点: {next_node}) # 执行当前节点 output app.invoke(current_state, config{configurable: {thread_id: 1}}) # 注意app.invoke会从entry_point运行到END。为了手动控制我们需要用另一种方式。 # 这里我们简化直接运行编译后的app但观察其内部步骤需要更精细的控制。 # 为了演示我们直接运行完整的app并接受其可能的多轮循环。 # 在实际调试中可以使用LangGraph的流式输出或设置中断点。 break # 此处为演示先跳出。实际应运行完整app。 # 简化版直接运行完整图 print(\n--- 开始执行完整工作流 ---) final_state app.invoke(initial_state) print(\n--- 工作流执行完毕 ---) if final_state.get(final_answer): print(f\n✅ 最终答案\n{final_state[final_answer]}) else: print(f\n❌ 未能在{max_cycles}轮内生成最终答案。) print(f最后状态{final_state}) return final_state # 测试一个复杂查询 if __name__ __main__: # 示例查询一个需要多源、多跳信息的问题 test_query LangChain和LlamaIndex这两个框架在构建RAG系统时各有什幺优缺点它们最新的版本有什么重要更新 result run_agentic_rag(test_query)5. 工程化与生产级部署考量一个可运行的Demo只是第一步。要将其变为“生产级可信AI Agent”我们必须考虑以下方面5.1 性能与可扩展性异步处理Search Fanout节点中的多个检索工具调用应改为异步asyncio以降低延迟。缓存策略对频繁出现的相似查询或子查询结果进行缓存减少对LLM和搜索API的调用。可以使用Redis或内存缓存如cachetools。限流与降级为外部API如OpenAI、Google Search设置限流和熔断机制。当某个服务不可用时系统能降级运行例如仅使用向量数据库。向量数据库优化生产环境需考虑ChromaDB的持久化、备份、分片和索引优化或迁移至Qdrant等支持集群的数据库。5.2 可观测性与监控结构化日志记录每个Agent的输入、输出、决策原因、耗时和Token使用量。使用structlog或loggingJSON格式。指标收集使用Prometheus等工具收集关键指标请求量、各阶段延迟、检索轮次分布、答案充分率、错误率。分布式追踪为每个用户请求生成唯一的trace_id在系统内传递便于在复杂工作流中定位问题。可集成OpenTelemetry。审计轨迹保存每一轮检索的查询、返回的上下文、质检结果和最终答案用于事后分析和模型优化。5.3 安全与合规输入/输出过滤对用户输入和LLM输出进行内容安全过滤防止恶意提示词注入和生成有害内容。数据脱敏确保从私有知识库检索出的上下文不包含敏感信息PII或在输出前进行脱敏处理。API密钥管理使用专业的密钥管理服务如AWS Secrets Manager, HashiCorp Vault而非硬编码在环境文件中。访问控制对系统API接口实施认证和授权确保只有合法用户和服务可以访问。5.4 配置管理与实验参数外部化将LLM温度、检索数量、质检Agent的评估标准提示词等参数抽取为配置文件支持动态更新。A/B测试能够将不同版本的Agent例如不同提示词的质检Agent路由给部分用户对比效果。版本控制对工作流定义、提示词模板进行版本控制便于回滚和迭代。5.5 部署模式微服务化可以将Orchestrator、Sufficient Context Agent等关键组件拆分为独立的微服务提高独立部署和扩展的能力。容器化使用Docker容器打包应用确保环境一致性。编排使用Kubernetes或Docker Compose进行容器编排和管理。API网关通过API网关如Kong, APISIX对外提供统一的API接口处理认证、限流、日志等横切关注点。6. 常见问题与排查思路在开发和部署Agentic RAG系统过程中你可能会遇到以下典型问题问题现象可能原因排查步骤与解决方案工作流陷入无限循环1.Sufficient Context Agent判断逻辑有误始终认为信息不足。2.New Search Planner生成的新查询与旧查询本质相同无法获取新信息。3. 数据源本身没有所需信息。1.检查质检Agent日志查看其reasoning字段判断评估标准是否过于严格。调整提示词或引入置信度阈值。2.设置最大循环次数如代码中的max_cycles强制跳出循环并返回当前最佳答案或告知用户信息不足。3.分析新查询对比新旧查询列表确保规划器引入了新的关键词或角度。响应延迟非常高1. 串行调用过多外部APILLM、搜索、向量DB。2. 检索的上下文片段过多导致后续LLM处理Token数剧增。3. 网络或外部服务延迟。1.实现异步并发将Search Fanout节点内的工具调用改为异步。2.优化检索参数减少k返回数量值或使用更精确的检索器如MMR。3.实施缓存对LLM响应和搜索结果进行缓存。4.监控与告警建立延迟监控定位瓶颈节点。答案质量不稳定有时出现幻觉1. 检索到的上下文质量差或无关。2.Synthesis Agent的提示词约束力不够。3. 在信息不足时系统被强制生成了答案。1.优化检索检查向量数据库的嵌入模型和分块策略优化查询重写逻辑。2.强化提示词在Synthesis Agent的提示词中强调“严格基于上下文”和“注明无法回答的部分”。3.利用质检结果将质检Agent的reasoning也传递给Synthesis Agent让其知晓哪些信息是经过验证的。Google Search API返回结果为空或错误1. API密钥或搜索引擎IDCSE ID错误或过期。2. 查询触发了Google的搜索限制或风控。3. 查询语句格式对搜索引擎不友好。1.验证凭证确认GOOGLE_API_KEY和GOOGLE_CSE_ID有效且未超过配额。2.处理异常在search_google函数中添加更完善的异常处理和重试机制。3.优化查询让Query RewriterAgent学习生成更符合搜索引擎语法的查询。系统内存或CPU占用过高1. 大语言模型调用频繁且上下文越来越长。2. 向量数据库检索未设置限制返回过多数据。3. 工作流状态对象在循环中不断膨胀。1.限制上下文长度在将上下文传递给LLM前进行截断或摘要。2.优化检索严格限制k值并使用similarity_threshold过滤低分结果。3.清理状态在循环中选择性清空或压缩state中不再需要的历史消息。7. 最佳实践与进阶方向7.1 提示词工程优化为每个Agent定制角色本文示例中的提示词是基础版本。生产系统中应为每个Agent精心设计系统提示词System Prompt明确其角色、职责、输出格式和边界。少样本学习Few-Shot在提示词中提供少量高质量的例子能显著提升Agent执行任务的准确性和一致性。输出结构化强制Agent以JSON等结构化格式输出便于程序解析提高系统稳定性。7.2 引入更复杂的Agent工具计算器/代码解释器对于需要数值计算或逻辑推理的问题可以赋予Agent调用Python解释器或计算器的能力。专业API集成企业内部系统API如CRM、ERP数据库让Agent能获取实时业务数据。验证工具增加一个“事实核查”Agent在答案生成后再次核对答案中的关键事实与原始上下文是否一致。7.3 评估与持续改进建立评估体系定义答案准确性、完整性、有用性等指标定期用测试集评估系统性能。收集反馈数据在用户界面提供“答案是否有用”的反馈按钮收集人工反馈用于微调Agent的决策。利用LLM进行评估可以使用一个更强的LLM如GPT-4作为“裁判”对其他LLM生成的答案和检索上下文进行自动评估低成本地扩大评估数据集。7.4 与现有MCPModel Context Protocol等生态集成MCPModel Context Protocol这是一个新兴标准用于标准化AI应用与数据源、工具之间的连接。未来可以考虑将你的向量数据库、搜索引擎等工具封装成MCP Server让你的Agentic RAG系统通过标准的MCP协议与它们交互提高系统的模块化和可插拔性。LangSmith/LangFuse利用这些LLM应用开发平台进行跟踪、监控、调试和评估可以极大提升开发运维效率。构建一个工程化的Agentic RAG系统是一个持续迭代的过程。从本文介绍的多智能体协作框架出发你可以根据实际业务需求不断优化各个Agent的能力引入更强大的工具并完善周边的监控、安全、部署设施。这条路虽然复杂但它是通向构建真正智能、可靠、能与业务深度结合的新一代AI应用的必经之路。 30款热门AI模型一站整合DeepSeek/GLM/Claude 随心用限时 5 折。 点击领海量免费额度