从RAG到Agentic RAG：智能检索增强生成的进化之路
本文将深入解析传统RAG与Agentic RAG的核心差异，通过对比架构设计、实现方式和实战案例，帮助你理解为什么Agentic RAG是下一代AI应用的关键技术。
·
【个人主页：玄同765】
大语言模型（LLM）开发工程师｜中国传媒大学·数字媒体技术（智能交互与游戏设计）
深耕领域：大语言模型开发 / RAG知识库 / AI Agent落地 / 模型微调
技术栈：Python / LangChain/RAG（Dify+Redis+Milvus）| SQL/NumPy | FastAPI+Docker ️
工程能力：专注模型工程化部署、知识库构建与优化，擅长全流程解决方案
「让AI交互更智能，让技术落地更高效」
欢迎技术探讨/项目合作！ 关注我，解锁大模型与智能交互的无限可能！
核心内容：本文将深入解析传统RAG与Agentic RAG的核心差异，通过对比架构设计、实现方式和实战案例，帮助你理解为什么Agentic RAG是下一代AI应用的关键技术。
一、RAG技术概述
1.1 什么是RAG
RAG（Retrieval-Augmented Generation，检索增强生成） 是一种将外部知识检索与大语言模型生成能力相结合的技术架构。它通过从外部知识库中检索相关信息，并将其作为上下文提供给LLM，从而生成更准确、更可靠的回答。
1.2 RAG的核心价值
┌─────────────────────────────────────────────────────────────┐
│ RAG核心价值 │
├─────────────────────────────────────────────────────────────┤
│ ✅ 解决知识截止问题 - 访问最新、私有数据 │
│ ✅ 减少幻觉生成 - 基于检索到的真实信息 │
│ ✅ 提供可溯源回答 - 引用具体文档来源 │
│ ✅ 降低微调成本 - 无需重新训练模型 │
│ ✅ 支持动态更新 - 知识库可实时更新 │
└─────────────────────────────────────────────────────────────┘
1.3 RAG的典型应用场景
| 场景 | 说明 | 示例 |
|---|---|---|
| 企业知识库问答 | 基于内部文档的智能客服 | 产品手册查询、HR政策咨询 |
| 法律文档分析 | 合同审查、案例检索 | 法规查询、判例匹配 |
| 医疗辅助诊断 | 医学文献检索 | 症状分析、治疗方案推荐 |
| 学术研究助手 | 论文检索与综述 | 文献综述、研究趋势分析 |
| 代码智能助手 | 技术文档查询 | API使用、最佳实践查询 |
二、传统RAG架构详解
2.1 传统RAG的工作流程
传统RAG采用线性管道架构，遵循固定的处理流程：
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ 用户查询 │ --> │ 向量检索 │ --> │ 生成回答 │
│ (Query) │ │ (Retrieve) │ │ (Generate) │
└──────────────┘ └──────────────┘ └──────────────┘
│
v
┌──────────────┐
│ 向量数据库 │
│ (Vector DB) │
└──────────────┘
标准流程：
- 索引阶段（离线）：
-
文档加载 → 文本分割 → 嵌入生成 → 向量存储
-
查询阶段（在线）：
- 查询向量化 → 相似度检索 → 上下文构建 → LLM生成
2.2 传统RAG的实现代码
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import TextLoader
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_classic.chains.combine_documents import create_stuff_documents_chain
from langchain_classic.chains.retrieval import create_retrieval_chain
from langchain_core.prompts import ChatPromptTemplate
# 1. 加载和分割文档
loader = TextLoader("knowledge_base.txt")
docs = loader.load()
splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
chunks = splitter.split_documents(docs)
# 2. 创建向量数据库
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(chunks, embeddings)
# 3. 创建检索器
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
# 4. 创建RAG链
prompt = ChatPromptTemplate.from_messages([
("system", "基于以下上下文回答问题:\n{context}"),
("human", "{input}")
])
document_chain = create_stuff_documents_chain(ChatOpenAI(), prompt)
rag_chain = create_retrieval_chain(retriever, document_chain)
# 5. 执行查询
response = rag_chain.invoke({"input": "什么是RAG技术？"})
print(response["answer"])
2.3 传统RAG的局限性
| 局限性 | 说明 | 影响 |
|---|---|---|
| 单次检索 | 只执行一次检索，无法迭代优化 | 复杂问题检索不充分 |
| 静态流程 | 固定流程，无法根据问题调整 | 简单问题过度检索 |
| 无自我纠错 | 无法验证生成内容的准确性 | 可能产生幻觉 |
| 缺乏推理 | 不能进行多步推理和规划 | 复杂查询处理能力弱 |
| 上下文限制 | 检索结果一次性输入LLM | 长文档处理能力受限 |
三、Agentic RAG：下一代RAG架构
3.1 什么是Agentic RAG
Agentic RAG（智能体化RAG） 是将AI Agent的智能决策能力引入RAG系统的下一代架构。它不再遵循固定的线性流程，而是通过智能体动态决策每个步骤的执行方式。
3.2 Agentic RAG vs 传统RAG
┌─────────────────────────────────────────────────────────────────────┐
│ 架构对比：传统RAG vs Agentic RAG │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ 传统RAG（线性管道） Agentic RAG（智能决策网络） │
│ ┌─────────┐ ┌─────────┐ │
│ │ 查询 │ │ 查询 │ │
│ └────┬────┘ └────┬────┘ │
│ │ │ │
│ v v │
│ ┌─────────┐ ┌───────────────┐ │
│ │ 检索 │ │ 查询分析Agent │ ← 判断是否需要检索 │
│ └────┬────┘ └───────┬───────┘ │
│ │ │ │
│ v v │
│ ┌─────────┐ ┌───────────────┐ │
│ │ 生成 │ │ 检索Agent │ ← 决定检索策略 │
│ └─────────┘ └───────┬───────┘ │
│ │ │
│ v │
│ ┌───────────────┐ │
│ │ 验证Agent │ ← 检查结果质量 │
│ └───────┬───────┘ │
│ │ │
│ v │
│ ┌───────────────┐ │
│ │ 生成Agent │ ← 生成最终回答 │
│ └───────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘
3.3 Agentic RAG的核心特征
| 特征 | 传统RAG | Agentic RAG |
|---|---|---|
| 决策能力 | 无，固定流程 | 有，动态决策 |
| 检索次数 | 单次 | 可多次迭代 |
| 查询改写 | 无 | 智能改写优化 |
| 自我验证 | 无 | 幻觉检测与纠正 |
| 工具使用 | 仅向量检索 | 多工具协同 |
| 推理能力 | 无 | 多步推理规划 |
四、Agentic RAG架构深度解析
4.1 Agentic RAG的核心组件
# Agentic RAG核心组件
class AgenticRAGComponents:
"""Agentic RAG的核心组件"""
def __init__(self):
# 1. 查询分析器 - 理解用户意图
self.query_analyzer = QueryAnalyzer()
# 2. 检索策略选择器 - 选择最佳检索方式
self.retrieval_selector = RetrievalSelector()
# 3. 文档评估器 - 评估检索结果质量
self.doc_evaluator = DocumentEvaluator()
# 4. 答案生成器 - 生成最终回答
self.answer_generator = AnswerGenerator()
# 5. 幻觉检测器 - 验证答案准确性
self.hallucination_checker = HallucinationChecker()
# 6. 查询改写器 - 优化检索查询
self.query_rewriter = QueryRewriter()
4.2 Agentic RAG的工作流程
┌─────────────────────────────────────────────────────────────────┐
│ Agentic RAG 工作流程 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────┐ │
│ │ 用户输入 │ │
│ └────┬─────┘ │
│ │ │
│ v │
│ ┌─────────────────────────────────────┐ │
│ │ Step 1: 查询分析 (Query Analysis) │ │
│ │ - 理解查询意图 │ │
│ │ - 判断复杂度 │ │
│ │ - 决定是否需要检索 │ │
│ └──────────────┬──────────────────────┘ │
│ │ │
│ ┌─────────┴─────────┐ │
│ │ │ │
│ v v │
│ ┌─────────┐ ┌─────────┐ │
│ │直接回答 │ │需要检索 │ │
│ └────┬────┘ └────┬────┘ │
│ │ │ │
│ v v │
│ ┌──────────┐ ┌─────────────────────────────────────┐ │
│ │ 生成答案 │ │ Step 2: 检索策略选择 │ │
│ └──────────┘ │ - 选择检索工具 │ │
│ │ - 确定检索参数 │ │
│ └──────────────┬──────────────────────┘ │
│ │ │
│ v │
│ ┌─────────────────────────────────────┐ │
│ │ Step 3: 执行检索 │ │
│ │ - 向量检索 │ │
│ │ - 关键词检索 │ │
│ │ - 网络搜索 │ │
│ └──────────────┬──────────────────────┘ │
│ │ │
│ v │
│ ┌─────────────────────────────────────┐ │
│ │ Step 4: 结果评估 │ │
│ │ - 评估文档相关性 │ │
│ │ - 判断是否足够 │ │
│ └──────────────┬──────────────────────┘ │
│ │ │
│ ┌────────────┴────────────┐ │
│ │ │ │
│ v v │
│ ┌──────────────┐ ┌──────────────┐ │
│ │ 结果充足 │ │ 结果不足 │ │
│ └──────┬───────┘ └──────┬───────┘ │
│ │ │ │
│ v v │
│ ┌──────────┐ ┌─────────────────────┐ │
│ │生成答案 │ │ Step 5: 查询改写 │ │
│ └──────────┘ │ - 优化查询词 │ │
│ │ - 重新检索 │ │
│ └──────────┬──────────┘ │
│ │ │
│ └──────→ (回到Step 3)│
│ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ Step 6: 答案验证 (Hallucination Check) │ │
│ │ - 验证答案是否基于检索内容 │ │
│ │ - 检测幻觉 │ │
│ └────────────────────────┬────────────────────────────────┘ │
│ │ │
│ ┌─────────┴─────────┐ │
│ │ │ │
│ v v │
│ ┌──────────┐ ┌──────────┐ │
│ │ 验证通过 │ │ 验证失败 │ │
│ └────┬─────┘ └────┬─────┘ │
│ │ │ │
│ v v │
│ ┌──────────┐ ┌──────────┐ │
│ │ 返回答案 │ │ 重新生成 │ │
│ └──────────┘ └──────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘
4.3 Agentic RAG的关键技术
4.3.1 查询分析与路由
from typing import Literal
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
class QueryAnalysis(BaseModel):
"""查询分析结果"""
query_type: Literal["factual", "analytical", "creative", "ambiguous"] = Field(
description="查询类型：事实性、分析性、创造性、模糊"
)
complexity: Literal["simple", "medium", "complex"] = Field(
description="复杂度级别"
)
needs_retrieval: bool = Field(
description="是否需要检索外部知识"
)
reasoning_steps: int = Field(
description="预估需要的推理步骤数"
)
def analyze_query(query: str) -> QueryAnalysis:
"""分析用户查询"""
prompt = ChatPromptTemplate.from_messages([
("system", """分析用户查询的特征，包括：
1. 查询类型：事实性(需要具体信息)、分析性(需要推理)、创造性(需要生成)、模糊(需要澄清)
2. 复杂度：简单(直接回答)、中等(需要一些推理)、复杂(需要多步推理)
3. 是否需要检索：判断是否需要外部知识
4. 推理步骤：预估需要的推理步骤数"""),
("human", "查询: {query}")
])
llm = ChatOpenAI(model="gpt-4o-mini")
structured_llm = llm.with_structured_output(QueryAnalysis)
chain = prompt | structured_llm
return chain.invoke({"query": query})
# 使用示例
result = analyze_query("比较Transformer和RNN在序列建模中的优缺点")
print(f"查询类型: {result.query_type}")
print(f"复杂度: {result.complexity}")
print(f"需要检索: {result.needs_retrieval}")
4.3.2 自适应检索策略
from typing import List, Optional
from langchain_core.documents import Document
class AdaptiveRetriever:
"""自适应检索器"""
def __init__(self, vectorstore, web_search_tool=None):
self.vectorstore = vectorstore
self.web_search = web_search_tool
def retrieve(self, query: str, strategy: str = "auto") -> List[Document]:
"""
根据策略执行检索
Args:
query: 查询字符串
strategy: 检索策略 ("vector", "keyword", "web", "hybrid", "auto")
"""
if strategy == "auto":
strategy = self._select_strategy(query)
if strategy == "vector":
return self._vector_search(query)
elif strategy == "keyword":
return self._keyword_search(query)
elif strategy == "web":
return self._web_search(query)
elif strategy == "hybrid":
return self._hybrid_search(query)
else:
return self._vector_search(query)
def _select_strategy(self, query: str) -> str:
"""自动选择检索策略"""
# 简单启发式规则
if any(word in query.lower() for word in ["latest", "recent", "news", "2024", "2025"]):
return "web" # 需要最新信息
elif any(word in query.lower() for word in ["compare", "difference", "vs"]):
return "hybrid" # 需要对比，使用混合检索
elif len(query.split()) < 5:
return "keyword" # 短查询，关键词检索
else:
return "vector" # 默认向量检索
def _vector_search(self, query: str) -> List[Document]:
"""向量相似度搜索"""
retriever = self.vectorstore.as_retriever(search_kwargs={"k": 5})
return retriever.invoke(query)
def _keyword_search(self, query: str) -> List[Document]:
"""关键词搜索（使用BM25）"""
from langchain_community.retrievers import BM25Retriever
# 实现关键词检索逻辑
pass
def _web_search(self, query: str) -> List[Document]:
"""网络搜索"""
if self.web_search:
results = self.web_search.invoke(query)
return [Document(page_content=r["content"], metadata={"source": r["url"]})
for r in results]
return []
def _hybrid_search(self, query: str) -> List[Document]:
"""混合检索：向量 + 关键词"""
vector_results = self._vector_search(query)
keyword_results = self._keyword_search(query)
# 融合两种检索结果
return self._merge_results(vector_results, keyword_results)
def _merge_results(self, results1: List[Document], results2: List[Document]) -> List[Document]:
"""合并检索结果并去重"""
seen = set()
merged = []
for doc in results1 + results2:
key = doc.page_content[:100] # 使用内容前100字符作为去重键
if key not in seen:
seen.add(key)
merged.append(doc)
return merged[:10] # 返回前10个结果
4.3.3 迭代检索与查询改写
class IterativeRetriever:
"""迭代检索器 - 支持多轮检索"""
def __init__(self, llm, retriever, max_iterations=3):
self.llm = llm
self.retriever = retriever
self.max_iterations = max_iterations
def retrieve_with_refinement(self, query: str) -> List[Document]:
"""
通过迭代优化进行检索
策略：
1. 初始检索
2. 评估结果质量
3. 如果质量不足，改写查询并重新检索
4. 重复直到满意或达到最大迭代次数
"""
all_docs = []
current_query = query
for i in range(self.max_iterations):
# 执行检索
docs = self.retriever.invoke(current_query)
all_docs.extend(docs)
# 评估检索质量
quality_score = self._evaluate_quality(docs, query)
if quality_score >= 0.8: # 质量足够
break
if i < self.max_iterations - 1: # 还有迭代次数
# 改写查询
current_query = self._rewrite_query(current_query, docs, query)
print(f"迭代 {i+1}: 查询改写为 -> {current_query}")
# 去重并返回
return self._deduplicate(all_docs)
def _evaluate_quality(self, docs: List[Document], query: str) -> float:
"""评估检索结果质量"""
if not docs:
return 0.0
# 使用LLM评估相关性
prompt = f"""评估以下文档与用户查询的相关性（0-1分）：
查询: {query}
文档:
{docs[0].page_content[:500]}
只返回一个0到1之间的数字，表示相关性得分。"""
try:
response = self.llm.invoke(prompt)
score = float(response.content.strip())
return min(max(score, 0.0), 1.0) # 确保在0-1范围内
except:
return 0.5 # 默认中等质量
def _rewrite_query(self, current_query: str, docs: List[Document], original_query: str) -> str:
"""基于已有结果改写查询"""
prompt = f"""基于原始查询和已检索到的信息，生成一个新的查询以获取更多相关信息。
原始查询: {original_query}
当前查询: {current_query}
已检索到的信息片段:
{chr(10).join([f"- {doc.page_content[:200]}..." for doc in docs[:3]])}
请生成一个更具体、更有针对性的查询，以获取缺失的信息。只返回查询字符串。"""
response = self.llm.invoke(prompt)
return response.content.strip()
def _deduplicate(self, docs: List[Document]) -> List[Document]:
"""文档去重"""
seen = set()
unique = []
for doc in docs:
key = hash(doc.page_content[:200])
if key not in seen:
seen.add(key)
unique.append(doc)
return unique
4.3.4 幻觉检测与自我纠正
from typing import Tuple
class HallucinationDetector:
"""幻觉检测器"""
def __init__(self, llm):
self.llm = llm
def check(self, answer: str, context: List[Document]) -> Tuple[bool, str]:
"""
检测答案中是否存在幻觉
Returns:
(是否存在幻觉, 解释说明)
"""
context_text = "\n\n".join([doc.page_content for doc in context])
prompt = f"""验证以下答案是否完全基于提供的上下文，是否存在幻觉（编造信息）。
上下文:
{context_text[:2000]}
答案:
{answer}
请分析：
1. 答案中的每个事实是否都能在上下文中找到依据？
2. 是否存在上下文之外的信息？
3. 是否存在过度推断？
以JSON格式返回：
{{
"has_hallucination": true/false,
"explanation": "详细解释",
"unsupported_claims": ["具体的不受支持的陈述"]
}}"""
try:
response = self.llm.invoke(prompt)
# 解析JSON响应
import json
result = json.loads(response.content)
return result.get("has_hallucination", False), result.get("explanation", "")
except:
# 如果解析失败，保守起见认为可能存在幻觉
return True, "无法验证答案"
def correct(self, answer: str, context: List[Document]) -> str:
"""纠正幻觉，基于上下文重新生成答案"""
context_text = "\n\n".join([doc.page_content for doc in context])
prompt = f"""基于以下上下文重新生成答案。确保答案完全基于上下文，不要添加外部信息。
上下文:
{context_text[:2000]}
原答案（可能包含幻觉）:
{answer}
请生成一个准确、基于上下文的答案："""
response = self.llm.invoke(prompt)
return response.content
class SelfCorrectingRAG:
"""自我纠正的RAG系统"""
def __init__(self, llm, retriever, max_retries=2):
self.llm = llm
self.retriever = retriever
self.detector = HallucinationDetector(llm)
self.max_retries = max_retries
def generate(self, query: str) -> dict:
"""生成答案，包含自我纠正机制"""
# 检索相关文档
docs = self.retriever.invoke(query)
context = "\n\n".join([doc.page_content for doc in docs])
# 初始生成
answer = self._generate_answer(query, context)
# 幻觉检测与纠正
for i in range(self.max_retries):
has_hallucination, explanation = self.detector.check(answer, docs)
if not has_hallucination:
break
print(f"检测到幻觉，正在纠正... (尝试 {i+1}/{self.max_retries})")
print(f"原因: {explanation}")
# 重新生成
answer = self.detector.correct(answer, docs)
return {
"answer": answer,
"context": docs,
"has_hallucination": has_hallucination if 'has_hallucination' in locals() else False
}
def _generate_answer(self, query: str, context: str) -> str:
"""基于上下文生成答案"""
prompt = f"""基于以下上下文回答问题。
上下文:
{context}
问题: {query}
答案:"""
response = self.llm.invoke(prompt)
return response.content
五、使用LangGraph构建Agentic RAG
5.1 LangGraph简介
LangGraph 是LangChain的扩展库，专门用于构建复杂的多智能体工作流。它使用图（Graph）结构来定义状态机，非常适合实现Agentic RAG。
5.2 Agentic RAG状态机设计
from typing import TypedDict, List, Annotated
from langgraph.graph import StateGraph, END
from langchain_core.documents import Document
import operator
class AgenticRAGState(TypedDict):
"""Agentic RAG的状态定义"""
query: str # 用户查询
query_type: str # 查询类型
documents: List[Document] # 检索到的文档
rewritten_queries: List[str] # 改写后的查询历史
iteration: int # 当前迭代次数
answer: str # 生成的答案
needs_retrieval: bool # 是否需要检索
quality_score: float # 答案质量评分
has_hallucination: bool # 是否存在幻觉
def create_agentic_rag_graph(llm, retriever):
"""创建Agentic RAG的状态图"""
# 1. 查询分析节点
def analyze_query_node(state: AgenticRAGState) -> AgenticRAGState:
query = state["query"]
# 分析查询
analysis = analyze_query(query)
return {
**state,
"query_type": analysis.query_type,
"needs_retrieval": analysis.needs_retrieval,
"iteration": 0,
"rewritten_queries": []
}
# 2. 检索节点
def retrieve_node(state: AgenticRAGState) -> AgenticRAGState:
query = state["query"]
# 如果有改写查询，使用最新的
if state["rewritten_queries"]:
query = state["rewritten_queries"][-1]
# 执行检索
docs = retriever.invoke(query)
return {
**state,
"documents": docs,
"iteration": state["iteration"] + 1
}
# 3. 评估节点
def evaluate_node(state: AgenticRAGState) -> AgenticRAGState:
docs = state["documents"]
query = state["query"]
# 评估检索质量
detector = HallucinationDetector(llm)
quality_score = detector._evaluate_quality(docs, query)
return {
**state,
"quality_score": quality_score
}
# 4. 查询改写节点
def rewrite_node(state: AgenticRAGState) -> AgenticRAGState:
query = state["query"]
docs = state["documents"]
# 改写查询
iterative_retriever = IterativeRetriever(llm, retriever)
rewritten = iterative_retriever._rewrite_query(
state["rewritten_queries"][-1] if state["rewritten_queries"] else query,
docs,
query
)
return {
**state,
"rewritten_queries": state["rewritten_queries"] + [rewritten]
}
# 5. 生成答案节点
def generate_node(state: AgenticRAGState) -> AgenticRAGState:
query = state["query"]
docs = state["documents"]
context = "\n\n".join([doc.page_content for doc in docs])
# 生成答案
prompt = f"""基于以下上下文回答问题。
上下文:
{context}
问题: {query}
请提供详细、准确的答案："""
response = llm.invoke(prompt)
return {
**state,
"answer": response.content
}
# 6. 幻觉检测节点
def hallucination_check_node(state: AgenticRAGState) -> AgenticRAGState:
answer = state["answer"]
docs = state["documents"]
detector = HallucinationDetector(llm)
has_hallucination, _ = detector.check(answer, docs)
return {
**state,
"has_hallucination": has_hallucination
}
# 7. 直接回答节点（无需检索）
def direct_answer_node(state: AgenticRAGState) -> AgenticRAGState:
query = state["query"]
response = llm.invoke(f"回答以下问题：{query}")
return {
**state,
"answer": response.content,
"documents": []
}
# 构建图
workflow = StateGraph(AgenticRAGState)
# 添加节点
workflow.add_node("analyze", analyze_query_node)
workflow.add_node("retrieve", retrieve_node)
workflow.add_node("evaluate", evaluate_node)
workflow.add_node("rewrite", rewrite_node)
workflow.add_node("generate", generate_node)
workflow.add_node("hallucination_check", hallucination_check_node)
workflow.add_node("direct_answer", direct_answer_node)
# 设置入口
workflow.set_entry_point("analyze")
# 添加条件边
workflow.add_conditional_edges(
"analyze",
lambda state: "retrieve" if state["needs_retrieval"] else "direct_answer"
)
workflow.add_edge("direct_answer", END)
workflow.add_edge("retrieve", "evaluate")
workflow.add_conditional_edges(
"evaluate",
lambda state: (
"generate" if state["quality_score"] >= 0.7 or state["iteration"] >= 3
else "rewrite"
)
)
workflow.add_edge("rewrite", "retrieve")
workflow.add_edge("generate", "hallucination_check")
workflow.add_conditional_edges(
"hallucination_check",
lambda state: (
END if not state["has_hallucination"]
else "generate" # 可以添加一个纠正节点
)
)
# 编译图
return workflow.compile()
# 使用示例
# graph = create_agentic_rag_graph(llm, retriever)
# result = graph.invoke({"query": "什么是Agentic RAG？"})
# print(result["answer"])
5.3 可视化工作流
def visualize_workflow(graph):
"""可视化Agentic RAG工作流"""
try:
from IPython.display import Image, display
# 生成Mermaid图
mermaid_code = """
graph TD
A[查询分析] -->|需要检索| B[检索文档]
A -->|直接回答| C[生成答案]
B --> D[评估质量]
D -->|质量不足| E[改写查询]
E --> B
D -->|质量足够| F[生成答案]
F --> G[幻觉检测]
G -->|存在幻觉| H[纠正答案]
H --> G
G -->|通过检测| I[返回答案]
C --> I
"""
print("Agentic RAG工作流:")
print(mermaid_code)
except ImportError:
print("请安装IPython以查看可视化图形")
# 可视化
# visualize_workflow(graph)
六、RAG vs Agentic RAG 实战对比
6.1 测试场景设计
# 测试用例
test_cases = [
{
"name": "简单事实查询",
"query": "什么是RAG技术？",
"expected_behavior": "传统RAG和Agentic RAG都应该能很好回答"
},
{
"name": "复杂多跳查询",
"query": "比较Transformer和RNN在序列建模中的优缺点，并说明为什么Transformer更适合并行计算",
"expected_behavior": "Agentic RAG应该进行多步检索和推理"
},
{
"name": "模糊查询",
"query": "那个新技术",
"expected_behavior": "Agentic RAG应该进行查询澄清或改写"
},
{
"name": "需要最新信息",
"query": "2024年最新的LLM模型有哪些？",
"expected_behavior": "Agentic RAG应该使用网络搜索工具"
}
]
6.2 性能对比
| 指标 | 传统RAG | Agentic RAG | 说明 |
|---|---|---|---|
| 响应时间 | 快（1-2秒） | 较慢（3-10秒） | Agentic需要多步推理 |
| 准确率 | 中等（70-80%） | 高（85-95%） | Agentic能处理更复杂问题 |
| 复杂问题处理能力 | 弱 | 强 | Agentic支持多步推理 |
| 幻觉率 | 中等（15-20%） | 低（5-10%） | Agentic有自我验证机制 |
| 成本 | 低 | 高 | Agentic需要更多LLM调用 |
| 可解释性 | 低 | 高 | Agentic有明确的推理步骤 |
6.3 选择建议
使用传统RAG的场景：
- ✅ 查询简单明确，不需要复杂推理
- ✅ 对响应时间要求高
- ✅ 预算有限，需要控制成本
- ✅ 知识库稳定，不需要频繁更新
使用Agentic RAG的场景：
- ✅ 查询复杂，需要多步推理
- ✅ 对答案准确性要求极高
- ✅ 需要处理模糊或不完整的查询
- ✅ 需要最新信息（结合网络搜索）
- ✅ 需要可解释的推理过程
七、最佳实践与优化建议
7.1 混合架构设计
class HybridRAG:
"""混合RAG架构 - 根据查询复杂度自动选择策略"""
def __init__(self, simple_rag, agentic_rag):
self.simple_rag = simple_rag
self.agentic_rag = agentic_rag
self.llm = ChatOpenAI(model="gpt-4o-mini")
def route_query(self, query: str) -> str:
"""路由查询到合适的RAG系统"""
prompt = f"""判断以下查询的复杂度，选择最适合的处理方式。
查询: {query}
选项:
1. "simple" - 简单查询，直接回答或单次检索即可
2. "agentic" - 复杂查询，需要多步推理、查询改写或工具使用
只返回 "simple" 或 "agentic"。"""
response = self.llm.invoke(prompt)
return response.content.strip().lower()
def query(self, query: str) -> dict:
"""执行查询"""
route = self.route_query(query)
if route == "simple":
print("使用传统RAG处理...")
return self.simple_rag.invoke({"input": query})
else:
print("使用Agentic RAG处理...")
return self.agentic_rag.invoke({"query": query})
7.2 成本优化策略
- 分层检索：先使用轻量级模型进行初步筛选
- 缓存机制：缓存常见查询的结果
- 智能路由：简单查询使用传统RAG，复杂查询才使用Agentic RAG
- 批量处理：合并多个查询进行批量处理
7.3 评估与监控
class RAGEvaluator:
"""RAG系统评估器"""
def __init__(self, llm):
self.llm = llm
def evaluate(self, query: str, answer: str, context: List[Document]) -> dict:
"""评估RAG回答质量"""
# 1. 相关性评估
relevance = self._evaluate_relevance(query, context)
# 2. 忠实度评估（是否有幻觉）
faithfulness = self._evaluate_faithfulness(answer, context)
# 3. 完整性评估
completeness = self._evaluate_completeness(query, answer)
return {
"relevance": relevance,
"faithfulness": faithfulness,
"completeness": completeness,
"overall": (relevance + faithfulness + completeness) / 3
}
def _evaluate_relevance(self, query: str, context: List[Document]) -> float:
"""评估检索文档的相关性"""
# 实现相关性评估逻辑
pass
def _evaluate_faithfulness(self, answer: str, context: List[Document]) -> float:
"""评估答案的忠实度"""
detector = HallucinationDetector(self.llm)
has_hallucination, _ = detector.check(answer, context)
return 0.0 if has_hallucination else 1.0
def _evaluate_completeness(self, query: str, answer: str) -> float:
"""评估答案的完整性"""
# 实现完整性评估逻辑
pass
八、总结与展望
8.1 核心要点回顾
- 传统RAG：线性管道架构，适合简单查询，成本低但能力有限
- Agentic RAG：智能决策架构，适合复杂查询，能力强但成本高
- 核心差异：Agentic RAG引入了查询分析、自适应检索、迭代优化和自我验证
- LangGraph：构建Agentic RAG的理想工具，支持状态机和工作流编排
8.2 技术演进趋势
| 阶段 | 特点 | 代表技术 |
|---|---|---|
| RAG 1.0 | 基础检索+生成 | 向量检索、简单prompt |
| RAG 2.0 | 高级检索策略 | Hybrid Search、Reranking |
| RAG 3.0 | Agentic RAG | LangGraph、多智能体协作 |
| RAG 4.0 | 自主智能体 | 完全自主规划、执行、验证 |
8.3 未来发展方向
- 多模态RAG：支持文本、图像、音频、视频的统一检索
- 联邦RAG：跨多个知识源的分布式检索
- 个性化RAG：基于用户画像的个性化检索策略
- 实时RAG：支持流式数据的实时检索与更新
- 可解释RAG：提供清晰的推理过程和决策依据
8.4 学习资源
结语：从传统RAG到Agentic RAG的演进，代表了AI系统从被动工具向主动智能体的转变。理解这两种架构的差异和适用场景，将帮助你构建更强大、更可靠的AI应用。
选择建议：
- 从传统RAG开始，满足基本需求
- 逐步引入Agentic特性，解决复杂场景
- 最终构建混合架构，实现最佳性价比
有“AI”的1024 = 2048，欢迎大家加入2048 AI社区
更多推荐
4
0- 0
已为社区贡献23条内容
所有评论(0)