企业官网建设流程全解析

【AI应用开发】RAG+Agent 组合架构怎么设计？什么时候先检索，什么时候先规划？

1. RAG + Agent 的本质关系

RAG 解决的是"知识从哪来"（检索增强） Agent 解决的是"事情怎么做"（自主决策与执行） RAG + Agent = 一个既知道怎么查资料、又知道怎么动手的 AI

简单比喻：

2. 四种组合架构模式

模式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_tool=ToolDefinition(name="search_knowledge_base",description="在知识库中搜索。适用：产品FAQ、操作指南、政策文档等",handler=self._rag_search,parameters={...})# 其他业务工具self.tools=[self.rag_tool,query_order_tool,create_ticket_tool,# ...]self.agent=AgentOrchestrator(self.tools,llm)def_rag_search(self,query,top_k=5):"""RAG 检索实现"""results=self.vector_store.search(query,k=top_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_result=awaitself._deep_retrieval(user_input)# 第二阶段：基于检索结果进行规划plan=awaitself._plan_based_on_context(user_input,retrieval_result)# 第三阶段：执行计划returnawaitself._execute_plan(plan,retrieval_result)asyncdef_deep_retrieval(self,user_input):"""深度检索：不是简单搜一次，而是多层次检索"""# 1. Query 改写（扩展同义词、拆分子问题）queries=awaitself._rewrite_queries(user_input)# 2. 多路召回all_results=[]forqinqueries:results=awaitself.vector_store.search(q,k=5)all_results.extend(results)# 3. Rerank 精排reranked=awaitself._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_prompt=f"""用户问题:{user_input}知识库检索结果:{json.dumps(retrieval['extracted_facts'],ensure_ascii=False)}制定执行计划: 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_task=asyncio.create_task(self._rag_channel(user_input))plan_task=asyncio.create_task(self._agent_channel(user_input))rag_result,plan_result=awaitasyncio.gather(rag_task,plan_task)# 融合两个通道的结果returnawaitself._fusion(user_input,rag_result,plan_result)asyncdef_rag_channel(self,user_input):"""通道1: RAG 检索 → 初步答案"""docs=awaitself.vector_store.search(user_input,k=5)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_prompt=f"""融合以下两个维度的信息给用户最终答案: 维度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():score=sum(1forpinrule["pattern"]ifpinuser_input)scores[category]=score best=max(scores,key=scores.get)returnself.ROUTING_RULES[best]["strategy"]ifscores[best]>0else"parallel"asyncdefdispatch(self,user_input:str):strategy=self.classify(user_input)ifstrategy=="rag_only":returnawaitself.rag.answer(user_input)elifstrategy=="rag_then_agent":returnawaitself.rag_driven_agent.run(user_input)elifstrategy=="agent_only":returnawaitself.agent.run(user_input)else:returnawaitself.parallel_agent.run(user_input)

7. 先检索 vs 先规划：决策框架

决策树: 问题类型是什么？ ├── 知识型问题（"XXX是什么？""YYY怎么做？"） │ └── 先检索 → 检索结果够了直接答，不够再规划 │ ├── 操作型问题（"帮我查XXX""创建YYY"） │ └── 先规划 → 根据规划按需检索 │ ├── 混合型问题（"查XXX然后对比YYY，最后给建议"） │ └── 并行 → 检索提供知识背景，Agent 规划执行 │ └── 未知类型 └── 默认并行 → 取两个通道最好的结果

决策原则：

8. 实战架构设计

classRAGAgentSystem:"""生产级 RAG + Agent 组合系统"""def__init__(self,llm,vector_store,tools):self.llm=llm self.vector_store=vector_store self.tools=tools# 子模块self.query_rewriter=QueryRewriter(llm)self.classifier=IntentClassifier(llm)self.rag=RAGModule(vector_store,llm)self.agent=AgentOrchestrator(tools,llm)self.reranker=Reranker(llm)self.router=AdaptiveRouter()# 工具集中包含 RAG 检索self.tools["search_knowledge_base"]=RAGSearchTool(self.rag)asyncdefprocess(self,user_input:str,session_id:str):"""完整处理流程"""# 1. 意图分类intent=awaitself.classifier.classify(user_input)# 2. Query 改写（总是执行，提升检索质量）rewritten=awaitself.query_rewriter.rewrite(user_input,session_id=session_id)# 3. 根据意图选择策略ifintent["type"]=="simple_fact":# 简单事实 → 只检索docs=awaitself.rag.search(rewritten,k=5)returnawaitself._synthesize_answer(user_input,docs)elifintent["type"]=="operation":# 操作型 → Agent 带 RAG 工具returnawaitself.agent.run(user_input,extra_context={"rewritten_query":rewritten})elifintent["type"]=="complex":# 复杂问题 → 检索 → 规划 → 执行retrieval=awaitself._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_task=asyncio.create_task(self.rag.answer(user_input,rewritten))agent_task=asyncio.create_task(self.agent.run(user_input))rag_result,agent_result=awaitasyncio.gather(rag_task,agent_task)returnawaitself._fusion(user_input,rag_result,agent_result)

9. OpenAI Assistants + RAG 参考实现

# 结合 OpenAI Assistants API 的 RAGfromopenaiimportOpenAIclassAssistantsRAGAgent:"""使用 OpenAI Assistants API 构建 RAG+Agent"""def__init__(self,vector_store_id:str):self.client=OpenAI()self.assistant=self.client.beta.assistants.create(name="智能客服 Agent",model="gpt-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:str=None):ifnotthread_id:thread=self.client.beta.threads.create()thread_id=thread.idself.client.beta.threads.messages.create(thread_id=thread_id,role="user",content=user_message)run=self.client.beta.threads.runs.create_and_poll(thread_id=thread_id,assistant_id=self.assistant.id)ifrun.status=="completed":messages=self.client.beta.threads.messages.list(thread_id=thread_id)returnmessages.data[0].content[0].text.valuereturnf"处理失败:{run.status}"

10. 总结

金句：RAG 告诉 Agent “知道什么”，Agent 决定"做什么"——两者不是二选一，而是 1+1>2 的关系。