痛点
传统 RAG 管道是一条"固定流水线"——用户提问 → 向量检索 → 拼接上下文 → LLM 生成。这种线性模式在简单问答场景能用，但遇到以下情况就力不从心：
- 多跳推理：问题需要先查 A 文档，从 A 的结论出发再查 B，传统 RAG 一次检索根本覆盖不了
- 查询意图模糊：用户问"我们系统上个月的故障和这个月的有什么关联"，需要先拆解时间范围、再分别检索、最后综合分析
- 检索质量不可控：召回的文档不相关时，LLM 照样硬编答案，没有"发现检索结果不够就再查一次"的能力
- 多数据源路由：知识可能分布在 Wiki、代码仓库、监控数据、工单系统中，固定管道无法动态选择数据源
Agentic RAG 的核心思路：把检索动作交给 Agent 自主决策——Agent 分析问题、规划检索策略、执行检索、评估结果质量、必要时迭代优化，直到获得满意的上下文再生成最终答案。
方案：Agentic RAG 架构设计
核心组件
| 组件 | 职责 | 实现方式 |
|---|---|---|
| Query Analyzer | 解析用户意图，拆解子问题 | LLM + Structured Output |
| Router | 根据问题类型选择数据源/检索策略 | Function Calling / Tool Use |
| Retriever Tools | 多个检索工具供 Agent 调用 | 向量检索、关键词检索、SQL 查询、API 调用 |
| Grader | 评估检索结果的相关性 | LLM 评分 / Cross-Encoder Rerank |
| Reflector | 判断是否需要重新检索或换策略 | Agent 自省循环 |
架构流程
用户提问
│
▼
┌─────────────────┐
│ Query Analyzer │ ← 拆解意图、识别子问题
└────────┬────────┘
│
▼
┌─────────────────┐
│ Router │ ← 选择数据源 + 检索策略
└────────┬────────┘
│
▼
┌─────────────────┐
│ Retriever Tools │ ← 执行检索（可能多轮）
└────────┬────────┘
│
▼
┌─────────────────┐
│ Grader │ ← 评估相关性，过滤噪声
└────────┬────────┘
│
相关性不足？──→ 回到 Router 换策略/改写查询
│
▼ (满足)
┌─────────────────┐
│ LLM 生成 │ ← 基于高质量上下文生成答案
└─────────────────┘
实操步骤
第1步：定义 Retriever Tools
将不同数据源封装为 Agent 可调用的工具：
from langchain_core.tools import tool
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
# 向量检索工具 — 适合语义模糊查询
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma(persist_directory="./chroma_db", embedding_function=embeddings)
@tool
def vector_search(query: str) -> str:
"""从知识库中语义检索相关文档片段。适合概念性、描述性问题。"""
docs = vectorstore.similarity_search_with_relevance_scores(query, k=5)
# 过滤低相关性结果
filtered = [(doc, score) for doc, score in docs if score > 0.75]
if not filtered:
return "未找到高相关性文档，建议改写查询或换用其他检索方式。"
return "\n---\n".join([f"[相关度:{score:.2f}] {doc.page_content}" for doc, score in filtered])
# 关键词检索工具 — 适合精确匹配（错误码、配置项名）
from elasticsearch import Elasticsearch
es = Elasticsearch("http://localhost:9200")
@tool
def keyword_search(query: str, index: str = "wiki") -> str:
"""使用关键词精确匹配检索。适合搜索错误码、配置项名称、特定命令。"""
resp = es.search(index=index, body={
"query": {"match_phrase": {"content": query}},
"size": 5
})
hits = resp["hits"]["hits"]
if not hits:
return "关键词检索无结果，建议尝试语义检索或调整关键词。"
return "\n---\n".join([h["_source"]["content"][:500] for h in hits])
# SQL 查询工具 — 适合结构化数据（监控指标、工单统计）
import sqlite3
@tool
def query_incidents(time_range: str, severity: str = "") -> str:
"""查询故障工单数据。time_range 格式: '2026-06-01,2026-06-30'。severity: P0/P1/P2。"""
start, end = time_range.split(",")
conn = sqlite3.connect("incidents.db")
sql = "SELECT title, severity, root_cause, created_at FROM incidents WHERE created_at BETWEEN ? AND ?"
params = [start.strip(), end.strip()]
if severity:
sql += " AND severity = ?"
params.append(severity)
rows = conn.execute(sql, params).fetchall()
conn.close()
if not rows:
return "该时间范围内无匹配工单。"
return "\n".join([f"[{r[1]}] {r[0]} | 根因: {r[2]} | {r[3]}" for r in rows])
第2步：构建带自省能力的 Agent
核心是让 Agent 在检索后评估结果质量，不满意则迭代：
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
class AgentState(TypedDict):
messages: Annotated[list, add_messages]
retrieved_context: str
retrieval_attempts: int
is_context_sufficient: bool
llm = ChatOpenAI(model="gpt-4o", temperature=0)
llm_with_tools = llm.bind_tools([vector_search, keyword_search, query_incidents])
def analyze_query(state: AgentState) -> AgentState:
"""分析用户问题，规划检索策略"""
system_prompt = """你是一个检索策略规划器。分析用户问题后：
1. 判断需要哪些数据源（知识库/工单系统/精确匹配）
2. 是否需要拆解为子问题
3. 调用合适的检索工具
优先级：精确匹配（错误码/配置项）→ 关键词检索，概念性问题 → 语义检索，数据统计 → SQL查询。"""
messages = [{"role": "system", "content": system_prompt}] + state["messages"]
response = llm_with_tools.invoke(messages)
return {"messages": [response], "retrieval_attempts": state.get("retrieval_attempts", 0) + 1}
def grade_context(state: AgentState) -> AgentState:
"""评估检索结果是否足够回答问题"""
grading_prompt = f"""评估以下检索结果是否足以回答用户问题。
用户问题: {state['messages'][0].content}
检索结果: {state['retrieved_context']}
回答 SUFFICIENT 或 INSUFFICIENT（附简要原因）。"""
response = llm.invoke([{"role": "user", "content": grading_prompt}])
is_sufficient = "SUFFICIENT" in response.content.upper()
return {"is_context_sufficient": is_sufficient}
def should_retry(state: AgentState) -> str:
"""决策节点：是否需要重试检索"""
if state.get("is_context_sufficient", False):
return "generate"
if state.get("retrieval_attempts", 0) >= 3:
return "generate" # 最多重试3次，避免死循环
return "retry"
def generate_answer(state: AgentState) -> AgentState:
"""基于充分的上下文生成最终答案"""
system_prompt = """基于检索到的上下文回答问题。如果上下文不足，明确告知用户哪些信息缺失。
不要编造不在上下文中的信息。"""
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": f"上下文:\n{state['retrieved_context']}\n\n问题: {state['messages'][0].content}"}
]
response = llm.invoke(messages)
return {"messages": [response]}
# 构建 LangGraph 工作流
workflow = StateGraph(AgentState)
workflow.add_node("analyze", analyze_query)
workflow.add_node("grade", grade_context)
workflow.add_node("generate", generate_answer)
workflow.set_entry_point("analyze")
workflow.add_edge("analyze", "grade")
workflow.add_conditional_edges("grade", should_retry, {
"generate": "generate",
"retry": "analyze"
})
workflow.add_edge("generate", END)
app = workflow.compile()
第3步：Query Routing 实现智能分流
对于多数据源场景，用 Structured Output 实现精确路由：
from pydantic import BaseModel, Field
from enum import Enum
class DataSource(str, Enum):
VECTOR_KB = "vector_kb" # 知识库语义检索
KEYWORD_SEARCH = "keyword_search" # 精确关键词匹配
INCIDENT_DB = "incident_db" # 故障工单数据库
MULTI_SOURCE = "multi_source" # 需要组合多个数据源
class QueryRoute(BaseModel):
"""查询路由决策"""
datasource: DataSource = Field(description="最适合的数据源")
sub_queries: list[str] = Field(default_factory=list, description="拆解后的子查询")
reasoning: str = Field(description="路由决策原因")
structured_llm = llm.with_structured_output(QueryRoute)
def route_query(user_question: str) -> QueryRoute:
routing_prompt = f"""分析以下问题，决定检索路由：
- 包含错误码、配置项名、具体命令 → keyword_search
- 概念解释、方案对比、架构设计 → vector_kb
- 涉及时间范围的故障/工单统计 → incident_db
- 复杂问题需要多步/多源 → multi_source
问题: {user_question}"""
return structured_llm.invoke([{"role": "user", "content": routing_prompt}])
# 使用示例
route = route_query("ERROR-5012 在什么情况下会触发？上个月出现了几次？")
# → DataSource.MULTI_SOURCE
# → sub_queries: ["ERROR-5012 触发条件", "2026-06 ERROR-5012 出现次数"]
第4步：Adaptive Retrieval — 根据上下文动态调整
@tool
def adaptive_vector_search(query: str, filter_metadata: str = "") -> str:
"""自适应向量检索：先宽泛检索，根据结果自动调整过滤条件。
filter_metadata 示例: 'source:wiki,date_after:2026-01-01'
"""
# 解析过滤条件
filters = {}
if filter_metadata:
for pair in filter_metadata.split(","):
k, v = pair.split(":")
filters[k.strip()] = v.strip()
# 第一次检索
where_filter = {f"${k}": v for k, v in filters.items()} if filters else None
docs = vectorstore.similarity_search_with_relevance_scores(
query, k=8, filter=where_filter
)
# 自动降级：如果过滤后结果太少，放宽条件重试
if len(docs) < 2 and filters:
docs = vectorstore.similarity_search_with_relevance_scores(query, k=8)
# 多样性去重：MMR
if len(docs) > 3:
docs_mmr = vectorstore.max_marginal_relevance_search(query, k=4, fetch_k=10)
return "\n---\n".join([doc.page_content for doc in docs_mmr])
return "\n---\n".join([f"[{score:.2f}] {doc.page_content}" for doc, score in docs])
避坑
1. 检索死循环
Agent 可能反复检索却始终不满意——必须设置 max_retrieval_attempts 上限（建议 3 次）。超限后强制进入生成阶段，并在回答中标注"信息可能不完整"。
# 在 should_retry 中硬性限制
if state.get("retrieval_attempts", 0) >= 3:
return "generate" # 强制跳出
2. Tool 描述决定路由质量
Agent 选择哪个 Tool 完全依赖 Tool 的 docstring 描述。描述模糊会导致路由错误。最佳实践：
- 明确写出每个工具"适合什么场景"和"不适合什么场景"
- 给出输入格式示例
- 用 few-shot 示例让 Agent 学会何时用哪个工具
3. 成本与延迟控制
Agentic RAG 每次请求可能触发 3-5 次 LLM 调用（分析 + 检索判断 + 评估 + 生成），成本和延迟都会翻倍。应对策略：
| 策略 | 做法 |
|---|---|
| 分层模型 | 路由/评估用 gpt-4o-mini，最终生成用 gpt-4o |
| 缓存 | 对相同查询的路由结果做缓存，TTL 设 1h |
| 并行检索 | 子查询拆解后并行调用多个 Tool，不串行等待 |
| 短路机制 | 简单问题直接走传统 RAG，复杂问题才触发 Agentic 流程 |
# 短路判断：简单问题直接走快速路径
def should_use_agentic(question: str) -> bool:
"""判断是否需要 Agentic RAG"""
complexity_prompt = f"判断以下问题是简单查询(SIMPLE)还是需要多步推理(COMPLEX): {question}"
resp = ChatOpenAI(model="gpt-4o-mini").invoke([{"role": "user", "content": complexity_prompt}])
return "COMPLEX" in resp.content.upper()
总结
Agentic RAG 将检索从"固定管道"升级为"自主决策"，核心收益：
- 多跳推理：Agent 可以基于第一轮检索结果生成新查询，逐步逼近答案
- 动态路由：根据问题类型自动选择最合适的数据源和检索策略
- 质量自控：通过 Grader 环节过滤低质量检索结果，避免 LLM 基于噪声生成幻觉
- 可扩展：新增数据源只需注册一个新 Tool，无需改架构
适用场景：企业知识库问答、运维故障诊断辅助、跨系统数据关联分析。对于简单 FAQ 类需求，传统 RAG 管道已足够，不必为了"高级"而引入额外复杂度。
落地建议：先在传统 RAG 基础上加 Query Routing（投入最小、收益最大），再逐步引入 Grader 和自省循环。用 LangSmith/LangFuse 全程追踪每次检索决策，方便调试和优化路由策略。