智能体驱动的RAG：让大语言模型决定何时以及从何处检索信息。
为什么静态RAG在多跳查询和时效性要求高的问题上会失效，以及LangGraph智能体如何在Qdrant与网络搜索之间进行路由，并附相关限制与注意事项。
来源: Your RAG Pipeline Is Lying to You. The Problem Is Not the Embeddings — Agentic RAG
“RAG基本上已经是一个解决的问题了。”在团队评审中常会听到这样的说法，而诚实的回答是：这取决于你对检索的理解。任何在实际环境中使用RAG流程处理医疗记录摘要、自然语言转SQL的工作流或多智能体研究系统的人，往往都会遇到同样的瓶颈。
这个瓶颈并非出在嵌入模型上，而在于那种认为检索只需在某个固定步骤中针对单一数据源进行一次的假设。
静态RAG（将文档嵌入，获取前k个结果，再将它们传递给大语言模型）擅长处理基于稳定内部知识的简单、独立的问题。一旦问题变得需要多步推理、对时间敏感，或既需要内部文档又需要实时数据，这种静态流程就会开始产生错误的答案。这样的流程甚至可能根据过时的内部指南推荐临床操作步骤，而实际上只需花半分钟在网络上搜索就能了解到相关变更。
更好的嵌入技术也无法解决这个问题。真正有效的办法是让大语言模型自行决定何时以及从何处获取信息。
“一次检索，永久回答”的弊端
想象一下，有一位图书管理员，他只能在你说完话之前就走到同一个书架前进行回答，而且每次都是同一个书架。
只要询问“二甲双胍的副作用有哪些？”，系统就能立即给出答案。但现在试问一个更难的问题：FDA最近针对二甲双胍与新型GLP-1类药物指出了哪些药物相互作用，现有的患者治疗方案是否涵盖了这些情况？这需要查阅内部文件和实时监管数据，还需分析两者之间的差异。静态RAG仅从单一来源获取信息，而大语言模型则会对收到的内容进行简单处理。
这并非假设情况。在某项针对医疗流程的内部评估中[1]，对于简单问题，静态RAG的答案相关性可达89%，但当问题需要将内部临床文档与外部治疗指南相互对照时，这一比例降至61%。初步评估往往因为答案看似合理就予以通过，但实际上这些答案早已过时。
这个解决方案说起来简单，但实现起来却很复杂：信息检索不再是一个固定步骤，而变成了一种决策过程。
代理式RAG的含义
在代理式RAG中，大语言模型不会在预设时间点自动进行信息检索。它将检索视为一种工具，类似于计算器或代码执行器，只有在判断现有证据不足时才会使用它。由此会产生三种静态流程无法实现的行为。
- 条件性检索。如果上下文已能回答问题，代理就会直接跳过检索步骤。
- 多轮检索。代理会先获取某些信息并阅读，发现还有缺失的部分后便再次去获取，每轮检索都会受到前一轮结果的制约。这正是人们进行研究的方式。
这些理念其实并不新鲜。Yao及其同事在2022年发表的ReAct论文提出了将推理步骤与工具调用相互交织的模式，而智能体RAG则是将ReAct应用到信息检索领域[2]。
近期的变化在于相关工具的改进。LangGraph终于使得带有分支的状态循环成为可能。Tavily则提供了一种专为智能体设计的网页搜索API，能够返回可直接供大语言模型使用的内容，据称其每月可处理超过1亿次请求，50%请求的搜索延迟为180毫秒[3]。详细信息可在Tavily 101文章中找到。
三个核心组成部分
不妨把侦探而非图书管理员放在心上。图书管理员是按指令取书，而侦探则会阅读手头资料，判断缺少什么，再去获取，再次阅读并持续下去，直到能得出可成立的结论。这个流程包含三个组成部分。
- 路由器负责确定查询需要何种检索方式。大语言模型会分析问题，判断现有上下文是否足够、是否需要文档、是否需要网络信息，或是三者皆需。
- 检索工具就是证据来源。最少需要两种工具：一种用于搜索内部文档的向量存储，另一种用于搜索网络上的实时信息。这两种都是普通函数，大语言模型通过查询字符串来调用它们。
可运行的实现方案
这是 LangGraph 中的一个混合式智能体，能够在 Qdrant 向量存储与 Tavily 网络搜索之间切换。它会自行选择工具，可以按顺序调用这两种工具，若首个搜索结果无法为答案提供依据，则会再次检索。Qdrant 部分遵循 LangChain 的集成指南。
from typing import Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langchain_qdrant import QdrantVectorStore
from langchain_openai import OpenAIEmbeddings
from tavily import TavilyClient
from qdrant_client import QdrantClient
import operator
from typing import TypedDict, Sequence
# --- State definition ---
class AgentState(TypedDict):
messages: Annotated[Sequence, operator.add]
retrieval_count: int # guard against infinite loops
# --- Tool 1: Vector store retrieval (internal docs) ---
qdrant_client = QdrantClient(url="http://localhost:6333")
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vector_store = QdrantVectorStore(
client=qdrant_client,
collection_name="internal_docs",
embedding=embeddings,
)
@tool
def retrieve_from_docs(query: str) -> str:
"""
Search internal document knowledge base.
Use for questions about internal policies, protocols, or proprietary knowledge.
"""
results = vector_store.similarity_search(query, k=4)
if not results:
return "No relevant internal documents found for this query."
return "\n\n---\n\n".join([
f"[Source: {doc.metadata.get('source', 'internal')}]\n{doc.page_content}"
for doc in results
])
# --- Tool 2: Web retrieval (live data) ---
tavily = TavilyClient(api_key="YOUR_TAVILY_API_KEY")
@tool
def retrieve_from_web(query: str) -> str:
"""
Search the live web for current information.
Use for recent news, regulatory updates, or anything that may have changed
since the internal knowledge base was last updated.
"""
results = tavily.search(
query=query,
max_results=5,
search_depth="advanced", # deep retrieval for better accuracy
)
if not results.get("results"):
return "No relevant web results found."
return "\n\n---\n\n".join([
f"[Source: {r['url']}]\n{r['content']}"
for r in results["results"]
])
# --- Agent node ---
tools = [retrieve_from_docs, retrieve_from_web]
tool_node = ToolNode(tools)
llm = ChatOpenAI(model="gpt-4o", temperature=0).bind_tools(tools)
SYSTEM_PROMPT = """You are a research assistant with access to two retrieval tools:
- retrieve_from_docs: internal document knowledge base (policies, protocols, proprietary data)
- retrieve_from_web: live web search (recent news, regulatory updates, real-time information)
For any question:
1. Decide whether you need to retrieve information or can answer from context.
2. If retrieval is needed, choose the right source - or call both if the question spans internal and live knowledge.
3. After retrieving, check if the evidence is sufficient. If not, retrieve again with a refined query.
4. Always cite your sources in the final answer.
Do not hallucinate. If you cannot find sufficient evidence, say so explicitly."""
def call_agent(state: AgentState) -> AgentState:
messages = [SystemMessage(content=SYSTEM_PROMPT)] + list(state["messages"])
response = llm.invoke(messages)
return {
"messages": [response],
"retrieval_count": state.get("retrieval_count", 0)
}
def update_retrieval_count(state: AgentState) -> AgentState:
return {
"messages": [],
"retrieval_count": state.get("retrieval_count", 0) + 1
}
# --- Routing logic ---
MAX_RETRIEVALS = 3 # prevent infinite loops
def should_continue(state: AgentState) -> Literal["tools", "end"]:
last_message = state["messages"][-1]
# If the LLM called a tool and we haven't hit the loop guard, continue
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
if state.get("retrieval_count", 0) < MAX_RETRIEVALS:
return "tools"
return "end"
# --- Build the graph ---
graph_builder = StateGraph(AgentState)
graph_builder.add_node("agent", call_agent)
graph_builder.add_node("tools", tool_node)
graph_builder.add_node("count_retrieval", update_retrieval_count)
graph_builder.add_edge(START, "agent")
graph_builder.add_conditional_edges(
"agent",
should_continue,
{"tools": "tools", "end": END}
)
graph_builder.add_edge("tools", "count_retrieval")
graph_builder.add_edge("count_retrieval", "agent")
agent = graph_builder.compile()
# --- Run it ---
if __name__ == "__main__":
result = agent.invoke({
"messages": [
HumanMessage(content=(
"What does our current patient protocol say about GLP-1 usage, "
"and are there any FDA safety updates issued in 2025 we should be aware of?"
))
],
"retrieval_count": 0,
})
print(result["messages"][-1].content)
有些细节值得注意。
路由逻辑中的 MAX_RETRIEVALS 限制并非装饰性设置。如果没有它，智能体在面对一个模棱两可的问题时可能会进行多达15次检索，从而浪费资源与时间。通常三次检索就足够了；如果不够，说明问题范围过大或工具返回的结果质量不佳。
状态中的 retrieval_count 字段使得循环能够被中断。LangGraph的 StateGraph 会在每次节点转换时都保留该值，因此它会在整个对话过程中持续存在，而不仅仅局限于单个循环。
系统提示同样非常重要。明确告知模型何时使用何种工具，而不仅仅是说明工具的存在，这样才能区分出那些会合理选择工具的智能体与那些出于懒惰而总是调用 retrieve_from_web 的智能体。
当智能体RAG反而让情况更糟时
它并非取代静态RAG，而是针对特定问题场景替代后者，但在四种情况下会适得其反。
- 延迟是主要限制因素。每次循环都会多一次模型调用以及一次工具调用。对于那些要求p95延迟必须控制在1秒以内的基础问答助手而言，可能需要循环两到三次的智能体方案根本不可行。经过良好调优的静态检索器速度更快且已足够使用。
- 知识库应当稳定且自包含。如果用户提出的所有问题都能从向量存储中准确获取答案，那么引入这种智能体方案只会徒增复杂性。首先使用RAGAS（相关论文）测试静态处理流程，若上下文召回率高于85%且答案准确性高于90%，则使用智能体封装层只会带来额外开销。
当知识来源于多个渠道且其中部分内容会发生变化时，应选择代理型RAG。可以先从静态RAG开始，一旦RAGAS评分显示出该设计存在的特定缺陷，再考虑升级。
从哪里开始
最小的步骤，耗时不到五分钟：使用 from langchain_community.tools.tavily_search import TavilySearchResults 将一个 Tavily 工具添加到现有的 LangChain 智能体中。此时还不需要 LangGraph。先暴露该工具，编写说明何时调用它的系统提示词，运行五个测试查询，观察它会选择什么结果。
更高阶的步骤，大约需要一个下午的时间：将上面的 LangGraph 示例指向你自己的向量存储，并保留这两个工具。最初先不添加检索计数器，看看智能体会多频繁地循环，之后再添加它。这种对比能带来很多启示。
高级模式：在每个工具处理结果之后、智能体下次响应之前插入一个评分步骤。首先通过一个规模更小、成本更低的LLM来判断所获取的上下文是否足够，只有得到肯定答复后，主模型才会生成回复。这一步骤将上下文调用的标准从“测试中足够好”提升到“生产环境中可靠”。
占比高出12%的工具路由错误是导致生产环境调试时间消耗最多的原因。应尽早对这类错误进行监控，让评分节点在用户发现之前就将其捕获。
参考资料
[1] 使用RAGAS 0.1.x对200个精心挑选的临床查询在医疗领域RAG流程上进行的内部评估，不属于已发表的研究。
[2] Yao等人（2022），《ReAct：语言模型中的推理与行动协同》，arXiv:2210.03629。
[3] Tavily博客，《Tavily 101》，数据为2026年5月产品页面上标注的延迟与处理量数值。
相关阅读
- 智能检索与充足上下文：当RAG信息不足时 ——传统RAG往往过早给出答案。通过Acme案例，学习如何规划搜索、筛选信息来源、评估证据并重新检索，直至获得足够上下文。
- Chroma Context-1：一款20B规模的检索子代理，助力更高效、成本更低的智能RAG系统 ——介绍Chroma的20B Context-1如何作为仅负责检索的子代理自行筛选上下文、其训练方式、缺少该组件时的缺陷以及存在的局限。