AI 新兴范式：Agentic RAG 技术详解

优势	说明
事实基础	将响应锚定在检索到的事实中，减少幻觉
领域专业化	无需模型再训练，即可提供特定领域知识
知识新近度	允许访问超出模型训练截止时间的最新信息
透明度	允许引用生成内容的信息来源
可控性	对模型可访问的信息进行细粒度控制

三、传统 RAG 的局限性

尽管传统 RAG 优势明显，但仍面临以下挑战：

单次检索步骤：初始检索结果质量差时，最终生成结果也会受到影响
查询-文档不匹配：用户问题（通常是疑问句）与包含答案的文档（通常是陈述句）语义上不够匹配
推理能力有限：简单的 RAG 流程无法进行多步推理或查询细化
上下文窗口约束：检索到的文档内容必须适合模型的上下文窗口大小

四、Agentic RAG：更强大的方法

什么是 Agentic RAG？

Agentic RAG 是一种新兴的 AI 范式，其中大型语言模型（LLM）自主规划下一步行动，同时从外部资源提取信息。

与静态的"先检索后阅读"模式不同，Agentic RAG 的核心特征：

对 LLM 进行迭代调用，穿插使用工具或函数调用并输出结构化结果
系统评估结果、优化查询，必要时调用更多工具，持续循环直到获得满意答案
采用迭代的"制造者-检查者"模式，提升正确性，处理格式错误的查询，确保结果质量

Agentic 系统能够：主动重写失败的查询、选择不同的检索方式、整合多种工具（向量搜索、SQL 数据库、自定义 API 等），最终再给出答案。

Agentic RAG 的主要优势

能力	说明
制定优化查询	将用户问题转换为易于检索的查询语句
执行多次检索	根据需要迭代检索信息
多源推理	从多个来源分析、综合并得出结论
自我批评与改进	评估检索结果并动态调整检索策略

这种方法天然实现了以下先进 RAG 技术：

Hypothetical Document Embedding (HyDE)：代理不直接使用用户查询，而是制定检索优化查询
Self-Query Refinement：代理分析初始检索结果，并使用细化查询执行后续检索

五、Agentic RAG 完整实现

下面以 LangGraph + 通义千问为例，展示 Agentic RAG 系统的完整构建过程。

环境安装

pip install -U --quiet langgraph "langchain[openai]" langchain-community langchain-text-splitters

提示：在 Jupyter Notebook 中建议使用 %pip install（魔法命令），而非 !pip install 或直接写 pip install，以确保安装到当前 Notebook 所在的 Python 环境。

步骤 1：预处理文档

从网页加载文档，并使用 RecursiveCharacterTextSplitter 进行分块：

import os
from langchain_community.document_loaders import WebBaseLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

urls = [
    "https://baike.baidu.com/item/大语言模型/62884793",
    "https://baike.baidu.com/item/深度学习/3729729",
    "https://baike.baidu.com/item/人工智能/9180",
    "https://sports.sina.com.cn/"
]

# 加载文档
docs = [WebBaseLoader(url).load() for url in urls]
docs_list = [item for sublist in docs for item in sublist]

# 分块处理
text_splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
    chunk_size=100, chunk_overlap=50
)
doc_splits = text_splitter.split_documents(docs_list)

步骤 2：创建检索工具

使用向量存储构建检索器，并将其封装为 LangChain 工具：

from langchain_core.vectorstores import InMemoryVectorStore
from langchain_community.embeddings import DashScopeEmbeddings
from langchain.tools.retriever import create_retriever_tool

# 构建向量存储与检索器
vectorstore = InMemoryVectorStore.from_documents(
    documents=doc_splits,
    embedding=DashScopeEmbeddings(model="text-embedding-v1")
)
retriever = vectorstore.as_retriever()

# 封装为工具
retriever_tool = create_retriever_tool(
    retriever,
    "百度百科",                        # 工具名称
    "搜索百度或新浪体育上的信息."      # 工具描述
)

步骤 3：生成查询节点

模型决策节点：根据用户输入，决定是直接回答还是调用检索工具：

from langgraph.graph import MessagesState
from langchain_community.chat_models import ChatTongyi

response_model = ChatTongyi(
    model="qwen-plus",
    temperature=0,
    api_key=os.getenv("DASHESCOPE_API_KEY")
)

def generate_query_or_respond(state: MessagesState):
    """根据当前状态生成响应：直接回答用户，或调用检索工具获取信息。"""
    response = (
        response_model.bind_tools([retriever_tool]).invoke(state["messages"])
    )
    return {"messages": [response]}

步骤 4：对检索结果进行评分

使用结构化输出模型，对检索文档与问题的相关性进行二元评分（yes / no）：

from pydantic import BaseModel, Field
from typing import Literal

GRADE_PROMPT = (
    "You are a grader assessing relevance of a retrieved document to a user question. \n "
    "Here is the retrieved document: \n\n {context} \n\n"
    "Here is the user question: {question} \n"
    "If the document contains keyword(s) or semantic meaning related to the user question, "
    "grade it as relevant. \n"
    "Give a binary score 'yes' or 'no' score to indicate whether the document is relevant."
)

class GradeDocuments(BaseModel):
    """文档相关性评分（二元）"""
    binary_score: str = Field(
        description="Relevance score: 'yes' if relevant, or 'no' if not relevant"
    )

grader_model = ChatTongyi(model="qwen-plus", temperature=0,
                          api_key=os.getenv("DASHESCOPE_API_KEY"))

def grade_documents(state: MessagesState) -> Literal["generate_answer", "rewrite_question"]:
    """判断检索到的文档是否与问题相关，决定下一步走向。"""
    question = state["messages"][0].content
    context = state["messages"][-1].content
    prompt = GRADE_PROMPT.format(question=question, context=context)
    response = grader_model.with_structured_output(GradeDocuments).invoke(
        [{"role": "user", "content": prompt}]
    )
    return "generate_answer" if response.binary_score == "yes" else "rewrite_question"

步骤 5：问题重写节点

当检索结果不相关时，对原始问题进行语义重写，优化后重新检索：

REWRITE_PROMPT = (
    "Look at the input and try to reason about the underlying semantic intent / meaning.\n"
    "Here is the initial question:\n ------- \n{question}\n ------- \n"
    "Formulate an improved question:"
)

def rewrite_question(state: MessagesState):
    """重写原始用户问题，提升检索质量。"""
    question = state["messages"][0].content
    prompt = REWRITE_PROMPT.format(question=question)
    response = response_model.invoke([{"role": "user", "content": prompt}])
    return {"messages": [{"role": "user", "content": response.content}]}

步骤 6：生成最终答案

基于原始问题与检索到的上下文，生成简洁准确的最终答案：

GENERATE_PROMPT = (
    "You are an assistant for question-answering tasks. "
    "Use the following pieces of retrieved context to answer the question. "
    "If you don't know the answer, just say that you don't know. "
    "Use three sentences maximum and keep the answer concise.\n"
    "Question: {question} \n"
    "Context: {context}"
)

def generate_answer(state: MessagesState):
    """根据问题和检索上下文生成最终答案。"""
    question = state["messages"][0].content
    context = state["messages"][-1].content
    prompt = GENERATE_PROMPT.format(question=question, context=context)
    response = response_model.invoke([{"role": "user", "content": prompt}])
    return {"messages": [response]}

步骤 7：构建状态图

使用 LangGraph 的 StateGraph 将所有节点与边组合成完整的 Agentic RAG 工作流：

from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode, tools_condition

workflow = StateGraph(MessagesState)

# 注册节点
workflow.add_node(generate_query_or_respond)
workflow.add_node("retrieve", ToolNode([retriever_tool]))
workflow.add_node(rewrite_question)
workflow.add_node(generate_answer)

# 入口：从 generate_query_or_respond 开始
workflow.add_edge(START, "generate_query_or_respond")

# 条件边：决定是检索还是直接回答
workflow.add_conditional_edges(
    "generate_query_or_respond",
    tools_condition,
    {"tools": "retrieve", END: END},
)

# 条件边：评分后决定生成答案还是重写问题
workflow.add_conditional_edges("retrieve", grade_documents)

# 固定边
workflow.add_edge("generate_answer", END)
workflow.add_edge("rewrite_question", "generate_query_or_respond")

# 编译
graph = workflow.compile()

整体工作流示意如下：

START
  └─► generate_query_or_respond
        ├─► END（无需检索，直接回答）
        └─► retrieve（调用检索工具）
              ├─► generate_answer ──► END（文档相关）
              └─► rewrite_question ──► generate_query_or_respond（文档不相关，重写后重试）

步骤 8：运行 Agentic RAG

for chunk in graph.stream(
    {
        "messages": [
            {"role": "user", "content": "国足刚刚赢比赛的比分是多少?"}
        ]
    }
):
    for node, update in chunk.items():
        print("Update from node", node)
        update["messages"][-1].pretty_print()
        print("\n\n")

系统会根据问题自动决策：若知识库中有相关内容则直接检索回答；若检索结果不相关则重写查询后重试；若完全无需检索则直接响应。