生产级 RAG 核心：动态检索路由而非一次性召回

大多数 RAG 教程在生产环境开始的地方就戛然而止了。

它们教你把文档分块、向量化、塞进提示词。在 demo 里效果拔群。但当你把它丢给一个需要回答多跳问题、跨文档对比、或者判断是否需要检索的 agent 时——整套方案就崩了。

问题不在于检索精度，而在于大多数 RAG 实现把检索当作一次性的预处理步骤，而 agent 实际上需要的是运行时的动态决策。

以下是我在反复构建和重构之后最终确定的方案：动态检索路由。

Tutorial RAG 与 Agentic RAG 之间的鸿沟

在标准 RAG 设置中，你在对话开始时检索一次。你把用户查询向量化，找到 top-k 块，然后交给模型。完成。

在 agentic 系统中，这样做的原因有三个：

Agent 不知道自己不知道什么——如果它已经从过时上下文回答了问题，它无法决定去检索更多。

不同的查询需要不同的策略——有些需要密集向量搜索，有些需要 BM25 关键词匹配，有些需要网络搜索。

检索时机很重要——调用得太早浪费 token；太晚则 agent 在凭猜测推理。

解决方案是把检索推向 agent 的决策循环，而不仅仅是它的开场白。

实践中的检索路由器

以下是我一直在运行的架构。Agent 获得的是一个 route_and_retrieve 工具，而不是原始检索工具。基于查询类型，它选择正确的检索策略。

from enum import Enum
from typing import Literal

class RetrievalStrategy(Enum):
    VECTOR = "vector"
    KEYWORD = "keyword"
    WEB = "web"
    HYBRID = "hybrid"

def route_query(query: str) -> RetrievalStrategy:
    """Classify the query to pick the right retrieval strategy."""
    classify_prompt = f"""Classify this query by retrieval need:

Query: {query}

Options:
- vector: factual questions about stored documents
- keyword: specific entity lookups (names, dates, numbers)
- web: current events, anything after knowledge cutoff
- hybrid: comparison or multi-hop questions spanning sources

Respond with only the strategy name."""

    result = llm.invoke(classify_prompt).content.strip().lower()

    if result not in [s.value for s in RetrievalStrategy]:
        return RetrievalStrategy.HYBRID
    return RetrievalStrategy(result)

def retrieve(query: str, top_k: int = 5) -> list[str]:
    strategy = route_query(query)

    if strategy == RetrievalStrategy.VECTOR:
        return vector_search(query, top_k)
    elif strategy == RetrievalStrategy.KEYWORD:
        return bm25_search(query, top_k)
    elif strategy == RetrievalStrategy.WEB:
        return web_search(query)
    else:  # HYBRID
        vector_results = vector_search(query, top_k)
        keyword_results = bm25_search(query, top_k)
        return merge_and_rerank(vector_results, keyword_results, query)

Agent 调用 retrieve(question) 作为工具，获取上下文返回，然后决定它是否有所需的内容，或者应该用更精准的查询再次检索。

当检索成为一个循环时会发生什么

一旦检索进入了 agent 的工具循环，之前不可能的几件事就变得可能了。

自我改进：Agent 可以查看它检索到的内容，判断这些内容不够具体。它可以用更有针对性的查询再次调用 retrieve()——这在一次性 RAG 设置中是做不到的。

策略切换中途：比较问题可能一半需要向量搜索，另一半需要关键词搜索。Agent 可以用不同的子查询两次调用路由器。

条件检索：有些问题根本不需要检索——模型的权重已经覆盖了。Agent 可以完全跳过事实性召回的检索，或者在之前的轮次中已经有了足够上下文的情况下跳过。

以下是一个最小化 agent 循环中的条件逻辑：

def agent_loop(query: str, max_retrievals: int = 3):
    context = []
    recent_queries = []

    for i in range(max_retrievals):
        prompt = build_prompt(query, context)
        response = llm.invoke(prompt)

        if response.tool_calls:
            for call in response.tool_calls:
                if call.name == "retrieve":
                    docs = retrieve(call.arguments["query"])
                    context.append({"role": "user", "content": format_docs(docs)}")
                    recent_queries.append(call.arguments["query"])
                elif call.name == "done":
                    return response.content
        else:
            return response.content

    return "Could not resolve in max retrieval steps"

done 工具是 agent 表示它对已有内容满意的方式。写起来可能觉得奇怪——你在让模型决定何时停止——但当提示词明确说明了停止条件时，这在实践中效果很好。

让我措手不及的失败模式

坑到我的不是检索质量。而是冗余检索带来的上下文窗口压力。

在循环中，agent 很容易在语义相似的查询上连续两个轮次调用 retrieve()，每次都追加重叠的上下文。几个轮次后，你已经烧掉了 60% 的上下文在几乎重复的块上。

我的修复是在追加新上下文之前做一个轻量级的去重步骤：

def deduplicate_context(existing: list, new_chunks: list[str], threshold: float = 0.85) -> list[str]:
    """Remove new chunks that are too similar to existing context."""
    existing_text = "\n".join(existing)
    filtered = []

    for chunk in new_chunks:
        similarity = compute_similarity(chunk, existing_text)
        if similarity < threshold:
            filtered.append(chunk)

    return filtered

这并不复杂——只是对完整文本做余弦相似度计算。但在测试中，它将每个对话轮次的平均 token 使用量减少了约 35%，而且没有损害答案质量。

从「RAG 作为预处理」到「RAG 作为 agent 工具」的跨越在架构上并不复杂。这主要是你关于「检索什么」这个决策的发生位置发生了变化。

一旦我停止把检索视为在对话开始时发生一次的事情，开始把它视为 agent 在需要时调用的能力，很多边界情况就自行解决了。

动态路由这部分是我如果要删东西会保留的部分。即使查询分类 10% 的时间是错误的，让 agent 用有意识的检索策略运作往往比盲目向量化然后碰运气产生更好的结果。

去重步骤是意外的收获。它不在原始设计中——我是在观察到更长对话中 token 计数飙升之后才加上的。这种东西在教程中很容易被跳过，但在实践中却很重要。

如果你在 agent 内部运行 RAG，而且你在顶部一次性做完，值得问一下：第二次检索、不同的策略、或者一点去重会改变什么。