LCEL链表达式的缺点与扩展

LCEL 与 AgentExecutor 的局限性

LangChain 提供的 LCEL 表达式，虽然可以很便捷的创建链应用，并且将 知识库、LLM、prompt、工具调用、输出解析器 等内容串联起来，构成一个有向无环图，例如下方这种结构：

虽然已经极大降低了 LLM 应用的开发难度，但是链式应用在处理复杂、动态的对话流程时存在着一些局限性：

1. 线性流程：链通常是线性的，这意味着它们只能按照预定义的顺序执行步骤，这种线性结构限制了在对话中进行动态路由和条件分支的能力。
2. 状态管理：链在处理多轮对话时，状态管理变得非常复杂，每次调用链时，都需要手动传递和更新状态，增加了代码的复杂性和出错的可能性。
3. 工具集成：虽然链可以调用外部工具，但在链的内部结构中集成和协调多个工具的使用并不直观，尤其是在需要根据对话上下文动态选择工具时。

例如在前面课时中学习的 问题分解策略 - 并行子问题优化策略 中，工具嵌套层级稍微多一些，链的构建就会变得很复杂（而且必须是无环结构才可以使用 LCEL 表达式构建），例如：

# 分解问题链
decomposition_chain = (
    {"question": RunnablePassthrough()}
    | decomposition_prompt
    | ChatOpenAI(model="gpt-4o-mini", temperature=0)
    | StrOutputParser()
    | (lambda x: x.strip().split("\n"))
)

# 子问题答案生成链
sub_question_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | sub_question_prompt
    | ChatOpenAI(model="gpt-4o-mini")
    | StrOutputParser()
)

# 组装链
chain = (
    {
        "question": RunnablePassthrough(),
        "context": decomposition_chain | {
            "questions": RunnablePassthrough(),
            "answers": sub_question_chain.map()
        } | RunnableLambda(format_qa_pairs)
    } | prompt | llm_output_str
)

AgentExecutor 的诞生解决了 LCEL 的部分缺陷，它允许智能体根据输入动态选择工具和操作，尽管 AgentExecutor 提供了一定的灵活性，但它也存在着一些局限性：

1. 复杂性：AgentExecutor 的配置和使用相对复杂，尤其是在处理复杂的对话流程和多轮对用，这增加了开发的难度。
2. 动态路由：AgentExecutor 虽然支持动态选择工具，但在处理复杂的条件分支和动态路由时，仍然不够灵活。缺乏一种直观的方式来定义和执行复杂的对话流程。
3. 状态持久性：AgentExecutor 在处理长时间运行的对话时，缺乏内置的状态持久性机制。每次对话重启时，都需要从头开始，无法恢复之前的对话状态。
4. 过度封装：AgentExecutor 要求被包装的 Agent 必须符合特定的要求才能使用，例如 输入变量固定、输入 prompt 固定、解析器固定等，想要二次开发 AgentExecutor 难度非常大。
5. 黑盒不可控：在构建复杂 Agent 时无法修改工具的使用顺序，无法在执行过程中添加人机交互。

面对 LCEL 链应用 与 AgentExecutor 的局限性，LangGraph 应运而生，LangGraph 的设计目标是解决这些局限性，提供一个更灵活、更强大的框架来构建复杂的智能体应用，在我们正式学习 LangGraph 之前，让我们先通过一个简单的概念来帮助大家认识 图 和 状态机。

从 “带娃状态图” 认识「图」与「状态机」

图 和 状态机 两个词看起来有点莫名其妙，为什么会和 LLM/Agent 应用开发扯上关系呢？其实图和状态机是计算机科学中非常重要的概念，在很多场合下都有它们的身影。

先来认识状态机：可以将状态机看作一个记录有限行为状态的盒子，例如一个婴儿的行为状态存在：饿 / 饱、睡觉 / 清醒、想晒太阳 / 想遮阴等状态，我们可以定义一个状态机记录这些事件，示例：

baby_status = {
    "饥饿程度": "饿",
    "是否睡觉": "清醒",
    "体感温度": "想遮阴",
}

而任何具有状态的事物，都有一个 初始状态，也有一个 最终状态，不同状态之间是可以通过 事件 进行转换的，并且 转换 和 事件 是确定性的，这意味着每个转换和事件总是指向相同的下一个状态，就像 哄宝宝入睡 这个 事件 在执行完成后，是否睡觉 这个状态肯定会变为睡觉，你永远不会把 宝宝哄入睡 后，它还醒着，或者起床后，还在睡着，这就是 有限状态。

有了 状态、N 个事件 之后，我们可以把相关联的 事件 连接起来，一个 事件 接受 状态 作为输入，同时也输出 状态，并将输出状态传递给连接的下一个实践，最后通过一定约束和条件设置好一个初始状态与最终状态，这样就构成了一张 图，在这张 graph (图) 中，每个节点就是 事件，边就是连接 节点 与 节点 的桥梁（通信逻辑），而 状态 就是 节点 与 边 的输入模式。

例如，我们有以下假设：

1. 以婴儿的行为 作为 状态，涵盖了 是否需要喂奶、是否需要换尿布、是否涨肚、是否需要晒太阳、是否想玩游戏、是否需要安抚睡觉 等。
2. 以 指导带娃的妈妈 (知道该执行什么操作)、判断带娃是否正确的奶奶 / 外婆 (可以判断操作是否合理)、执行带娃的爸爸 (执行操作) 作为 事件节点。
3. 由妈妈下命令带娃（判断婴儿需要什么），奶奶 / 外婆检测妈妈的指挥是否正确进行判断，由爸爸执行命令，直到带娃成功（婴儿不哭不闹），整个流程结束。

有上述的3个假设后，我们就可以基于这些信息构建一个状态图，如下：

在上述的状态图中，妈妈主要负责决定需要执行什么事件，例如 换尿布、游戏互动，亦或者做出决定 不用带娃了；老大人主要对妈妈做的事件决定进行判断，检测是否可执行；而 爸爸 主要负责执行具体的事件，并且执行完特定事件后（更新婴儿状态），由 奶奶 检测婴儿状态 继续做决定，直到符合特定的状态，整个带娃流程结束。

学习到这里，其实你就已经通过 带娃状态图 理解 图 和 状态机 究竟是什么了，一个 状态 经过特定的 节点 处理后，会得到一个符合需求的 最终状态，有没有发现和 LLM/Agent 应用很接近，对于一个 Agent 应用来说，我们提出一个 初始状态 (需求)，让 Agent 经过一系列的复杂操作得到一个 最终状态 (答案)，简化下整个流程，其实就变成了：

而利用 LangGraph 就可以快速构建流程图中 图结构 的部分，对比 LCEL 链应用和 AgentExecutor，图结构的优点也非常多：

1. 图结构：LangGraph 采用图（Graph）结构来表示对话流程，允许开发者定义复杂的非线性流程和条件分支。这种图结构提供了更大的灵活性，使得动态路由和条件分支变得直观和简单。
2. 状态管理：LangGraph 内置了强大的状态管理机制，可以无缝地管理多轮对话的状态。开发者无需手动传递和更新状态，框架会自动处理状态的持久化和恢复。
3. 工具集成：LangGraph 简化了工具集成和使用，可以轻松地将多个工具集成到对话流程中，并根据对话上下文动态选择和调用工具。
4. 持久性：LangGraph 提供了内置状态持久性机制，支持长时间运行的对话。开发者可以随时暂停和恢复对话，无需担心状态丢失。

并且 LangGraph 不是一个颠覆性的框架，并且利用 LangGraph 构建的组件仍然是一个 Runnable 可运行组件，所以它是 LCEL 的扩展，不仅可以单独使用，还可以和 LCEL 链应用、原始的 LangChain Runnable 可运行组件、丰富的大量第三方集成、甚至与其他图结构应用进行嵌套结合，从而让 LLM/Agent 应用的开发变得非常简单。

LangGraph介绍与基础组件上手

LangGraph介绍与功能

LangGraph 是一个构建具有 状态、多角色 应用程序的库，用于创建智能体和多智能体工作流，与其他 LLM 框架相比，LangGraph 提供了以下核心优势：循环、可控制性和持久性。LangGraph 允许定义设计循环、条件判断的流程，这对于高级 Agent 非常重要，这和传统的有向无环图（DAG）解决方案区分开。

因为 LangGraph 作为最底层的框架，所以涉及的组件都是最基础的，并没有过度封装，允许开发者实现对应用程序流程和状态的精细控制（自由度极高），而且 LangGraph 可以便捷集成持久化方案、任意节点中断交互、修改状态等特性，该框架虽然由 LangChain 团队构建，但是却可以在没有 LangChain 的情况下单独使用。

对比 LCEL 表达式，LangGraph 的主要功能：

1. 循环和分支：在 LLM/Agent 应用程序中使用便捷的方式实现循环和条件语句，而无需配置额外的程序。
2. 持久化：在图的每个步骤之后自动保存状态，在运行的任意一个阶段都支持暂停和恢复图执行，以支持错误恢复、人机交互工作流、时间旅行等。
3. 人机交互：中断图的执行以批准或者编辑状态计划去执行下一计划。
4. 流支持：图结构的每个节点都支持流式输出（包括 token 流式输出）。
5. 与 Langchain 集成：LangGraph 可以和 LangChain 和 LangSmith 无缝集成。

在一个最基础 LangGraph 应用程序中，涵盖了 3 种基本组件：

1. 状态：状态是图应用程序处理与交互的基础，是图中所有 节点 和 边 的输入，它可以是一个 dict (字典) 或者 Pydantic 模型，在 LangGraph 中，状态包括 图的模式 (数据结构) 以及如何更新状态的 归纳函数，如果没有设置 归纳函数，则每次节点都会覆盖 状态的原始数据。
2. 节点：节点通常是 Python 函数 (同步或异步)，其节点函数的第一个参数是 state (状态)，第二个参数是 config (Runnable 运行的配置)，节点函数的返回值一般都是 状态，整个图的起点被称为开始节点，最后的终点被称为结束节点。
3. 边：边在图中定义了路由逻辑，即不同节点之间是如何传递的，传递给谁，以及图节点从哪里开始，从哪里结束，并且一个节点可以设置多条边，如果有多条边，则下一条边连接的所有节点都会并行运行。

pip install -U langgraph

由于 LangGraph 是一个独立的框架，使用前必须先安装，命令如下：

LangGraph基础组件使用

通过上面的图示例，其实可以很容易知道一个 LangGraph 应用程序的组成部分：节点、边、状态，接下来我们来利用这 3 个组件来构建一个最最基础的 图架构应用，仅包含 开始节点、中间节点 和 结束节点 的聊天机器人应用，其图结构如下：

在这个 图架构 应用中，开始节点连接 LLM 大语言模型 节点，并且该节点接收 状态数据，并生成对应内容，随后传递给 结束节点。简单来说，节点完成工作。边指示下一步要做什么。

在 LangGraph 中，开始 / 结束节点 作为特殊节点，并预定义了，导入后即可使用：

from langgraph.graph import START, END

对于状态的声明，LangGraph 推荐使用的是 TypedDict 或者 Pydantic 模型，并且在 LangGraph 中，还支持对声明的状态使用 归纳函数，归纳函数可以修改数据的更新方式，例如下方是一个没有为任何键添加归纳函数的状态：

from typing import TypedDict

class State(TypedDict):
    foo: int
    bar: list[str]

假设在上述图结构中，输入是 {“foo”: 1, “bar”: [“hi”]}，并且第一个节点返回 {“foo”: 2}，LangGraph 会根据返回的数据对状态进行更新，并且节点不需要返回所有数据，这样下一个节点接收到的状态数据为 {“foo”: 2, “bar”: [“hi”]}，可以看到默认更新方式是覆盖更新。

如果要使用 归纳函数，我们可以使用 Annotated 类型来为特定的键添加，例如为第二个键添加 operator.add 函数，代码更新如下：

from typing import TypedDict, Annotated
from operator import add

class State(TypedDict):
    foo: int
    bar: Annotated[list[str], add]

这个时候，假设图的输入是 {“foo”: 1, “bar”: [“hi”]}，然后，第一个节点返回 {“bar”: [“bye”]}，此时状态数据变为 {“foo”: 2, “bar”: [“hi”, “bye”]}，即数据会使用 Annotated 标注的函数来进行更新（add 函数会将两个列表的数据相加）。

创建好状态后，就可以根据状态来实例化图结构了，在 LangGraph 中有两种类型的图，一种是 stateGraph，另外一种是 MessageGraph，前者的状态是自定义字典，后者的状态是消息列表，使用示例：

from langgraph.graph import StateGraph

graph_builder = StateGraph(State)

创建好图结构后，即可为图添加节点与边，使用的函数也非常简单：

• add_node (节点名称，节点对应函数)：将一个函数添加到图上，并设置对应的节点名称。
• add_edge (开始节点名称，结束节点名称)：将两个节点进行拼接，第一个参数是拼接的开始，第二个参数是拼接的结束

完成上述的步骤后，这个时候的 图结构 仍然不可使用，我们需要将其转换成 Runnable 可运行组件，这个时候就可以调用 .compile() 函数进行编译，编译返回的数据就是 Runnable 可运行组件，可以调用统一的接口，例如：invoke、stream、batch 等。

通过上述的分析，可以知道在 LangGraph 中，无论多么复杂的项目，都可以拆分成对应的 6 个步骤，只需按照这 6 个步骤来执行即可：

1. 初始化大语言模型和工具（ChatOpenAI、tools）。
2. 用状态初始化图架构（StateGraph 状态图）。
3. 为图定义每一个节点（add_node 函数为图添加节点）。
4. 定义图的起点、终点和节点边（add_edge 函数为图添加边）。
5. 编译图架构为 Runnable 可运行组件（graph.compile 函数编译图）。
6. 调用编译后的 Runnable 可运行组件执行图（graph.invoke 函数调用图）。

例如将上述的 3 节点聊天机器人 实现后，示例代码如下：

from typing import TypedDict, Annotated, Any

import dotenv
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

dotenv.load_dotenv()

llm = ChatOpenAI(model="gpt-4o-mini")


# 1.创建状态图，并使用GraphState作为状态数据
class State(TypedDict):
    """图结构的状态数据"""
    messages: Annotated[list, add_messages]
    use_name: str


def chatbot(state: State, config: dict) -> Any:
    """聊天机器人节点，使用大语言模型根据传递的消息列表生成内容"""
    ai_message = llm.invoke(state["messages"])
    return {"messages": [ai_message], "use_name": "chatbot"}


graph_builder = StateGraph(State)

# 2.添加节点
graph_builder.add_node("llm", chatbot)

# 3.添加边
graph_builder.add_edge(START, "llm")
graph_builder.add_edge("llm", END)

# 4.编译图为Runnable可运行组件
graph = graph_builder.compile()

# 5.调用图架构应用
print(graph.invoke({"messages": [("human", "你好，你是谁，我叫慕小课，我喜欢打篮球游泳")], "use_name": "graph"}))

得益于LangSmith日志信息的采集，并且编译后的图架构应用也是一个Runnable可运行组件，所以可以完整观察图的整个运行流程，为后续复杂图架构应用开发调试流程可观察提供了基础。

条件边与循环流程实现工具调用Agent

条件边与循环流程

在 LangChain 中，边定义了节点之间是如何工作的，以及图结构从哪里开始，从哪里结束，在 LangGraph 中，边的种类有 4 种：

• 普通边：直接从一个节点到下一个节点。
• 条件边：调用一个函数来确定下一个需要跳转的节点。
• 入口边：用户输入到达时首先调用的节点，即确定图结构的开始节点。
• 条件入口点：调用一个函数来确定用户输入到达时首先调用的节点，即通过函数来确定图结构的开始节点。

并且一个节点是可以拥有多个输出边的，如果一个节点同时拥有多条输出边，则所有目标节点将在下一个步骤中并行执行，将这些边的类型转换到图结构中如下：

普通边可以直接使用 add_edge() 函数即可，如果想为某个节点添加条件边，可以使用 add_conditional_edge() 函数，该函数的返回值为字符串或者列表，代表需要执行节点的名称（一个或多个），函数共有 4 个参数，其中前 2 个参数为必填：

• source：条件边的起始节点名称，该节点运行结束后会执行条件边。
• path：确定下一个节点是什么的可运行对象或者函数。
• path_map：可选参数，类型为一个字典，用于表示返回的 path 和节点名称的映射关系，如果不设置的话，path 的返回值应该是节点名称。
• then：可选参数，在执行 path 节点之后统一选择节点，通过该设置就不需要为后续的每一个节点都设置一个统一的关联节点。

对于循环流程而言，在 LangGraph 并没有单独设置函数，只需要通过 add_edge() 将两个节点串联起来即可。

注意下，对于循环流程，一般都会有一个条件边用于跳出循环，否则 LangGraph 框架会检测到没有跳出循环的条件，应用程序会崩溃，并且极大消耗系统资源。

实现基于工具调用的Agent

通过上面的内容了解条件边与循环流程，接下来我们就可以利用这些组件来实现一个基于工具调用的Agent,其运行流程如下：

示例代码：

import json
from typing import TypedDict, Annotated, Any, Literal

import dotenv
from langchain_community.tools import GoogleSerperRun
from langchain_community.tools.openai_dalle_image_generation import OpenAIDALLEImageGenerationTool
from langchain_community.utilities import GoogleSerperAPIWrapper
from langchain_community.utilities.dalle_image_generator import DallEAPIWrapper
from langchain_core.messages import ToolMessage
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages

dotenv.load_dotenv()


class GoogleSerperArgsSchema(BaseModel):
    query: str = Field(description="执行谷歌搜索的查询语句")


class DallEArgsSchema(BaseModel):
    query: str = Field(description="输入应该是生成图像的文本提示(prompt)")


# 1.定义工具与工具列表
google_serper = GoogleSerperRun(
    name="google_serper",
    description=(
        "一个低成本的谷歌搜索API。"
        "当你需要回答有关时事的问题时，可以调用该工具。"
        "该工具的输入是搜索查询语句。"
    ),
    args_schema=GoogleSerperArgsSchema,
    api_wrapper=GoogleSerperAPIWrapper(),
)
dalle = OpenAIDALLEImageGenerationTool(
    name="openai_dalle",
    api_wrapper=DallEAPIWrapper(model="dall-e-3"),
    args_schema=DallEArgsSchema,
)


class State(TypedDict):
    """图状态数据结构，类型为字典"""
    messages: Annotated[list, add_messages]


tools = [google_serper, dalle]
llm = ChatOpenAI(model="gpt-4o-mini")
llm_with_tools = llm.bind_tools(tools)


def chatbot(state: State, config: dict) -> Any:
    """聊天机器人函数"""
    # 1.获取状态里存储的消息列表数据并传递给LLM
    ai_message = llm_with_tools.invoke(state["messages"])
    # 2.返回更新/生成的状态
    return {"messages": [ai_message]}


def tool_executor(state: State, config: dict) -> Any:
    """工具执行节点"""
    # 1.提取数据状态中的tool_calls
    tool_calls = state["messages"][-1].tool_calls

    # 2.根据找到的tool_calls去获取需要执行什么工具
    tools_by_name = {tool.name: tool for tool in tools}

    # 3.执行工具得到对应的结果
    messages = []
    for tool_call in tool_calls:
        tool = tools_by_name[tool_call["name"]]
        messages.append(ToolMessage(
            tool_call_id=tool_call["id"],
            content=json.dumps(tool.invoke(tool_call["args"])),
            nane=tool_call["name"]
        ))

    # 4.将工具的执行结果作为工具消息更新到数据状态机中
    return {"messages": messages}


def route(state: State, config: dict) -> Literal["tool_executor", "__end__"]:
    """通过路由来取检测下后续的返回节点是什么，返回的节点有2个，一个是工具执行，一个是结束节点"""
    ai_message = state["messages"][-1]
    if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:
        return "tool_executor"
    return END


# 1.创建状态图，并使用GraphState作为状态数据
graph_builder = StateGraph(State)

# 2.添加节点
graph_builder.add_node("llm", chatbot)
graph_builder.add_node("tool_executor", tool_executor)

# 3.添加边
graph_builder.set_entry_point("llm")
graph_builder.add_conditional_edges("llm", route)
graph_builder.add_edge("tool_executor", "llm")

# 4.编译图为Runnable可运行组件
graph = graph_builder.compile()

# 5.调用图架构应用
state = graph.invoke({"messages": [("human", "2024年北京半程马拉松的前3名成绩是多少")]})

for message in state["messages"]:
    print("消息类型: ", message.type)
    if hasattr(message, "tool_calls") and len(message.tool_calls) > 0:
        print("工具调用参数: ", message.tool_calls)
    print("消息内容: ", message.content)
    print("=====================================")

并行调用边使用示例

在 LangGraph 中，一个节点可以同时连接多条边，被连接到的所有节点全部都会并行执行，直到再次关联到一起，或者图运行结束。

例如下方左右有两个并行运行流程，其中左侧的两个并行节点均有连接到 END 节点，右侧的只有一个，但是最终结果是一模一样的，只要不把 状态 看成是 传递，而是整个图的全局变量，每个节点执行的都是 修改 操作即可。

from typing import Any

from langchain_core.messages import AIMessage, HumanMessage
from langgraph.graph.message import StateGraph, MessagesState

graph_builder = StateGraph(MessagesState)


def chatbot(state: MessagesState, config: dict) -> Any:
    return {"messages": [AIMessage(content="你好，我是OpenAI开发的聊天机器人")]}


def parallel1(state: MessagesState, config: dict) -> Any:
    print("并行1: ", state)
    return {"messages": [HumanMessage(content="这是并行1函数")]}


def parallel2(state: MessagesState, config: dict) -> Any:
    print("并行2: ", state)
    return {"messages": [HumanMessage(content="这是并行2函数")]}


def chat_end(state: MessagesState, config: dict) -> Any:
    print("聊天结束: ", state)
    return {"messages": [HumanMessage(content="这是聊天结束函数")]}


graph_builder.add_node("chat_bot", chatbot)
graph_builder.add_node("parallel1", parallel1)
graph_builder.add_node("parallel2", parallel2)
graph_builder.add_node("chat_end", chat_end)

graph_builder.set_entry_point("chat_bot")
graph_builder.set_finish_point("chat_end")
graph_builder.add_edge("chat_bot", "parallel1")
graph_builder.add_edge("chat_bot", "parallel2")
graph_builder.add_edge("parallel2", "chat_end")

graph = graph_builder.compile()

print(graph.invoke({"messages": [HumanMessage(content="你好，你是")]}))

LangGraph实现ReACT架构Agent

预构建的ReACT智能体

在 LangGraph 中除了能使用基础组件（节点、边、数据状态）来构建 Agent 智能体，这也是 LangGraph 自由性高的一个优点，我们还可以使用 LangGraph 预构建的代理来快速创建智能体，例如：ReACT 智能体 亦或者 工具调用智能体。

使用技巧也非常简单，导入对应的 预构建函数 然后调用函数即可，例如：

from langgraph.prebuilt.chat_agent_executor import create_react_agent, create_tool_calling_executor

不过在 LangGraph 底层，目前预构建的 ReACT 智能体目前也是基于 函数调用 的，并且在 0.3.0 版本会被剔除（后续会更新优化预构建智能体，可以持续留意关注），但是其封装思路仍然非常值得借鉴。

完整示例如下：

import dotenv
from langchain_community.tools import GoogleSerperRun
from langchain_community.tools.openai_dalle_image_generation import OpenAIDALLEImageGenerationTool
from langchain_community.utilities import GoogleSerperAPIWrapper
from langchain_community.utilities.dalle_image_generator import DallEAPIWrapper
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent

dotenv.load_dotenv()


class GoogleSerperArgsSchema(BaseModel):
    query: str = Field(description="执行谷歌搜索的查询语句")


class DallEArgsSchema(BaseModel):
    query: str = Field(description="输入应该是生成图像的文本提示(prompt)")


# 1.定义工具与工具列表
google_serper = GoogleSerperRun(
    name="google_serper",
    description=(
        "一个低成本的谷歌搜索API。"
        "当你需要回答有关时事的问题时，可以调用该工具。"
        "该工具的输入是搜索查询语句。"
    ),
    args_schema=GoogleSerperArgsSchema,
    api_wrapper=GoogleSerperAPIWrapper(),
)
dalle = OpenAIDALLEImageGenerationTool(
    name="openai_dalle",
    api_wrapper=DallEAPIWrapper(model="dall-e-3"),
    args_schema=DallEArgsSchema,
)
tools = [google_serper, dalle]

# 2.创建大语言模型
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 3.使用预构建的函数创建ReACT智能体
agent = create_react_agent(model=model, tools=tools)

# 4.调用智能体并输出内容
print(agent.invoke({"messages": [("human", "请帮我绘制一幅鲨鱼在天上飞的图片")]}))

LangGraph其他预构建组件

除了 create_react_agent() 预构建组件，在 LangGraph 中其实还内置了一些高频使用的预构建组件，涵盖了 MessagesState、ToolNode、ValidationNode 和 injectedState 等。

其中 MessagesState 就是我们一直在定义的自定义状态，在 LangGraph 内部已经帮我们封装好了，导入后可以直接使用，如果需要除了消息外的其他信息，可以继承该类进行重写：

# langgraph/graph/message.py -> MessagesState
class MessagesState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

# 导入示例
from langgraph.graph import MessagesState

而 ToolNode 节点其实就是工具执行节点，该节点会自动执行数据状态中最后一条消息中的工具调用信息，其实就是我们上节课所使用的 tool_executor() 函数，输出一个 ToolMessages 列表，使用技巧也非常简单，实例化该类并传递工具列表即可。

该类等价于以下函数：

tools_by_name = {tool.name: tool for tool in tools}
def tool_node(state: dict):
    result = []
    for tool_call in state["messages"][-1].tool_calls:
        tool = tools_by_name[tool_call["name"]]
        observation = tool.invoke(tool_call["args"])
        result.append(ToolMessage(content=observation, tool_call_id=tool_call["id"]))
    return {"messages": result}

ValidationNode 这个类用来校验最后一个 AIMessage 中所有工具请求是否正确，通常用于校验 LLM 的结构化输出是否正确，该类在实例化的时候，需要传递 Pydantic 模型 作为需要校验的类，使用示例如下

class SelectNumber(BaseModel):
    a: int

    @validator("a")
    def a_must_be_meaningful(cls, v):
        if v != 37:
            raise ValueError("only 37 is allowed")
        return v

builder = MessageGraph()
llm = ChatAnthropic(model="claude-3-haiku-20240307").bind_tools([SelectNumber])
builder.add_node("model", llm)
builder.add_node("validation", ValidationNode([SelectNumber]))
builder.add_edge(START, "model")

def should_validate(state: list) -> Literal["validation", "__end__"]:
    if state[-1].tool_calls:
        return "validation"
    return END

builder.add_conditional_edges("model", should_validate)

而 InjectedState 其实和我们在 LLMOPs 项目中使用的 injector 依赖注入包非常接近，主要用于在工具上注入数据状态，这样在工具中也可以实际获取到 图架构应用 的数据状态，并且使用 Annotated 装饰的参数不会被视为是工具的参数，LLM 在执行函数调用的过程中，不会生成该参数，使用示例如下：

from typing_extensions import Annotated, TypedDict

class AgentState(TypedDict):
    messages: List[BaseMessage]
    foo: str

@tool
def state_tool(x: int, state: Annotated[dict, InjectedState]) -> str:
    '''Do something with state.'''
    if len(state["messages"]) > 2:
        return state["foo"] + str(x)
    else:
        return "not enough messages"

图结构应用程序删除消息的使用技巧

更新删除与归纳函数

在图结构应用程序中，消息列表是一种高频使用的状态，通常情况下我们只会往状态中添加消息。但是在某些特殊的情况下，我们可能希望删除消息列表中的某一条消息（亦或者是修改消息列表中的某一条数据）。

这个时候就需要使用 LangGraph 为我们提供的 RemoveMessage 装饰符配合 add_messages() 函数一起来实现这个功能。其核心思想是归纳函数 add_messages() 底层针对更新的消息类型做了检测，如果检测到是 RemoveMessage 类型，则不会新增数据，而是执行删除数据的操作。

所以对于需要删除的消息，只需要在节点返回的时候，创建 RemoveMessage 实例并传递 消息 id 即可，例如下方提问后删除人类消息：

from typing import Any

import dotenv
from langchain_core.messages import RemoveMessage, AIMessage
from langchain_core.runnables import RunnableConfig
from langchain_openai import ChatOpenAI
from langgraph.graph import MessagesState, StateGraph

dotenv.load_dotenv()

llm = ChatOpenAI(model="gpt-4o-mini")


def chatbot(state: MessagesState, config: RunnableConfig) -> Any:
    """聊天机器人节点"""
    return {"messages": [llm.invoke(state["messages"])]}


def delete_human_message(state: MessagesState, config: RunnableConfig) -> Any:
    """删除状态中的人类消息"""
    human_message = state["messages"][0]
    return {"messages": [RemoveMessage(id=human_message.id)]}


def update_ai_message(state: MessagesState, config: RunnableConfig) -> Any:
    """更新AI的消息，为AI消息添加上前缀"""
    ai_message = state["messages"][-1]
    return {"messages": [AIMessage(id=ai_message.id, content="更新后的AI消息:" + ai_message.content)]}


# 1.创建图构建器
graph_builder = StateGraph(MessagesState)

# 2.添加节点
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_node("delete_human_message", delete_human_message)
graph_builder.add_node("update_ai_message", update_ai_message)

# 3.添加边
graph_builder.set_entry_point("chatbot")
graph_builder.add_edge("chatbot", "delete_human_message")
graph_builder.add_edge("delete_human_message", "update_ai_message")
graph_builder.set_finish_point("update_ai_message")

# 4.编译图
graph = graph_builder.compile()

# 5.调用图应用程序
print(graph.invoke({"messages": [("human", "你好，你是")]}))

在上述的代码中，其运行逻辑是由 归纳函数 进行处理判断的，核心代码如下：

for m in right:
    if (existing_idx := left_idx_by_id.get(m.id)) is not None:
        if isinstance(m, RemoveMessage):
            ids_to_remove.add(m.id)
        else:
            merged[existing_idx] = m
    else:
        if isinstance(m, RemoveMessage):
            raise ValueError(
                f"Attempting to delete a message with an ID that doesn't exist ('{m.id}')"
            )
        merged.append(m)

merged = [m for m in merged if m.id not in ids_to_remove]

在这段代码中，会检测右侧传递的消息列表，并检测类型是否为 RemoveMessage，如果是的话，则在合并的时候会剔除数据，如果不是的话则会覆盖，所以利用这个技巧也可以修改原来的消息数据，例如添加如下代码：

def update_ai_message(state: MessagesState, config: RunnableConfig) -> Any:
    """修改AI消息节点"""
    ai_message = state["messages"][-1]
    return {"messages": [AIMessage(id=ai_message.id, content="我是被修改过后的AI消息:" + ai_message.content)]}

graph_builder.add_node("update_ai_message", update_ai_message)

graph_builder.add_edge("delete_human_message", "update_ai_message")
graph_builder.set_finish_point("update_ai_message")

这也是为什么使用 LangGraph 提供的 add_messages 而不是 operator.add 来实现，operator.add 虽然能实现对列表的相加，但没有针对修改或者删除的逻辑，仍然需要手动去实现，因为本质上这里的删除 / 更新逻辑是归纳函数实现的，所以对于没有配置归纳函数，或者归纳函数没有该逻辑的则无法实现。

删除消息一定要特别注意，因为绝大部分模型期望消息列表存在某些规则。例如，有些模型期望它们以 user 消息开头，其他模型期望所有带有工具调用的消息后面都跟着工具消息。删除消息时，需要确保不会违反这些规则。

过滤与修剪消息

在 LangGraph 中 状态 可以很便捷管理整个过程中产生的所有消息信息，但是随着持续对话，亦或者图结构组件的增加，对话历史会不断累积，并占用越来越多的上下文窗口，这通常是不可取的，因为它会导致对 LLM 的调用变得非常昂贵和耗时，并降低 LLM 生成内容的正确性，所以在 LangGraph 中一般还需要对消息进行过滤和修剪。

过滤 / 修剪一般不会更改状态，而是在调用 LLM 时，只传递特定条数的消息或者按照 token 长度进行修剪。

例如使用过滤消息可以单独创建一个函数（非节点），在调用 LLM 前，对消息进行过滤，使用固定条数的消息列表：

def filter_messages(state: MessagesState) -> Any:
    """过滤数据状态并返回最后一条消息"""
    return state["messages"][-1:]

def chatbot(state: MessagesState, config: RunnableConfig) -> Any:
    """聊天机器人节点"""
    messages = filter_messages(state)
    return {"messages": llm.invoke(messages)}

这样在使用 LLM 时就可以避免全部将消息传递过去，并且在图架构程序内，状态仍然会保存最完整的信息。

除此之外，还可以依据 Token 长度限制 对消息列表进行修剪，在 LangChain 中对于该需求还封装了特定的函数 trim_messages，该函数的参数如下：

1. messages：需要修剪的消息列表。
2. max_tokens：修剪消息的最大 Token 数。
3. strategy：修剪策略，first 代表从前往后修剪消息，last 代表从后往前修剪消息，默认为 last。
4. token_counter：计算 Token 数的函数，或者传递大语言模型（使用大语言模型的 .get_num_tokens_from_messages() 计算 Token 数）。
5. allow_partial：如果只能拆分消息的一部分，是否拆分消息，默认为 False。
6. end_on：修剪消息结束的类型，如果执行，则在这种类型的最后一次出现时将被忽略，类型为列表或者单个值（支持传递消息的类型字符串，例如：system、human、ai、tool 等，亦或者传递消息类）。
7. start_on：修剪消息开始的类型，如果执行，则在这种类型的最后一次出现时将被忽略，类型为列表或者单个值（支持传递消息的类型字符串，例如：system、human、ai、tool 等，亦或者传递消息类）。
8. include_system：是否保留系统消息，只有在 strategy=“last” 时设置才有效。
9. text_splitter：文本分割器，默认为空，当设置 allow_partial=True 时才有用，用于对某个消息类型中的大文本进行分割。

例如实现对消息列表进行修剪，使其 Token 数不超过 80，保留前置消息，允许部分分割，使用的模型为 gpt-4o-mini，代码如下：

import dotenv
from langchain_core.messages import HumanMessage, AIMessage, trim_messages
from langchain_openai import ChatOpenAI
from langchain_text_splitters import RecursiveCharacterTextSplitter

dotenv.load_dotenv()

messages = [
    HumanMessage(content="你好，我叫慕小课，我喜欢游泳打篮球，你喜欢什么呢？"),
    AIMessage([
        {"type": "text", "text": "你好，慕小课！我对很多话题感兴趣，比如探索新知识和帮助解决问题。你最喜欢游泳还是篮球呢？"},
        {
            "type": "text",
            "text": "你好，慕小课！我喜欢探讨各种话题和帮助解答问题。你对游泳和篮球的兴趣很广泛，有没有特别喜欢的运动方式或运动员呢？"
        },
    ]),
    HumanMessage(content="如果我想学习关于天体物理方面的知识，你能给我一些建议么？"),
    AIMessage(
        content="当然可以！你可以从基础的天文学和物理学入手，然后逐步深入到更具体的天体物理领域。阅读相关的书籍，如《宇宙的结构》或《引力的秘密》，也可以关注一些优秀的天体物理学讲座和课程。你对哪个方面最感兴趣？"
    ),
]

llm = ChatOpenAI(model="gpt-4o-mini")

update_messages = trim_messages(
    messages,
    max_tokens=80,
    token_counter=llm,
    strategy="first",
    end_on="human",
    allow_partial=False,
    text_splitter=RecursiveCharacterTextSplitter(),
)

print(update_messages)

并且在 LangChain 中 trim_messages() 函数使用 @_runnable_support 装饰器进行装饰，所以该函数也是一个 Runnable 可运行组件，可以直接拼接到 LCEL 表达式构建的链应用中。

langGraph检查点实现记忆持久化功能

检查点与线程

在 LCEL 表达式构建的链应用中，我们将 Memory组件 通过 .with_listen() 函数绑定到整个链的 运行结束生命周期 上，从而去实现链记忆功能的自动管理，在 LangGraph 中也有类似的功能，不过该功能是 检查点，在编程中 检查点 通常用于记录或标记程序在某个阶段的状态，以便在程序运行过程中出现问题时，可以回溯到特定的状态，亦或者在图执行的过程中将任意一个节点的状态进行保存。

理解 检查点 其实很简单，想象一下你在玩一个需要多个任务的游戏，比如一个冒险游戏，你的角色需要完成许多关卡和任务。如果你在某个关卡中遇到困难或游戏崩溃，你不想从游戏的开头重新开始。于是，游戏就会在你完成每个关卡后保存一个 检查点 / 存档点，这样你就可以从不同 检查点 / 存档点 重新启动游戏继续玩。

并且在很多开源项目中都可以看到 检查点 的身影，例如：TensorFlow、PyTorch、Flink、Spark、Celery 等。

在 LangGraph 中，持久化使用的就是 检查点，并且除了这个功能，每个 检查点 还和 线程 ID 有关，检查点 存储的并不是整个图结构应用程序的 节点状态，而是存储 特定线程 的 数据状态，这是因为 LangGraph 在设计的时候就考虑到一个应用的多次独立对话功能。

就好比游戏存档中，每个家庭成员玩同一款游戏，可以保留独属于自己的不同存档，通过不同的 线程 和 检查点 实现共用一套程序，并实现完全隔离，这个时候 LangGraph 图结构的流程就变成如下：

可以看到这个时候 数据状态 和 检查点 在不同 线程 / 游戏账号 下都是相互独立的，不会互相干扰，在 LangGraph 中于是这样设计的，例如在不添加 检查点 的情况下，在同一份程序中，对 图结构应用程序 发起多次提问，可以发现 图 并没有记忆，如下：

# 1.第一次提问
print(agent.invoke({"messages": [("human", "你好，我叫慕小课，我喜欢游泳打球，你喜欢什么呢?")]}))

# 2.二次调用
print(agent.invoke({"messages": [("human", "你知道我叫什么吗？")]}))

可以发现，在没有设置 检查点 与 线程 的时候，图应用程序每次运行都会管理新的 数据状态，并不会持久化，要想使用 检查点 来为图提供持久化记忆，操作技巧也非常简单，共两步：

1. 实例化一个检查点，例如 AsyncSqliteSaver 或者 MemorySaver()，亦或者自定义检查点。
2. 在图编译的时候传递检查点，例如 compile(checkpointer=my_checkpointer)。

接下来在和图程序交互时传递 config，并配置 thread_id 即可记住以往的历史记忆 / 存档，更新代码如下：

# 使用预构建create_react_agent并传递检查点
checkpointer = MemorySaver()
agent = create_react_agent(
    model=model,
    tools=tools,
    checkpointer=checkpointer,
)

# 调用智能体并输出内容
print(agent.invoke(
    {"messages": [("human", "你好，我叫慕小课，我喜欢游泳打球，你喜欢什么呢?")]},
    config={"configurable": {"thread_id": 1}}
))

# 二次调用
print(agent.invoke(
    {"messages": [("human", "你知道我叫什么吗?")]},
    config={"configurable": {"thread_id": 1}}
))

langGraph其他检查点

在 LangGraph 中，除了封装了 MemorySaver 基于临时内存的检查点，还封装了基于 Postgres、MongoDB 和 Redis 的检查点。

LangGraph 持久化文档: https://langchain-ai.github.io/langgraph/how-tos/persistence/

这些检查点的运行流程都一模一样，只是持久化 / 存储的介质不一样而已，根据存储方式的不同使用不同的实例化方式。

例如使用 Postgres 作为存储介质时，在实例化 PostgresSaver 时，传递 postgres 的连接句柄即可，示例如下：

from psycopg_pool import ConnectionPool

pool = ConnectionPool(
    # 示例配置
    conninfo=DB_URI,
    max_size=20,
    kwargs=connection_kwargs
)

with pool.connection() as conn:
    checkpointer = PostgresSaver(conn)
    # 注意：您需要在第一次使用检查点时调用 .setup()
    checkpointer.setup()

graph = create_react_agent(model, tools=tools, checkpointer=checkpointer)
config = {"configurable": {"thread_id": "1"}}
res = graph.invoke({"messages": [("human", "旧金山的天气怎么样?")]}, config)
checkpoint = checkpointer.get(config)

不过在 LangGraph 中封装的检查点绝大部分场合都不太适合我们的业务，特别是 Postgres 这类持久化的检查点，会在数据库中创建一张表（预设好特定字段），有时候这些持久化数据我们希望能够按照自定义的规则进行存

自定义检查点总共要实现 4 种方法:

• .put()：使用其配置和元数据存储检查点。
• .put_writes()：存储与检查点相关联的中间写入（即挂起的写入）。
• .get_tuple()：使用给定配置（thread_id 和 checkpoint_id）获取检查点元组。
• .list()：列出与给定配置和筛选条件匹配的检查点。

由于 checkpoint 检查点目前在 LangGraph 下发布时间不长，并且该功能目前仍然处于 beta 状态，接口随时可能发生更改，所以自定义一个检查点相对麻烦，而且不稳定。

创建自定检查点的相关使用技巧：https://langchain-ai.github.io/langgraph/how-tos/persistence_redis/