【AI应用开发】RAGAgent 组合架构怎么设计什么时候先检索什么时候先规划1. RAG Agent 的本质关系RAG 解决的是知识从哪来检索增强 Agent 解决的是事情怎么做自主决策与执行 RAG Agent 一个既知道怎么查资料、又知道怎么动手的 AI简单比喻角色能力对应图书管理员会查书、找资料RAG私人助理会思考、规划、执行Agent两者结合能查资料 能做事RAG Agent2. 四种组合架构模式模式A: Agent驱动RAG ──── RAG 是 Agent 工具箱里的一个工具 模式B: RAG驱动Agent ──── 先用检索结果决定 Agent 的规划 模式C: 并行双通道 ──── 检索和规划同时进行结果融合 模式D: 自适应路由 ──── 根据问题类型动态选择执行路径3. 模式一Agent 驱动 RAG检索为工具最常用的模式RAG 被视为 Agent 的一个 Tool用户: iPhone 15 的退货流程是什么 │ ▼ Agent 规划: Step1: search_knowledge_base(iPhone 15 退货流程) ← RAG 作为工具 Step2: 根据检索结果回答用户classAgentWithRAG:Agent 驱动 RAG —— RAG 作为工具def__init__(self,llm,vector_store):# RAG 检索工具self.rag_toolToolDefinition(namesearch_knowledge_base,description在知识库中搜索。适用产品FAQ、操作指南、政策文档等,handlerself._rag_search,parameters{...})# 其他业务工具self.tools[self.rag_tool,query_order_tool,create_ticket_tool,# ...]self.agentAgentOrchestrator(self.tools,llm)def_rag_search(self,query,top_k5):RAG 检索实现resultsself.vector_store.search(query,ktop_k)return{source:knowledge_base,results:[{content:r.content,score:r.score,metadata:r.metadata}forrinresults]}asyncdefrun(self,user_input):returnawaitself.agent.run(user_input)# 执行流程示例 用户: 订单到哪了顺便问下退货怎么操作 → Agent 自主决定: 1. query_order() 获取订单状态 2. search_knowledge_base(退货流程) 获取退货规则 3. 综合两个结果回答用户 适用场景Agent 需要同时使用多个工具RAG API 数据库且步骤间可能有依赖。4. 模式二RAG 驱动 Agent检索决定规划检索结果直接影响 Agent 的规划方向classRAGDrivenAgent:RAG 驱动 Agent —— 检索结果决定规划asyncdefrun(self,user_input):# 第一阶段先检索获取关键信息retrieval_resultawaitself._deep_retrieval(user_input)# 第二阶段基于检索结果进行规划planawaitself._plan_based_on_context(user_input,retrieval_result)# 第三阶段执行计划returnawaitself._execute_plan(plan,retrieval_result)asyncdef_deep_retrieval(self,user_input):深度检索不是简单搜一次而是多层次检索# 1. Query 改写扩展同义词、拆分子问题queriesawaitself._rewrite_queries(user_input)# 2. 多路召回all_results[]forqinqueries:resultsawaitself.vector_store.search(q,k5)all_results.extend(results)# 3. Rerank 精排rerankedawaitself._rerank(user_input,all_results)# 4. 结构化提取return{relevant_docs:reranked[:5],extracted_facts:awaitself._extract_facts(reranked,user_input),missing_info:self._identify_gaps(reranked,user_input),}asyncdef_plan_based_on_context(self,user_input,retrieval):基于检索结果制定计划plan_promptf用户问题:{user_input}知识库检索结果:{json.dumps(retrieval[extracted_facts],ensure_asciiFalse)}制定执行计划: 1. 如果检索结果已充分回答了用户问题 → 直接基于检索结果回答 2. 如果检索结果不够 → 确定还需要哪些信息选择合适工具 3. 如果需要多步操作 → 制定步骤顺序 输出 JSON 格式计划: {{ can_answer_directly: true/false, direct_answer: ..., // 如果可以直接回答 steps: [ // 否则列出步骤 {{tool: 工具名, args: {{}}, purpose: 目的}} ] }}returnawaitself.llm.chat(plan_prompt)适用场景用户问题需要深度理解背景知识才能规划时如技术排障、医疗问诊。5. 模式三并行双通道检索和规划同时进行最后融合importasyncioclassParallelAgent:并行双通道: RAG 和规划同时进行asyncdefrun(self,user_input):# 双通道并行rag_taskasyncio.create_task(self._rag_channel(user_input))plan_taskasyncio.create_task(self._agent_channel(user_input))rag_result,plan_resultawaitasyncio.gather(rag_task,plan_task)# 融合两个通道的结果returnawaitself._fusion(user_input,rag_result,plan_result)asyncdef_rag_channel(self,user_input):通道1: RAG 检索 → 初步答案docsawaitself.vector_store.search(user_input,k5)return{docs:docs,draft_answer:awaitself._generate_draft(user_input,docs)}asyncdef_agent_channel(self,user_input):通道2: Agent 规划 → 工具调用# Agent 在规划时也能调用 search_knowledge_base# 但这里的检索是另一个独立路径returnawaitself.agent.run(user_input)asyncdef_fusion(self,user_input,rag_result,plan_result):融合双通道结果fusion_promptf融合以下两个维度的信息给用户最终答案: 维度1 - 知识库检索结果:{rag_result[draft_answer]}维度2 - Agent 工具执行结果:{plan_result}融合原则: - 信息一致 → 合并给出统一答案 - 信息冲突 → 优先采用工具执行的实际数据 - 信息互补 → 整合呈现完整信息 - 标注每个信息来源知识库 / 实时数据returnawaitself.llm.chat(fusion_prompt)适用场景对响应速度要求高且检索和规划可以独立进行的场景。6. 模式四自适应路由根据问题类型动态选择处理路径classAdaptiveRouter:根据问题类型自适应路由ROUTING_RULES{simple_fact:{pattern:[什么是,定义,多少,什么时间],strategy:rag_only,# 只检索reason:简单事实查询不需要规划},procedure:{pattern:[怎么做,如何,流程,步骤],strategy:rag_then_agent,# 先检索再规划reason:需要参考文档后制定步骤},data_query:{pattern:[订单,库存,我的,状态],strategy:agent_only,# 只规划reason:需要调用API/DB不依赖知识库},complex_analysis:{pattern:[分析,对比,总结,评估],strategy:parallel,# 并行reason:需要知识背景数据分析}}defclassify(self,user_input:str)-str:对用户问题分类scores{}forcategory,ruleinself.ROUTING_RULES.items():scoresum(1forpinrule[pattern]ifpinuser_input)scores[category]score bestmax(scores,keyscores.get)returnself.ROUTING_RULES[best][strategy]ifscores[best]0elseparallelasyncdefdispatch(self,user_input:str):strategyself.classify(user_input)ifstrategyrag_only:returnawaitself.rag.answer(user_input)elifstrategyrag_then_agent:returnawaitself.rag_driven_agent.run(user_input)elifstrategyagent_only:returnawaitself.agent.run(user_input)else:returnawaitself.parallel_agent.run(user_input)7. 先检索 vs 先规划决策框架决策树: 问题类型是什么 ├── 知识型问题XXX是什么YYY怎么做 │ └── 先检索 → 检索结果够了直接答不够再规划 │ ├── 操作型问题帮我查XXX创建YYY │ └── 先规划 → 根据规划按需检索 │ ├── 混合型问题查XXX然后对比YYY最后给建议 │ └── 并行 → 检索提供知识背景Agent 规划执行 │ └── 未知类型 └── 默认并行 → 取两个通道最好的结果决策原则情况策略原因问题可能已有文档答案先检索避免 Agent 重复造轮子需要调用外部 API先规划确定需要什么数据需要知识操作结合并行最大化效率操作依赖检索结果先检索再规划检索是规划的前提8. 实战架构设计classRAGAgentSystem:生产级 RAG Agent 组合系统def__init__(self,llm,vector_store,tools):self.llmllm self.vector_storevector_store self.toolstools# 子模块self.query_rewriterQueryRewriter(llm)self.classifierIntentClassifier(llm)self.ragRAGModule(vector_store,llm)self.agentAgentOrchestrator(tools,llm)self.rerankerReranker(llm)self.routerAdaptiveRouter()# 工具集中包含 RAG 检索self.tools[search_knowledge_base]RAGSearchTool(self.rag)asyncdefprocess(self,user_input:str,session_id:str):完整处理流程# 1. 意图分类intentawaitself.classifier.classify(user_input)# 2. Query 改写总是执行提升检索质量rewrittenawaitself.query_rewriter.rewrite(user_input,session_idsession_id)# 3. 根据意图选择策略ifintent[type]simple_fact:# 简单事实 → 只检索docsawaitself.rag.search(rewritten,k5)returnawaitself._synthesize_answer(user_input,docs)elifintent[type]operation:# 操作型 → Agent 带 RAG 工具returnawaitself.agent.run(user_input,extra_context{rewritten_query:rewritten})elifintent[type]complex:# 复杂问题 → 检索 → 规划 → 执行retrievalawaitself._deep_retrieve(rewritten)ifretrieval[sufficiency]0.8:# 检索结果足够 → 直接回答returnawaitself._synthesize_answer(user_input,retrieval[docs])else:# 检索不够 → Agent 进一步处理returnawaitself.agent.run(user_input,extra_context{retrieval_context:retrieval[context],gaps:retrieval[missing]})else:# 未知类型 → 并行rag_taskasyncio.create_task(self.rag.answer(user_input,rewritten))agent_taskasyncio.create_task(self.agent.run(user_input))rag_result,agent_resultawaitasyncio.gather(rag_task,agent_task)returnawaitself._fusion(user_input,rag_result,agent_result)9. OpenAI Assistants RAG 参考实现# 结合 OpenAI Assistants API 的 RAGfromopenaiimportOpenAIclassAssistantsRAGAgent:使用 OpenAI Assistants API 构建 RAGAgentdef__init__(self,vector_store_id:str):self.clientOpenAI()self.assistantself.client.beta.assistants.create(name智能客服 Agent,modelgpt-4o,instructions 你是智能客服助手可以 1. 使用 file_search 检索知识库 2. 使用 function 工具查询订单/库存等业务数据 3. 综合分析后给出答案 ,tools[{type:file_search},# ← RAG 能力{type:function,function:{# ← Agent 工具name:query_order,description:查询订单信息,parameters:{...}}},{type:function,function:{name:query_inventory,description:查询库存,parameters:{...}}},],tool_resources{file_search:{vector_store_ids:[vector_store_id]}})asyncdefchat(self,user_message:str,thread_id:strNone):ifnotthread_id:threadself.client.beta.threads.create()thread_idthread.idself.client.beta.threads.messages.create(thread_idthread_id,roleuser,contentuser_message)runself.client.beta.threads.runs.create_and_poll(thread_idthread_id,assistant_idself.assistant.id)ifrun.statuscompleted:messagesself.client.beta.threads.messages.list(thread_idthread_id)returnmessages.data[0].content[0].text.valuereturnf处理失败:{run.status}10. 总结模式先检索/先规划适用场景延迟Agent 驱动 RAG先规划Agent 需多工具的复杂任务中RAG 驱动 Agent先检索需背景知识的操作中高并行双通道同时对速度要求高的混合问题低自适应路由动态选择通用客服系统灵活金句RAG 告诉 Agent “知道什么”Agent 决定做什么——两者不是二选一而是 112 的关系。