Agentic_RAG实战：动态检索策略与自适应推理完整指南

Agentic RAG 实战：动态检索策略与自适应推理完整指南作者：Crown_22 | AI Agent RAG 系统开发者 | 技术分享前言传统 RAG（Retrieval-Augmented Generation）系统采用固定的"检索-生成"流水线，查询进来后先向量检索 top-k 文档，再拼接上下文交给 LLM 生成答案。这种架构简单直接，但在复杂场景下暴露了严重瓶颈：检索粒度单一：用户问"什么是 Transformer"和"Transformer 的注意力机制在长序列上的时间复杂度是多少"用的是同一套检索策略无法动态调整：检索到的内容不够精准时，系统不会自动重试或换一种检索方式缺乏推理能力：面对需要多步推理的复杂问题，单次检索往往无法提供足够信息Agentic RAG 的核心思想是：让 AI Agent 来决定如何检索、检索什么、是否需要多次检索，而不是用固定流水线一刀切。本文将从零构建一个生产级 Agentic RAG 系统，涵盖动态检索策略、自适应推理链、查询路由和质量评估。一、传统 RAG vs Agentic RAG 架构对比1.1 传统 RAG 的问题# 传统 RAG：固定流水线，无决策能力deftraditional_rag(query:str)-str:# 步骤固定，无法根据查询复杂度调整docs=vector_store.similarity_search(query,k=4)# 固定 top-kcontext="\n".join([doc.page_contentfordocindocs])prompt=f"根据以下资料回答：\n{context}\n\n问题：{query}"returnllm.invoke(prompt)实际生产中遇到的问题：# 问题场景1：查询太模糊，向量检索命中率低query="那个性能问题怎么解决的"# 向量检索返回的都是泛泛而谈的性能优化文章，缺少具体上下文# 问题场景2：需要多步推理的复杂问题query="对比 LangChain 和 LlamaIndex 在多文档推理场景下的实现差异"# 单次检索只能获取一个框架的信息，无法同时获取两个框架的对比# 问题场景3：查询需要不同粒度的检索query="Python asyncio 的事件循环实现原理"# 文档级检索太粗，段落级检索太细，需要自适应选择1.2 Agentic RAG 的核心设计Agentic RAG 将传统流水线替换为 Agent 驱动的动态决策系统：用户查询 │ ▼ ┌─────────────────┐ │ 查询分析 Agent │ ← 分析查询类型、复杂度、所需信息 └────────┬────────┘ │ ▼ ┌─────────────────┐ │ 查询路由决策 │ ← 选择检索策略（向量/关键词/混合/多跳） └────────┬────────┘ │ ┌────┼────┬────────┐ ▼ ▼ ▼ ▼ 向量 关键词 SQL Web API 检索 检索 查询 调用 │ │ │ │ └────┴────┴────────┘ │ ▼ ┌─────────────────┐ │ 结果评估 Agent │ ← 判断信息是否足够，是否需要重新检索 └────────┬────────┘ │ (不够？重试) ▼ ┌─────────────────┐ │ 答案生成 Agent │ ← 综合所有信息生成最终答案 └─────────────────┘二、从零构建 Agentic RAG 系统2.1 项目结构agentic-rag/ ├── core/ │ ├── agent.py # 主 Agent 逻辑 │ ├── retrievers/ # 多种检索器 │ │ ├── vector.py # 向量检索 │ │ ├── keyword.py # 关键词检索（BM25） │ │ ├── hybrid.py # 混合检索 │ │ └── multi_hop.py # 多跳检索 │ ├── evaluators/ # 质量评估 │ │ ├── relevance.py # 相关性评估 │ │ └── completeness.py # 完整性评估 │ └── memory/ # 对话记忆 ├── config.py ├── main.py └── requirements.txt2.2 依赖安装# requirements.txtlangchain=0.3.0langchain-openai=0.2.0langchain-community=0.3.0faiss-cpu=1.8.0rank-bm25=0.2.2pydantic=2.0tiktoken=0.7.0pipinstall-rrequirements.txt2.3 核心 Agent 实现# core/agent.pyfromtypingimportLiteralfrompydanticimportBaseModel,Fieldfromlangchain_openaiimportChatOpenAIfromlangchain_core.messagesimportHumanMessage,SystemMessagefromlangchain_core.toolsimporttoolclassQueryAnalysis(BaseModel):"""查询分析结果"""query_type:Literal["factual","analytical","comparative","procedural"]=Field(description="查询类型：事实性、分析性、对比性、过程性")complexity:Literal["simple","moderate","complex"]=Field(description="查询复杂度")sub_queries:list[str]=Field(default_factory=list,description="拆解后的子查询（复杂查询才需要）")recommended_strategy:Literal["vector","keyword","hybrid","multi_hop"]=Field(description="推荐的检索策略")reasoning:str=Field(description="策略选择的理由")classRetrievalResult(BaseModel):"""检索结果评估"""documents:list[dict]relevance_scores:list[float]is_sufficient:bool=Field(description="信息是否足够回答问题")missing_info:str=Field(default="",description="缺失的信息")suggested_action:Literal["answer","retry_vector","retry_keyword","multi_hop"]=Field(description="建议的下一步操作")classAgenticRAG:"""Agentic RAG 主系统"""def__init__(self,vector_store,bm25_index,llm_model="gpt-4o"):self.vector_store=vector_store self.bm25_index=bm25_index self.llm=ChatOpenAI(model=llm_model,temperature=0)self.structured_llm=self.llm.with_structured_output(QueryAnalysis)defanalyze_query(self,query:str)-QueryAnalysis:"""分析查询，决定检索策略"""messages=[SystemMessage(content="""你是一个查询分析专家。分析用户查询并决定最佳检索策略。 查询类型： - factual：事实性查询，如"什么是X"、"X的定义" - analytical：分析性查询，如"为什么X会导致Y" - comparative：对比性查询，如"X和Y的区别" - procedural：过程性查询，如"如何实现X" 检索策略选择规则： - simple + factual → vector（向量检索即可） - moderate + analytical → hybrid（混合检索） - complex + comparative → multi_hop（多跳检索） - simple + procedural → keyword（关键词检索，代码相关） - complex查询需要拆解为sub_queries"""),HumanMessage(content=f"请分析以下查询：\n\n{query}")]returnself.structured_llm.invoke(messages)defvector_retrieve(self,query:str,k:int=4)-list[dict]:"""向量检索"""docs=self.vector_store.similarity_search_with_score(query,k=k)return[{"content":doc.page_content,"metadata":doc.metadata,"score":score}fordoc,scoreindocs]defkeyword_retrieve(self,query:str,k:int=4)-list[dict]:"""BM25 关键词检索"""fromrank_bm25importBM25Okapi tokenized_query=list(query)scores=self.bm25_index.get_scores(tokenized_query)top_indices=sorted(range(len(scores)),key=lambdai:scores[i],reverse=True)[:k]return[{"content":self.corpus[i],"metadata":self.metadata[i],"score":scores[i]}foriintop_indices]defhybrid_retrieve(self,query:str,k:int=6)-list[dict]:"""混合检索：向量 + BM25，RRF 融合"""vector_results=self.vector_retrieve(query,k=k)keyword_results=self.keyword_retrieve(query,k=k)# Reciprocal Rank Fusionrrf_scores={}forrank,docinenumerate(vector_results):key=doc["content"][:100]# 用内容前100字符作为keyrrf_scores[key]=rrf_scores.get(key,0)+1/(60+rank)forrank,docinenumerate(keyword_results):key=doc["content"][:100]rrf_scores[key]=rrf_scores.get(key,0)+1/(60+rank)# 合并去重并按RRF分数排序all_docs={doc["content"][:100]:docfordocinvector_results+keyword_results}sorted_keys=sorted(rrf_scores.keys(),key=lambdak:rrf_scores[k],reverse=True)results=[]forkeyinsorted_keys[:k]:doc=all_docs[key]doc["rrf_score"]=rrf_scores[key]results.append(doc)returnresultsdefmulti_hop_retrieve(self,sub_queries:list[str])-list[dict]:"""多跳检索：对每个子查询分别检索，去重合并"""all_results=[]seen_contents=set()forsqinsub_queries:results=self.hybrid_retrieve(sq,k=3)fordocinresults: