RAG 技术的下一站Agentic RAG 和自主检索的未来一、从一次检索到自主检索RAG 技术的进化2026 年初某法律 AI 平台对 RAG 系统进行了一次升级从固定检索 5 个文档升级到Agentic RAGAgent 自主决定检索几次、检索什么。结果回答准确率从 72% 提升到 89%但成本也增加了 40%。这个案例揭示了 RAG 技术的下一个进化方向从被动检索到主动、多步、自适应的检索。本文将深入分析 RAG 技术的未来趋势。二、阶段一Naive RAG当前主流典型实现# Naive RAG简单但不是最优 class NaiveRAG: 朴素 RAG 实现 def __init__(self, vector_db, llm): self.vector_db vector_db self.llm llm def query(self, user_query: str) - str: # 1. 向量检索固定 Top-5 query_embedding self.embed(user_query) docs self.vector_db.search(query_embedding, top_k5) # 2. 拼接上下文 context \n.join([doc[content] for doc in docs]) # 3. 生成回答 prompt f基于以下内容回答问题\n{context}\n\n问题{user_query} answer self.llm.generate(prompt) return answer问题三、阶段二Advanced RAG当前最佳实践核心改进混合检索向量 关键词Rerank重排序查询改写Query RewritingHyDE假设文档生成生产级实现class AdvancedRAG: 进阶 RAG 实现 def __init__(self, vector_db, keyword_index, llm): self.vector_db vector_db self.keyword_index keyword_index self.llm llm self.reranker CrossEncoder(BAAI/bge-reranker-v1.5) def query(self, user_query: str, history: List[Dict] None) - str: # 1. 查询改写基于历史 rewritten_query self._rewrite_query(user_query, history) # 2. 混合检索 vector_results self._vector_search(rewritten_query) keyword_results self._keyword_search(rewritten_query) # 3. 结果融合 merged self._merge_results(vector_results, keyword_results) # 4. Rerank reranked self._rerank(rewritten_query, merged) # 5. 上下文压缩 context self._compress_context(reranked[:5]) # 6. 生成回答 answer self._generate_answer(user_query, context, history) return answer def _rewrite_query(self, query: str, history: List[Dict]) - str: 查询改写 if not history: return query # 使用 LLM 改写考虑上下文 prompt f 基于以下对话历史改写用户的当前问题使其更明确 历史 {self._format_history(history)} 当前问题{query} 改写后的问题 rewritten self.llm.generate(prompt, max_tokens100) return rewritten.strip() def _vector_search(self, query: str, top_k: int 20) - List[Dict]: 向量检索 query_embedding self.embed(query) return self.vector_db.search(query_embedding, top_ktop_k) def _keyword_search(self, query: str, top_k: int 20) - List[Dict]: 关键词检索BM25 return self.keyword_index.search(query, top_ktop_k) def _merge_results(self, list1: List[Dict], list2: List[Dict]) - List[Dict]: 融合结果Reciprocal Rank Fusion scores {} for rank, doc in enumerate(list1, 1): doc_id doc[id] if doc_id not in scores: scores[doc_id] {doc: doc, score: 0} scores[doc_id][score] 1 / (60 rank) for rank, doc in enumerate(list2, 1): doc_id doc[id] if doc_id not in scores: scores[doc_id] {doc: doc, score: 0} scores[doc_id][score] 1 / (60 rank) merged sorted(scores.values(), keylambda x: x[score], reverseTrue) return [item[doc] for item in merged] def _rerank(self, query: str, docs: List[Dict]) - List[Dict]: 重排序 pairs [[query, doc[content]] for doc in docs] scores self.reranker.predict(pairs) # 排序 doc_score_pairs list(zip(docs, scores)) doc_score_pairs.sort(keylambda x: x[1], reverseTrue) return [doc for doc, _ in doc_score_pairs]Advanced RAG 的效果实测数据某客服场景方法准确率平均延迟成本tokens/queryNaive RAG72%1.5s800Advanced RAG85%2.8s1500结论Advanced RAG 明显更准但更慢、更贵。四、阶段三Agentic RAG未来趋势核心思想让 Agent 控制检索过程对比维度Advanced RAGAgentic RAG检索次数固定 1 次动态1-N 次检索策略预设Agent 决策适应性低高成本可控不可控可能多次调用 LLM生产级实现class AgenticRAG: Agentic RAGAgent 控制检索 def __init__(self, tools: List[Dict], llm): self.tools tools self.llm llm def query(self, user_query: str) - str: # 使用 ReAct 模式Reasoning Acting max_iterations 5 context [] for i in range(max_iterations): # 1. Reasoning判断是否已足够回答 prompt f 用户问题{user_query} 当前已检索的信息 {self._format_context(context)} 请判断 1. 如果信息已足够回答输出: FINAL_ANSWER: 答案 2. 如果还需要检索输出: SEARCH: 检索查询 response self.llm.generate(prompt) # 2. Acting执行检索如果需要 if response.startswith(FINAL_ANSWER:): # 生成最终答案 answer response.replace(FINAL_ANSWER:, ).strip() return answer elif response.startswith(SEARCH:): # 执行检索 search_query response.replace(SEARCH:, ).strip() # 调用检索工具 search_result self._call_tool(search, {query: search_query}) # 添加到上下文 context.append({ query: search_query, result: search_result }) else: # 解析失败重试 continue # 达到最大迭代次数基于当前上下文回答 final_prompt f 用户问题{user_query} 检索到的信息 {self._format_context(context)} 请基于以上信息回答问题 return self.llm.generate(final_prompt) def _call_tool(self, tool_name: str, params: Dict) - str: 调用工具 if tool_name search: # 混合检索 query params[query] vector_results self._vector_search(query) keyword_results self._keyword_search(query) merged self._merge_results(vector_results, keyword_results) return self._format_docs(merged[:3]) return def _format_context(self, context: List[Dict]) - str: 格式化上下文 parts [] for i, ctx in enumerate(context, 1): parts.append(f检索 {i}: {ctx[query]}) parts.append(f结果 {i}: {ctx[result]}) return \n\n.join(parts)Agentic RAG 的优势案例用户问对比 iPhone 15 和 Samsung S24 的摄像头Naive RAG检索 1 次iPhone 15 Samsung S24 摄像头可能检索到不相关的文档Agentic RAG第 1 次检索iPhone 15 摄像头规格第 2 次检索Samsung S24 摄像头规格第 3 次检索iPhone 15 vs Samsung S24 摄像头对比基于 3 次检索结果生成答案更准确成本优化Agentic RAG 的主要问题是成本高多次调用 LLM。优化方案class CostOptimizedAgenticRAG(AgenticRAG): 成本优化的 Agentic RAG def __init__(self, tools: List[Dict], llm, cheap_llm): super().__init__(tools, llm) self.cheap_llm cheap_llm # 用便宜的模型做推理 def query(self, user_query: str) - str: # 使用便宜模型做检索决策 # 使用贵模型生成最终答案 max_iterations 3 # 限制迭代次数 context [] for i in range(max_iterations): # 用便宜模型判断 should_continue self._should_continue_cheap(query, context) if not should_continue: break # 执行检索 search_query self._generate_search_query_cheap(query, context) result self._call_tool(search, {query: search_query}) context.append({query: search_query, result: result}) # 用贵模型生成最终答案 final_answer self.llm.generate( self._build_final_prompt(query, context) ) return final_answer def _should_continue_cheap(self, query: str, context: List[Dict]) - bool: 用便宜模型判断是否继续检索 prompt f基于当前信息能否回答 {query}(yes/no) response self.cheap_llm.generate(prompt, max_tokens10) return no in response.lower()结论RAG 技术的下一站Agentic RAG✅ 更准确自适应检索⚠️ 成本更高多次 LLM 调用 优化方向用便宜模型做决策未来方向2027-2028端到端优化的 RAG训练一个检索策略网络类似强化学习自动学习何时检索、检索什么多模态 RAG检索图片、视频、音频跨模态检索文本 → 图片实时 RAG支持流式数据更新检索延迟 100ms实施建议现在用 Advanced RAG混合检索 Rerank近期试点 Agentic RAG高价值场景长期关注端到端优化方案关键指标准确率 85%平均检索次数 3成本 $0.05/query
RAG 技术的下一站：Agentic RAG 和自主检索的未来
RAG 技术的下一站Agentic RAG 和自主检索的未来一、从一次检索到自主检索RAG 技术的进化2026 年初某法律 AI 平台对 RAG 系统进行了一次升级从固定检索 5 个文档升级到Agentic RAGAgent 自主决定检索几次、检索什么。结果回答准确率从 72% 提升到 89%但成本也增加了 40%。这个案例揭示了 RAG 技术的下一个进化方向从被动检索到主动、多步、自适应的检索。本文将深入分析 RAG 技术的未来趋势。二、阶段一Naive RAG当前主流典型实现# Naive RAG简单但不是最优 class NaiveRAG: 朴素 RAG 实现 def __init__(self, vector_db, llm): self.vector_db vector_db self.llm llm def query(self, user_query: str) - str: # 1. 向量检索固定 Top-5 query_embedding self.embed(user_query) docs self.vector_db.search(query_embedding, top_k5) # 2. 拼接上下文 context \n.join([doc[content] for doc in docs]) # 3. 生成回答 prompt f基于以下内容回答问题\n{context}\n\n问题{user_query} answer self.llm.generate(prompt) return answer问题三、阶段二Advanced RAG当前最佳实践核心改进混合检索向量 关键词Rerank重排序查询改写Query RewritingHyDE假设文档生成生产级实现class AdvancedRAG: 进阶 RAG 实现 def __init__(self, vector_db, keyword_index, llm): self.vector_db vector_db self.keyword_index keyword_index self.llm llm self.reranker CrossEncoder(BAAI/bge-reranker-v1.5) def query(self, user_query: str, history: List[Dict] None) - str: # 1. 查询改写基于历史 rewritten_query self._rewrite_query(user_query, history) # 2. 混合检索 vector_results self._vector_search(rewritten_query) keyword_results self._keyword_search(rewritten_query) # 3. 结果融合 merged self._merge_results(vector_results, keyword_results) # 4. Rerank reranked self._rerank(rewritten_query, merged) # 5. 上下文压缩 context self._compress_context(reranked[:5]) # 6. 生成回答 answer self._generate_answer(user_query, context, history) return answer def _rewrite_query(self, query: str, history: List[Dict]) - str: 查询改写 if not history: return query # 使用 LLM 改写考虑上下文 prompt f 基于以下对话历史改写用户的当前问题使其更明确 历史 {self._format_history(history)} 当前问题{query} 改写后的问题 rewritten self.llm.generate(prompt, max_tokens100) return rewritten.strip() def _vector_search(self, query: str, top_k: int 20) - List[Dict]: 向量检索 query_embedding self.embed(query) return self.vector_db.search(query_embedding, top_ktop_k) def _keyword_search(self, query: str, top_k: int 20) - List[Dict]: 关键词检索BM25 return self.keyword_index.search(query, top_ktop_k) def _merge_results(self, list1: List[Dict], list2: List[Dict]) - List[Dict]: 融合结果Reciprocal Rank Fusion scores {} for rank, doc in enumerate(list1, 1): doc_id doc[id] if doc_id not in scores: scores[doc_id] {doc: doc, score: 0} scores[doc_id][score] 1 / (60 rank) for rank, doc in enumerate(list2, 1): doc_id doc[id] if doc_id not in scores: scores[doc_id] {doc: doc, score: 0} scores[doc_id][score] 1 / (60 rank) merged sorted(scores.values(), keylambda x: x[score], reverseTrue) return [item[doc] for item in merged] def _rerank(self, query: str, docs: List[Dict]) - List[Dict]: 重排序 pairs [[query, doc[content]] for doc in docs] scores self.reranker.predict(pairs) # 排序 doc_score_pairs list(zip(docs, scores)) doc_score_pairs.sort(keylambda x: x[1], reverseTrue) return [doc for doc, _ in doc_score_pairs]Advanced RAG 的效果实测数据某客服场景方法准确率平均延迟成本tokens/queryNaive RAG72%1.5s800Advanced RAG85%2.8s1500结论Advanced RAG 明显更准但更慢、更贵。四、阶段三Agentic RAG未来趋势核心思想让 Agent 控制检索过程对比维度Advanced RAGAgentic RAG检索次数固定 1 次动态1-N 次检索策略预设Agent 决策适应性低高成本可控不可控可能多次调用 LLM生产级实现class AgenticRAG: Agentic RAGAgent 控制检索 def __init__(self, tools: List[Dict], llm): self.tools tools self.llm llm def query(self, user_query: str) - str: # 使用 ReAct 模式Reasoning Acting max_iterations 5 context [] for i in range(max_iterations): # 1. Reasoning判断是否已足够回答 prompt f 用户问题{user_query} 当前已检索的信息 {self._format_context(context)} 请判断 1. 如果信息已足够回答输出: FINAL_ANSWER: 答案 2. 如果还需要检索输出: SEARCH: 检索查询 response self.llm.generate(prompt) # 2. Acting执行检索如果需要 if response.startswith(FINAL_ANSWER:): # 生成最终答案 answer response.replace(FINAL_ANSWER:, ).strip() return answer elif response.startswith(SEARCH:): # 执行检索 search_query response.replace(SEARCH:, ).strip() # 调用检索工具 search_result self._call_tool(search, {query: search_query}) # 添加到上下文 context.append({ query: search_query, result: search_result }) else: # 解析失败重试 continue # 达到最大迭代次数基于当前上下文回答 final_prompt f 用户问题{user_query} 检索到的信息 {self._format_context(context)} 请基于以上信息回答问题 return self.llm.generate(final_prompt) def _call_tool(self, tool_name: str, params: Dict) - str: 调用工具 if tool_name search: # 混合检索 query params[query] vector_results self._vector_search(query) keyword_results self._keyword_search(query) merged self._merge_results(vector_results, keyword_results) return self._format_docs(merged[:3]) return def _format_context(self, context: List[Dict]) - str: 格式化上下文 parts [] for i, ctx in enumerate(context, 1): parts.append(f检索 {i}: {ctx[query]}) parts.append(f结果 {i}: {ctx[result]}) return \n\n.join(parts)Agentic RAG 的优势案例用户问对比 iPhone 15 和 Samsung S24 的摄像头Naive RAG检索 1 次iPhone 15 Samsung S24 摄像头可能检索到不相关的文档Agentic RAG第 1 次检索iPhone 15 摄像头规格第 2 次检索Samsung S24 摄像头规格第 3 次检索iPhone 15 vs Samsung S24 摄像头对比基于 3 次检索结果生成答案更准确成本优化Agentic RAG 的主要问题是成本高多次调用 LLM。优化方案class CostOptimizedAgenticRAG(AgenticRAG): 成本优化的 Agentic RAG def __init__(self, tools: List[Dict], llm, cheap_llm): super().__init__(tools, llm) self.cheap_llm cheap_llm # 用便宜的模型做推理 def query(self, user_query: str) - str: # 使用便宜模型做检索决策 # 使用贵模型生成最终答案 max_iterations 3 # 限制迭代次数 context [] for i in range(max_iterations): # 用便宜模型判断 should_continue self._should_continue_cheap(query, context) if not should_continue: break # 执行检索 search_query self._generate_search_query_cheap(query, context) result self._call_tool(search, {query: search_query}) context.append({query: search_query, result: result}) # 用贵模型生成最终答案 final_answer self.llm.generate( self._build_final_prompt(query, context) ) return final_answer def _should_continue_cheap(self, query: str, context: List[Dict]) - bool: 用便宜模型判断是否继续检索 prompt f基于当前信息能否回答 {query}(yes/no) response self.cheap_llm.generate(prompt, max_tokens10) return no in response.lower()结论RAG 技术的下一站Agentic RAG✅ 更准确自适应检索⚠️ 成本更高多次 LLM 调用 优化方向用便宜模型做决策未来方向2027-2028端到端优化的 RAG训练一个检索策略网络类似强化学习自动学习何时检索、检索什么多模态 RAG检索图片、视频、音频跨模态检索文本 → 图片实时 RAG支持流式数据更新检索延迟 100ms实施建议现在用 Advanced RAG混合检索 Rerank近期试点 Agentic RAG高价值场景长期关注端到端优化方案关键指标准确率 85%平均检索次数 3成本 $0.05/query
