Agentic RAG

Agentic RAG 将 Agent 的规划能力和 RAG 的检索能力融合，让系统能够自主决定何时检索、检索什么、以及如何整合信息来回答复杂问题。

从传统 RAG 到 Agentic RAG

对比分析

Agentic RAG 实现

"""
Agentic RAG 引擎
"""
from dataclasses import dataclass, field
from enum import Enum
from abc import ABC, abstractmethod
class ToolType(Enum):
VECTOR_SEARCH = "vector_search"
KEYWORD_SEARCH = "keyword_search"
SQL_QUERY = "sql_query"
WEB_SEARCH = "web_search"
CALCULATOR = "calculator"
@dataclass
class RetrievalResult:
"""检索结果"""
source: str
content: str
relevance: float
tool_used: ToolType
@dataclass
class AgentStep:
"""Agent 执行步骤"""
thought: str
action: str
result: str
step_num: int
@dataclass
class AgenticRAGState:
"""Agentic RAG 状态"""
query: str
sub_questions: list[str] = field(default_factory=list)
retrieved_docs: list[RetrievalResult] = field(default_factory=list)
steps: list[AgentStep] = field(default_factory=list)
is_sufficient: bool = False
final_answer: str = ""
class RetrievalTool(ABC):
"""检索工具基类"""
@abstractmethod
def search(self, query: str, top_k: int = 5) -> list[RetrievalResult]:
pass
class VectorSearchTool(RetrievalTool):
"""向量检索工具"""
def __init__(self, index, embedder):
self.index = index
self.embedder = embedder
def search(self, query: str, top_k: int = 5) -> list[RetrievalResult]:
embedding = self.embedder.encode(query)
results = self.index.search(embedding, top_k)
return [
RetrievalResult(
source=r["id"],
content=r["text"],
relevance=r["score"],
tool_used=ToolType.VECTOR_SEARCH,
)
for r in results
]
class AgenticRAGEngine:
"""Agentic RAG 引擎"""
MAX_ITERATIONS = 5
def __init__(self, llm_client, tools: dict[str, RetrievalTool]):
self.llm = llm_client
self.tools = tools
def answer(self, query: str) -> AgenticRAGState:
"""回答复杂问题"""
state = AgenticRAGState(query=query)
# Step 1: 分析问题复杂度
plan = self._plan(query)
state.sub_questions = plan["sub_questions"]
# Step 2: 迭代检索
for iteration in range(self.MAX_ITERATIONS):
# 选择下一步动作
action = self._decide_action(state)
if action["type"] == "search":
results = self._execute_search(
action["tool"], action["query"]
)
state.retrieved_docs.extend(results)
state.steps.append(AgentStep(
thought=action["reasoning"],
action=f"search:{action['tool']}",
result=f"找到 {len(results)} 条结果",
step_num=iteration + 1,
))
elif action["type"] == "answer":
state.is_sufficient = True
break
# Step 3: 综合生成
state.final_answer = self._synthesize(state)
return state
def _plan(self, query: str) -> dict:
"""规划检索策略"""
prompt = (
f"分析以下问题，分解为子问题并选择检索策略：\n"
f"问题：{query}\n"
f"可用工具：{list(self.tools.keys())}\n"
f"输出 JSON：{{\"sub_questions\": [...], \"strategy\": \"...\"}}"
)
result = self.llm.generate(prompt)
# 简化：返回分解后的子问题
return {"sub_questions": [query], "strategy": "iterative"}
def _decide_action(self, state: AgenticRAGState) -> dict:
"""决定下一步动作"""
# 检查已有信息是否充分
if len(state.retrieved_docs) >= 10:
return {"type": "answer"}
# 选择尚未回答的子问题
answered = {s.action for s in state.steps}
for sq in state.sub_questions:
for tool_name in self.tools:
key = f"search:{tool_name}"
if key not in answered:
return {
"type": "search",
"tool": tool_name,
"query": sq,
"reasoning": f"使用 {tool_name} 搜索: {sq}",
}
return {"type": "answer"}
def _execute_search(self, tool_name: str, query: str) -> list[RetrievalResult]:
"""执行检索"""
tool = self.tools.get(tool_name)
if tool is None:
return []
return tool.search(query, top_k=5)
def _synthesize(self, state: AgenticRAGState) -> str:
"""综合生成答案"""
context = "\n".join(
f"[{r.source}] ({r.tool_used.value}, 相关度:{r.relevance:.2f}) {r.content[:200]}"
for r in sorted(state.retrieved_docs, key=lambda x: x.relevance, reverse=True)[:10]
)
prompt = (
f"基于以下检索结果回答问题。\n"
f"问题：{state.query}\n"
f"检索结果：\n{context}\n"
f"请生成准确、完整的回答。如果信息不足，说明哪些部分需要更多研究。"
)
return self.llm.generate(prompt)

典型应用场景

本章小结

下一章：代码 Agent