Langgraph小入门

米落枫 · 发表于 2 小时前

作者：CSDN博客
langgraph官方中文文档
LangGraph

不同的Agent架构赋予LLM不同程度的控制权。在一个极端情况下，路由器允许LLMLLM充当路由器从指定的一组选项中选择一个步骤，而在另一个极端情况下，完全自主的长期运行代理可以只有选择任何它希望为给定问题选择的步骤序列。

许多Agnet框架都使用了以下一些概念：

图

LangGraph 的核心是将代理工作流建模为图。您可以使用三个关键组件定义代理的行为

通过组合节点和边，您可以创建随着时间推移而演变状态的复杂循环工作流。但是，真正的力量来自于 LangGraph 如何管理状态。强调一点：节点和边仅仅是 Python 函数 - 它们可以包含 LLM 或普通的 Python 代码。
简而言之：节点完成工作，边指示下一步要做什么。
消息传递

在LangGraph中，图由多个节点（代表操作或任务）和连接这些节点的边组成。消息传递是图中节点之间进行通信的方式。当一个节点完成操作时，他会将结果作为一条消息，沿着连接的边发送给其他节点。接收到这些消息的节点则会使用该信息执行自己的操作，然后继续讲处理后的消息传递给下一个节点。这个过程类似于分布式系统中的通信方式。
状态图（StateGraph）

StateGraph类是使用的主要图类。它由用户定义的State对象参数化。
消息图（MessageGraph）

MessageGraph类是一种特殊图类型。MessageGraph的State仅为消息列表。除了聊天机器人外，很少使用此类，因为大多数应用程序都需要State比消息列表更复杂。
编译图

要构建图，首先定义状态，然后添加节点和边，最后编译它。
编译是一个非常简单的步骤。它对图的结构进行一些基本检查（没有孤立节点等）。它也是您可以在其中指定运行时参数的地方，例如检查点和断点。您可以通过调用 .compile 方法来编译您的图

graph = graph_builder.compile(...)

复制代码

必须在使用前编译它。
State（状态）

定义图时，首先要做的就是定义图的State。State包含图的模式以及化简器函数，这些函数指定如何将更新应用于状态。State的模式将成为图中所有节点和边的输入模式，并且可以是TypedDict或Pydantic模型。所有节点都将发出对State的更新，这些更新随后使用指定的化简器函数进行应用。

状态的模式将成为图中所有节点和边的输入模式，这意味着节点和边的所有操作都是基于这个模式的状态进行的。换句话说，图的所有组件在处理数据时都会遵循这个模式的定义。

模式

在LangGraph中，定义状态的主要方式是使用TypedDict。TypedDict是Python中的一种工具，允许为字典中的每个键指定明确类型。例如

classMyState(TypedDict):
key1:int
key2:str

复制代码

除了TypedDict外，LangGraph也支持使用Pydantic的BaseModel来定义状态。
使用Pydantic，可以为状态中的每个字段设置默认值，并且Pydantic会自动进行类型检查和验证。

from pydantic import BaseModel
class MyState(BaseModel):
key1: int = 0 # 默认值
key2: str

复制代码

默认的输入和输出模式

默认情况下爱，LangGraph的图会使用相同的模式来处理输入和输出，也就是说，图的输入状态结构和输出状态结构是一样的。这适用于大多数场景，因为输入和输出通常是一致的。
显式指定输入和输出模式

在某些情况下，希望图的输入模式和输出模式有所不同（例如，当图的输入和输出有不同的键，或者有些键旨在输入时有意义，而输出时不需要），则可以显式地指定不同的输入和输出模式。
多个模式的使用

不同的输入/输出模式

代码讲解
1.定义状态模式
首先，代码定义了四个不同的状态模式 (InputState、OutputState、OverallState 和 PrivateState)，这些模式用于描述图中不同部分的状态结构。

InputState：包含一个键user_input，用于表示图的输入，即用户提供的输入数据。OutputState：包含一个键graph_output，表示图的输出结果，图处理后的最终输出。

OverallState ：是图的整体状态，包含三个字段

foo：用来存储中间的计算结果。user_input：用于记录用户输入。graph_output：记录图的最终输出结果。

PrivateState：包含一个字段 bar，是图内部使用的私有状态，用来在节点间传递数据，不会直接暴露给外部。

3.构建图

StateGraph(OverallState, input=InputState, output=OutputState)：

这里初始化了一个图，定义了该图的整体状态为 OverallState，输入模式为 InputState，输出模式为 OutputState。也就是说，图将从 InputState 获取输入，并最终输出 OutputState。

builder.add_node(...)：这些行代码将节点函数添加到图中，节点按顺序执行：

node_1 处理用户输入并返回 OverallState。node_2 处理 OverallState 并返回 PrivateState。node_3 处理 PrivateState 并返回最终的 OutputState。

builder.add_edge(...)：这些行代码定义了图的执行顺序，规定了节点之间的执行顺序：

图从 START 开始，先执行 node_1。node_1 执行完后，结果传递给 node_2。然后执行 node_3，并在 END 节点结束。

class InputState(TypedDict):
user_input: str
class OutputState(TypedDict):
graph_output: str
class OverallState(TypedDict):
foo: str
user_input: str
graph_output: str
class PrivateState(TypedDict):
bar: str
def node_1(state: InputState) -> OverallState:
# Write to OverallState
return {"foo": state["user_input"] + " name"}
def node_2(state: OverallState) -> PrivateState:
# Read from OverallState, write to PrivateState
return {"bar": state["foo"] + " is"}
def node_3(state: PrivateState) -> OutputState:
# Read from PrivateState, write to OutputState
return {"graph_output": state["bar"] + " Lance"}
builder = StateGraph(OverallState,input=InputState,output=OutputState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
builder.add_edge("node_2", "node_3")
builder.add_edge("node_3", END)
graph = builder.compile()
graph.invoke({"user_input":"My"})
{'graph_output': 'My name is Lance'}

复制代码

化简器（Reducer）

化简器概念

化简器（Reducer）是用来管理状态更新的函数。它的作用是在节点返回部分状态更新时，决定如何将这些更新应用到图的整体状态。

示例A：默认化简器

在这个例子中，没有为状态键显式指定reducer，所以会使用默认reducer，即覆盖当前状态。

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

复制代码

State 包含两个字段：

假设输入状态为：

{
"foo":1,"bar":["hi"]}

复制代码

假设第一个节点返回：

{
"foo":2}

复制代码

结果状态变为：
1. {
3. "foo":2,"bar":["hi"]}
复制代码

假设第二个节点返回：

{
"bar":["bye"]}

复制代码

最终状态变为：

{
"foo":2,"bar":["bye"]}

复制代码

示例B：指定化简器函数

在这个示例中，为 bar 字段指定了一个化简器函数 operator.add，这意味着更新时将执行 累加操作 而不是覆盖操作。

from typing import Annotated
from typing_extensions import TypedDict
from operator import add
classState(TypedDict):
foo:int
bar: Annotated[list[str], add]

复制代码

合并

假设输入状态：

{
"foo":1,"bar":["hi"]}

复制代码

假设第一个节点返回：

{
"foo":2}

复制代码

结果状态变为
1. {
3. "foo":2,"bar":["hi"]}
复制代码

假设第二个节点返回：

{
"bar":["bye"]}

复制代码

最终状态变为：
1. {
3. "foo":2,"bar":["hi","bye"]}
复制代码

在状态图中使用消息

1.为什么要使用消息？

LLM模型，尤其对话模型，通常接受一个消息列表作为输入。每个消息对象可能有不同的类型，比如：

这些消息对象能够帮助模型理解对话的上下文。因此，使用消息对象列表在图中存储和管理对话历史非常有用。
2.在图状态中使用消息

在LangGraph中，可以将消息历史存储在图状态中，这样可以跟踪与LLM的交互历史。

使用 reducer 管理消息更新：当状态更新时，reducer 函数会告诉图如何处理这些更新。如果你不指定 reducer，每次状态更新都会覆盖之前的消息列表。

示例代码说明

from langchain_core.messages import AnyMessage
from langgraph.graph.message import add_messages
from typing import Annotated
from typing_extensions import TypedDict
class GraphState(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]

复制代码

GraphState

add_messages 函数

3.手动更新消息列表

如果你想手动更新图中的消息（例如人工干预），使用 operator.add 时会将你的更新追加到现有消息列表中，而不是覆盖。为了避免这种情况，你可以使用 add_messages 函数，它会检查消息的 ID：

4.序列化与反序列化

add_messages 函数的序列化功能：该函数还具备将消息反序列化为 LangChain 消息对象的功能，这意味着你可以以不同的格式更新消息。例如，以下两种格式都可以被图状态接受：
1. {
3. "messages":[
复制代码

原文地址：https://blog.csdn.net/xnuscd/article/details/143474722

25位大厂高管转战生成式AI创业！吸金猛，不

Langgraph小入门

浏览过的版块

大模型时代下向量数据库的创新与变革

关于我们

产品与服务

解决方案

产品与服务