Agentic RAG 实战：让检索增强学会自主决策

传统 RAG（检索增强生成）把「向量检索 + 大模型生成」串成一条固定流水线，确实能缓解幻觉，但面对多跳问题、跨知识库场景时常常力不从心。Agentic RAG 的核心，是把智能体（Agent）的决策能力注入检索链路：让模型自己判断「要不要检索」「去哪个知识库检索」「答案够不够好」。本文用三段可运行代码，带你把 RAG 从「被动管道」升级成「自主决策系统」。

过去两年，RAG 几乎成了「给大模型接外部知识」的默认答案：把文档切块、向量化、存进向量库，提问时先搜再答。它确实把模型从「凭记忆胡说」拉回了「有依据地回答」，所以在企业知识库、客服、文档问答里迅速普及。但用久了你会发现，它的「固定流程」本身就是天花板——模型没有机会说「我不需要查」「我该去另一个库查」「这个答案我还不满意，再查一轮」。Agentic RAG 正是在这几个关键节点上，把决策权交还给模型。

一、传统 RAG 的三道天花板

先看传统 RAG 的骨架。它在一个问题进来后，只做一次向量检索、拼进 prompt、再生成一次答案，流程完全写死：

# 传统 RAG：一次性检索 + 一次性生成
def naive_rag(question: str) -> str:
    ctx = vector_search(question, top_k=3)      # 只检索一次
    prompt = f"根据以下资料回答：\n{ctx}\n问题：{question}"
    return llm.generate(prompt)                 # 只生成一次

这套链路有三个绕不开的瓶颈：① 检索时机固定，明明靠模型自身知识就能答的问题也被迫检索；② 检索范围固定，一个问句只能打一个向量库，跨库问题直接抓瞎；③ 答案不可反思，生成出来就交付，错了也无人纠错。下面这张对照表一目了然：

二、Agentic RAG 的三种主流范式

1. 路由式（Routing）：先分类，再检索

先用一个小模型把问题分到对应知识库，再精准检索。适合「产品文档 / 技术报错 / 闲聊」等多库并存的客服场景。

2. 查询分解（Decomposition）：先拆解，并行查

把复杂问题拆成若干子问题，分别检索后合并证据。适合「对比 A 和 B」「某流程第一步怎么做」这类多跳问题。

3. 反思式（Self-RAG）：先判断，再反思

生成前判断是否需要检索，生成后再判断答案是否充分，不充分就回到检索环节再来一轮。质量最高，但调用次数也最多。

三、环境准备：统一你的 LLM 入口

三种范式都依赖同一个 LLM 调用。为了「本地 Ollama、云端模型、多个 Key」随意切换，建议用 自建 One API 网关 暴露一个 OpenAI 兼容端点，业务代码完全不用关心后端是谁：

from openai import OpenAI

# 通过 One API 网关统一接入，本地 Ollama / 云端模型随意切换
client = OpenAI(
    base_url="http://localhost:3000/v1",   # One API 暴露的 OpenAI 兼容端点
    api_key="sk-xxxx",
)

def ask(system: str, user: str) -> str:
    resp = client.chat.completions.create(
        model="qwen-max",
        messages=[
            {"role": "system", "content": system},
            {"role": "user", "content": user},
        ],
    )
    return resp.choices[0].message.content.strip()

四、实战一：路由式 Agentic RAG

路由式的关键是「先分类、后检索」。让模型只输出知识库类名，再用字典映射到真实的向量索引：

ROUTES = {
    "产品文档": "product_index",
    "技术报错": "error_kb",
    "闲聊": None,          # 不需要检索，直接走模型自身知识
}

def route(question: str):
    label = ask(
        "判断用户问题属于哪类知识库，只返回类名",
        question,
    )
    return ROUTES.get(label.strip())

def routing_rag(question: str) -> str:
    index = route(question)
    if index is None:                       # 闲聊，跳过检索
        return ask("你是贴心助手", question)
    ctx = vector_search(question, index)    # 精准打到对应库
    return ask(f"根据资料回答：{ctx}", question)

这套做法把「检索范围」从写死变成模型决策，跨库问答的准确率通常会明显提升。想快速搭一个带知识库的底座，可以参考 Dify + Ollama 私有知识库教程。

五、实战二：查询分解并行检索

当问题需要多份证据时，先把问题拆开，再用协程并行检索，最后合并成一段上下文交给模型综合：

import asyncio

def decompose(question: str) -> list[str]:
    text = ask(
        "把复杂问题拆成 2-4 个可独立检索的子问题，每行一个",
        question,
    )
    return [q.strip() for q in text.splitlines() if q.strip()]

async def agentic_rag(question: str) -> str:
    subs = decompose(question)
    chunks = await asyncio.gather(*[retrieve(s) for s in subs])  # 并行检索
    merged = "\n".join(chunks)
    return ask(f"综合以下证据回答问题：\n{merged}", question)

合并策略：不要简单拼接

直接把各子问题的检索结果首尾相接，容易让模型在长上下文里「迷路」。更稳的做法是给每段证据打上来源标签（如 [子问题1]），或在合并前先做一次去重。向量检索本身的质量决定了上限，关于 Embedding 选型与语义检索落地，可以看 Embeddings 向量嵌入实战。

六、实战三：Self-RAG 自我反思

反思式把「检索」和「生成」都变成可条件的步骤。两个判断函数让模型自己把关：

def need_retrieval(question: str) -> bool:
    return ask("判断回答该问题是否需要外部知识，只回答 yes 或 no", question).lower() == "yes"

def is_sufficient(answer: str, question: str) -> bool:
    return ask(
        "评估回答是否充分覆盖问题，只回答 yes 或 no",
        f"问题：{question}\n回答：{answer}",
    ).lower() == "yes"

def self_rag(question: str, max_loop: int = 2) -> str:
    ctx = vector_search(question) if need_retrieval(question) else ""
    answer = ask(f"根据资料回答：{ctx}", question)
    for _ in range(max_loop):
        if is_sufficient(answer, question):
            break
        ctx += "\n" + vector_search(question)   # 不够好，再查一轮
        answer = ask(f"根据资料重新回答：{ctx}", question)
    return answer

这种「判断—行动—反思」的闭环，正是 ReAct 智能体 的标配节拍；若想让模型在反思中调用工具，还可叠加 Function Calling 工具调用。

七、落地清单与常见坑

把范式落到生产环境，下面这几条是踩得最多的坑：

八、何时该上 Agentic RAG：一个简单的选型清单

并不是所有问答场景都需要 Agentic RAG。如果你的知识库单薄、问题高度同质（比如只查一份固定 API 文档），传统 RAG 已经够用，强行加智能体只会增加延迟和成本。一条好用的经验法则是：单一知识库、问题平铺 → 传统 RAG 即可；多库并存、需要分类 → 路由式；问题常带「对比 / 步骤 / 原因」→ 查询分解；答案容错率低、必须可解释 → 反思式。先把复杂度压到刚好够用，再按需加码。

还有一个常被忽略的维度是「可观测性」。智能体每多一次检索、多一轮反思，就多一个可能出错的环节。上线后务必把「分类结果、检索命中数、反思轮次」打点记录，否则一旦答错，你根本分不清是路由歪了、检索空了，还是模型飘了。先小流量灰度，再逐步放开，是踩坑最少的一条路；这部分链路完全可以用 GitHub Actions 接自动化回归，把典型问题固化成测试集。

九、把三种范式组合成一个编排器

真实项目里，三种范式往往组合出现：先用路由决定去哪个库，再用分解处理复杂子问题，最后用反思兜底质量。下面是一个最小可用的编排骨架，把前面三段代码串起来，每一步都可独立替换：

def orchestrate(question: str) -> str:
    index = route(question)                 # 路由式：选知识库
    if index is None:
        return ask("你是贴心助手", question)
    subs = decompose(question)              # 分解式：拆子问题
    chunks = [retrieve(s, index) for s in subs]
    ctx = "\n".join(chunks)
    answer = ask(f"根据资料回答：{ctx}", question)
    if is_sufficient(answer, question):     # 反思式：兜底
        return answer
    ctx += "\n" + vector_search(question, index)
    return ask(f"根据资料重新回答：{ctx}", question)

这套编排不追求一次到位，而是把「决策—检索—反思」拆成可独立优化的零件：哪一步弱了，就单独优化那一步。它本质上就是一个用检索替代记忆、用反思替代人工校验的轻量智能体，与 ReAct 智能体 的节拍完全一致。当你把这条链路沉淀进服务，再配合 One API 网关 的配额与限流，一个生产可用的 Agentic RAG 就立起来了。

部署上，把网关、向量库、业务服务都放进同一张 Docker Compose 网络，再用 GitHub Actions 跑自动化发布，就能得到一个稳定可复现的 Agentic RAG 服务。从「被动管道」到「自主决策」，差的不是模型大小，而是这一层会思考的检索编排。