AI创想

标题: Langgraph小入门 [打印本页]

---

作者: 米落枫    时间: 4 小时前
标题: Langgraph小入门
作者：CSDN博客
langgraph官方中文文档
LangGraph

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

(, 下载次数: 0)

上传

点击文件名下载附件

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

工具调用：这通常是LLM做出决策的方式采取行动：通过，LLM的输出用作操作的输入内存：可靠的系统需要了解发生的事情计划：计划步骤（显式或隐式）有助于确保LLM在做出决策时以最高保真度做出决策。

图

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

状态：一个共享数据结构，表示应用程序的当前快照。它可以是任何 Python 类型，但通常是 TypedDict 或 Pydantic BaseModel。节点：Python 函数，用于编码代理的逻辑。它们以当前 状态 作为输入，执行一些计算或副作用，并返回更新后的 状态。边：Python 函数，根据当前 状态 确定要执行的下一个 节点。它们可以是条件分支或固定转换。

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

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

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

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

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

1. graph = graph_builder.compile(...)

复制代码

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

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

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

模式

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

1. classMyState(TypedDict):
2.     key1:int
3.     key2:str

复制代码

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

1. from pydantic import BaseModel
3. class MyState(BaseModel):
4.     key1: int = 0  # 默认值
5.     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 节点结束。
  
  
  

1. class InputState(TypedDict):
2.     user_input: str
4. class OutputState(TypedDict):
5.     graph_output: str
7. class OverallState(TypedDict):
8.     foo: str
9.     user_input: str
10.     graph_output: str
12. class PrivateState(TypedDict):
13.     bar: str
15. def node_1(state: InputState) -> OverallState:
16.     # Write to OverallState
17.     return {"foo": state["user_input"] + " name"}
19. def node_2(state: OverallState) -> PrivateState:
20.     # Read from OverallState, write to PrivateState
21.     return {"bar": state["foo"] + " is"}
23. def node_3(state: PrivateState) -> OutputState:
24.     # Read from PrivateState, write to OutputState
25.     return {"graph_output": state["bar"] + " Lance"}
27. builder = StateGraph(OverallState,input=InputState,output=OutputState)
28. builder.add_node("node_1", node_1)
29. builder.add_node("node_2", node_2)
30. builder.add_node("node_3", node_3)
31. builder.add_edge(START, "node_1")
32. builder.add_edge("node_1", "node_2")
33. builder.add_edge("node_2", "node_3")
34. builder.add_edge("node_3", END)
36. graph = builder.compile()
37. graph.invoke({"user_input":"My"})
38. {'graph_output': 'My name is Lance'}
复制代码

化简器（Reducer）

化简器概念

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

每个状态键可以有自己的化简器函数。如果没有明确指定化简器，默认行为是**覆盖（replace）**旧的值，即节点返回的更新会直接替换现有的状态值。

示例A：默认化简器

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

1. from typing_extensions import TypedDict
3. class State(TypedDict):
4.     foo: int
5.     bar: list[str]

复制代码

• State 包含两个字段：
  
  foo: 整数类型。bar: 字符串列表类型。
  
  
  

假设输入状态为：

1. {
2.    
3.    "foo":1,"bar":["hi"]}

复制代码

假设第一个节点返回：

1. {
2.    
3.    "foo":2}

复制代码

根据默认reducer的规则，只更新foo字段.

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

  复制代码

假设第二个节点返回：

1. {
2.    
3.    "bar":["bye"]}

复制代码

最终状态变为：

1. {
2.    
3.    "foo":2,"bar":["bye"]}

复制代码

示例B：指定化简器函数

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

1. from typing import Annotated
2. from typing_extensions import TypedDict
3. from operator import add
5. classState(TypedDict):
6.     foo:int
7.     bar: Annotated[list[str], add]

复制代码

在这个 State 中，foo 没有化简器，因此仍然使用默认的覆盖行为。bar 使用 operator.add 作为化简器，这意味着每次更新 bar 时，它会将新的列表与现有的列表 合并（通过 add）。

假设输入状态：

1. {
2.    
3.    "foo":1,"bar":["hi"]}

复制代码

假设第一个节点返回：

1. {
2.    
3.    "foo":2}

复制代码

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

  复制代码

假设第二个节点返回：

1. {
2.    
3.    "bar":["bye"]}

复制代码

由于 bar 使用了 operator.add 作为化简器，["bye"] 将与 ["hi"] 合并。

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

  复制代码

在状态图中使用消息

1.为什么要使用消息？

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

**HumanMessage：**代表用户输入的消息。**AIMessage：**代表LLM生成的响应。**SystemMessage：**代表系统给出的说明或指令。

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

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

**消息存储：**可以通过向图状态添加一个键（例如message），将消息对象的列表存储在其中。

• 使用 reducer 管理消息更新：当状态更新时，reducer 函数会告诉图如何处理这些更新。如果你不指定 reducer，每次状态更新都会覆盖之前的消息列表。
  
  如果想要将新的消息追加到现有的消息列表中，可以使用 operator.add 作为 reducer，这样每次状态更新时，新消息都会追加到现有消息中。
  
  
  

示例代码说明

1. from langchain_core.messages import AnyMessage
2. from langgraph.graph.message import add_messages
3. from typing import Annotated
4. from typing_extensions import TypedDict
6. class GraphState(TypedDict):
7.     messages: Annotated[list[AnyMessage], add_messages]

复制代码

GraphState：这里定义了一个 GraphState，它包含一个 messages 键，该键使用 add_messages 作为其 reducer 函数。add_messages 函数：这个函数不仅可以处理消息的追加，还可以跟踪消息的唯一 ID，确保更新已经存在的消息而不是重复添加。

3.手动更新消息列表

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

如果是全新的消息，它会将消息追加到列表中。如果消息已存在，它会更新该消息，而不是简单地追加。

4.序列化与反序列化

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

  复制代码

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

---

欢迎光临 AI创想 (https://llms-ai.com/)

Powered by Discuz! X3.4