传统 RAG 的天花板
Retrieval-Augmented Generation（RAG）自 2020 年被提出以来，已经成为企业 AI 应用的标配架构。核心思路简单直接：把用户问题作为查询，从知识库中检索相关文档，然后把文档作为上下文送入 LLM 生成答案。
这个”检索→生成”的线性管道在简单场景下效果不错，但在实际企业部署中频繁碰壁：
问题 1：单次检索的召回率瓶颈
用户的问题往往不是”一个查询就能找到答案”的。比如：
“我们公司的 Q1 营收比竞争对手 A 高多少？各产品线的贡献比例分别是多少？”
这个问题至少需要三次检索：公司 Q1 营收、竞争对手 A 的 Q1 营收、各产品线的营收分布。传统 RAG 的单次检索根本无法覆盖。
问题 2：查询与文档的语义鸿沟
用户用自然语言提问，但文档里的表述可能完全不同。比如用户问”怎么退款”，但文档中写的是”订单撤销流程”。向量相似度可能不够高，导致检索不到正确文档。
问题 3：无法验证检索质量
传统 RAG 不知道检索到的文档是否真的相关。如果检索结果质量差，LLM 要么基于错误信息生成幻觉答案，要么拒绝回答——两种结果都不理想。
问题 4：无法跨源整合
企业知识通常分散在多个系统中：文档库、数据库、API、Wiki、工单系统。传统 RAG 通常只连接一个向量数据库，无法在多个数据源之间协调检索。
Agentic RAG 的出现正是为了突破这些天花板。
Agentic RAG 的核心架构
Agentic RAG 的核心思想是：用 AI Agent 替代固定的检索管道，让 Agent 自主决定何时检索、检索什么、以及是否需要继续检索。
架构从线性管道变为决策循环：
传统 RAG:
Query → Retriever → Generator → Answer
Agentic RAG:
Query → Agent ──→ 分析问题
├──→ 改写查询
├──→ 选择数据源
├──→ 执行检索
├──→ 评估结果质量
├──→ 需要更多信息? → 回到改写查询
└──→ 信息充分 → 生成答案核心组件
1. 查询分析器（Query Analyzer）
Agent 首先分析用户问题的复杂度，决定检索策略：
def analyze_query(query: str) -> QueryPlan:
"""Agent 分析问题并生成检索计划"""
# 判断问题类型
# - 简单事实查询 → 单次检索
# - 多维度分析 → 分解为子查询
# - 比较型问题 → 并行检索多个实体
# - 时序型问题 → 按时间范围分段检索
pass2. 查询改写器（Query Rewriter）
当初次检索结果不理想时，Agent 会自动改写查询：
def rewrite_query(original_query: str, retrieval_results: list, feedback: str) -> str:
"""基于检索结果和反馈改写查询"""
# 策略 1: 扩展关键词（同义词替换）
# 策略 2: 缩小范围（添加过滤条件）
# 策略 3: 分解子查询（拆分复合问题）
# 策略 4: 切换数据源（从文档库切换到数据库）
pass3. 检索路由器（Retrieval Router）
根据查询特征选择最合适的数据源和检索方式：
| 查询类型 | 数据源 | 检索方式 |
|---|---|---|
| 概念性问题 | 文档库 | 向量检索 |
| 精确数据查询 | 数据库 | SQL/API |
| 最新信息 | 搜索引擎 | 网页检索 |
| 流程性问题 | Wiki/知识库 | 关键词 + 向量混合 |
| 跨域问题 | 多源 | 并行检索 + 合并 |
4. 结果评估器（Result Evaluator）
每次检索后，Agent 评估结果是否足够回答问题：
def evaluate_results(query: str, results: list[Document]) -> EvalResult:
"""评估检索结果的充分性"""
return EvalResult(
relevance_score=0.85, # 结果与问题的相关性
coverage_score=0.60, # 覆盖了问题的多少方面
confidence=0.72, # Agent 对答案质量的信心
missing_aspects=["竞品数据", "时间范围确认"],
should_continue=True # 是否需要继续检索
)五种核心设计模式
模式 1：自适应检索（Adaptive Retrieval）
最基础的模式。Agent 根据问题复杂度决定检索次数和深度：
用户: "公司的休假政策是什么？"
Agent 判断: 简单事实查询 → 单次检索即可
执行: 检索 1 次 → 生成答案
用户: "对比我们和竞品在亚太市场的定价策略差异"
Agent 判断: 复杂对比查询 → 需要多次检索
执行: 检索 4 次（我方定价、竞品定价、亚太市场报告、行业基准）→ 整合生成模式 2：查询分解（Query Decomposition）
将复杂问题分解为多个简单子查询，分别检索后合并：
# 原始问题: "2025年我们的客户流失率和行业平均相比如何？主要流失原因是什么？"
sub_queries = [
"2025年公司客户流失率数据",
"2025年SaaS行业平均客户流失率",
"客户流失原因分析报告",
"客户反馈和投诉汇总"
]
# 并行检索所有子查询
results = await asyncio.gather(*[retrieve(q) for q in sub_queries])
# 合并结果后生成综合答案
answer = generate(original_query, merged_results)模式 3：反思式检索（Reflective Retrieval）
Agent 生成初步答案后，自我审查是否存在信息缺口或不一致：
Step 1: 检索 → 生成初步答案
Step 2: 自我审查
- "答案中提到了 Q3 数据，但用户问的是 Q4" → 补充检索
- "引用了 2023 年的报告，可能有更新版本" → 检索最新版本
- "提到了两个矛盾的数字" → 交叉验证
Step 3: 修正答案
Step 4: 再次审查 → 满意 → 输出最终答案模式 4：工具增强检索（Tool-Augmented Retrieval）
Agent 不只是检索文档，还能调用工具获取实时数据：
tools = [
VectorSearchTool(index="company_docs"),
SQLQueryTool(database="analytics_db"),
WebSearchTool(engine="tavily"),
APICallTool(endpoints=["crm_api", "erp_api"]),
CalculatorTool()
]
# Agent 自主选择工具组合
# 例如: 先从文档库获取产品规格，再从数据库查询销售数据，
# 然后调用计算器计算增长率模式 5：多 Agent 协作检索（Multi-Agent Retrieval）
对于超复杂的查询，多个专业 Agent 分工协作：
┌─── 财务 Agent ──→ 检索财报数据
用户问题 → 编排 Agent ──┤─── 市场 Agent ──→ 检索市场分析
├─── 技术 Agent ──→ 检索技术文档
└─── 合规 Agent ──→ 检查数据使用合规性
↓
汇总 Agent → 整合所有结果 → 生成报告用 LangGraph 实现 Agentic RAG
LangGraph 是实现 Agentic RAG 最成熟的框架，原生支持循环图和状态管理：
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
class RAGState(TypedDict):
query: str
sub_queries: list[str]
retrieved_docs: list[dict]
evaluation: dict
answer: str
iteration: int
def analyze_query(state: RAGState) -> RAGState:
"""分析问题并生成子查询"""
sub_queries = llm.invoke(
f"将以下问题分解为独立的检索子查询:\n{state['query']}"
)
return {"sub_queries": sub_queries, "iteration": 0}
def retrieve(state: RAGState) -> RAGState:
"""执行检索"""
docs = []
for sq in state["sub_queries"]:
results = vector_store.similarity_search(sq, k=3)
docs.extend(results)
return {"retrieved_docs": docs}
def evaluate(state: RAGState) -> RAGState:
"""评估检索质量"""
eval_result = llm.invoke(
f"评估以下检索结果是否足够回答问题:\n"
f"问题: {state['query']}\n"
f"检索结果: {state['retrieved_docs']}"
)
return {"evaluation": eval_result}
def should_continue(state: RAGState) -> str:
"""决定是否继续检索"""
if state["evaluation"]["confidence"] > 0.8:
return "generate"
if state["iteration"] >= 3:
return "generate"
return "rewrite"
def rewrite(state: RAGState) -> RAGState:
"""改写查询"""
new_queries = llm.invoke(
f"检索结果不够充分，请改写查询:\n"
f"原始问题: {state['query']}\n"
f"已有结果: {state['retrieved_docs']}\n"
f"缺失信息: {state['evaluation']['missing']}"
)
return {
"sub_queries": new_queries,
"iteration": state["iteration"] + 1
}
def generate(state: RAGState) -> RAGState:
"""生成最终答案"""
answer = llm.invoke(
f"基于以下检索结果回答问题:\n"
f"问题: {state['query']}\n"
f"参考资料: {state['retrieved_docs']}"
)
return {"answer": answer}
# 构建图
graph = StateGraph(RAGState)
graph.add_node("analyze", analyze_query)
graph.add_node("retrieve", retrieve)
graph.add_node("evaluate", evaluate)
graph.add_node("rewrite", rewrite)
graph.add_node("generate", generate)
graph.set_entry_point("analyze")
graph.add_edge("analyze", "retrieve")
graph.add_edge("retrieve", "evaluate")
graph.add_conditional_edges("evaluate", should_continue, {
"generate": "generate",
"rewrite": "rewrite"
})
graph.add_edge("rewrite", "retrieve")
graph.add_edge("generate", END)
app = graph.compile()性能与成本权衡
延迟分析
| 环节 | 传统 RAG | Agentic RAG（平均 2.3 轮） |
|---|---|---|
| 查询分析 | 0ms（无） | 200-500ms |
| 检索 | 100-300ms × 1 | 100-300ms × 2.3 |
| LLM 生成 | 1-3s × 1 | 1-3s × 3.3 |
| 总延迟 | 1.5-3.5s | 4-12s |
端到端延迟增加约 3-4 倍。对于实时对话场景，可以用流式输出缓解等待感。
成本控制策略
- 设置迭代上限：限制最大检索轮数（推荐 3-5 轮）
- 缓存检索结果：相似查询复用之前的检索结果
- 分层模型调用：查询分析用小模型（如 Haiku），最终生成用大模型（如 Sonnet）
- 提前终止：如果第一轮检索置信度已经够高，直接生成答案
# 成本控制配置示例
config = {
"max_iterations": 3,
"confidence_threshold": 0.8,
"analysis_model": "claude-haiku-4-5",
"generation_model": "claude-sonnet-4-6",
"cache_ttl": 3600,
"parallel_retrieval": True
}评估框架
推荐使用 RAGAS（Retrieval Augmented Generation Assessment）框架进行自动化评估：
from ragas import evaluate
from ragas.metrics import (
answer_relevancy,
faithfulness,
context_recall,
context_precision
)
# 准备测试数据
test_dataset = {
"question": ["问题1", "问题2", ...],
"ground_truth": ["标准答案1", "标准答案2", ...],
"answer": ["模型答案1", "模型答案2", ...],
"contexts": [["上下文1-1", "上下文1-2"], ...]
}
# 运行评估
results = evaluate(
test_dataset,
metrics=[
answer_relevancy,
faithfulness,
context_recall,
context_precision
]
)关键指标的健康区间：
| 指标 | 定义 | 健康区间 |
|---|---|---|
| Answer Relevancy | 答案与问题的相关性 | >0.85 |
| Faithfulness | 答案是否忠实于检索内容 | >0.90 |
| Context Recall | 检索到了多少相关文档 | >0.75 |
| Context Precision | 检索结果中相关文档的占比 | >0.70 |
生产环境部署建议
监控与可观测性
Agentic RAG 的调试比传统 RAG 复杂得多，因为每次查询的检索路径不同。必须做好全链路追踪：
# 使用 LangSmith 或 OpenTelemetry 追踪
with tracer.start_span("agentic_rag") as span:
span.set_attribute("query", user_query)
span.set_attribute("iterations", num_iterations)
span.set_attribute("sources_used", sources)
span.set_attribute("confidence", final_confidence)
span.set_attribute("total_tokens", token_count)回退机制
Agent 的决策不总是正确的。设计合理的回退策略：
- 超时回退：如果 Agent 超过 30 秒仍未生成答案，回退到传统 RAG
- 置信度回退：如果多轮检索后置信度仍低于阈值，诚实告知用户”信息不足”
- 成本回退：如果单次查询的 token 消耗超过预算，强制终止并输出当前最佳结果
总结
Agentic RAG 代表了 RAG 架构从”管道式”到”智能体式”的范式转变。它的核心价值不在于单项技术的突破，而在于将多个能力（检索、推理、工具调用、自我评估）有机编排在一起，形成自适应的知识获取系统。
对于大多数企业 AI 应用来说，建议渐进式迁移：
- 先做好传统 RAG 的基础：向量化质量、分块策略、检索召回率
- 加入查询改写：这一步投入小、收益大
- 引入结果评估：让系统知道检索结果是否足够好
- 完整的 Agentic RAG：多步检索、工具调用、多 Agent 协作
不要试图一步到位——传统 RAG 做到 80 分再考虑 Agentic 升级。