🚀 从传统 RAG 到 Agentic RAG：端到端完全指南

---

📑 目录

1. 什么是 RAG？
2. 传统 RAG 的实现方式
3. 传统 RAG 的五大局限性
4. 什么是 Agentic RAG？
5. Agentic RAG 核心架构
6. 四大 Agentic 设计模式
7. 完整代码实现
8. 传统 RAG vs Agentic RAG 对比
9. 最佳实践与调优
10. 总结与展望

---

1. 什么是 RAG？

RAG (Retrieval-Augmented Generation) 检索增强生成，是一种将外部知识检索与大语言模型生成相结合的技术架构。

🎯 RAG 解决的核心问题

问题	RAG 如何解决
LLM 知识过时	从实时知识库检索最新信息
LLM 幻觉问题	基于真实文档生成，有据可查
领域知识不足	接入企业私有知识库
无法引用来源	可追溯到原始文档

📊 RAG 三大核心组件

┌─────────────────────────────────────────────────────────────┐
│                      RAG 系统架构                            │
├─────────────────┬─────────────────┬─────────────────────────┤
│   📚 知识库      │   🔍 检索器      │   🧠 生成器 (LLM)        │
│                 │                 │                         │
│ • 文档存储       │ • Embedding     │ • GPT-4 / Claude        │
│ • 向量数据库     │ • 相似度计算     │ • 上下文理解             │
│ • 索引管理       │ • Top-K 召回    │ • 答案生成               │
└─────────────────┴─────────────────┴─────────────────────────┘

---

2. 传统 RAG 的实现方式

📈 传统 RAG 流程图

用户查询 ──→ Query Embedding ──→ 向量检索 ──→ Top-K 文档 ──→ Prompt 组装 ──→ LLM 生成 ──→ 输出答案
   │              │                 │              │              │              │
   ▼              ▼                 ▼              ▼              ▼              ▼
"什么是     [0.12, 0.85,      Cosine        [Doc1,        "Context:      "RAG是一种
 RAG?"      0.33, ...]       Similarity     Doc2,         {docs}         结合检索与
                                            Doc3]         Q: ..."        生成的技术"

🔧 传统 RAG 代码实现

Step 1: 环境准备

# 安装依赖
# pip install langchain langchain-openai chromadb tiktoken

from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

Step 2: 文档处理与索引构建

# 1. 加载文档
documents = [
    "RAG（检索增强生成）是一种AI架构，结合了信息检索和文本生成...",
    "向量数据库用于存储文档的embedding向量，支持高效的相似度搜索...",
    "LangChain是一个用于构建LLM应用的框架，提供了丰富的组件..."
]

# 2. 文档切分
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,      # 每个chunk的最大字符数
    chunk_overlap=50,    # chunk之间的重叠字符数
    separators=["\n\n", "\n", "。", "，", " "]
)
chunks = text_splitter.create_documents(documents)

# 3. 创建向量存储
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

# 4. 创建检索器
retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 3}  # 返回最相似的3个文档
)

Step 3: RAG Chain 构建

# 5. 定义 Prompt 模板
template = """你是一个专业的AI助手。请根据以下上下文回答问题。
如果上下文中没有相关信息，请说"我没有找到相关信息"。

上下文：
{context}

问题：{question}

回答："""

prompt = ChatPromptTemplate.from_template(template)

# 6. 初始化 LLM
llm = ChatOpenAI(model="gpt-4", temperature=0)

# 7. 构建 RAG Chain
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs)

rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

# 8. 执行查询
response = rag_chain.invoke("什么是RAG？")
print(response)

---

3. 传统 RAG 的五大局限性

❌ 局限性详解

┌────────────────────────────────────────────────────────────────────┐
│                    传统 RAG 的五大痛点                              │
├────────────────────┬───────────────────────────────────────────────┤
│ 🔴 单次检索        │ 检索一次就结束，失败无法重试                    │
├────────────────────┼───────────────────────────────────────────────┤
│ 🔴 固定策略        │ 所有查询用同样的检索方式，无法适应不同场景        │
├────────────────────┼───────────────────────────────────────────────┤
│ 🔴 无质量评估      │ 不知道检索到的文档是否真正相关                   │
├────────────────────┼───────────────────────────────────────────────┤
│ 🔴 单一数据源      │ 只能访问向量库，无法查询数据库、API、网络         │
├────────────────────┼───────────────────────────────────────────────┤
│ 🔴 简单推理        │ 无法处理需要多步推理的复杂问题                   │
└────────────────────┴───────────────────────────────────────────────┘

📋 具体案例分析

案例：用户问 “2024年诺贝尔物理学奖得主的主要研究成果是什么？”

步骤	传统 RAG 的处理	问题
1	在知识库中搜索	知识库可能没有2024年数据
2	返回空结果或旧数据	无法判断数据是否过时
3	LLM 基于错误上下文回答	可能产生幻觉
4	输出错误答案	没有纠错机制

Agentic RAG 的处理方式：

1. Agent 判断：这是时效性问题，需要联网搜索
2. 调用 Web Search 工具获取最新信息
3. 评估搜索结果的可靠性
4. 如果信息不完整，继续搜索补充
5. 综合多源信息生成答案

---

4. 什么是 Agentic RAG？

🤖 定义

Agentic RAG 是将 自主 AI Agent 嵌入 RAG 流程的新一代架构。Agent 能够：

• 🎯 规划：分析问题，制定检索策略
• 🔧 工具调用：动态选择和使用多种工具
• 🔄 反思：评估结果质量，决定是否重试
• 📚 记忆：保持上下文，支持多轮交互

📊 Agentic RAG 架构图

                           ┌─────────────────────────────────────┐
                           │         🤖 AI Agent (核心)           │
                           │  ┌─────────────────────────────────┐ │
                           │  │  🧠 推理引擎 (LLM)              │ │
                           │  │  • 理解用户意图                 │ │
                           │  │  • 制定检索策略                 │ │
                           │  │  • 评估结果质量                 │ │
                           │  │  • 决定下一步行动               │ │
                           │  └─────────────────────────────────┘ │
                           └──────────────┬──────────────────────┘
                                          │
              ┌───────────────────────────┼───────────────────────────┐
              │                           │                           │
              ▼                           ▼                           ▼
    ┌─────────────────┐       ┌─────────────────┐       ┌─────────────────┐
    │ 📚 向量检索工具   │       │ 🌐 Web搜索工具   │       │ 🗄️ SQL查询工具   │
    │                 │       │                 │       │                 │
    │ • 语义相似搜索   │       │ • 实时网络搜索   │       │ • 结构化数据查询  │
    │ • 私有知识库     │       │ • 新闻/论文检索  │       │ • 数据库操作      │
    └─────────────────┘       └─────────────────┘       └─────────────────┘
              │                           │                           │
              └───────────────────────────┼───────────────────────────┘
                                          │
                                          ▼
                           ┌─────────────────────────────────────┐
                           │         📝 响应生成 & 质量检查        │
                           │  • 综合多源信息                      │
                           │  • 幻觉检测                         │
                           │  • 必要时迭代优化                    │
                           └─────────────────────────────────────┘

---

5. Agentic RAG 核心架构

🏗️ 四层架构模型

┌─────────────────────────────────────────────────────────────────────┐
│                        Layer 4: 编排层                              │
│   ┌─────────────┐  ┌─────────────┐  ┌─────────────┐                │
│   │ 工作流管理   │  │  状态追踪   │  │  错误恢复   │                │
│   └─────────────┘  └─────────────┘  └─────────────┘                │
├─────────────────────────────────────────────────────────────────────┤
│                        Layer 3: Agent层                             │
│   ┌─────────────┐  ┌─────────────┐  ┌─────────────┐                │
│   │  路由Agent  │  │  检索Agent  │  │  评估Agent  │                │
│   │ (分发任务)   │  │ (执行检索)   │  │ (质量把控)  │                │
│   └─────────────┘  └─────────────┘  └─────────────┘                │
├─────────────────────────────────────────────────────────────────────┤
│                        Layer 2: 工具层                              │
│   ┌───────┐ ┌───────┐ ┌───────┐ ┌───────┐ ┌───────┐               │
│   │向量检索│ │Web搜索│ │SQL查询│ │API调用│ │代码执行│               │
│   └───────┘ └───────┘ └───────┘ └───────┘ └───────┘               │
├─────────────────────────────────────────────────────────────────────┤
│                        Layer 1: 数据层                              │
│   ┌─────────────┐  ┌─────────────┐  ┌─────────────┐                │
│   │  向量数据库  │  │  关系数据库  │  │  文档存储   │                │
│   │  (Chroma)   │  │  (MySQL)    │  │  (S3/本地)  │                │
│   └─────────────┘  └─────────────┘  └─────────────┘                │
└─────────────────────────────────────────────────────────────────────┘

🔄 Agentic RAG 工作流程

┌──────────┐     ┌──────────┐     ┌──────────┐     ┌──────────┐
│ 用户查询  │────▶│ Agent    │────▶│ 工具选择  │────▶│ 执行检索  │
│          │     │ 规划     │     │ & 路由   │     │          │
└──────────┘     └──────────┘     └──────────┘     └────┬─────┘
                                                        │
     ┌──────────────────────────────────────────────────┘
     │
     ▼
┌──────────┐     ┌──────────┐     ┌──────────┐     ┌──────────┐
│ 文档评估  │────▶│ 是否足够? │──N─▶│ 查询重写  │────▶│ 补充检索  │
│          │     │          │     │ / 换工具  │     │          │
└──────────┘     └────┬─────┘     └──────────┘     └──────────┘
                      │Y
                      ▼
               ┌──────────┐     ┌──────────┐     ┌──────────┐
               │ 生成回答  │────▶│ 幻觉检查  │────▶│ 输出结果  │
               │          │     │          │     │          │
               └──────────┘     └──────────┘     └──────────┘

---

6. 四大 Agentic 设计模式

📐 设计模式详解

模式 1: 🔄 Reflection (反思)

# Agent 自我评估和改进
def reflection_pattern(query, response):
    evaluation = llm.invoke(f"""
    评估以下回答的质量：
    问题：{query}
    回答：{response}
    
    检查：
    1. 是否完整回答了问题？
    2. 是否有事实错误？
    3. 是否需要补充信息？
    
    输出：PASS 或 NEEDS_IMPROVEMENT + 改进建议
    """)
    return evaluation

模式 2: 🛠️ Tool Use (工具使用)

# Agent 动态选择和调用工具
tools = {
    "vector_search": vector_retriever,
    "web_search": tavily_search,
    "sql_query": sql_executor,
    "calculator": math_tool
}

def tool_use_pattern(query):
    # LLM 决定使用哪个工具
    tool_choice = llm.invoke(f"""
    根据问题选择最合适的工具：
    问题：{query}
    可用工具：{list(tools.keys())}
    """)
    return tools[tool_choice](query)

模式 3: 📋 Planning (规划)

# Agent 制定多步执行计划
def planning_pattern(complex_query):
    plan = llm.invoke(f"""
    为以下复杂问题制定分步执行计划：
    问题：{complex_query}
    
    输出格式：
    Step 1: [行动]
    Step 2: [行动]
    ...
    """)
    
    results = []
    for step in parse_steps(plan):
        result = execute_step(step)
        results.append(result)
    
    return synthesize(results)

模式 4: 🤝 Multi-Agent (多Agent协作)

# 多个专业Agent协同工作
class MultiAgentSystem:
    def __init__(self):
        self.router_agent = RouterAgent()      # 路由分发
        self.retrieval_agent = RetrievalAgent() # 检索执行
        self.grader_agent = GraderAgent()       # 质量评估
        self.writer_agent = WriterAgent()       # 答案生成
    
    def process(self, query):
        # 1. 路由Agent分析并分发
        task_type = self.router_agent.analyze(query)
        
        # 2. 检索Agent执行检索
        docs = self.retrieval_agent.retrieve(query, task_type)
        
        # 3. 评估Agent检查质量
        graded_docs = self.grader_agent.grade(docs, query)
        
        # 4. 写作Agent生成答案
        response = self.writer_agent.generate(query, graded_docs)
        
        return response

---

7. 完整代码实现

🛠️ 使用 LangGraph 实现 Agentic RAG

完整项目结构

agentic_rag/
├── requirements.txt
├── config.py
├── main.py
├── agents/
│   ├── __init__.py
│   ├── retriever.py
│   ├── grader.py
│   └── generator.py
├── tools/
│   ├── __init__.py
│   ├── vector_search.py
│   └── web_search.py
└── graph/
    ├── __init__.py
    ├── state.py
    └── workflow.py

Step 1: 安装依赖

pip install langchain langgraph langchain-openai chromadb tavily-python

Step 2: 定义状态 (state.py)

from typing import TypedDict, List, Annotated
from langchain_core.documents import Document
import operator

class AgentState(TypedDict):
    """Agentic RAG 的状态定义"""
    question: str                                    # 用户问题
    generation: str                                  # 生成的答案
    documents: Annotated[List[Document], operator.add]  # 检索到的文档
    retry_count: int                                 # 重试次数
    web_search_needed: bool                          # 是否需要网络搜索

Step 3: 实现各个节点

# retriever.py - 检索节点
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

def create_retriever():
    embeddings = OpenAIEmbeddings()
    vectorstore = Chroma(
        persist_directory="./chroma_db",
        embedding_function=embeddings
    )
    return vectorstore.as_retriever(search_kwargs={"k": 4})

def retrieve_node(state: AgentState) -> dict:
    """检索相关文档"""
    print("📚 执行向量检索...")
    retriever = create_retriever()
    documents = retriever.invoke(state["question"])
    return {"documents": documents}

# grader.py - 文档评估节点
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

def grade_documents_node(state: AgentState) -> dict:
    """评估文档相关性"""
    print("🔍 评估文档相关性...")
    
    llm = ChatOpenAI(model="gpt-4", temperature=0)
    
    grading_prompt = ChatPromptTemplate.from_template("""
    你是一个文档相关性评估专家。
    
    用户问题：{question}
    文档内容：{document}
    
    该文档是否与问题相关？只回答 'yes' 或 'no'。
    """)
    
    chain = grading_prompt | llm
    
    relevant_docs = []
    for doc in state["documents"]:
        result = chain.invoke({
            "question": state["question"],
            "document": doc.page_content
        })
        if "yes" in result.content.lower():
            relevant_docs.append(doc)
            print(f"  ✅ 相关文档: {doc.page_content[:50]}...")
        else:
            print(f"  ❌ 不相关文档: {doc.page_content[:50]}...")
    
    web_search_needed = len(relevant_docs) < 2
    
    return {
        "documents": relevant_docs,
        "web_search_needed": web_search_needed
    }

# web_search.py - 网络搜索节点
from langchain_community.tools.tavily_search import TavilySearchResults

def web_search_node(state: AgentState) -> dict:
    """执行网络搜索补充信息"""
    print("🌐 执行网络搜索...")
    
    search_tool = TavilySearchResults(max_results=3)
    results = search_tool.invoke(state["question"])
    
    web_docs = []
    for result in results:
        from langchain_core.documents import Document
        doc = Document(
            page_content=result["content"],
            metadata={"source": result["url"]}
        )
        web_docs.append(doc)
        print(f"  🔗 找到: {result['url']}")
    
    return {"documents": web_docs}

# generator.py - 生成节点
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

def generate_node(state: AgentState) -> dict:
    """生成最终答案"""
    print("✍️ 生成答案...")
    
    llm = ChatOpenAI(model="gpt-4", temperature=0)
    
    prompt = ChatPromptTemplate.from_template("""
    你是一个专业的AI助手。请根据以下上下文回答问题。
    
    上下文：
    {context}
    
    问题：{question}
    
    请提供详细、准确的回答。如果上下文信息不足，请明确说明。
    """)
    
    context = "\n\n".join([doc.page_content for doc in state["documents"]])
    
    chain = prompt | llm | StrOutputParser()
    
    response = chain.invoke({
        "context": context,
        "question": state["question"]
    })
    
    return {"generation": response}

Step 4: 构建工作流图 (workflow.py)

from langgraph.graph import StateGraph, END
from state import AgentState
from agents.retriever import retrieve_node
from agents.grader import grade_documents_node
from agents.generator import generate_node
from tools.web_search import web_search_node

def decide_to_search(state: AgentState) -> str:
    """条件边：决定是否需要网络搜索"""
    if state.get("web_search_needed", False):
        print("🔀 路由到: 网络搜索")
        return "web_search"
    print("🔀 路由到: 生成答案")
    return "generate"

def check_hallucination(state: AgentState) -> str:
    """检查是否存在幻觉"""
    # 简化版：实际应该用LLM检查
    retry_count = state.get("retry_count", 0)
    if retry_count < 2 and len(state.get("generation", "")) < 50:
        print("⚠️ 答案可能不完整，重试...")
        return "retry"
    return "end"

def build_agentic_rag_graph():
    """构建 Agentic RAG 工作流"""
    
    # 创建状态图
    workflow = StateGraph(AgentState)
    
    # 添加节点
    workflow.add_node("retrieve", retrieve_node)
    workflow.add_node("grade_documents", grade_documents_node)
    workflow.add_node("web_search", web_search_node)
    workflow.add_node("generate", generate_node)
    
    # 设置入口点
    workflow.set_entry_point("retrieve")
    
    # 添加边
    workflow.add_edge("retrieve", "grade_documents")
    
    # 条件边：根据文档质量决定下一步
    workflow.add_conditional_edges(
        "grade_documents",
        decide_to_search,
        {
            "web_search": "web_search",
            "generate": "generate"
        }
    )
    
    workflow.add_edge("web_search", "generate")
    
    # 条件边：检查生成质量
    workflow.add_conditional_edges(
        "generate",
        check_hallucination,
        {
            "retry": "retrieve",
            "end": END
        }
    )
    
    return workflow.compile()

# 可视化工作流
"""
         ┌─────────────┐
         │   retrieve  │
         └──────┬──────┘
                │
                ▼
      ┌──────────────────┐
      │ grade_documents  │
      └────────┬─────────┘
               │
        ┌──────┴──────┐
        │ 文档足够?    │
        └──────┬──────┘
          Y/   │   \N
           /   │    \
          ▼    │     ▼
  ┌──────────┐ │ ┌──────────┐
  │ generate │ │ │web_search│
  └────┬─────┘ │ └────┬─────┘
       │       │      │
       │       └──────┘
       │
  ┌────┴────┐
  │质量检查? │
  └────┬────┘
    Y/ │ \N
     / │  \
    ▼  │   ▼
  END  │  retry
       └───→ retrieve
"""

Step 5: 主程序 (main.py)

from graph.workflow import build_agentic_rag_graph

def main():
    # 构建工作流
    app = build_agentic_rag_graph()
    
    # 测试查询
    questions = [
        "什么是 Agentic RAG？它和传统RAG有什么区别？",
        "LangGraph 如何实现状态管理？",
        "2024年最新的AI发展趋势是什么？"  # 需要网络搜索
    ]
    
    for question in questions:
        print("\n" + "="*60)
        print(f"❓ 问题: {question}")
        print("="*60)
        
        # 执行工作流
        result = app.invoke({
            "question": question,
            "documents": [],
            "retry_count": 0,
            "web_search_needed": False
        })
        
        print("\n📝 最终答案:")
        print(result["generation"])

if __name__ == "__main__":
    main()

---

8. 传统 RAG vs Agentic RAG 对比

📊 全面对比表

维度	传统 RAG	Agentic RAG
检索方式	单次固定检索	多轮动态检索
数据源	单一向量库	多源异构（向量库+Web+SQL+API）
错误处理	无	自动重试、查询重写
质量控制	无	文档评估 + 幻觉检测
推理能力	简单直接	多步推理、任务分解
适应性	静态	根据场景动态调整策略
复杂度	低	中-高
延迟	低（单次调用）	较高（多轮迭代）
成本	低	较高（更多LLM调用）
适用场景	简单FAQ、知识检索	复杂问答、研究分析、多源整合

🎯 如何选择？

                    任务复杂度
                        ↑
                        │
        ┌───────────────┼───────────────┐
        │               │               │
        │   Agentic     │   Agentic     │
        │   RAG         │   RAG         │
        │   (推荐)      │   (必须)       │
        │               │               │
低 ←────┼───────────────┼───────────────┼────→ 高
数据源   │               │               │    数据源
单一     │  传统 RAG     │   Agentic     │    多样
        │  (足够)       │   RAG         │
        │               │   (推荐)      │
        │               │               │
        └───────────────┼───────────────┘
                        │
                        ↓

---

9. 最佳实践与调优

✅ 设计最佳实践

1. Chunking 策略选择

# 根据内容类型选择不同策略
chunking_strategies = {
    "技术文档": {
        "chunk_size": 200,      # 小chunk，精确检索
        "chunk_overlap": 20,
        "separators": ["\n## ", "\n### ", "\n\n", "\n"]
    },
    "叙事内容": {
        "chunk_size": 800,      # 大chunk，保持上下文
        "chunk_overlap": 100,
        "separators": ["\n\n", "。", "\n"]
    },
    "代码文档": {
        "chunk_size": 500,
        "chunk_overlap": 50,
        "separators": ["\nclass ", "\ndef ", "\n\n", "\n"]
    }
}

2. 检索策略优化

# 混合检索：结合稀疏和稠密检索
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever

# 创建混合检索器
ensemble_retriever = EnsembleRetriever(
    retrievers=[
        bm25_retriever,      # 关键词匹配
        vector_retriever     # 语义相似
    ],
    weights=[0.4, 0.6]       # 权重分配
)

3. 监控与可观测性

# 添加 LangSmith 追踪
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "agentic-rag"

# 自定义回调记录关键指标
from langchain.callbacks import BaseCallbackHandler

class MetricsCallback(BaseCallbackHandler):
    def __init__(self):
        self.retrieval_count = 0
        self.llm_calls = 0
        self.total_tokens = 0
    
    def on_retriever_end(self, documents, **kwargs):
        self.retrieval_count += 1
        print(f"📊 检索次数: {self.retrieval_count}, 文档数: {len(documents)}")
    
    def on_llm_end(self, response, **kwargs):
        self.llm_calls += 1
        self.total_tokens += response.llm_output.get("token_usage", {}).get("total_tokens", 0)

⚠️ 常见陷阱与解决方案

陷阱	症状	解决方案
无限循环	Agent 不断重试	设置最大重试次数
检索过度	延迟过高	优化检索策略，缓存结果
上下文溢出	Token 超限	压缩文档，设置最大长度
幻觉检测失效	错误答案通过	使用更严格的验证prompt

---

10. 总结与展望

🎓 核心要点回顾

1. 传统 RAG 适合简单场景，但存在单次检索、固定策略、无纠错等局限

2. Agentic RAG 通过引入 Agent 实现：

   • 动态检索策略
   • 多工具协作
   • 自我反思与纠错
   • 复杂推理能力

3. 四大设计模式：Reflection、Tool Use、Planning、Multi-Agent

4. 选型建议：

   • 简单FAQ → 传统 RAG
   • 复杂分析/多源数据 → Agentic RAG

🔮 未来趋势

• 更智能的 Agent：更强的规划和推理能力
• 多模态 RAG：图像、音频、视频的检索与生成
• 实时学习：从用户反馈中持续优化
• 分布式 Agent：大规模多Agent协作系统

---

📚 参考资源

• LangGraph 官方文档
• Agentic RAG Survey Paper (arXiv)
• Weaviate: What is Agentic RAG
• LangChain RAG Tutorial

---