第 15 篇结尾留了一扇门："这条从管道到 Agent 的路，正是第 17 篇 Agentic RAG 的全部剧情。"现在兑现。系列两条线在本篇会师：模块二搭好了 Agent 的循环——第 9 篇手写 ReAct、第 10 篇拆解 create_agent；模块三攒齐了检索的家底——第 14 篇嵌入与向量库、第 15 篇六步流水线。会师的产物就是 Agentic RAG，当前生产环境 RAG 的主流形态：检索不再是每问必走的管道，而是工具箱里的一件工具——要不要用、用什么词、用完够不够，全部由模型在运行时决定。
本篇是《AI Agent 开发》系列第 17 篇，代码基于 langchain-core 1.6.x 与 langgraph 1.2.x（2026-10 基线），知识库沿用第 15 篇建好的 Chroma。全文按"分野 → 检索即工具 → 最小升级 → 多工具 → 自纠错 → 多跳 → 成本 → 总结"推进；自纠错一节会回到手搭图，第 8、9 篇的条件边与回边将全部复用。
1、固定管道与 Agentic 的分野
先把两种形态拆干净。第 15 篇的流水线是固定管道：问题进来，检索、生成两步焊死在代码里，一万次提问走同一条路，可以叫它"RAG-forced"——不管问题需不需要，每次必检索。Agentic RAG 把决策权交还给模型：问题先过模型的脑子，由它判断要不要查库、查什么词、查回来的够不够、要不要换个问法再查。
差异逐维看：
| 维度 | 固定管道 | Agentic RAG |
|---|---|---|
| 检索决策 | 无决策，每问必查 | 模型运行时判断 |
| 查询词 | 原问题照搬 | 模型可改写、可拆解 |
| 检索次数 | 恒定一次 | 零到多次 |
| 延迟 | 稳定两段式 | 随决策轮次浮动 |
| token 成本 | 可精确预估 | 随轮次放大 |
落到场景上就是取舍：
| 场景 | 推荐 | 判断依据 |
|---|---|---|
| 客服 FAQ、条款查询 | 固定管道 | 问题高度集中，每问必查，决策是白开销 |
| 问答与闲聊混合的助手 | Agentic | 大量问题根本不需要检索 |
| 调研问答、复杂归因 | Agentic | 需要改写、多跳、自纠错 |
| 高 QPS 低延迟接口 | 固定管道 | 延迟方差比均值更致命 |
判断只看一条："要不要检索"这个问题本身有没有信息量。九成问题都要查库，就别让模型决策；一半问题与库无关，固定管道就是在白白付检索与上下文的税。
2、检索即工具
会师只差一个函数。第 5 篇立过规矩：Agent 世界里的每样能力都是工具，由名字、参数 schema、描述三样东西构成。检索器想进这个世界，就得穿上这层皮——create_retriever_tool 干的就是包裹这一件事。挂载第 15 篇的库并包出工具：
from langchain.tools.retriever import create_retriever_tool
from langchain.embeddings import init_embeddings
from langchain_chroma import Chroma
vectorstore = Chroma(
collection_name="my_notes",
persist_directory="./kb", # 第 15 篇建好的库
embedding=init_embeddings("openai:text-embedding-3-small"),
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})
search_notes = create_retriever_tool(
retriever,
name="search_notes", # 唯一、无空格、≤64 字符
description=(
"在个人技术笔记知识库中检索，返回相关片段。"
"问题涉及笔记里的技术细节、配置、代码用法时调用；"
"闲聊、翻译、常识问题不要调用。"
),
)
解读三个要点。其一，retriever 是第 15 篇的原样复用，一行没改，变化全在包装层。其二，工具执行时内部就是 retriever.invoke(query)，返回的 Document 列表拍平成纯文本回填给模型；想让结果带出处，可以传 document_prompt 参数，用模板把每块 metadata 里的 source 拼进输出。其三也是最要紧的：description 是模型决定用不用这件工具的唯一依据——模型看不见里面是 Chroma 还是 FAISS，也看不见 k 是多少，它眼里只有名字、描述和一个 query 字符串的参数 schema。
所以描述要写双向：库里有什么、什么问题该查、什么问题别查。反面教材是"检索文档"这四个字：没给任何使用时机，遇到"你好"它可能也去查一遭，遇到真正该查的反而犹豫。name 也别随手起：它会出现在模型的调用决策、日志与 LangSmith trace 里，search_notes 这种动宾结构一眼能懂，比 retriever_1 强得多。完整参数见 create_retriever_tool 参考。
3、最小 Agentic RAG
工具备好，把第 15 篇的问答链换成第 10 篇的 create_agent——五行核心代码，五分钟完成升级：
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
llm = init_chat_model("openai:gpt-5")
agent = create_agent(
llm,
tools=[search_notes],
system_prompt=(
"你是技术笔记助手：问题可能涉及笔记内容时，"
"先调用 search_notes 检索再回答；笔记里没有的直说不知道。"
),
)
def ask(q):
result = agent.invoke({"messages": [{"role": "user", "content": q}]})
calls = [c["name"] for m in result["messages"] if m.type == "ai"
for c in m.tool_calls] # 抽出全部工具调用
print(f"问题: {q}")
print(f"工具轨迹: {calls if calls else '（未调用）'}")
print(f"回答: {result['messages'][-1].text}\n")
解读：create_agent 内部就是第 9 篇手写的那张图，检索工具在其中与任何普通工具平起平坐；ask 把每轮模型的 tool_calls 抽出来打印，用它观察两种典型对话。先看"不需要检索"的一问：
问题: 把"检索增强生成"翻译成英文
工具轨迹: （未调用）
回答: Retrieval-Augmented Generation。
模型判断这是翻译任务，与笔记无关，跳过检索直接作答，全程一次模型调用。再看"需要检索"的一问：
问题: asyncio.wait 的 return_when 有哪些取值？
工具轨迹: ['search_notes']
回答: 可取 FIRST_COMPLETED、FIRST_EXCEPTION、ALL_COMPLETED，
默认 ALL_COMPLETED，出自 notes/python-async.md。
这次模型先调 search_notes，工具把四个片段回填进历史，模型再依据片段作答——两次模型调用加一次检索。对照第 15 篇：前一问在固定管道里同样会照查一次库，白付一次检索和四个无关片段的上下文；后一问两种形态殊途同归。要不要检索，第一次由模型说了算，这就是五分钟升级换来的全部。顺带一提：这里的 agent 没配 checkpointer，ask 每次都是独立会话；要做连续追问，挂上第 12 篇的 InMemorySaver 与 thread_id 即可，循环本身不用改一行。
4、多工具并存与路由
检索一旦是工具，工具箱就能继续装东西——这正是第 15 篇预告的"知识问答 Agent"雏形：笔记问题走检索、算账走计算器、业务数据查订单表，模型按问题性质选路。三件工具并存：
from langchain_core.tools import tool
import numexpr # uv add numexpr
@tool
def calc(expression: str) -> str:
"""计算算术表达式，支持 + - * / 与括号，如 '880+1299'。"""
return str(numexpr.evaluate(expression)) # 安全求值，不用 eval
@tool
def query_orders(status: str) -> str:
"""按状态查询订单，返回订单号、客户与金额列表。"""
data = { # 演示数据，生产换 SQL
"待支付": "ORD-1042 张三 880 元；ORD-1055 李四 1299 元",
"已发货": "ORD-1038 王五 640 元",
}
return data.get(status, f"未知状态：{status}")
kb_agent = create_agent(
llm,
tools=[search_notes, calc, query_orders],
system_prompt=(
"你是团队内部助手：笔记相关查 search_notes，"
"计算用 calc，查订单用 query_orders。"
),
)
解读：三件工具对模型完全平级，路由依据仍然只有各自的描述；系统提示词里的分工说明是路标，最终选择权在模型。带三个问题跑 kb_agent：
"return_when 有哪些取值" → search_notes
"待支付订单一共多少钱" → query_orders → calc
"把'协程'翻译成英文" → 不调工具，直接回答
第一问走检索，第三问谁都不找，第二问最值得看：模型先查回两笔待支付订单，再自己拼出 880+1299 调计算器——两件工具接力，中间没有任何硬编码编排。这也是工具规模的分水岭：工具一多选择质量会掉，收拢的办法是第 10 篇的 agent as tool 子代理，第 24 篇展开。
5、自纠错检索循环
到目前为止，检索对不对全凭模型一次查准。但向量检索有个软肋：问题与库的用词对不上时，召回质量塌方——用户问"绿色线程"，笔记里通篇写"协程"，嵌入模型对这种别名映射并不稳定。深水区的答案是自纠错：检索结果先质检，不合格就改写重查，合格才生成。这套模式学名 Corrective RAG，出自 2024 年的 CRAG 论文，近亲 Self-RAG 论文把反思做进了模型训练本身。
这次不用 create_agent，回到手搭图。理由正是第 10 篇四条判断之首：质检要插在循环中段，还要有一条"不合格就退回去"的条件回边，标准 ReAct 循环装不下。State 与打分节点先行：
5.1、检索结果打分
State 只要四个普通键，不需要 add_messages——这张图不靠消息历史传递，query 单独拎出来是因为改写节点只更新它：
from typing import Literal
from typing_extensions import TypedDict
from pydantic import BaseModel, Field
class CragState(TypedDict):
question: str # 用户原始问题
query: str # 当前查询词，可能被改写过
documents: list # 检索回的 Document 列表
answer: str # 最终回答
class GradeDocuments(BaseModel):
"""对单篇文档与问题的相关性打分。"""
binary_score: str = Field(description="相关为 'yes'，不相关为 'no'")
grader = llm.with_structured_output(GradeDocuments) # 第 4 篇
GRADE_PROMPT = "你是相关性评审，判断文档对回答问题是否有用，只回 yes 或 no。"
def grade_documents(state: CragState) -> dict:
keep = []
for doc in state["documents"]:
verdict = grader.invoke([
{"role": "system", "content": GRADE_PROMPT},
{"role": "user", "content":
f"问题：{state['question']}\n\n文档：{doc.page_content}"},
])
if verdict.binary_score == "yes":
keep.append(doc)
return {"documents": keep} # 只留相关的，覆盖式更新
解读：打分器是 with_structured_output 包出的"只回 yes 或 no"的专职调用，质检独立于生成，比让主模型一边读文档一边自评更可控。逐篇打分、逐篇过滤，grade 节点只干过滤这一件事；"去重查还是去生成"的判断留给路由。
5.2、改写回边与组装
其余三个节点都很薄，一次写完：
def retrieve(state: CragState) -> dict:
query = state.get("query") or state["question"] # 首轮用原问题
return {"documents": retriever.invoke(query), "query": query}
REWRITE_PROMPT = ("把问题改写成更适合向量检索的查询："
"口语与别名换成通用术语，保留原意，只输出查询本身。")
def rewrite_query(state: CragState) -> dict:
resp = llm.invoke([
{"role": "system", "content": REWRITE_PROMPT},
{"role": "user", "content": state["question"]},
])
return {"query": resp.text}
GEN_PROMPT = "依据资料回答问题；资料里没有的直说不知道。\n\n资料：\n{context}"
def generate(state: CragState) -> dict:
context = "\n\n".join(d.page_content for d in state["documents"])
resp = llm.invoke([
{"role": "system", "content": GEN_PROMPT.format(context=context)},
{"role": "user", "content": state["question"]},
])
return {"answer": resp.text}
解读：retrieve 首轮拿 question 当查询词，改写轮拿更新过的 query——回边转一圈，状态里变的只有这个键；generate 是第 15 篇纪律条款的原样复刻。最后按第 8、9 篇的手法组装：
from langgraph.graph import StateGraph, START, END
def decide(state: CragState) -> Literal["generate", "rewrite"]:
if state["documents"]: # 还有相关文档 → 生成
return "generate"
return "rewrite" # 全不相关 → 改写重查
builder = StateGraph(CragState)
builder.add_node("retrieve", retrieve)
builder.add_node("grade", grade_documents)
builder.add_node("rewrite", rewrite_query)
builder.add_node("generate", generate)
builder.add_edge(START, "retrieve")
builder.add_edge("retrieve", "grade")
builder.add_conditional_edges("grade", decide, ["generate", "rewrite"])
builder.add_edge("rewrite", "retrieve") # 条件回边：退回去重查
builder.add_edge("generate", END)
crag = builder.compile()
print(crag.invoke(
{"question": "绿色线程在 Python 里怎么用？"})["answer"])
解读：这张图与第 9 篇的 ReAct 图同构但职责重排——回边从"工具执行完回到模型"变成"质检不合格退回检索"，路由依据从 tool_calls 变成文档列表空不空。给四个节点各加一行 print，运行轨迹如下：
retrieve: query="绿色线程在 Python 里怎么用？" → 4 篇
grade: 0 篇相关（笔记通篇用"协程"，没有"绿色线程"字样）
rewrite: query="Python 协程 asyncio 使用方法"
retrieve: → 4 篇；grade: 留下 3 篇相关
generate: 协程用 async def 定义，asyncio.run 启动事件循环……（出自 notes/python-async.md）
第一跳全军覆没，条件回边把改写后的查询送回 retrieve，第二跳命中，图终止在 generate。固定管道遇到这种问题只能硬拿四篇无关文档生成，正是幻觉的温床；CRAG 多付几轮便宜的打分与改写调用，换回"查不到就换姿势再查"的韧性。注意这张图的回边没有上限，兜底只有 recursion_limit（默认 25）——生产上应该显式熔断，练习里你来加。完整实现可对照官方 Agentic RAG 教程。
6、多跳检索
复杂问题往往一步查不全："A 公司的投资方的 CEO 是谁"得先查 A 的投资方，再拿结果去查 CEO，第二次的查询词由第一次的结果拼出来。固定管道一次检索的天花板就卡在这里；而 Agent 循环天然支持多跳——模型调一次工具看一次结果，不够就再调。第 9 篇结尾那道"《三体》的作者拿过哪个科幻最高奖"的练习，你已经手做过一次多跳。
把知识库换成公司内部 wiki（照第 2 节包一个 search_wiki 即可，代码略），跑一个两跳问题：
问题: A 项目用的消息队列是哪个？谁负责它的运维？
模型调用: search_wiki(query="A 项目 技术栈 消息队列")
工具返回: ……A 项目核心链路使用 RocketMQ……
模型调用: search_wiki(query="RocketMQ 运维负责人")
工具返回: ……中间件组王五负责 RocketMQ 集群……
回答: A 项目使用 RocketMQ，运维负责人是中间件组的王五。
解读：第二跳的查询词里，"RocketMQ" 来自第一跳的工具返回——问题拆解与接续由模型在循环里自己完成，不需要任何多跳专用代码。代价同样清晰：每一跳都是一轮完整的模型调用加一次检索，成本按跳数线性涨。多跳质量取决于两点：模型会不会拆问题，弱模型常在第二跳丢上下文；工具返回的信息够不够它构造下一跳，这也是第 2 节建议给检索结果带出处的原因。recursion_limit 同时是跳数的天花板，默认 25 对绝大多数多跳问题绰绰有余。
再多想一步：检索工具旁边再加一件 web 搜索工具，Agent 就能在"库里有"与"库外"之间自主切换——库里查不到的事实去网上核实，两边的来源都进答案。这正是 Deep Research 类产品的最小内核，第 26 篇会正面展开。
7、成本与延迟的账
天下没有免费的决策。Agentic 的灵活性每一分都来自额外的模型调用——固定管道一次问答是"一次嵌入 + 一次生成"；Agentic 带检索的问答至少"两次模型调用 + 一次嵌入"，且每轮模型调用都全量重发历史与工具 schema。单次问答的开销构成：
| 开销项 | 固定管道 | Agentic（带一次检索） |
|---|---|---|
| 模型调用 | 1 次 | ≥2 次（决策 + 生成） |
| 嵌入调用 | 1 次 | 1 次（工具内部） |
| 输入 token | 问题 + 资料 | 逐轮累加历史与工具结果 |
| 延迟 | 稳定两段式 | 随轮次浮动，多跳可达数倍 |
工程上真正扎手的是方差：固定管道的 p99 与 p50 接近，Agentic 的尾延迟由最坏轮次决定，对延迟敏感的接口，这比均值更麻烦。什么时候回归固定管道？三个信号：
- trace 统计显示九成以上问题都调了检索工具，决策成了纯开销
- 延迟 SLA 卡死，流式场景容不下"先想想要不要查"的那一拍
- 模型能力吃紧，弱模型的错误路由比不路由更伤答案
生产上更常见的不是二选一，而是混合形态：前置一个便宜的路由（规则或小模型分类）判断"要不要检索"，简单问题直进固定管道，复杂问题才放行给 Agentic，决策成本付在刀刃上。多跳与长对话场景下，第 10 篇提过的 SummarizationMiddleware 压历史，是省 token 的第一件工具。
口诀：灵活性付的是延迟与 token 的税。先把固定管道跑上线，用真实 trace 证明"模型决策"物有所值，再升级 Agentic。
8、总结
会师完成。本篇主线一句话：create_retriever_tool 把第 15 篇的检索器包成工具，第 9、10 篇的 Agent 循环接住它，RAG 从"每问必走的管道"变成"模型按需取用的能力"。四块内容各归其位：create_agent 一行完成最小升级；多工具并存让检索、计算、查库同场路由；CRAG 用条件回边实现"打分 → 改写 → 重查"的自纠错；多跳是循环的自然副产品。最后一节算了账——决策轮次就是延迟与 token，混合路由是生产上最常见的折中。
练习：给第 5 节的 CRAG 图加"最多重查 2 次"的熔断。三步：State 加 retries 计数；rewrite 节点里加一；decide 改成三分支——有相关文档去 generate，没有且未超限去 rewrite，超限也去 generate 并让模型诚实说明资料不足。小心一条新路径：超限进 generate 时 documents 是空列表，拼装上下文的代码要兜得住。跑通后故意问一个库里绝没有的问题，验证它两次重查后果断放弃，而不是转圈到 recursion_limit。
对于准备上生产知识库问答的同学，建议的路径是：固定管道先行，摸清真实问题分布；trace 里出现大量"无效检索"或"一查不中"时，再按本篇顺序升级——先 Agentic 化解决"要不要查"，再上 CRAG 解决"查不对"。下一篇第 18 篇：GraphRAG 与知识图谱，当文档之间的关系比内容本身更重要时，检索换一种组织方式。
