LLM之Agent（四十一）｜AI Agents（十）：Agentic RAG介绍

检索增强生成（RAG）代表了人工智能领域的一项重大进步，它将大型语言模型（LLM）的生成能力与实时数据检索相结合。尽管 LLM 在自然语言处理方面展现出了卓越的能力，但它们对静态预训练数据的依赖常常导致响应过时或不完整。RAG 通过动态地从外部来源检索相关信息并将其融入生成过程来解决这一局限性，从而实现上下文准确且最新的输出。

wshzd

807人浏览 · 2025-12-20 18:34:48

wshzd · 2025-12-20 18:34:48 发布

推荐阅读列表：

LLM之Agent（三十二）｜AI Agents（一）简介

LLM之Agent（三十三）｜AI Agents（二）：从零开始构建Agent

LLM之Agent（三十四）｜AI Agents（三）：AI Agents框架介绍

LLM之Agent（三十五）｜AI Agents（四）：AI Agent类型

LLM之Agent（三十六）｜AI Agents（五）：Workflow vs Agent

LLM之Agent（三十七）｜AI Agents（六）：AI Agents架构

LLM之Agent（三十八）｜AI Agents（七）：Multi-Agent架构

LLM之Agent（三十九）｜AI Agents（八）：构建Multi-Agent系统

LLM之Agent（四十）｜AI Agents（九）：Agentic Memory介绍

一、引言

从 2024 年到 2025 年，人工智能的焦点正在从检索增强生成 (RAG) 转移到Agentic RAG。

本文将介绍 Agentic RAG 的概念、实现方式、优点和局限性。

1.1 Agentic RAG 概述

1.2 传统RAG 与 Agentic RAG

传统RAG系统的架构，包括如下三个主要组件：

Retrieval: 负责查询外部数据源，例如知识库、API 或向量数据库。高级检索器利用密集向量搜索和基于 Transformer 的模型来提高检索精度和语义相关性；
Augmentation: 处理检索到的数据，提取并汇总最相关的信息，以符合查询上下文；
Generation: 将检索到的信息与 LLM 的预训练知识相结合，生成连贯的、符合上下文的响应。

Agentic RAG 引入了基于 AI Agent的自主决策和编排机制，从而实现了更强大、更灵活的检索生成工作流程。其流程主要包括以下步骤：

Step 1: Agent-Oriented Query Analysis

用户查询会被路由到AI agent，然后AI agent解读查询的意图和上下文。

Step 2: Memory and Strategy

智能体利用短期（会话）记忆和长期（历史）记忆来跟踪上下文。然后，制定动态的检索和推理策略。

Step 3: Tool Selection & Data Gathering

能够智能地选择工具（例如向量搜索、API 连接器，甚至其他智能体），并从相关的知识库（例如 MCP 服务器、图数据库、文档存储）中检索数据。

Step 4: Prompt Construction

检索到的内容会以结构化形式添加提示词模板，然后再输入给 LLM。

Step 5: LLM Response Generation

LLM 基于结构化的提示词生成相关高、可解释且适应性强的响应。

二、RAG 范式的演变

为了应对日益复杂的现实世界应用，检索增强生成（RAG）领域取得了显著发展。在这些应用中，上下文准确性、可扩展性和多步骤推理至关重要。最初基于关键词的简单检索已经演变为复杂、模块化和自适应的系统，能够整合各种数据源并进行自主决策。这种发展凸显了 RAG 系统高效处理复杂查询的迫切需求。

本节探讨 RAG 范式的演进历程，介绍了关键的发展阶段：朴素 RAG、高级 RAG、模块化 RAG、Graph RAG 和 Agentic RAG。

2.1 朴素RAG

朴素 RAG 代表了RAG的基础实现，图 3 展示了朴素 RAG 简单的检索-读取工作流程，重点关注基于关键词的检索和静态数据集。这些系统依赖于简单的基于关键词的检索技术，例如 TF-IDF 和 BM25，从静态数据集中获取文档。然后，检索到的文档用于增强语言模型的生成能力。

朴素RAG算法的特点是简单易行，因此适用于涉及事实查询且上下文复杂度极低的任务。然而，它也存在一些局限性：

缺乏上下文意识：由于依赖词汇匹配而不是语义理解，检索到的文档往往无法捕捉查询的语义细微差别；
输出碎片化：缺乏高级预处理或上下文整合往往会导致反应不连贯或过于笼统；
可扩展性问题：基于关键词的检索技术难以处理大型数据集，常常无法识别最相关的信息。

尽管存在这些局限性，但朴素的 RAG 系统为检索与生成相结合提供了重要的概念验证，为更复杂的范式奠定了基础。

2.2 高级 RAG

高级 RAG 系统通过融合语义理解和增强检索技术，克服了朴素 RAG 的局限性。图 4 展示了高级 RAG 中语义增强的检索机制以及迭代式、上下文感知的流程。这些系统利用密集检索模型（例如密集段落检索 (DPR)）和神经排序算法来提高检索精度。

2.3 模块化 RAG

模块化 RAG 代表了 RAG 范式的最新发展，强调灵活性和定制化。这些系统将检索和生成流程分解为独立的、可重用的组件，从而实现特定领域的优化和任务适应性。

图 5 展示了模块化架构，展示了混合检索策略、可组合管道和外部工具集成。

2.4 Graph RAG

如图 6 所示，Graph RAG 通过集成基于图的数据结构，扩展了传统的检索增强生成系统。这些系统利用图数据中的关系和层级结构来增强多跳推理和上下文丰富性。通过整合基于图的检索，Graph RAG 能够生成更丰富、更准确的生成式输出，尤其适用于需要关系理解的任务。

然而，Graph RAG 也存在一些局限性：

可扩展性有限：对图结构的依赖可能会限制可扩展性，尤其是在拥有大量数据源的情况下；
数据依赖性：高质量的图数据对于输出至关重要，这限制了其在非结构化或标注不良的数据集中的适用性；
集成的复杂性：将图数据与非结构化检索系统集成会增加设计和实现的复杂性。

Graph RAG 非常适合医疗保健诊断、法律研究以及其他对结构化关系进行推理至关重要的领域等应用。

2.5 Agentic RAG

Agentic RAG 通过引入能够进行动态决策和工作流优化的自主智能体，代表了一种范式转变。与静态系统不同，Agentic RAG 采用迭代优化和自适应检索策略来处理复杂、实时和跨领域的查询。这种范式充分利用了检索和生成过程的模块化特性，同时引入了基于智能体的自主性。

在将Agentic RAG系统扩展到解决实际业务问题时，并没有单一的、僵化的模板。架构师需要根据独特的用例、数据环境和领域需求而演进。

为了实现这一点，我们先来看一下 Agentic RAG 流水线各个模块化组件之间的交互方式：

🧠 1. 分析查询

第一步是将原始用户查询传递给一个基于LLM的智能体，该智能体负责进行深度分析和转换。以下是此阶段的具体步骤：

🔁 查询重写：智能体可以重新表述输入，生成更精细或替代的版本，以便更好地提取相关知识；
🧭 决策逻辑：它评估查询是否可以直接回答，或者是否需要其他数据。

🔍 2. 其他数据来源

如果需要更多上下文信息，系统会激活一个或多个检索智能体，负责获取最相关的数据。这些智能体可能会获取：

📍 实时用户数据（例如，用户的当前位置或状态）；
📂 私有/内部文档（项目文件、报告、手册）；
🌐 公共或基于网络的内容（API 数据、索引文章等）

每个检索路径都是动态的，通常根据查询的意图和性质来选择。

🧩 3. 重排序搜索结果

候选文档检索完成后，会通过重排序模型进行处理——该模型通常比基础嵌入模型功能更强大，且更能感知上下文。

🧪 4. 生成答案

如果 LLM 智能体判断其已掌握足够的信息，则直接进行答案合成，这可能生成以下结果：

✅ 单一的事实性回答；
🧮 建议操作列表；
🤖 或者甚至是执行智能体驱动的工作流程

📊 5. 分析答案

最后，生成的答案将由另一个智能体（或再次调用的同一智能体）进行严格评估：

🔄 如果准确且相关，则与最终用户分享；
♻️ 如果缺少数据，系统可能会重新构建原始查询并重新启动循环—— 但会设置预定义的重试次数限制，以避免无限循环。

本质上，Agentic RAG 系统更像是自主决策者，而非简单的检索工具。它们能够动态地适应、推理和演进，并随着业务环境的变化而不断调整和发展。

三、传统 RAG 系统的挑战和局限性

检索增强生成（RAG）通过将生成与实时数据相结合，提高了 LLM 的性能，但仍有一些关键挑战限制了其在复杂的现实世界场景中的性能：

3.1 上下文整合

即使 RAG 系统成功检索到相关信息，它们也常常难以将其无缝整合到生成的响应中。检索流程的静态特性和有限的上下文感知能力会导致输出结果零散、不一致或过于笼统。

例子：

当被问及 “阿尔茨海默病研究的最新进展及其对早期治疗的意义是什么？” 时，RAG 可能会检索到相关的研究，但却无法将这些发现转化为可操作的患者护理方案。同样，对于 “干旱地区小规模农业的最佳可持续实践是什么？” 这个问题，它可能会检索到一些宽泛的耕作方法，但却忽略了针对干旱地区的具体可持续发展技术。

3.2 多步骤推理

复杂的问题通常需要经过多个步骤的推理，但传统的 RAG 通常只通过一次检索就能获取信息，错失了更深层次的综合分析。

例子：

诸如 “欧洲可再生能源政策的哪些经验可以应用于发展中国家，以及潜在的经济影响是什么？” 这样的问题需要结合政策数据、当地情况和经济预测。RAG 模型通常无法将所有这些信息整合起来，得出连贯的答案。

3.3 可扩展性和延迟问题

随着外部数据的增长，搜索和排名大型数据集会减慢响应时间，这对于实时用例来说是个问题。

例子：

在实时金融分析或客户支持中，获取和处理大型数据集的延迟可能会造成损失——例如，由于检索速度慢而错过高频交易中的市场趋势。

四、Agentic RAG 的分类

4.1 Single Agent Agentic RAG: Router

单智能体Agentic RAG：作为一个集中式决策系统，其中单个智能体负责信息的检索、路由和集成（如图所示）。这种架构通过将这些任务整合到一个统一的智能体中来简化系统，使其在工具或数据源数量有限的场景中尤为有效。

工作流程如下所示：

1）查询提交与评估：用户提交查询；协调智能体分析并确定合适的数据源。

2）知识来源选择：

结构化数据库（例如，PostgreSQL、MySQL，Text2SQL）；
语义搜索（例如，基于向量的 PDF、书籍、内部记录搜索）；
网络搜索（实时在线数据）；
推荐系统（基于用户个人资料的个性化建议）。

3）数据集成和 LLM 合成：检索到的数据被传递给 LLM 进行合成，从而得到连贯的响应。

4）输出生成：系统为用户生成简洁、可操作的响应。

4.2 多智能体Agentic RAG 系统

多个专业智能体协同工作，每个智能体专注于特定的数据源或任务，从而实现对复杂查询的可扩展、模块化处理。

工作流程如下所示：

1）查询提交：协调智能体接收查询并将其分发给专业智能体。

2）专业检索代理：

智能体 1：结构化数据（PostgreSQL、MySQL 上的 SQL 查询）；
智能体 2：语义搜索（PDF、书籍、内部非结构化数据）；
智能体 3：实时网络搜索（新闻、政策更新）；
智能体 4：推荐系统（个性化内容、专家建议）。

3）工具访问和数据检索：并行利用矢量搜索、Text2SQL、Web 搜索和外部 API。

4）数据集成和 LLM 合成：将汇总数据传递给 LLM 进行合成，生成综合输出。

5. 输出生成：向用户提供最终的、综合的响应。

挑战：

各智能体间的协调复杂性增加；
更高的计算资源消耗；
跨异构输出的复杂数据集成。

4.3 分层Agentic RAG 系统

分层Agentic RAG：系统采用结构化的多层信息检索和处理方法，如图 18 所示，可提高效率并增强战略决策能力。智能体以层级结构组织，高层智能体监督和指导低层智能体。这种结构支持多级决策，确保查询由最合适的资源处理。

工作流程如下所示：

1）查询接收：用户提交查询，由负责初步评估和任务分配的高级智能体接收。

2）战略决策：顶级智能体评估查询的复杂性，并根据领域相关性和数据可靠性选择优先考虑哪些下级代理、数据库、API 或检索工具。

3）委托给下级智能体：任务分配给专门的低级智能体（例如，SQL 数据库、网络搜索、专有 API），这些智能体独立执行检索。

4）信息综合：下级智能体将结果返回给上级智能体，上级智能体将信息整合和综合成一个连贯、统一的响应。

5）最终响应：向用户呈现最终的综合答案，确保答案的完整性和上下文准确性。

挑战：

协调复杂性：跨多个层级的稳健智能体间通信会增加编排开销；
资源分配：跨层级的有效任务分配十分复杂，需要精心管理以避免瓶颈。

4.4 Agentic Corrective RAG

Corrective RAG 引入了自我纠正机制，通过将智能体嵌入工作流程来迭代地改进检索结果，从而提高文档利用率和响应质量。

工作流程：

1）上下文检索智能体：从向量数据库中检索初始上下文文档。

2）相关性评估智能体：评估检索到的文档的相关性；标记不相关或含糊不清的文档以便进行更正。

3）查询优化智能体：利用语义理解重写和优化查询，以提高检索结果。

4）外部知识检索智能体：如果上下文仍然不足，则进行网络搜索或访问其他数据源。

5）响应合成智能体：将所有经过验证和完善的信息合成为连贯、准确的响应。

矫正性 RAG 的核心理念：

文档相关性评估：对检索到的文档进行评估；相关性低的文档会触发纠正措施。
查询优化和增强：查询由查询优化代理重写，以提高检索效率，确保为响应生成提供更高质量的输入。

4.5 Adaptive Agentic RAG

自适应 RAG 通过根据每个查询的复杂性调整检索策略，引入动态查询处理，从而增强灵活性和效率。

自适应 RAG 的核心思想：

直接查询：无需检索即可直接生成基于事实的问题（例如，“水的沸点是多少？”）。
简单查询：对中等复杂程度的任务进行单步检索（例如，“我的最新电费账单状态是什么？”）。
复杂查询：多步骤检索，通过迭代细化进行分层推理（例如，“过去十年中 X 市的人口发生了怎样的变化，其影响因素是什么？”）。

工作流程：

1）分类器角色：

较小的 LLM 使用基于历史结果的自动标记数据集来分析查询复杂性。

2）动态策略选择：

对于简单的查询，LLM 会直接回复。
简单查询的单步检索。
针对需要迭代推理的复杂查询进行多步骤检索。

3）LLM 整合：

检索到的信息被综合成连贯的响应。
LLM 和分类器之间的迭代交互可以改进复杂任务的输出。

4.6 基于图的Agentic RAG

4.6.1 Agent-G：为Graph RAG 的Agentic框架

Agent-G 引入了一种新颖的架构，将图知识库与非结构化文档检索相结合，提高了 RAG 系统的推理和检索准确性。

Agent-G 的核心思想：

图知识库：提取结构化关系（例如，医疗保健中的疾病到症状的映射）。
非结构化文档：利用来自文本源的上下文信息补充结构化数据。
评论模块：评估检索数据的相关性和质量，并在需要时触发纠正措施。
反馈循环：通过验证和重新查询迭代地改进检索和合成。

工作流程如下所示：

1）检索库：模块化智能体专门用于图或非结构化数据检索，根据查询需求动态选择数据源。

2）评论模块：验证检索到的信息的相关性和置信度，标记不确定的结果以便改进。

3）动态智能体交互：特定任务智能体协作，将结构化和非结构化数据源集成到统一的工作流程中。

4）LLM 整合：将已验证的数据综合成连贯的响应，并通过批评反馈循环不断改进。

4.6.2 GeAR：用于 RAG 的图增强智能体

GeAR 通过集成基于图的检索和智能体控制来增强传统的 RAG，从而改进复杂查询的多跳检索。

GeAR 的核心理念：

图扩展：通过整合图结构数据来扩展基本检索器（例如 BM25），捕获复杂的实体关系和依赖关系。
智能体框架：利用智能体动态管理检索策略，根据查询复杂性自主选择图扩展路径。

工作流程如下所示：

1）图扩展模块：将图数据集成到检索中，扩展搜索空间以包含连通实体，从而更好地进行多跳推理。

2）基于智能体的检索：智能体动态选择和组合检索策略，并在必要时利用图扩展检索。

3）LLM 集成：使用 LLM 将图丰富和非结构化数据合成为连贯、上下文丰富的响应。

4.7 Agentic RAG 中的 Agentic 文档工作流 (ADW)

ADW 通过在以文档为中心的流程中协调端到端的知识工作自动化来扩展传统的 RAG，将解析、检索、推理和结构化输出与智能代理集成在一起。

工作流程如下所示：

1）文档解析与信息结构化：

使用企业工具（例如 LlamaParse）解析文档，提取发票编号、日期、供应商、行项目和付款条款等字段。
提取用于下游任务的数据结构。

2）跨流程的状态维护：维护文档上下文并跟踪跨多个处理阶段的工作流进度。

3）知识检索：

从外部知识库（例如 LlamaCloud）和向量索引中检索相关参考文献。
包含实时、特定领域的决策支持指南。

4）智能体编排：智能体应用业务规则，执行多跳推理，并协调解析器、检索器和 API 等组件。

5. 可操作的输出生成：生成结构化的输出，其中包含简洁、可操作的报告中的综合建议和见解。

五、Agentic RAG 框架的比较分析

表 2 对三种架构框架进行了全面的比较分析：传统 RAG、Agentic RAG 和代理文档工作流 (ADW)。该分析突出了它们各自的优势、劣势和最佳适用场景，为理解它们在不同用例中的适用性提供了宝贵的见解。

六、构建一个Agentic RAG 系统

在本教程中，我们将构建一个检索智能体。当您希望语言学习模型 (LLM) 决定是从向量存储中检索上下文还是直接响应用户时，检索智能体非常有用。

在本教程结束时，我们将完成以下内容：

获取并预处理将用于检索的文档。
为这些文档建立语义搜索索引，并为智能体创建检索工具。
构建一个能够决定何时使用检索工具的智能 RAG 系统。

环境设置

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

import getpass
import os
def _set_env(key: str):
    if key not in os.environ:
        os.environ[key] = getpass.getpass(f"{key}:")
_set_env("OPENAI_API_KEY")

6.1 预处理文档

1）获取要用于 RAG 系统的文档。

我们将使用 Lilian Weng 的优秀博客[1]中最新的三个页面。首先，我们将使用 WebBaseLoader 工具获取页面内容：

from langchain_community.document_loaders import WebBaseLoader
urls = [
    "https://lilianweng.github.io/posts/2024-11-28-reward-hacking/",
    "https://lilianweng.github.io/posts/2024-07-07-hallucination/",
    "https://lilianweng.github.io/posts/2024-04-12-diffusion-video/",
]
docs = [WebBaseLoader(url).load() for url in urls]

docs[0][0].page_content.strip()[:1000]

2. 将获取的文档拆分成更小的块，以便索引到我们的矢量存储库中：

from langchain_text_splitters import RecursiveCharacterTextSplitter
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)

doc_splits[0].page_content.strip()

6.2 创建检索工具

现在我们已经有了拆分后的文档，我们可以将它们索引到向量存储中，以便进行语义搜索。

1）使用内存向量存储和 OpenAI 嵌入：

from langchain_core.vectorstores import InMemoryVectorStore
from langchain_openai import OpenAIEmbeddings
vectorstore = InMemoryVectorStore.from_documents(
    documents=doc_splits, embedding=OpenAIEmbeddings()
)
retriever = vectorstore.as_retriever()

2）使用 LangChain 预构建的 create_retriever_tool 创建一个检索工具：

from langchain.tools.retriever import create_retriever_tool
retriever_tool = create_retriever_tool(
    retriever,
    "retrieve_blog_posts",
    "Search and return information about Lilian Weng blog posts.",
)

3）测试该工具：

retriever_tool.invoke({"query": "types of reward hacking"})

6.3 生成查询

现在我们将开始构建agentic RAG 图的组件（节点和边）。请注意，这些组件将操作 MessagesState 一个包含 messages 键和聊天消息列表的图状态。

1）构建一个 generate_query_or_respond 节点。该节点将调用 LLM 来根据当前图状态（消息列表）生成响应。给定输入消息，它将决定使用检索工具进行检索，还是直接回复用户。请注意，我们通过 .bind_tools 为聊天模型提供了对之前创建的 retriever_tool 访问权限：

from langgraph.graph import MessagesState
from langchain.chat_models import init_chat_model
response_model = init_chat_model("openai:gpt-4.1", temperature=0)
def generate_query_or_respond(state: MessagesState):
    """Call the model to generate a response based on the current state. Given
    the question, it will decide to retrieve using the retriever tool, or simply respond to the user.
    """
    response = (
        response_model
        .bind_tools([retriever_tool]).invoke(state["messages"])
    )
    return {"messages": [response]}

2）简单测试一下

input = {"messages": [{"role": "user", "content": "hello!"}]}
generate_query_or_respond(input)["messages"][-1].pretty_print()

6.4 Grade文件

1）添加一条条件边 grade_documents 以确定检索到的文档是否与问题相关。我们将使用具有结构化输出模式 GradeDocuments grade_documents 进行文档评分。grade_documents 函数将根据评分决策（ generate_answer 或 rewrite_question ）返回要跳转到的节点名称：

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 to the question."
)
class GradeDocuments(BaseModel):
    """Grade documents using a binary score for relevance check."""
    binary_score: str = Field(
        description="Relevance score: 'yes' if relevant, or 'no' if not relevant"
    )
grader_model = init_chat_model("openai:gpt-4.1", temperature=0)
def grade_documents(
    state: MessagesState,
) -> Literal["generate_answer", "rewrite_question"]:
    """Determine whether the retrieved documents are relevant to the 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}]
        )
    )
    score = response.binary_score
    if score == "yes":
        return "generate_answer"
    else:
        return "rewrite_question"

2）使用工具响应中的无关文档运行此命令：

from langchain_core.messages import convert_to_messages
input = {
    "messages": convert_to_messages(
        [
            {
                "role": "user",
                "content": "What does Lilian Weng say about types of reward hacking?",
            },
            {
                "role": "assistant",
                "content": "",
                "tool_calls": [
                    {
                        "id": "1",
                        "name": "retrieve_blog_posts",
                        "args": {"query": "types of reward hacking"},
                    }
                ],
            },
            {"role": "tool", "content": "meow", "tool_call_id": "1"},
        ]
    )
}
grade_documents(input)

3）确认相关文件已按此分类：

input = {
    "messages": convert_to_messages(
        [
            {
                "role": "user",
                "content": "What does Lilian Weng say about types of reward hacking?",
            },
            {
                "role": "assistant",
                "content": "",
                "tool_calls": [
                    {
                        "id": "1",
                        "name": "retrieve_blog_posts",
                        "args": {"query": "types of reward hacking"},
                    }
                ],
            },
            {
                "role": "tool",
                "content": "reward hacking can be categorized into two types: environment or goal misspecification, and reward tampering",
                "tool_call_id": "1",
            },
        ]
    )
}
grade_documents(input)

6.5 重写问题

1）构建 rewrite_question 节点。检索工具可能会返回一些不相关的文档，这表明需要改进原始用户问题。为此，我们将调用 rewrite_question 节点：

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):
    """Rewrite the original user question."""
    messages = state["messages"]
    question = messages[0].content
    prompt = REWRITE_PROMPT.format(question=question)
    response = response_model.invoke([{"role": "user", "content": prompt}])
    return {"messages": [{"role": "user", "content": response.content}]}

2）测试一下

input = {
    "messages": convert_to_messages(
        [
            {
                "role": "user",
                "content": "What does Lilian Weng say about types of reward hacking?",
            },
            {
                "role": "assistant",
                "content": "",
                "tool_calls": [
                    {
                        "id": "1",
                        "name": "retrieve_blog_posts",
                        "args": {"query": "types of reward hacking"},
                    }
                ],
            },
            {"role": "tool", "content": "meow", "tool_call_id": "1"},
        ]
    )
}
response = rewrite_question(input)
print(response["messages"][-1]["content"])

6.6 生成答案

1）构建 generate_answer 节点：如果通过评分器检查，我们可以根据原始问题和检索到的上下文生成最终答案：

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):
    """Generate an answer."""
    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]}

2）再测试一下

input = {
    "messages": convert_to_messages(
        [
            {
                "role": "user",
                "content": "What does Lilian Weng say about types of reward hacking?",
            },
            {
                "role": "assistant",
                "content": "",
                "tool_calls": [
                    {
                        "id": "1",
                        "name": "retrieve_blog_posts",
                        "args": {"query": "types of reward hacking"},
                    }
                ],
            },
            {
                "role": "tool",
                "content": "reward hacking can be categorized into two types: environment or goal misspecification, and reward tampering",
                "tool_call_id": "1",
            },
        ]
    )
}
response = generate_answer(input)
response["messages"][-1].pretty_print()

6.7 组装Graph

首先调用 generate_query_or_respond 函数，然后确定是否需要调用 retriever_tool

使用 tools_condition 跳转到下一步：

如果 generate_query_or_respond 返回 tool_calls ，则调用 retriever_tool 来检索上下文；
否则，直接回复用户。

根据检索到的文档内容与问题的相关性对其进行评分（ grade_documents ），并跳转到下一步：

如果问题不相关，请使用 rewrite_question 重写问题，然后再次调用 generate_query_or_respond；
如果相关，则继续执行 generate_answer 操作，并使用 ToolMessage 和检索到的文档上下文生成最终响应。

from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
from langgraph.prebuilt import tools_condition
workflow = StateGraph(MessagesState)
# Define the nodes we will cycle between
workflow.add_node(generate_query_or_respond)
workflow.add_node("retrieve", ToolNode([retriever_tool]))
workflow.add_node(rewrite_question)
workflow.add_node(generate_answer)
workflow.add_edge(START, "generate_query_or_respond")
# Decide whether to retrieve
workflow.add_conditional_edges(
    "generate_query_or_respond",
    # Assess LLM decision (call `retriever_tool` tool or respond to the user)
    tools_condition,
    {
        # Translate the condition outputs to nodes in our graph
        "tools": "retrieve",
        END: END,
    },
)
# Edges taken after the `action` node is called.
workflow.add_conditional_edges(
    "retrieve",
    # Assess agent decision
    grade_documents,
)
workflow.add_edge("generate_answer", END)
workflow.add_edge("rewrite_question", "generate_query_or_respond")
# Compile
graph = workflow.compile()

可视化Graph

from IPython.display import Image, display
display(Image(graph.get_graph().draw_mermaid_png()))

6.8 运行 Agentic RAG

for chunk in graph.stream(
    {
        "messages": [
            {
                "role": "user",
                "content": "What does Lilian Weng say about types of reward hacking?",
            }
        ]
    }
):
    for node, update in chunk.items():
        print("Update from node", node)
        update["messages"][-1].pretty_print()
        print("\n\n")

输出：

Update from node generate_query_or_respond
================================== Ai Message ==================================
Tool Calls:
  retrieve_blog_posts (call_cZehDWOxAfSV1RQdw1RHzKwG)
 Call ID: call_cZehDWOxAfSV1RQdw1RHzKwG
  Args:
    query: types of reward hacking
Update from node retrieve
================================= Tool Message =================================
Name: retrieve_blog_posts
Detecting Reward Hacking#
In-Context Reward Hacking#
(Note: Some work defines reward tampering as a distinct category of misalignment behavior from reward hacking. But I consider reward hacking as a broader concept here.)
At a high level, reward hacking can be categorized into two types: environment or goal misspecification, and reward tampering.
Why does Reward Hacking Exist?#
Update from node generate_answer
================================== Ai Message ==================================
Lilian Weng categorizes reward hacking into two types: environment or goal misspecification, and reward tampering. She considers reward hacking as a broader concept that includes both of these categories. Some work defines reward tampering separately, but Weng includes it under the umbrella of reward hacking.

七、结论

从传统的检索增强生成（RAG）系统到Agentic RAG 的演进，标志着我们在处理复杂实际应用中的推理、检索和响应生成方式发生了范式转变。正如我们所探讨的，Agentic RAG 架构通过赋予智能体迭代推理、任务委派、自我纠错和动态适应的能力，开启了智能的新维度——而这正是传统 RAG 流程难以实现的。

从单智能体路由到分层纠错的多智能体工作流，再到通过 Agent-G 和 GeAR 等框架将图结构与智能体推理融合，我们现在拥有了构建真正模块化、弹性且上下文感知的检索系统的基础模块。从自适应智能体到文档工作流代理，每一层抽象都为 RAG 生命周期增添了一种新的自主性和决策技巧。

然而，Agentic RAG 并非万能灵药。它需要精心策划、稳健的提示工程、周密的分类设计以及持续的监控，以确保其可靠性和可扩展性。随着该领域的不断成熟，未来在评估协议标准化、构建可互操作的智能体框架以及利用多模态和记忆增强功能拓展应用边界方面，都蕴藏着巨大的机遇。