LangChain Agentic RAG：让 Agent 自己决定要不要检索、检索几次

个人主页：
> for_ever_love__ <（欢迎各位大佬莅临😊）

其他栏目:
> 大模型开发从0到1 <

其他栏目:
> iOS项目总结大全 <

其他栏目:
> 我想学python了 <

其他栏目:
> iOS UI <

文章目录

• LangChain Agentic RAG：让 Agent 自己决定要不要检索、检索几次
• • 一、两种 RAG 的结构对比
  • 二、最小实现：把 retriever 包成工具
  • 三、进阶：加上"反思"节点
  • 四、什么时候该用 Agentic RAG
  • 五、成本对比
  • 六、让 Agent 知道"检索不到就别硬找"
  • 七、完整示例：带反思的客服 Agent
  • 八、常见坑
  • 九、小结

LangChain Agentic RAG：让 Agent 自己决定要不要检索、检索几次

传统 RAG 是"问一次，检索一次，答一次"的固定流程。 但很多问题根本不是一次检索能解决的： “对比一下 2023 和 2024 年的差旅制度有什么变化”—— 这需要检索两次，还得分别定位两个版本。

Agentic RAG 把"检索"变成一个工具， 让 Agent 自己决定：要不要查、查什么、查几次、够了没有。

一、两种 RAG 的结构对比

传统 RAG（固定链）：

问题 → 检索 → 拼 prompt → 生成 → 答案

流程写死，无论什么问题都走一遍。

Agentic RAG：

问题 → [Agent 思考] → 需要查吗？
├─ 不需要 → 直接回答（你好、谢谢这类）
├─ 需要 → 调用检索工具 → 结果满意吗？
│ ├─ 不满意 → 换个查询再查
│ └─ 满意 → 生成答案

核心区别：检索变成了 Agent 可支配的一个动作。

二、最小实现：把 retriever 包成工具

from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

@tool
def search_knowledge_base(query: str) –> str:
"""在企业知识库中搜索相关内容。

当需要查找公司制度、流程、产品规格、历史记录等信息时使用。
query 应该是简洁的检索词，不要写完整句子。
如果没找到有用内容，返回空字符串即可。
"""
docs = retriever.invoke(query)
if not docs:
return "没有找到相关内容。"
return "\\n\\n".join(
f"[{i}] {d.page_content}" for i, d in enumerate(docs, 1)
)

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
agent = create_react_agent(
llm,
[search_knowledge_base],
prompt="""你是企业知识库助手。

规则：
1. 需要查资料时必须调用 search_knowledge_base，不要凭记忆回答；
2. 可以多次调用，每次用不同的检索词；
3. 检索不到时就说"没有找到相关信息"，不要编造；
4. 回答时在句末标注 [编号]。
""",
)

result = agent.invoke({"messages": [("user", "对比 2023 和 2024 年的差旅制度变化")]})

Agent 会自己跑出这样的轨迹：

🤔 需要分别查两个版本
🔧 search_knowledge_base("2023 差旅制度")
📄 [1] 2023版：一线城市 500 元…
🤔 还需要 2024 版
🔧 search_knowledge_base("2024 差旅制度")
📄 [1] 2024版：一线城市 600 元…
🤔 信息够了
💬 主要变化：一线城市从 500 元上调到 600 元[2]…

传统 RAG 做不到这一点——它只有一次检索机会。

三、进阶：加上"反思"节点

让 Agent 显式判断"检索到的内容够不够"：

from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

class State(TypedDict):
messages: Annotated[list, add_messages]
documents: list
retries: int

def retrieve(state: State):
question = state["messages"][–1].content
docs = retriever.invoke(question)
return {"documents": docs}

def grade_documents(state: State) –> Literal["generate", "rewrite"]:
"""判断检索结果是否和问题相关"""
docs = state["documents"]
if not docs:
return "rewrite"

# 让 LLM 打分
scores = grade_chain.batch([
{"question": state["messages"][–1].content, "doc": d.page_content}
for d in docs
], config={"max_concurrency": 5})

relevant = [d for d, s in zip(docs, scores) if s == "yes"]
if len(relevant) >= 2:
return "generate"
if state.get("retries", 0) >= 2:
return "generate" # 重试够了，硬着头皮答
return "rewrite"

def rewrite_query(state: State):
"""换个说法再检索"""
new_q = rewrite_chain.invoke({
"question": state["messages"][–1].content
})
return {"messages": [HumanMessage(new_q)],
"retries": state.get("retries", 0) + 1}

def generate(state: State):
ctx = format_docs(state["documents"])
ans = (prompt | llm | StrOutputParser()).invoke({
"context": ctx,
"question": state["messages"][0].content,
})
return {"messages": [AIMessage(ans)]}

graph = StateGraph(State)
graph.add_node("retrieve", retrieve)
graph.add_node("rewrite", rewrite_query)
graph.add_node("generate", generate)

graph.add_edge(START, "retrieve")
graph.add_conditional_edges("retrieve", grade_documents, {
"generate": "generate",
"rewrite": "rewrite",
})
graph.add_edge("rewrite", "retrieve")
graph.add_edge("generate", END)

app = graph.compile()

三个关键节点：

retries 上限很重要，否则会无限循环。

四、什么时候该用 Agentic RAG

判断依据：你的问题是否需要"多步"。 如果 90% 的问题都是单跳问答，别上 Agentic RAG， 那 10% 单独处理就行。

五、成本对比

Agentic RAG 明显更贵更慢。 所以只在需要多跳的场景用。

一个折中方案：路由——简单问题走传统链，复杂问题走 Agent：

def route(question: str):
if is_multi_hop(question): # 规则或轻量分类
return agent.invoke({"messages": [("user", question)]})
return fast_chain.invoke(question)

六、让 Agent 知道"检索不到就别硬找"

这是 Agentic RAG 最容易出问题的地方： Agent 可能连续换 5 个查询词反复搜，最后还是编一个答案。

防护三招：

① 工具返回明确的空结果

@tool
def search_knowledge_base(query: str) –> str:
"""…"""
docs = retriever.invoke(query)
if not docs:
return "知识库中没有找到相关内容。请不要反复重试不同的检索词，" \\
"直接告诉用户没有找到。"
...

在工具返回值里直接下指令，比在 system prompt 里说更有效 （因为它出现在模型"刚看到"的位置）。

② 限制步数

agent.invoke({"messages": [...]}, config={"recursion_limit": 8})

③ 显式的"放弃"出口

prompt = """
…
4. 如果两次检索都没找到，直接回答"没有找到相关信息"，
并说明已尝试过哪些检索词。不要编造，不要继续重试。
"""

七、完整示例：带反思的客服 Agent

from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent

@tool
def search_faq(query: str) –> str:
"""搜索常见问题库。用于查找退款、物流、发票等标准问题的答案。"""
return "\\n".join(d.page_content for d in faq_retriever.invoke(query)) or "无结果"

@tool
def search_order(order_id: str) –> str:
"""根据订单号查询订单状态和物流。订单号形如 SO20240115。"""
return api.query_order(order_id)

@tool
def escalate_to_human(reason: str) –> str:
"""转人工客服。当无法解决用户问题，或涉及退款/投诉时使用。"""
create_ticket(reason)
return "已创建人工工单，客服将在 1 个工作日内联系用户。"

agent = create_react_agent(
ChatOpenAI(model="gpt-4o-mini", temperature=0),
[search_faq, search_order, escalate_to_human],
prompt="""你是电商客服助手。

流程：
1. 先判断问题类型：标准问题 → search_faq；具体订单 → search_order
2. 找不到答案时，最多再换一次检索词
3. 仍然找不到 → 调用 escalate_to_human 转人工
4. 涉及退款、投诉的，直接转人工，不要自行处理

不要编造物流信息、退款金额、政策条款。
""",
)

for chunk in agent.stream(
{"messages": [("user", "我的订单 SO20240115 什么时候到")]},
stream_mode="values",
config={"recursion_limit": 8},
):
...

给 Agent 一个"转人工"的工具是很实用的设计—— 它给模型一个明确的"我搞不定"的出口，比让它硬答好得多。

八、常见坑

九、小结

• Agentic RAG 把检索变成工具，Agent 自己决定查不查、查几次；
• 最小实现：@tool 包住 retriever + create_react_agent， 工具描述要写清楚什么时候用；
• 进阶加反思节点：retrieve → grade → (generate | rewrite → retrieve)， 必须设重试上限；
• 适合多跳问答、跨数据源；单跳问答用传统链更快更省；
• 成本是 2~4 倍，可以用路由做快慢分离；
• 防"反复检索 + 硬编"三招：工具返回值里下指令、限制步数、 给一个 escalate 类的明确出口。

下一篇讲 Corrective RAG——检索质量不好时怎么补救。