欢迎光临
我们一直在努力

RAG 技术的下一站:Agentic RAG 和自主检索的未来

RAG 技术的下一站:Agentic RAG 和自主检索的未来

一、从"一次检索"到"自主检索":RAG 技术的进化

2026 年初,某法律 AI 平台对 RAG 系统进行了一次升级:从"固定检索 5 个文档"升级到"Agentic RAG"(Agent 自主决定检索几次、检索什么)。

结果:回答准确率从 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_k=5)

# 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 Rewriting)
  • HyDE(假设文档生成)
  • 生产级实现

    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_tokens=100)
    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_k=top_k)

    def _keyword_search(self, query: str, top_k: int = 20) -> List[Dict]:
    """关键词检索(BM25)"""
    return self.keyword_index.search(query, top_k=top_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(), key=lambda x: x["score"], reverse=True)
    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(key=lambda x: x[1], reverse=True)

    return [doc for doc, _ in doc_score_pairs]

    Advanced RAG 的效果

    实测数据(某客服场景):

    方法准确率平均延迟成本(tokens/query)
    Naive RAG 72% 1.5s 800
    Advanced RAG 85% 2.8s 1500

    结论:Advanced RAG 明显更准,但更慢、更贵。

    四、阶段三:Agentic RAG(未来趋势)

    核心思想:让 Agent 控制检索过程

    对比:

    维度Advanced RAGAgentic RAG
    检索次数 固定 1 次 动态(1-N 次)
    检索策略 预设 Agent 决策
    适应性
    成本 可控 不可控(可能多次调用 LLM)

    生产级实现

    class AgenticRAG:
    """Agentic RAG(Agent 控制检索)"""

    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_tokens=10)
    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
    赞(0)
    未经允许不得转载:171主机测评 » RAG 技术的下一站:Agentic RAG 和自主检索的未来
    分享到: 更多 (0)

    评论 抢沙发

    • 昵称 (必填)
    • 邮箱 (必填)
    • 网址