30款热门AI模型一站整合DeepSeek/GLM/Claude 随心用限时 5 折。 点击领海量免费额度在实际企业级AI应用开发中单纯依赖传统检索增强生成RAG系统已显不足。当用户提出一个复杂、需要跨多个数据源或进行多轮推理多跳查询的问题时传统RAG的单次检索模式极易因信息碎片化或检索词不精准而失败要么给出不完整答案要么直接返回“未找到”。这种局限性在高风险的医疗、法律、金融等场景下尤为致命。因此业界开始探索一种更智能、更具自主性的架构——Agentic RAG。它通过引入多个具备特定功能的AI智能体Agent协同工作动态规划、执行和验证检索过程从而显著提升回答的准确性和完整性。本文旨在为希望将Agentic RAG从概念验证推进到生产环境的开发者提供一套工程化实践指南。我们将从理解其核心机制开始逐步构建一个具备“质检”能力的Agentic RAG原型并深入探讨其部署、监控与优化最终形成一套可落地、可信赖的生产级AI Agent系统。1. 理解Agentic RAG超越传统检索的智能工作流传统RAG的工作流程相对线性用户提问 - 向量化查询 - 检索相关文档片段 - 将片段与问题拼接后送入大语言模型LLM生成答案。这个流程的瓶颈在于它假设单次检索就能获得所有必要信息且LLM能完美地从提供的上下文中合成答案。Agentic RAG的核心思想是将检索过程“Agent化”。它不再是一个静态的管道而是一个由多个智能体组成的动态协作系统。每个智能体负责一个子任务例如理解用户意图、规划检索步骤、重写查询、执行并行检索、判断信息是否充足、综合答案等。这种架构带来了几个关键优势动态规划与迭代检索系统能够判断当前信息是否足够回答问题。如果不够它会自动生成新的、更精确的查询进行补充检索形成一个“规划-执行-评估”的循环直到信息充足或达到迭代上限。多源与并行处理可以同时向多个知识库如内部文档库、数据库、搜索引擎API发起查询并高效整合结果。意图理解与查询优化专门的Agent负责解析用户模糊或复杂的请求将其拆解或重写为更适合底层检索系统如向量数据库、全文搜索引擎的查询语句。可信度与可解释性由于过程被分解系统可以记录每个Agent的决策、检索到的文档以及判断依据为最终答案提供审计线索这在合规要求高的场景下至关重要。搜索材料中提到的Google Agentic RAG框架其核心组件“Sufficient Context Agent”质检Agent正是这一思想的体现。它扮演了“质量检查员”的角色评估已检索到的上下文是否足以支撑LLM生成一个可靠、完整的答案。如果发现信息缺口它会明确指出缺失了什么并引导系统进行针对性补搜。对于开发者而言工程化Agentic RAG意味着我们需要设计一个稳定、可观测、可维护的智能体协作系统而不仅仅是调用几个LLM API。2. 工程化架构设计与核心组件构建一个生产级的Agentic RAG系统需要从单体应用思维转向微服务或工作流编排思维。下面是一个可参考的工程化架构设计。2.1 系统总体架构一个典型的工程化Agentic RAG系统包含以下层次和组件用户请求 | v [API网关 / 负载均衡] | v [Orchestrator Agent (编排器)] — 核心调度中枢 | |——————————————————————————————————————————————— | | | | v v v v [Query理解] [Planner] [检索执行层] [Synthesis] [与重写Agent] (规划Agent) (Search Fanout) (综合Agent) | | | | | | v | | | [向量DB] [全文ES] [Google Search API]... | | | | | | v | | | [检索结果集] | | | | | | | v | | |——[Sufficient Context Agent (质检Agent)]——| | | | | (信息充足?) (信息不足?) | | | | v v | [生成最终答案] [反馈缺口触发新一轮规划/检索] | | v v [日志、监控、评估] [返回给用户]2.2 核心组件详解Orchestrator Agent (编排器)职责接收用户原始查询初始化工作流调用并管理其他Agent的执行顺序和交互。它是整个系统的“大脑”。工程实现可以使用专门的工作流引擎如Apache Airflow, Prefect, Temporal或利用LLM本身的能力通过提示工程进行动态编排。对于复杂逻辑后者可能不稳定建议采用可编程的工作流引擎。关键输出工作流执行ID、当前状态、下一步要调用的Agent。Query Understanding Rewriter Agent (查询理解与重写Agent)职责分析用户原始查询的深层意图可能进行查询扩展、同义词替换、纠错或将其拆解成多个子查询。例如将“苹果公司最新财报和iPhone销量如何”拆解为“苹果公司2023Q4财报摘要”和“iPhone 15 2023年销量数据”两个查询。工程实现一个封装了LLM调用的微服务输入是原始查询输出是一个或多个优化后的查询语句列表。需要配置合适的系统提示词System Prompt来约束其行为。Planner Agent (规划Agent)职责根据重写后的查询制定检索策略。决定需要查询哪些数据源如客户数据库、产品手册向量库、外部搜索引擎以及查询的先后或并行顺序。工程实现可以基于规则如查询中包含“股价”则调用金融数据API或再次利用LLM根据元数据数据源描述进行规划。其输出是一个可执行的检索计划JSON结构。Search Fanout / Retriever (检索执行层)职责并行或串行地执行Planner制定的检索计划。与各种数据源连接器交互。工程实现这是一个相对“笨”但高并发的组件。它接收查询列表和数据源标识调用对应的客户端如chromadb客户端、elasticsearch客户端、自定义API客户端进行检索并收集所有返回的文档片段Chunks。关键技术点连接池管理、超时控制、失败重试、结果去重。Sufficient Context Agent (质检Agent)职责这是Agentic RAG区别于传统RAG的核心。它评估当前检索到的所有上下文文档判断它们是否足够回答用户问题。如果不够它需要生成明确的“信息缺口描述”。工程实现一个LLM调用其提示词模板至关重要。模板应要求LLM以结构化格式如JSON输出判断结果和缺口描述。示例提示词骨架你是一个信息质量评估员。请根据以下用户问题和检索到的上下文判断信息是否足够生成一个准确、完整的答案。 用户问题{question} 检索到的上下文{context} 请按以下JSON格式输出 { is_sufficient: boolean, reason: string, // 解释为何足够或不足 missing_information: string | null // 如果不足描述具体缺失什么信息 }Synthesis Agent (综合Agent)职责当质检Agent判定信息充足时由该Agent负责阅读所有上下文生成最终面向用户的自然语言答案。它需要引用来源并确保答案不包含未在上下文中出现的信息避免幻觉。工程实现标准的RAG生成步骤但上下文是经过质检的。同样需要精心设计提示词强调基于上下文、引用来源。2.3 数据流与状态管理整个系统是一个有状态的工作流。需要一个中央状态存储如Redis、数据库来跟踪每个会话Session或请求Request的原始查询和会话ID当前工作流步骤历次检索的查询、数据源和结果质检Agent的历史判断记录最终答案和生成来源这对于调试、监控和实现“迭代检索”循环至关重要。当质检Agent判断信息不足时Orchestrator会根据missing_information更新状态并可能触发新一轮的规划Planner和检索。3. 从零搭建一个简易Agentic RAG原型我们将使用Python和LangChain框架来快速构建一个包含核心“质检Agent”环节的简易原型。这个原型将模拟多轮检索决策过程。3.1 环境准备与依赖安装首先确保你的Python环境建议3.9并安装必要库。我们使用OpenAI的GPT模型作为LLMChroma作为向量数据库。# 创建并激活虚拟环境可选 python -m venv agentic_rag_env source agentic_rag_env/bin/activate # Linux/Mac # agentic_rag_env\Scripts\activate # Windows # 安装核心依赖 pip install langchain langchain-openai langchain-chroma pip install chromadb tiktoken pydantic # 用于示例的文档加载和分词 pip install pypdf langchain-community设置你的OpenAI API密钥或其他兼容API的密钥和基地址export OPENAI_API_KEYyour-api-key-here # 或者在代码中通过os.environ设置3.2 构建知识库与基础检索器我们首先创建一个简单的向量知识库模拟内部文档源。# build_knowledge_base.py import os from langchain_chroma import Chroma from langchain_openai import OpenAIEmbeddings from langchain_community.document_loaders import TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter # 初始化嵌入模型 embeddings OpenAIEmbeddings(modeltext-embedding-3-small) # 模拟一些文档内容 documents [ 项目Alpha于2023年启动主要目标是开发下一代智能客服系统。目前团队有15名成员负责人是张三。, 项目Beta专注于区块链技术在供应链溯源中的应用。该项目在2024年获得了创新奖技术负责人是李四。, 公司的年度技术峰会将于2024年10月在上海举行预计有超过500名参与者。 keynote演讲将涉及AI Agent和元宇宙。, 张三资深后端工程师拥有8年Java和微服务架构经验。他于2020年加入公司。, 李四前端架构师擅长React和Vue框架是公司UI组件库的主要维护者。 ] # 将文档分割成片段 text_splitter RecursiveCharacterTextSplitter(chunk_size200, chunk_overlap50) docs text_splitter.create_documents(documents) # 创建并持久化向量库 vectorstore Chroma.from_documents( documentsdocs, embeddingembeddings, persist_directory./chroma_db # 数据将保存到本地目录 ) vectorstore.persist() print(知识库构建完成保存在 ./chroma_db)3.3 实现核心Agent编排器、检索器与质检员现在我们实现系统的几个核心Agent。为了简化我们将Orchestrator的逻辑写在一个主函数里。# agentic_rag_core.py import os from typing import List, Dict, Any, Optional from langchain_openai import ChatOpenAI from langchain.chains import LLMChain from langchain.prompts import ChatPromptTemplate, HumanMessagePromptTemplate from langchain_chroma import Chroma from langchain_openai import OpenAIEmbeddings from pydantic import BaseModel, Field # 1. 初始化LLM和向量检索器 llm ChatOpenAI(modelgpt-4o-mini, temperature0) # 使用小模型以控制成本 embeddings OpenAIEmbeddings(modeltext-embedding-3-small) vectorstore Chroma( persist_directory./chroma_db, embedding_functionembeddings ) retriever vectorstore.as_retriever(search_kwargs{k: 3}) # 每次检索top-3片段 # 2. 定义质检Agent的输出结构 class SufficiencyCheck(BaseModel): is_sufficient: bool Field(description信息是否足够回答问题) reason: str Field(description判断理由) missing_information: Optional[str] Field(description若不足缺失的信息是什么) # 绑定结构化输出到LLM structured_llm llm.with_structured_output(SufficiencyCheck) # 3. 质检Agent的函数实现 def sufficient_context_agent(question: str, retrieved_docs: List[str]) - SufficiencyCheck: 质检Agent评估检索到的文档是否足够回答问题。 context_text \n\n---\n\n.join(retrieved_docs) prompt ChatPromptTemplate.from_messages([ (system, 你是一个严格的信息质量评估员。请仅根据提供的上下文来评估是否能回答问题。如果上下文缺失关键信息必须明确指出。), (human, 用户问题{question}\n\n检索到的上下文{context}) ]) chain prompt | structured_llm result chain.invoke({question: question, context: context_text}) return result # 4. 查询重写Agent简易版 def query_rewriter_agent(question: str, missing_info: Optional[str] None) - str: 查询重写Agent根据原始问题或缺失信息生成更优的检索查询。 if missing_info: # 如果是补充检索则聚焦于缺失的信息 input_text f原始问题{question}\n\n已知信息缺口{missing_info}\n请生成一个用于检索缺失信息的查询语句。 else: # 初次检索尝试优化原始问题 input_text f请将以下用户问题优化为更适合进行文档检索的查询语句{question} prompt ChatPromptTemplate.from_messages([ (system, 你是一个查询优化专家。请输出一个简洁、关键的搜索查询语句不要添加任何解释。), (human, {input}) ]) chain prompt | llm new_query chain.invoke({input: input_text}).content.strip() return new_query # 5. 综合答案生成Agent def synthesis_agent(question: str, retrieved_docs: List[str]) - str: 综合Agent基于充足的上下文生成最终答案。 context_text \n\n---\n\n.join(retrieved_docs) prompt ChatPromptTemplate.from_messages([ (system, 请严格根据以下上下文回答问题。如果上下文不包含答案请说‘根据已知信息无法回答’。请保持答案简洁、准确。), (human, 上下文{context}\n\n问题{question}) ]) chain prompt | llm answer chain.invoke({context: context_text, question: question}).content return answer # 6. 主编排逻辑 def orchestrator(original_question: str, max_iterations: int 3) - Dict[str, Any]: 简易编排器协调检索、质检、重写循环。 all_retrieved_docs [] current_query original_question history [] for iteration in range(max_iterations): print(f\n--- 第 {iteration 1} 轮迭代 ---) print(f检索查询: {current_query}) # 执行检索 docs retriever.invoke(current_query) doc_contents [doc.page_content for doc in docs] all_retrieved_docs.extend(doc_contents) # 简单去重根据内容 unique_docs list(set(all_retrieved_docs)) print(f检索到 {len(docs)} 个片段。) # 调用质检Agent check_result sufficient_context_agent(original_question, unique_docs) history.append({ iteration: iteration, query: current_query, sufficiency_check: check_result.dict() }) print(f质检结果: 是否充足{check_result.is_sufficient}, 理由{check_result.reason[:100]}...) if check_result.is_sufficient: # 信息充足生成最终答案 final_answer synthesis_agent(original_question, unique_docs) return { status: success, answer: final_answer, iterations: iteration 1, retrieved_docs: unique_docs, history: history } else: # 信息不足准备下一轮检索 if iteration max_iterations - 1: return { status: max_iterations_reached, answer: 经过多轮检索仍无法获取足够信息来完整回答问题。, iterations: iteration 1, retrieved_docs: unique_docs, history: history } # 根据缺失信息重写查询 current_query query_rewriter_agent(original_question, check_result.missing_information) print(f信息不足缺失: {check_result.missing_information}) print(f重写后的查询用于下一轮: {current_query}) # 理论上不会走到这里因为循环内已返回 return {status: error, message: Unexpected loop exit} # 7. 运行示例 if __name__ __main__: # 示例问题一个需要多跳推理的问题 question 张三负责的项目是什么这个项目有什么成就 print(f用户问题: {question}) result orchestrator(question) print(\n 最终结果 ) print(f状态: {result[status]}) print(f答案: {result.get(answer, N/A)}) print(f总迭代轮次: {result.get(iterations, N/A)}) print(f总共检索到唯一片段数: {len(result.get(retrieved_docs, []))})3.4 运行与结果分析运行上述脚本你会看到类似以下的输出展示了Agentic RAG的迭代思考过程用户问题: 张三负责的项目是什么这个项目有什么成就 --- 第 1 轮迭代 --- 检索查询: 张三负责的项目是什么这个项目有什么成就 检索到 3 个片段。 质检结果: 是否充足False, 理由上下文提到了张三负责项目Alpha但没有提及该项目有任何成就或奖项... 信息不足缺失: 项目Alpha所取得的成就、奖项或关键成果。 重写后的查询用于下一轮: 项目Alpha的成就或奖项 --- 第 2 轮迭代 --- 检索查询: 项目Alpha的成就或奖项 检索到 3 个片段。 质检结果: 是否充足False, 理由检索到的片段中提到了项目Beta获得了创新奖但没有提到项目Alpha的成就... 信息不足缺失: 项目Alpha的具体成就例如是否获奖、有何里程碑。 重写后的查询用于下一轮: 项目Alpha 获奖 成就 里程碑 --- 第 3 轮迭代 --- 检索查询: 项目Alpha 获奖 成就 里程碑 检索到 3 个片段。 质检结果: 是否充足True, 理由上下文明确说明张三负责项目Alpha且项目Beta获得了创新奖。虽然未直接说项目Alpha获奖但通过对比可以推断项目Alpha可能没有类似奖项或者成就信息未在上下文中。结合已有信息可以回答。 最终结果 状态: success 答案: 张三负责的项目是Alpha这是一个于2023年启动的、旨在开发下一代智能客服系统的项目。根据已知信息上下文中没有提及项目Alpha所获得的具体成就或奖项。不过上下文提到了另一个项目Beta在2024年获得了创新奖。 总迭代轮次: 3 总共检索到唯一片段数: 5这个原型清晰地演示了Agentic RAG的工作流程第一轮用原始问题检索找到了“张三是项目Alpha负责人”但质检Agent发现缺少“成就”信息。第二轮根据缺失信息重写查询为“项目Alpha的成就或奖项”检索结果提到了“项目Beta获奖”但依然不是Alpha的成就。第三轮查询进一步重写质检Agent最终判定信息“足够”因为它能推断出“上下文中没有Alpha的成就信息”这一事实本身也是一个有效答案从而触发综合Agent生成最终回复。4. 生产级工程化考量与部署将上述原型转化为生产级系统需要解决稳定性、性能、可观测性和成本等一系列问题。4.1 架构升级从脚本到服务原型是单脚本生产环境需要拆分为独立的、可伸缩的服务。API网关处理用户请求进行认证、限流、路由。Orchestrator 服务一个常驻服务如FastAPI/Spring Boot应用负责管理工作流状态、调用其他Agent服务。它需要持久化工作流状态到Redis或DB。Agent 服务将每个Agent功能查询重写、质检、综合封装为独立的微服务或Lambda函数。这允许独立扩缩容和更新。向量数据库/检索服务作为独立基础设施可能是一个集群。消息队列在Agent之间使用消息队列如RabbitMQ, Kafka进行异步通信提高系统的解耦性和韧性。4.2 关键配置与参数调优组件关键参数说明与调优建议检索器(Retriever)search_kwargs: {“k”: n}n是返回的文档片段数。太小可能信息不足太大会增加LLM处理开销和成本。通常从5-10开始根据质检Agent的反馈动态调整。质检AgentLLM温度(temperature)必须设置为0或接近0以保证判断的确定性和一致性。提示词模板这是核心。需明确指令其“严格基于上下文”并定义清晰的输出格式如JSON Schema。需通过大量测试用例进行迭代优化。编排器max_iterations最大检索迭代次数防止无限循环。通常设置为2-4。生产环境需结合超时设置。超时控制为每个Agent调用和检索操作设置单独的超时如LLM调用30s检索5s。缓存查询/结果缓存对频繁出现的相同或相似查询缓存最终答案或中间检索结果大幅降低成本和延迟。可使用Redis。4.3 可观测性与监控没有监控的AI系统是黑盒无法运维。日志记录结构化记录每个请求的完整轨迹。session_id,request_id每个Agent的输入/输出检索的查询词、数据源、返回片段ID质检Agent的每次判断结果is_sufficient,missing_information最终答案各步骤耗时指标监控请求量/成功率/延迟API网关层面。Agent调用指标各Agent的调用次数、成功/失败率、平均耗时。检索指标检索耗时、召回片段数量、缓存命中率。质检指标is_sufficient为True的比例即一轮检索即成功的比例、平均迭代次数。这是衡量系统效率的关键。成本指标估算每个请求消耗的Token数特别是输入给LLM的上下文长度和API调用费用。追踪与调试集成OpenTelemetry等分布式追踪工具可视化一个请求流经所有Agent和服务的完整路径便于定位性能瓶颈和错误。4.4 安全与合规输入输出过滤对用户输入和Agent生成的查询进行内容安全过滤防止Prompt注入或检索到不当内容。数据访问控制检索层需要实现基于用户/角色的数据权限过滤确保用户只能访问其有权访问的文档。审计日志记录谁、在什么时候、问了什么问题、系统检索了哪些文档、生成了什么答案。这对法律、医疗等合规场景是必须的。幻觉抑制在综合Agent的提示词中强化“仅基于上下文回答”的指令并可在后端对生成的答案进行事实一致性检查通过另一个LLM调用验证答案中的关键主张是否能在上下文中找到支持。5. 常见问题排查与优化实践在开发和运维Agentic RAG系统时你会遇到一些典型问题。5.1 问题一质检Agent始终判断信息不足导致无限循环或达到最大迭代现象系统陷入检索循环每次质检Agent都返回is_sufficient: false。可能原因与排查知识库根本不存在答案这是正常情况。需要让质检Agent学会判断“问题在知识库覆盖范围外”。在提示词中加入明确指令例如“如果上下文明确显示该信息不存在于知识库中也视为‘信息充足’并请在reason中说明‘信息在知识库中不存在’。”检索器召回率低向量搜索的相似度阈值太高或者查询重写效果差导致根本检索不到相关文档。检查检索器返回的片段是否与问题相关。可以尝试调整search_kwargs如增加k值或优化查询重写Agent的提示词。质检Agent提示词过于严格它可能要求“完美答案”所需的所有细节而知识库只有部分信息。调整提示词让其接受“部分答案”或“基于现有信息的推断”。例如“如果上下文包含了回答问题的核心事实即使缺少一些次要细节也应判定为信息充足。”解决与优化收集一批“循环案例”人工分析质检Agent的判断是否合理。针对性地修改质检Agent的提示词或引入少量示例Few-shot到提示词中。设置一个“置信度阈值”如果连续两轮检索到的文档相似度都很低则提前终止循环返回“信息未找到”。5.2 问题二系统响应延迟过高现象单个请求耗时长达数十秒。可能原因与排查串行调用Agent和检索步骤如果是同步串行执行延迟会累加。LLM调用慢特别是使用大模型或网络不佳时。检索慢向量数据库未优化或查询复杂。迭代次数多默认进行了多轮检索。解决与优化异步与并行将可以并行的步骤并行化。例如查询重写后可以向多个数据源发起并行检索。缓存对频繁出现的查询和中间结果进行缓存。模型优化在非关键路径如查询重写、质检使用更小、更快的模型如GPT-4o-mini。设置超时和降级为每个步骤设置严格的超时超时后使用默认值或跳过该步骤。限制迭代合理设置max_iterations如2次。5.3 问题三答案质量不稳定有时包含幻觉现象相同问题在不同时间得到不同答案或答案中包含了上下文中没有的信息。可能原因与排查LLM温度参数生成答案的Synthesis Agent温度参数可能不为0导致随机性。上下文过长或混乱迭代检索积累的上下文可能包含冗余或矛盾信息干扰LLM。提示词不明确Synthesis Agent的指令不够强硬。解决与优化确保Synthesis Agent的temperature0。在将上下文传递给Synthesis Agent前进行去重和排序优先保留与问题最相关的片段。强化提示词使用“你必须严格引用上下文中的句子来支持你的答案”、“如果信息不在上下文中请直接说‘我不知道’”等指令。实现后处理验证增加一个“答案验证Agent”检查最终答案中的每个关键事实是否都能在上下文中找到出处。5.4 生产环境检查清单在将Agentic RAG系统部署上线前请对照此清单进行检查[ ]基础设施服务是否容器化Docker是否有部署编排K8s配置是否外置[ ]依赖管理LLM API密钥、数据库连接串等敏感信息是否通过秘钥管理[ ]弹性设计是否有重试机制特别是LLM API调用是否有熔断和降级策略[ ]监控告警核心指标延迟、错误率、迭代次数是否有监控面板是否设置了异常告警[ ]日志与追踪是否每个请求都有唯一ID贯穿全链路日志是否结构化并集中收集[ ]成本控制是否有Token使用量和API调用量的监控与预算告警[ ]测试覆盖是否有针对核心Agent尤其是质检Agent的单元测试和集成测试是否有回归测试集[ ]版本管理提示词、模型版本、Agent逻辑是否有版本控制并能快速回滚工程化Agentic RAG系统的核心价值在于将智能的“决策”能力何时检索、检索什么与稳定的“执行”能力高效检索、可靠生成相结合。它不是一个固定的管道而是一个可观测、可调控、可迭代的智能工作流。从Google Search API集成到企业内部知识库的复杂查询其架构思想是相通的通过多智能体的分工与协作让RAG系统具备了动态规划和自我验证的能力从而在复杂、高要求的场景下提供更可信的回答。 30款热门AI模型一站整合DeepSeek/GLM/Claude 随心用限时 5 折。 点击领海量免费额度