AI
人工智能相关文章
Agentic RAG实战指南：构建能自主规划检索策略的AI问答系统
Agentic RAG实战指南：构建能自主规划检索策略的AI问答系统
# Agentic RAG实战指南：构建能自主规划检索策略的AI问答系统
## 摘要
传统RAG是被动的"检索-生成"流水线，无法根据中间结果动态调整检索策略。Agentic RAG将Agent的自主规划能力引入RAG流程，让系统像人类研究员一样，自主决定检索什么、何时检索、如何组合信息。
## 一、RAG进化论：从Naive到Agentic
### 1.1 三代RAG对比
| 维度 | Naive RAG | Adaptive RAG | Agentic RAG |
|-----|-----------|-------------|-------------|
| 检索策略 | 固定Top-K | 查询改写+自适应 | Agent自主规划 |
| 检索轮数 | 单轮 | 1-2轮 | 多轮（自主决定） |
| 工具使用 | 仅向量检索 | 多检索器 | 向量+SQL+API+搜索 |
| 纠错能力 | 无 | 有限 | 自我反思+重检索 |
| 适用场景 | 简单FAQ | 中等复杂度 | 复杂研究型任务 |
### 1.2 Agentic RAG的核心思想
Agentic RAG的关键在于**将检索过程本身变成一个Agent决策问题**：
```
用户问题 → Agent思考("需要什么信息？")
→ 选择工具（向量检索/SQL查询/Web搜索）
→ 获取结果
→ Agent思考("信息是否充分？")
→ 不充分 → 重新检索（换关键词/换工具）
→ 充分 → 综合生成答案
```
## 二、基于LangGraph的Agentic RAG实现
### 2.1 定义状态与工具
```python
from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage
import operator
class AgentState(TypedDict):
messages: Annotated[Sequence[BaseMessage], operator.add]
original_question: str
search_history: list[str]
retrieved_docs: list[dict]
reflection_count: int
is_sufficient: bool
# 定义可用工具
from langchain_core.tools import tool
@tool
def vector_search(query: str, top_k: int = 5) -> str:
"""从向量数据库检索相关文档片段"""
from langchain_community.vectorstores import Chroma
vectorstore = Chroma(persist_directory="./chroma_db",
embedding_function=embeddings)
docs = vectorstore.similarity_search(query, k=top_k)
return "\n\n---\n\n".join([d.page_content for d in docs])
@tool
def sql_query(question: str) -> str:
"""将自然语言问题转为SQL查询结构化数据"""
from langchain_community.utilities import SQLDatabase
db = SQLDatabase.from_uri("mysql+pymysql://root:Renjnet@1@59.110.223.195:7889/info")
from langchain.chains import create_sql_query_chain
chain = create_sql_query_chain(llm, db)
sql = chain.invoke({"question": question})
return db.run(sql)
@tool
def web_search(query: str) -> str:
"""搜索互联网获取最新信息"""
from langchain_community.tools.tavily import TavilySearchResults
search = TavilySearchResults(max_results=5)
return search.invoke(query)
tools = [vector_search, sql_query, web_search]
```
### 2.2 构建Agent工作流
```python
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
# Agent节点：决策下一步
def agent_node(state: AgentState):
system_prompt = """你是一个智能研究助手。对于用户的问题，你需要：
1. 分析当前已获取的信息是否足够回答问题
2. 如果不够，选择合适的工具获取更多信息
3. 如果已充分，生成最终答案
可用工具：
- vector_search: 检索内部文档库
- sql_query: 查询结构化数据库
- web_search: 搜索互联网
已检索轮数：{reflection_count}/5
已获取文档数：{retrieved_docs_count}
如果信息充分，直接给出最终答案。"""
messages = state["messages"]
response = llm_with_tools.invoke(
[{"role": "system", "content": system_prompt.format(
reflection_count=state["reflection_count"],
retrieved_docs_count=len(state["retrieved_docs"])
)}] + list(messages)
)
return {"messages": [response]}
# 反思节点：检查答案质量
def reflection_node(state: AgentState):
last_msg = state["messages"][-1]
reflection_prompt = """评估当前信息是否足够回答原始问题。
原始问题：{question}
当前已获取信息摘要：{summary}
信息是否充分？回答 YES 或 NO，并说明缺少什么。"""
response = llm.invoke(reflection_prompt.format(
question=state["original_question"],
summary=str(last_msg.content)[:1000]
))
is_sufficient = "YES" in response.content
return {
"is_sufficient": is_sufficient,
"reflection_count": state["reflection_count"] + 1
}
# 路由决策
def should_continue(state: AgentState):
if state["is_sufficient"] or state["reflection_count"] >= 5:
return "end"
if hasattr(state["messages"][-1], "tool_calls") and state["messages"][-1].tool_calls:
return "tools"
return "agent"
# 构建图
workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("tools", ToolNode(tools))
workflow.add_node("reflection", reflection_node)
workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_continue, {
"tools": "tools",
"end": END
})
workflow.add_edge("tools", "reflection")
workflow.add_edge("reflection", "agent")
app = workflow.compile()
```
### 2.3 运行示例
```python
result = app.invoke({
"messages": [{"role": "user", "content": "公司2025年Q4的AI产品营收同比增长多少？主要驱动因素是什么？"}],
"original_question": "公司2025年Q4的AI产品营收同比增长多少？主要驱动因素是什么？",
"search_history": [],
"retrieved_docs": [],
"reflection_count": 0,
"is_sufficient": False
})
print(result["messages"][-1].content)
```
**执行流程示例**：
```
1. Agent → "需要先查数据库获取营收数据" → sql_query
2. Reflection → "有营收数据但缺少增长因素分析" → 不充分
3. Agent → "搜索内部文档了解增长分析" → vector_search
4. Reflection → "有定性分析但缺少市场对比" → 不充分
5. Agent → "搜索行业报告" → web_search
6. Reflection → "信息充分" → YES
7. Agent → 生成最终答案
```
## 三、高级优化技巧
### 3.1 查询分解（Query Decomposition）
对于复杂问题，先分解为子问题再逐个检索：
```python
decompose_prompt = """将以下复杂问题分解为2-5个独立的子问题。
原问题：{question}
输出JSON数组格式：
["子问题1", "子问题2", ...]"""
def decompose_query(question: str) -> list[str]:
response = llm.invoke(decompose_prompt.format(question=question))
return json.loads(response.content)
```
### 3.2 答案融合（Answer Fusion）
多个子问题的答案需要融合去重：
```python
fusion_prompt = """以下是多个子问题的检索结果和答案，请融合为一个完整、连贯的回答。
原始问题：{question}
子答案：
{sub_answers}
要求：
1. 去除重复信息
2. 保留所有关键数据点
3. 逻辑连贯
4. 标注信息来源"""
```
### 3.3 检索策略自适应
```python
def select_strategy(question: str) -> str:
"""根据问题类型自动选择检索策略"""
classify_prompt = """判断以下问题属于哪类：
- factual: 事实型，需要精确数据（适合SQL查询）
- analytical: 分析型，需要综合信息（适合多轮向量检索）
- current: 时效型，需要最新信息（适合Web搜索）
问题：{question}
只输出类型关键词。"""
return llm.invoke(classify_prompt.format(question=question)).content.strip()
```
## 四、生产部署注意事项
1. **限制最大检索轮数**：防止Agent陷入无限循环（建议3-5轮）
2. **Token预算控制**：每轮检索结果截断，避免上下文溢出
3. **缓存策略**：相似问题的中间检索结果可复用
4. **超时机制**：单次查询总耗时不超过30秒
5. **降级方案**：Agent失败时退回传统RAG
## 总结
Agentic RAG将RAG从"被动检索"升级为"主动研究"，通过Agent的自主规划能力实现多轮检索、工具组合和自我纠错。虽然增加了系统复杂度，但在需要深度分析的复杂问答场景中，其回答质量远超传统RAG。
---
*本文由北科信息日采集系统自动生成，发布日期：2026-05-05*