Agentic RAG實戰：構建自主推理檢索Agent的5大核心模式
AI与大数据
傳統RAG已經不夠用了
你精心搭建的RAG系統，使用者問了一句「RAG和微調哪個更適合我的場景？」，結果傳回了一堆關於RAG的基礎介紹和微調的定義——完全沒回答問題。這不是檢索不準，而是傳統RAG根本不具備推理能力。它只能做「問什麼搜什麼」，遇到需要多步推理、跨知識源綜合、自我糾正的複雜問題，就只能原地打轉。
2026年，Agentic RAG成為破局關鍵。它不再是被動的「檢索-生成」管道，而是讓Agent自主決定：要不要檢索？去哪檢索？檢索結果夠不夠？要不要換個策略重新搜？這種「自主決策+迴圈迭代」的範式，才是複雜問題的正確開啟方式。
核心概念速查
| 概念 | 說明 | 典型實作 |
|---|---|---|
| Agentic RAG | Agent自主決策檢索策略的RAG範式 | LangGraph StateGraph、LlamaIndex Agent |
| 自主檢索 | Agent根據問題自主決定是否檢索、檢索什麼 | ReAct迴圈、工具呼叫 |
| 多跳推理 | 將複雜問題分解為多個子問題逐步檢索 | 問題分解、鏈式檢索 |
| 自我反思 | Agent評估檢索結果品質，決定是否重新搜尋 | 文件評分、查詢重寫 |
| 工具呼叫 | Agent呼叫外部工具（搜尋、計算、資料庫）增強能力 | Function Calling、Tool Use |
| LangGraph | 構建有狀態Agent工作流的框架 | StateGraph、條件邊、檢查點 |
| ReAct模式 | 推理（Reasoning）+行動（Acting）交替進行 | Thought-Action-Observation迴圈 |
問題分析：傳統RAG的5大侷限
| # | 侷限 | 具體表現 | 影響 |
|---|---|---|---|
| 1 | 單次檢索 | 只能做一次向量搜尋，無法迭代最佳化 | 檢索不全面，遺漏關鍵資訊 |
| 2 | 無推理能力 | 無法分解複雜問題為子問題 | 多跳問題回答品質差 |
| 3 | 無法自我糾正 | 檢索到無關文件也無法識別和重試 | 幻覺嚴重，答非所問 |
| 4 | 缺乏規劃 | 沒有檢索策略，無法決定先搜什麼後搜什麼 | 效率低下，資源浪費 |
| 5 | 無法處理多跳問題 | 「A和B哪個更適合C場景？」類問題完全無解 | 使用者體驗極差，棄用率高 |
傳統RAG的本質問題：它是一個無狀態的管道，而不是一個有智慧的Agent。Agentic RAG的核心轉變，就是從「管道」到「Agent」——讓檢索過程具備自主決策、迴圈迭代和自我糾錯的能力。
模式1：路由Agent — 智慧分發
最簡單的Agentic RAG模式：根據問題型別，將查詢路由到不同的知識庫或檢索策略。比統一向量搜尋精準得多。
from enum import Enum
from typing import Optional
class QueryType(Enum):
TECHNICAL = "technical"
BUSINESS = "business"
COMPARISON = "comparison"
TUTORIAL = "tutorial"
class RouterAgent:
def __init__(self, llm, retrievers: dict[str, object]):
self.llm = llm
self.retrievers = retrievers
def classify(self, question: str) -> QueryType:
prompt = f"""Classify this question into one of:
- technical: code/API/implementation questions
- business: pricing/ROI/strategy questions
- comparison: comparing two or more options
- tutorial: how-to/step-by-step guides
Question: {question}
Type:"""
result = self.llm.invoke(prompt).strip().lower()
return QueryType(result)
def route(self, question: str) -> list[dict]:
query_type = self.classify(question)
retriever = self.retrievers.get(query_type.value)
if not retriever:
retriever = self.retrievers["technical"]
docs = retriever.similarity_search(question, k=5)
return [{"content": d.page_content, "source": d.metadata.get("source", "")} for d in docs]
retrievers = {
"technical": tech_vector_store,
"business": biz_vector_store,
"comparison": compare_vector_store,
"tutorial": tutorial_vector_store,
}
router = RouterAgent(llm, retrievers)
result = router.route("TiDB和CockroachDB在分散式事務方面有什麼區別？")
路由Agent適合知識庫分類明確的場景，實作簡單但效果立竿見影。
模式2：多跳推理Agent
複雜問題需要分解。比如「比較TiDB和CockroachDB在分散式事務方面的差異，給出效能基準資料」，需要先分別檢索兩個資料庫的資訊，再檢索效能基準，最後綜合回答。
from langchain.agents import create_react_agent
tools = [
search_knowledge_base,
search_web,
calculate_metrics,
query_database,
]
agent = create_react_agent(llm, tools, prompt)
result = agent.invoke({
"input": "比較TiDB和CockroachDB在分散式事務方面的差異，給出效能基準資料"
})
更細粒度的多跳實作：
from dataclasses import dataclass
from typing import Optional
@dataclass
class SubQuestion:
question: str
answer: Optional[str] = None
sources: list[str] = None
class MultiHopAgent:
def __init__(self, llm, retriever):
self.llm = llm
self.retriever = retriever
def decompose(self, question: str) -> list[SubQuestion]:
prompt = f"""Break down this complex question into 2-4 sub-questions.
Each sub-question should be independently answerable.
Question: {question}
Sub-questions (one per line):"""
result = self.llm.invoke(prompt)
lines = [l.strip() for l in result.strip().split("\n") if l.strip()]
return [SubQuestion(question=line) for line in lines]
def answer_sub_question(self, sub_q: SubQuestion) -> SubQuestion:
docs = self.retriever.similarity_search(sub_q.question, k=3)
context = "\n".join([d.page_content for d in docs])
answer = self.llm.invoke(
f"Based on:\n{context}\n\nAnswer: {sub_q.question}"
)
sub_q.answer = answer
sub_q.sources = [d.metadata.get("source", "") for d in docs]
return sub_q
def synthesize(self, question: str, sub_answers: list[SubQuestion]) -> str:
evidence = "\n".join([f"- {sq.question}: {sq.answer}" for sq in sub_answers])
return self.llm.invoke(
f"Based on the following evidence, answer the original question.\n\n"
f"Original: {question}\n\nEvidence:\n{evidence}\n\nAnswer:"
)
def run(self, question: str) -> dict:
sub_questions = self.decompose(question)
answered = [self.answer_sub_question(sq) for sq in sub_questions]
final = self.synthesize(question, answered)
return {
"answer": final,
"sub_questions": [
{"q": sq.question, "a": sq.answer} for sq in answered
],
}
agent = MultiHopAgent(llm, vector_store)
result = agent.run("RAG和微調哪個更適合我的場景？請從成本、效果、適用資料量三個維度分析")
模式3：自我反思搜尋Agent
Agent檢索到文件後，自己評估品質——如果不夠好，就重寫查詢再搜。這是Agentic RAG最核心的能力：自我糾錯。
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class ReflectionState:
question: str
documents: list = field(default_factory=list)
filtered_docs: list = field(default_factory=list)
answer: str = ""
iterations: int = 0
max_iterations: int = 3
need_more_search: bool = False
class SelfReflectiveAgent:
def __init__(self, llm, retriever, min_relevant: int = 2):
self.llm = llm
self.retriever = retriever
self.min_relevant = min_relevant
def retrieve(self, state: ReflectionState) -> dict:
docs = self.retriever.similarity_search(state.question, k=5)
return {"documents": docs}
def grade_documents(self, state: ReflectionState) -> dict:
filtered = []
for doc in state.documents:
score = self.llm.invoke(
f"Is this document relevant to '{state.question}'? "
f"Answer yes or no:\n{doc.page_content[:500]}"
)
if "yes" in score.lower():
filtered.append(doc)
return {
"filtered_docs": filtered,
"need_more_search": len(filtered) < self.min_relevant,
}
def rewrite_query(self, state: ReflectionState) -> dict:
better = self.llm.invoke(
f"Rewrite this question for better search results: {state.question}"
)
return {"question": better, "iterations": state.iterations + 1}
def generate(self, state: ReflectionState) -> dict:
context = "\n".join([d.page_content for d in state.filtered_docs])
answer = self.llm.invoke(
f"Based on:\n{context}\n\nAnswer: {state.question}"
)
return {"answer": answer}
def run(self, question: str) -> dict:
state = ReflectionState(question=question)
self.retrieve(state)
while state.iterations < state.max_iterations:
self.grade_documents(state)
if not state.need_more_search:
break
self.rewrite_query(state)
self.retrieve(state)
self.generate(state)
return {"answer": state.answer, "iterations": state.iterations}
agent = SelfReflectiveAgent(llm, vector_store)
result = agent.run("如何在生產環境中部署Agentic RAG？")
模式4：工具增強Agent
Agent不只檢索文件，還能呼叫搜尋引擎、計算器、資料庫等外部工具。這極大擴展了Agent的能力邊界。
from typing import Annotated
def search_web(query: str) -> str:
"""Search the web for up-to-date information."""
results = web_search_engine.search(query, max_results=3)
return "\n".join([r.snippet for r in results])
def query_database(sql: str) -> str:
"""Execute SQL query against the product database."""
rows = db.execute(sql)
return str(rows[:20])
def calculate(expression: str) -> str:
"""Safely evaluate a mathematical expression."""
allowed = set("0123456789+-*/.() ")
if not all(c in allowed for c in expression):
return "Error: invalid expression"
return str(eval(expression))
def search_knowledge_base(query: str) -> str:
"""Search internal knowledge base for documentation."""
docs = vector_store.similarity_search(query, k=3)
return "\n".join([d.page_content for d in docs])
tools = [search_web, query_database, calculate, search_knowledge_base]
tool_descriptions = "\n".join(
[f"- {t.__name__}: {t.__doc__}" for t in tools]
)
class ToolAugmentedAgent:
def __init__(self, llm, tools: list):
self.llm = llm
self.tools = {t.__name__: t for t in tools}
def run(self, question: str, max_steps: int = 5) -> dict:
messages = [{"role": "user", "content": question}]
steps = []
for _ in range(max_steps):
response = self.llm.invoke(
f"Available tools:\n{tool_descriptions}\n\n"
f"To use a tool, respond: USE: tool_name(arg)\n"
f"To give final answer, respond: ANSWER: your answer\n\n"
f"Question: {messages[-1]['content']}"
)
if response.startswith("ANSWER:"):
return {"answer": response[7:].strip(), "steps": steps}
if response.startswith("USE:"):
tool_call = response[4:].strip()
tool_name = tool_call.split("(")[0]
tool_arg = tool_call.split("(")[1].rstrip(")")
tool_fn = self.tools.get(tool_name)
if tool_fn:
observation = tool_fn(tool_arg)
steps.append({
"tool": tool_name,
"arg": tool_arg,
"result": observation[:200],
})
messages.append({
"role": "assistant",
"content": f"Used {tool_name}, got: {observation}",
})
return {"answer": "Max steps reached", "steps": steps}
agent = ToolAugmentedAgent(llm, tools)
result = agent.run("我們產品的月活使用者是多少？和競品相比如何？")
模式5：LangGraph工作流編排
最完整的Agentic RAG實作——用LangGraph的StateGraph定義完整的檢索-評估-重寫-生成迴圈，支援條件分支和檢查點。
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, List
import operator
class AgentState(TypedDict):
question: str
documents: Annotated[list, operator.add]
answer: str
iterations: int
need_more_search: bool
def retrieve(state: AgentState) -> dict:
docs = vector_store.similarity_search(state["question"], k=5)
return {"documents": docs}
def grade_documents(state: AgentState) -> dict:
question = state["question"]
docs = state["documents"]
filtered = []
for doc in docs:
score = llm.invoke(f"Is this doc relevant to '{question}'? Answer yes/no: {doc.page_content}")
if "yes" in score.lower():
filtered.append(doc)
return {"documents": filtered, "need_more_search": len(filtered) < 2}
def generate(state: AgentState) -> dict:
context = "\n".join([d.page_content for d in state["documents"]])
answer = llm.invoke(f"Based on:\n{context}\n\nAnswer: {state['question']}")
return {"answer": answer}
def decide_to_search(state: AgentState) -> str:
if state["need_more_search"] and state["iterations"] < 3:
return "rewrite"
return "generate"
def rewrite_query(state: AgentState) -> dict:
better_query = llm.invoke(f"Rewrite this question for better search: {state['question']}")
return {"question": better_query, "iterations": state["iterations"] + 1}
workflow = StateGraph(AgentState)
workflow.add_node("retrieve", retrieve)
workflow.add_node("grade", grade_documents)
workflow.add_node("generate", generate)
workflow.add_node("rewrite", rewrite_query)
workflow.set_entry_point("retrieve")
workflow.add_edge("retrieve", "grade")
workflow.add_conditional_edges("grade", decide_to_search, {
"rewrite": "rewrite",
"generate": "generate",
})
workflow.add_edge("rewrite", "retrieve")
workflow.add_edge("generate", END)
app = workflow.compile()
result = app.invoke({"question": "RAG和微調哪個更適合我的場景？", "iterations": 0})
LangGraph的優勢在於：視覺化工作流、支援檢查點恢復、條件分支靈活、易於除錯。生產環境首選。
避坑指南：5大常見陷阱
| # | 陷阱 | 表現 | 解決方案 |
|---|---|---|---|
| 1 | 無限迴圈 | Agent不斷重寫查詢，永遠不滿意 | 設定max_iterations硬上限（3-5次） |
| 2 | 過度檢索 | 每個問題都觸發多輪搜尋，延遲飆升 | 簡單問題走快速路徑，複雜問題才走Agentic |
| 3 | 工具幻覺 | Agent編造不存在的工具名呼叫 | 嚴格校驗工具名，白名單機制 |
| 4 | 上下文爆炸 | 多輪檢索的文件全部塞進Prompt | 每輪只保留評分最高的Top-K文件 |
| 5 | 評分偏差 | LLM對文件相關性評分不穩定 | 用結構化輸出（JSON Schema）約束評分格式 |
報錯排查：10大常見錯誤
| # | 錯誤資訊 | 原因 | 解決方案 |
|---|---|---|---|
| 1 | GraphRecursionError: Recursion limit reached |
Agent迴圈超過LangGraph預設步數限制 | 設定recursion_limit引數，或最佳化條件邊邏輯 |
| 2 | KeyError: 'documents' |
State中缺少預期欄位 | 檢查節點傳回值是否包含所有必需欄位 |
| 3 | ToolCallingError: Unknown tool |
Agent呼叫了未註冊的工具 | 檢查tools列表，確保工具名與描述一致 |
| 4 | TokenLimitExceeded |
多輪檢索累積文件超過上下文視窗 | 在grade節點過濾文件，限制傳入generate的文件數 |
| 5 | ValidationError: State schema mismatch |
State型別定義與實際資料不匹配 | 檢查TypedDict欄位型別，Annotated用法是否正確 |
| 6 | TimeoutError: LLM invocation timed out |
評分/生成步驟LLM回應慢 | 設定timeout引數，考慮非同步呼叫 |
| 7 | ImportError: No module named 'langgraph' |
未安裝LangGraph | pip install langgraph langchain-core |
| 8 | JSONDecodeError in structured output |
LLM傳回非標準JSON | 使用with_structured_output()強制結構化 |
| 9 | RateLimitError: Too many requests |
多輪檢索觸發API限流 | 加exponential backoff重試，或換用本地模型 |
| 10 | AttributeError: 'NoneType' has no attribute |
某節點傳回None | 確保每個節點都傳回dict，而非None |
進階最佳化技巧
1. 混合路由：簡單問題快速路徑
不是所有問題都需要Agentic流程。加一個分類器，簡單問題走Naive RAG，複雜問題才走Agentic RAG，平均延遲降低60%。
class HybridRouter:
def __init__(self, llm, naive_rag, agentic_rag):
self.llm = llm
self.naive_rag = naive_rag
self.agentic_rag = agentic_rag
def is_simple(self, question: str) -> bool:
result = self.llm.invoke(
f"Is this a simple factual question? Answer yes/no: {question}"
)
return "yes" in result.lower()
def run(self, question: str) -> dict:
if self.is_simple(question):
return self.naive_rag.query(question)
return self.agentic_rag.invoke({"question": question, "iterations": 0})
2. 檢查點恢復：長時間工作流不遺失
LangGraph支援檢查點（Checkpoint），Agent中斷後可從上次狀態恢復。
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
app = workflow.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "user-123"}}
result = app.invoke({"question": "分析我們的技術選型", "iterations": 0}, config=config)
3. 平行檢索：多知識源同時搜尋
import asyncio
async def parallel_retrieve(question: str, stores: list) -> list:
tasks = [store.asimilarity_search(question, k=3) for store in stores]
results = await asyncio.gather(*tasks)
return [doc for batch in results for doc in batch]
4. 快取熱點查詢
from functools import lru_cache
@lru_cache(maxsize=1000)
def cached_retrieve(query_hash: str, question: str) -> list:
return vector_store.similarity_search(question, k=5)
5. 可觀測性：追蹤Agent每一步決策
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("agentic_rag")
def traced_node(func):
def wrapper(state):
logger.info(f"Node: {func.__name__}, Input: {str(state)[:200]}")
result = func(state)
logger.info(f"Node: {func.__name__}, Output: {str(result)[:200]}")
return result
return wrapper
對比分析：Agentic RAG vs Naive RAG vs Advanced RAG vs Modular RAG
| 維度 | Naive RAG | Advanced RAG | Modular RAG | Agentic RAG |
|---|---|---|---|---|
| 檢索策略 | 單次向量搜尋 | 查詢重寫+混合檢索 | 可插拔檢索模組 | Agent自主決策檢索 |
| 推理能力 | ✗ | ✗ | ✗ | ✓ 多跳推理 |
| 自我糾正 | ✗ | △ 重排序 | ✗ | ✓ 反思+重寫 |
| 工具呼叫 | ✗ | ✗ | △ | ✓ 多工具協同 |
| 工作流 | 線性管道 | 線性+預處理 | 模組化管道 | 有狀態圖 |
| 實作複雜度 | ★☆☆ | ★★☆ | ★★★ | ★★★★ |
| 延遲 | ~1s | ~2s | ~2s | ~3-8s |
| 準確率 | 60-70% | 75-85% | 80-88% | 85-95% |
| 適用場景 | 簡單FAQ | 標準問答 | 客製化場景 | 複雜多跳問題 |
★越多表示實作複雜度越高；✓支援 △部分支援 ✗不支援；延遲和準確率為參考值
線上工具推薦
- JSON格式化 — 格式化Agent State和檢索結果的JSON結構，除錯必備
- 雜湊計算 — 計算文件去重和查詢快取的MD5/SHA雜湊值
- Curl轉程式碼 — 將LLM API除錯curl命令轉為Python程式碼
總結與展望
"The future of RAG is not better retrieval — it's smarter agents that know when and how to retrieve."
"RAG的未來不是更好的檢索，而是更聰明的Agent——它知道何時檢索、如何檢索。"
Agentic RAG正在從「實驗性架構」走向「生產標配」。2026年的關鍵趨勢：
- 原生Agent支援：LangGraph、LlamaIndex等框架將Agentic RAG作為一等公民
- 多Agent協作：多個專業Agent協同完成複雜檢索任務
- 自適應策略：Agent根據問題難度自動選擇檢索深度
- 可解釋性增強：每一步檢索決策都有清晰的推理鏈可追溯
- 成本最佳化：智慧路由+快取，讓Agentic RAG的成本接近Naive RAG
選擇Agentic RAG方案的原則：從路由Agent開始，逐步升級。先用路由解決知識庫分發問題，再加自我反思提升檢索品質，最後用LangGraph編排完整工作流。別一上來就搞最複雜的——先讓Agent跑起來，再讓它變聰明。
延伸閱讀
本站提供瀏覽器本地工具，免註冊即可試用 →
#Agentic RAG#Agent RAG#自主检索Agent#LLM Agent推理#RAG Agent工作流#2026#LangGraph Agentic RAG