这是用户在 2024-7-29 10:35 为 https://langchain-ai.github.io/langgraph/concepts/low_level/ 保存的双语快照页面,由 沉浸式翻译 提供双语支持。了解如何保存?
Skip to content  跳到内容

Low Level Conceptual Guide
低级概念指南

Graphs 图表

At its core, LangGraph models agent workflows as graphs. You define the behavior of your agents using three key components:
从本质上讲,LangGraph 将座席工作流建模为图形。您可以使用三个关键组件来定义代理的行为:

  1. State: A shared data structure that represents the current snapshot of your application. It can be any Python type, but is typically a TypedDict or Pydantic BaseModel.
    状态:表示应用程序的当前快照的共享数据结构。它可以是任何 Python 类型,但通常是 TypedDict 或 Pydantic BaseModel

  2. Nodes: Python functions that encode the logic of your agents. They receive the current State as input, perform some computation or side-effect, and return an updated State.
    节点:对代理的逻辑进行编码的 Python 函数。它们接收当前状态作为输入,执行一些计算或副作用,并返回更新的状态

  3. Edges: Python functions that determine which Node to execute next based on the current State. They can be conditional branches or fixed transitions.
    Edges:根据当前状态确定接下来要执行哪个节点的 Python 函数。它们可以是条件分支,也可以是固定过渡。

By composing Nodes and Edges, you can create complex, looping workflows that evolve the State over time. The real power, though, comes from how LangGraph manages that State. To emphasize: Nodes and Edges are nothing more than Python functions - they can contain an LLM or just good ol' Python code.
通过组合节点,您可以创建复杂的循环工作流,这些工作流会随着时间的推移而演变状态。然而,真正的力量来自于 LangGraph 如何管理该状态。需要强调的是:节点只不过是 Python 函数 - 它们可以包含或LLM只是好的 Python 代码。

In short: nodes do the work. edges tell what to do next.
简而言之:节点完成工作。边缘告诉下一步该做什么

LangGraph's underlying graph algorithm uses message passing to define a general program. When a Node completes its operation, it sends messages along one or more edges to other node(s). These recipient nodes then execute their functions, pass the resulting messages to the next set of nodes, and the process continues. Inspired by Google's Pregel system, the program proceeds in discrete "super-steps."
LangGraph 的底层图算法使用消息传递来定义通用程序。当节点完成其操作时,它会沿着一条或多条边向其他节点发送消息。然后,这些接收方节点执行其功能,将生成的消息传递给下一组节点,然后该过程继续进行。受谷歌Pregel系统的启发,该程序以离散的“超级步骤”进行。

A super-step can be considered a single iteration over the graph nodes. Nodes that run in parallel are part of the same super-step, while nodes that run sequentially belong to separate super-steps. At the start of graph execution, all nodes begin in an inactive state. A node becomes active when it receives a new message (state) on any of its incoming edges (or "channels"). The active node then runs its function and responds with updates. At the end of each super-step, nodes with no incoming messages vote to halt by marking themselves as inactive. The graph execution terminates when all nodes are inactive and no messages are in transit.
可以将超级步骤视为对图形节点的单次迭代。并行运行的节点属于同一超步骤,而按顺序运行的节点属于单独的超步骤。在图形执行开始时,所有节点都以非活动状态开始。当节点在其任何传入边缘(或“通道”)上接收到新消息(状态)时,它就会变得活动。然后,活动节点运行其函数并使用更新进行响应。在每个超级步骤结束时,没有传入消息的节点通过将自己标记为非活动投票停止。当所有节点都处于非活动状态并且没有消息在传输中时,图形执行将终止。

StateGraph StateGraph(状态图)¶

The StateGraph class is the main graph class to uses. This is parameterized by a user defined State object.
StateGraph 类是要使用的主要图形类。这是由用户定义的 State 对象参数化的。

MessageGraph
MessageGraph(消息图)¶

The MessageGraph class is a special type of graph. The State of a MessageGraph is ONLY a list of messages. This class is rarely used except for chatbots, as most applications require the State to be more complex than a list of messages.
MessageGraph 类是一种特殊类型的图形。MessageGraph 的状态只是一个消息列表。除了聊天机器人之外,此类很少使用,因为大多数应用程序要求 State 比消息列表更复杂。

Compiling your graph
编译你的图表

To build your graph, you first define the state, you then add nodes and edges, and then you compile it. What exactly is compiling your graph and why is it needed?
要构建图形,首先要定义状态,然后添加节点,然后编译它。究竟是什么在编译你的图表,为什么需要它?

Compiling is a pretty simple step. It provides a few basic checks on the structure of your graph (no orphaned nodes, etc). It is also where you can specify runtime args like checkpointers and breakpoints. You compile your graph by just calling the .compile method:
编译是一个非常简单的步骤。它提供了一些关于图形结构的基本检查(没有孤立节点等)。您还可以在此处指定运行时参数,例如检查点断点。只需调用 .compile 方法即可编译图形:

graph = graph_builder.compile(...)

You MUST compile your graph before you can use it.
您必须先编译您的图表,然后才能使用它。

State 状态

The first thing you do when you define a graph is define the State of the graph. The State consists of the schema of the graph as well as reducer functions which specify how to apply updates to the state. The schema of the State will be the input schema to all Nodes and Edges in the graph, and can be either a TypedDict or a Pydantic model. All Nodes will emit updates to the State which are then applied using the specified reducer function.
定义图形时要做的第一件事就是定义图形的状态State图形的架构以及 reducer 函数组成,这些函数指定如何对状态应用更新。State 的架构将是图中所有 NodesEdge 的输入架构,可以是 TypedDictPydantic 模型。所有节点都将向 State 发出更新,然后使用指定的 reducer 函数应用这些更新。

Schema 架构

The main documented way to specify the schema of a graph is by using TypedDict. However, we also support using a Pydantic BaseModel as your graph state to add default values and additional data validation.
指定图形架构的主要记录方法是使用 TypedDict。但是,我们也支持使用 Pydantic BaseModel 作为图形状态,以添加默认值和其他数据验证。

By default, the graph will have the same input and output schemas. If you want to change this, you can also specify explicit input and output schemas directly. This is useful when you have a lot of keys, and some are explicitly for input and others for output. See the notebook here for how to use.
默认情况下,图形将具有相同的输入和输出架构。如果要更改此设置,还可以直接指定显式输入和输出架构。当您有很多键时,这很有用,其中一些键明确用于输入,而另一些键用于输出。有关如何使用,请参阅此处的笔记本

By default, all nodes in the graph will share the same state. This means that they will read and write to the same state channels. It is possible to have nodes write to private state channels inside the graph for internal node communication - see this notebook for how to do that.
默认情况下,图形中的所有节点将共享相同的状态。这意味着它们将读取和写入相同的状态通道。可以让节点写入图内的私有状态通道以进行内部节点通信 - 请参阅此笔记本了解如何执行此操作。

Reducers 减速器

Reducers are key to understanding how updates from nodes are applied to the State. Each key in the State has its own independent reducer function. If no reducer function is explicitly specified then it is assumed that all updates to that key should override it. Let's take a look at a few examples to understand them better.
Reducer 是理解如何将节点的更新应用于 State 的关键。State 中的每个键都有其自己的独立 reducer 函数。如果未明确指定 reducer 函数,则假定对该键的所有更新都应覆盖它。让我们看几个例子来更好地理解它们。

Example A: 示例 A:

from typing import TypedDict

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

In this example, no reducer functions are specified for any key. Let's assume the input to the graph is {"foo": 1, "bar": ["hi"]}. Let's then assume the first Node returns {"foo": 2}. This is treated as an update to the state. Notice that the Node does not need to return the whole State schema - just an update. After applying this update, the State would then be {"foo": 2, "bar": ["hi"]}. If the second node returns {"bar": ["bye"]} then the State would then be {"foo": 2, "bar": ["bye"]}
在此示例中,未为任何键指定 reducer 函数。假设图形的输入是 {“foo”: 1, “bar”: [“hi”]}。然后,我们假设第一个节点返回 {“foo”: 2}。这被视为对状态的更新。请注意,Node 不需要返回整个 State 架构 - 只需返回一次更新。应用此更新后,状态将为 {“foo”: 2, “bar”: [“hi”]}。如果第二个节点返回 {“bar”: [“bye”]},状态将为 {“foo”: 2, “bar”: [“bye”]}

Example B: 示例 B:

from typing import TypedDict, Annotated
from operator import add

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

In this example, we've used the Annotated type to specify a reducer function (operator.add) for the second key (bar). Note that the first key remains unchanged. Let's assume the input to the graph is {"foo": 1, "bar": ["hi"]}. Let's then assume the first Node returns {"foo": 2}. This is treated as an update to the state. Notice that the Node does not need to return the whole State schema - just an update. After applying this update, the State would then be {"foo": 2, "bar": ["hi"]}. If the second node returns {"bar": ["bye"]} then the State would then be {"foo": 2, "bar": ["hi", "bye"]}. Notice here that the bar key is updated by adding the two lists together.
在此示例中,我们使用 Annotated 类型为第二个键 (bar) 指定 reducer 函数 (operator.add)。请注意,第一个键保持不变。假设图形的输入是 {“foo”: 1, “bar”: [“hi”]}。然后,我们假设第一个节点返回 {“foo”: 2}。这被视为对状态的更新。请注意,Node 不需要返回整个 State 架构 - 只需返回一次更新。应用此更新后,状态将为 {“foo”: 2, “bar”: [“hi”]}。如果第二个节点返回 {“bar”: [“bye”]},State 将为 {“foo”: 2, “bar”: [“hi”, “bye”]}。请注意,此处的条形键是通过将两个列表相加来更新的。

Working with Messages in Graph State
处理 Graph 状态下的消息

Why use messages?
为什么要使用消息?

Most modern LLM providers have a chat model interface that accepts a list of messages as input. LangChain's ChatModel in particular accepts a list of Message objects as inputs. These messages come in a variety of forms such as HumanMessage (user input) or AIMessage (LLM response). To read more about what message objects are, please refer to this conceptual guide.
大多数现代LLM提供商都有一个聊天模型接口,该接口接受消息列表作为输入。特别是LangChain的ChatModel接受一系列Message对象作为输入。这些消息以多种形式出现,例如 HumanMessage(用户输入)或 AIMessage(LLM响应)。要了解有关消息对象的更多信息,请参阅概念指南。

Using Messages in your Graph
在 Graph 中使用消息

In many cases, it is helpful to store prior conversation history as a list of messages in your graph state. To do so, we can add a key (channel) to the graph state that stores a list of Message objects and annotate it with a reducer function (see messages key in the example below). The reducer function is vital to telling the graph how to update the list of Message objects in the state with each state update (for example, when a node sends an update). If you don't specify a reducer, every state update will overwrite the list of messages with the most recently provided value. If you wanted to simply append messages to the existing list, you could use operator.add as a reducer.
在许多情况下,将之前的对话历史记录存储为图形状态中的消息列表会很有帮助。为此,我们可以向图形状态添加一个键(通道),该键存储 Message 对象列表,并使用 reducer 函数对其进行注释(请参阅以下示例中的消息键)。reducer 函数对于告诉图形如何使用每次状态更新来更新处于状态的 Message 对象列表至关重要(例如,当节点发送更新时)。如果未指定 reducer,则每次状态更新都将覆盖具有最近提供的值的消息列表。如果您只想将消息追加到现有列表中,则可以使用 operator.add 作为 reducer。

However, you might also want to manually update messages in your graph state (e.g. human-in-the-loop). If you were to use operator.add, the manual state updates you send to the graph would be appended to the existing list of messages, instead of updating existing messages. To avoid that, you need a reducer that can keep track of message IDs and overwrite existing messages, if updated. To achieve this, you can use the prebuilt add_messages function. For brand new messages, it will simply append to existing list, but it will also handle the updates for existing messages correctly.
但是,您可能还希望手动更新处于图形状态(例如,人机在环)的消息。如果要使用 operator.add,则发送到图形的手动状态更新将追加到现有消息列表,而不是更新现有消息。为了避免这种情况,您需要一个 reducer,它可以跟踪消息 ID 并覆盖现有消息(如果已更新)。为此,您可以使用预构建的 add_messages 函数。对于全新的消息,它将简单地附加到现有列表,但它也会正确处理现有消息的更新。

Serialization
序列化

In addition to keeping track of message IDs, the add_messages function will also try to deserialize messages into LangChain Message objects whenever a state update is received on the messages channel. See more information on LangChain serialization/deserialization here. This allows sending graph inputs / state updates in the following format:
除了跟踪消息 ID 外,add_messages函数还会在消息通道上收到状态更新时,尝试将消息反序列化为 LangChain Message 对象。在此处查看有关LangChain序列化/反序列化的更多信息。这允许以以下格式发送图形输入/状态更新:

# this is supported
{"messages": [HumanMessage(content="message")]}

# and this is also supported
{"messages": [{"type": "human", "content": "message"}]}

Since the state updates are always deserialized into LangChain Messages when using add_messages, you should use dot notation to access message attributes, like state["messages"][-1].content. Below is an example of a graph that uses add_messages as it's reducer function.
由于在使用add_messages时,状态更新总是被反序列化为 LangChain Messages,因此您应该使用点表示法来访问消息属性,例如 state[“messages”][-1].content。下面是一个使用 add_messages 作为其 reducer 函数的图表示例。

from langchain_core.messages import AnyMessage
from langgraph.graph.message import add_messages
from typing import Annotated, TypedDict

class GraphState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

MessagesState

Since having a list of messages in your state is so common, there exists a prebuilt state called MessagesState which makes it easy to use messages. MessagesState is defined with a single messages key which is a list of AnyMessage objects and uses the add_messages reducer. Typically, there is more state to track than just messages, so we see people subclass this state and add more fields, like:
由于在您的状态中拥有消息列表非常普遍,因此存在一个名为 MessagesState 的预构建状态,这使得使用消息变得容易。MessagesState 使用单个 messages 键定义,该键是 AnyMessage 对象的列表,并使用 add_messages reducer。通常,要跟踪的状态不仅仅是消息,因此我们看到人们将此状态细分并添加更多字段,例如:

from langgraph.graph import MessagesState

class State(MessagesState):
    documents: list[str]

Nodes 节点

In LangGraph, nodes are typically python functions (sync or async) where the first positional argument is the state, and (optionally), the second positional argument is a "config", containing optional configurable parameters (such as a thread_id).
在 LangGraph 中,节点通常是 python 函数(sync 或 async),其中第一个位置参数是状态,(可选)第二个位置参数是“config”,包含可选的可配置参数(例如 thread_id)。

Similar to NetworkX, you add these nodes to a graph using the add_node method:
NetworkX 类似,您可以使用 add_node 方法将这些节点添加到图形中:

from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph

builder = StateGraph(dict)


def my_node(state: dict, config: RunnableConfig):
    print("In node: ", config["configurable"]["user_id"])
    return {"results": f"Hello, {state['input']}!"}


# The second argument is optional
def my_other_node(state: dict):
    return state


builder.add_node("my_node", my_node)
builder.add_node("other_node", my_other_node)
...

Behind the scenes, functions are converted to RunnableLambda's, which add batch and async support to your function, along with native tracing and debugging.
在后台,函数将转换为 RunnableLambda 的函数,从而为您的函数添加批处理和异步支持,以及本机跟踪和调试。

If you add a node to graph without specifying a name, it will be given a default name equivalent to the function name.
如果在未指定名称的情况下将节点添加到图形中,则将为其提供与函数名称等效的默认名称。

builder.add_node(my_node)
# You can then create edges to/from this node by referencing it as `"my_node"`

START Node
开始节点

The START Node is a special node that represents the node sends user input to the graph. The main purpose for referencing this node is to determine which nodes should be called first.
START 节点是一个特殊节点,表示节点将用户输入发送到图形。引用此节点的主要目的是确定应首先调用哪些节点。

from langgraph.graph import START

graph.add_edge(START, "node_a")

END Node
结束节点

The END Node is a special node that represents a terminal node. This node is referenced when you want to denote which edges have no actions after they are done.
END 节点是一个特殊节点,表示终端节点。当您想要表示哪些边在完成后没有操作时,将引用此节点。

from langgraph.graph import END

graph.add_edge("node_a", END)

Edges 边缘

Edges define how the logic is routed and how the graph decides to stop. This is a big part of how your agents work and how different nodes communicate with each other. There are a few key types of edges:
边定义了逻辑的路由方式以及图形决定停止的方式。这是代理工作方式以及不同节点之间通信方式的重要组成部分。有几种关键类型的边缘:

  • Normal Edges: Go directly from one node to the next.
    普通边缘:直接从一个节点转到下一个节点。
  • Conditional Edges: Call a function to determine which node(s) to go to next.
    条件边缘:调用函数以确定接下来要转到哪个节点。
  • Entry Point: Which node to call first when user input arrives.
    入口点:当用户输入到达时首先调用哪个节点。
  • Conditional Entry Point: Call a function to determine which node(s) to call first when user input arrives.
    条件入口点:调用一个函数以确定在用户输入到达时首先调用哪些节点。

A node can have MULTIPLE outgoing edges. If a node has multiple out-going edges, all of those destination nodes will be executed in parallel as a part of the next superstep.
一个节点可以有多个传出边。如果一个节点有多个传出边,则所有这些目标节点将作为下一个超级步骤的一部分并行执行。

Normal Edges
法线边缘

If you always want to go from node A to node B, you can use the add_edge method directly.
如果始终想从节点 A 转到节点 B,可以直接使用 add_edge 方法。

graph.add_edge("node_a", "node_b")

Conditional Edges
条件边

If you want to optionally route to 1 or more edges (or optionally terminate), you can use the add_conditional_edges method. This method accepts the name of a node and a "routing function" to call after that node is executed:
如果要选择性地路由到 1 个或多个 Edge(或选择性地终止),可以使用 add_conditional_edges 方法。此方法接受节点的名称和在执行该节点后要调用的“路由函数”:

graph.add_conditional_edges("node_a", routing_function)

Similar to nodes, the routing_function accept the current state of the graph and return a value.
与节点类似,routing_function接受图形的当前状态并返回一个值。

By default, the return value routing_function is used as the name of the node (or a list of nodes) to send the state to next. All those nodes will be run in parallel as a part of the next superstep.
默认情况下,返回值 routing_function 用作要将状态发送到下一个的节点(或节点列表)的名称。所有这些节点将作为下一个超级步骤的一部分并行运行。

You can optionally provide a dictionary that maps the routing_function's output to the name of the next node.
您可以选择提供一个字典,用于将routing_function的输出映射到下一个节点的名称。

graph.add_conditional_edges("node_a", routing_function, {True: "node_b", False: "node_c"})

Entry Point
入口点

The entry point is the first node(s) that are run when the graph starts. You can use the add_edge method from the virtual START node to the first node to execute to specify where to enter the graph.
入口点是图形启动时运行的第一个节点。您可以使用 add_edge 方法从虚拟 START 节点到要执行的第一个节点,以指定输入图形的位置。

from langgraph.graph import START

graph.add_edge(START, "node_a")

Conditional Entry Point
有条件入场点

A conditional entry point lets you start at different nodes depending on custom logic. You can use add_conditional_edges from the virtual START node to accomplish this.
通过条件入口点,您可以根据自定义逻辑从不同的节点开始。您可以使用虚拟 START 节点中的add_conditional_edges来实现此目的。

from langgraph.graph import START

graph.add_conditional_edges(START, routing_function)

You can optionally provide a dictionary that maps the routing_function's output to the name of the next node.
您可以选择提供一个字典,用于将routing_function的输出映射到下一个节点的名称。

graph.add_conditional_edges(START, routing_function, {True: "node_b", False: "node_c"})

Send
发送

By default, Nodes and Edges are defined ahead of time and operate on the same shared state. However, there can be cases where the exact edges are not known ahead of time and/or you may want different versions of State to exist at the same time. A common of example of this is with map-reduce design patterns. In this design pattern, a first node may generate a list of objects, and you may want to apply some other node to all those objects. The number of objects may be unknown ahead of time (meaning the number of edges may not be known) and the input State to the downstream Node should be different (one for each generated object).
默认情况下,节点Edge 是提前定义的,并在相同的共享状态下运行。但是,在某些情况下,事先不知道确切的边缘,并且/或者您可能希望同时存在不同版本的状态。一个常见的示例是map-reduce设计模式。在此设计模式中,第一个节点可能会生成对象列表,您可能希望将其他节点应用于所有这些对象。对象的数量可能事先是未知的(这意味着边的数量可能是未知的),并且下游节点的输入状态应该不同(每个生成的对象一个)。

To support this design pattern, LangGraph supports returning Send objects from conditional edges. Send takes two arguments: first is the name of the node, and second is the state to pass to that node.
为了支持这种设计模式,LangGraph 支持从条件边返回 Send 对象。Send 有两个参数:第一个是节点的名称,第二个是要传递给该节点的状态。

def continue_to_jokes(state: OverallState):
    return [Send("generate_joke", {"subject": s}) for s in state['subjects']]

graph.add_conditional_edges("node_a", continue_to_jokes)

Checkpointer
检查点

LangGraph has a built-in persistence layer, implemented through checkpointers. When you use a checkpointer with a graph, you can interact with the state of that graph. When you use a checkpointer with a graph, you can interact with and manage the graph's state. The checkpointer saves a checkpoint of the graph state at every super-step, enabling several powerful capabilities:
LangGraph 有一个内置的持久化层,通过检查点实现。当您将检查点与图形一起使用时,您可以与该图形的状态进行交互。当您将检查点与图形一起使用时,您可以与图形的状态进行交互并管理其状态。检查点在每个超级步骤中保存图形状态的检查点,从而启用几个强大的功能:

First, checkpointers facilitate human-in-the-loop workflows workflows by allowing humans to inspect, interrupt, and approve steps.Checkpointers are needed for these workflows as the human has to be able to view the state of a graph at any point in time, and the graph has to be to resume execution after the human has made any updates to the state.
首先,检查点允许人工检查、中断和批准步骤,从而促进人机交互工作流。这些工作流需要检查点,因为人类必须能够在任何时间点查看图形的状态,并且图形必须在人类对状态进行任何更新后恢复执行。

Second, it allows for "memory" between interactions. You can use checkpointers to create threads and save the state of a thread after a graph executes. In the case of repeated human interactions (like conversations) any follow up messages can be sent to that checkpoint, which will retain its memory of previous ones.
其次,它允许交互之间的“记忆”。您可以使用检查点来创建线程,并在图形执行后保存线程的状态。在重复的人类交互(如对话)的情况下,任何后续消息都可以发送到该检查点,该检查点将保留其先前的记忆。

See this guide for how to add a checkpointer to your graph.
请参阅本指南,了解如何向图表添加检查点。

Threads 线程

Threads enable the checkpointing of multiple different runs, making them essential for multi-tenant chat applications and other scenarios where maintaining separate states is necessary. A thread is a unique ID assigned to a series of checkpoints saved by a checkpointer. When using a checkpointer, you must specify a thread_id or thread_ts when running the graph.
线程支持对多个不同的运行进行检查点,这使得它们对于多租户聊天应用程序和其他需要维护单独状态的方案至关重要。线程是分配给检查点器保存的一系列检查点的唯一 ID。使用检查点时,必须在运行图形时指定thread_idthread_ts

thread_id is simply the ID of a thread. This is always required
thread_id 只是线程的 ID。这始终是必需的

thread_ts can optionally be passed. This identifier refers to a specific checkpoint within a thread. This can be used to kick of a run of a graph from some point halfway through a thread.
可以选择性地传递thread_ts。此标识符是指线程中的特定检查点。这可以用来从线程中途的某个点开始运行图形。

You must pass these when invoking the graph as part of the configurable part of the config.
在将图形作为配置的可配置部分的一部分调用时,必须传递这些内容。

config = {"configurable": {"thread_id": "a"}}
graph.invoke(inputs, config=config)

See this guide for how to use threads.
有关如何使用线程的信息,请参阅本指南

Checkpointer state
检查点状态

When interacting with the checkpointer state, you must specify a thread identifier.Each checkpoint saved by the checkpointer has two properties:
与检查点状态交互时,必须指定线程标识符。检查点保存的每个检查点都有两个属性:

  • values: This is the value of the state at this point in time.
    values:这是此时状态的值。
  • next: This is a tuple of the nodes to execute next in the graph.
    next:这是图中接下来要执行的节点元组。

Get state 获取状态

You can get the state of a checkpointer by calling graph.get_state(config). The config should contain thread_id, and the state will be fetched for that thread.
您可以通过调用 graph.get_state(config) 来获取检查点的状态。配置应包含thread_id,并且将获取该线程的状态。

Get state history
获取状态历史记录

You can also call graph.get_state_history(config) to get a list of the history of the graph. The config should contain thread_id, and the state history will be fetched for that thread.
您也可以调用 graph.get_state_history(config) 来获取图表的历史记录列表。配置应包含thread_id,并且将获取该线程的状态历史记录。

Update state
更新状态

You can also interact with the state directly and update it. This takes three different components:
您还可以直接与状态交互并更新它。这需要三个不同的组件:

  • config 配置
  • values 
  • as_node

config 配置

The config should contain thread_id specifying which thread to update.
配置应包含指定要更新的线程thread_id

values 

These are the values that will be used to update the state. Note that this update is treated exactly as any update from a node is treated. This means that these values will be passed to the reducer functions that are part of the state. So this does NOT automatically overwrite the state. Let's walk through an example.
这些是将用于更新状态的值。请注意,此更新的处理方式与处理来自节点的任何更新完全相同。这意味着这些值将被传递给属于状态的 reducer 函数。因此,这不会自动覆盖状态。让我们看一个例子。

Let's assume you have defined the state of your graph as:
假设您已将图形的状态定义为:

from typing import TypedDict, Annotated
from operator import add

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

Let's now assume the current state of the graph is
现在让我们假设图形的当前状态为

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

If you update the state as below:
如果按如下方式更新状态:

graph.update_state(config, {"foo": 2, "bar": ["b"]})

Then the new state of the graph will be:
那么图形的新状态将是:

{"foo": 2, "bar": ["a", "b"]}

The foo key is completely changed (because there is no reducer specified for that key, so it overwrites it). However, there is a reducer specified for the bar key, and so it appends "b" to the state of bar.
foo 键已完全更改(因为没有为该键指定 reducer,因此它会覆盖它)。但是,为 bar 键指定了一个 reducer,因此它将“b”附加到 bar 的状态。

as_node

The final thing you specify when calling update_state is as_node. This update will be applied as if it came from node as_node. If as_node is not provided, it will be set to the last node that updated the state, if not ambiguous.
调用 update_state 时指定的最后一件事是as_node。此更新将应用,就好像它来自节点 as_node一样。如果未提供 as_node,则将设置为更新状态的最后一个节点(如果不是不明确)。

The reason this matters is that the next steps in the graph to execute depend on the last node to have given an update, so this can be used to control which node executes next.
这很重要的原因是,图中接下来要执行的步骤取决于最后一个提供更新的节点,因此这可以用来控制下一个节点执行。

Graph Migrations
图迁移

LangGraph can easily handle migrations of graph definitions (nodes, edges, and state) even when using a checkpointer to track state.
LangGraph 可以轻松处理图形定义(节点、边缘和状态)的迁移,即使在使用检查点跟踪状态时也是如此。

  • For threads at the end of the graph (i.e. not interrupted) you can change the entire topology of the graph (i.e. all nodes and edges, remove, add, rename, etc)
    对于图形末尾的线程(即未中断),您可以更改图形的整个拓扑结构(即所有节点和边、删除、添加、重命名等)
  • For threads currently interrupted, we support all topology changes other than renaming / removing nodes (as that thread could now be about to enter a node that no longer exists) -- if this is a blocker please reach out and we can prioritize a solution.
    对于当前中断的线程,我们支持除重命名/删除节点以外的所有拓扑更改(因为该线程现在可能即将进入不再存在的节点)——如果这是一个阻碍因素,请伸出援手,我们可以优先考虑解决方案。
  • For modifying state, we have full backwards and forwards compatibility for adding and removing keys
    对于修改状态,我们在添加和删除键方面具有完全的向后和向前兼容性
  • State keys that are renamed lose their saved state in existing threads
    重命名的状态键将丢失其在现有线程中的保存状态
  • State keys whose types change in incompatible ways could currently cause issues in threads with state from before the change -- if this is a blocker please reach out and we can prioritize a solution.
    如果状态键的类型以不兼容的方式更改,则当前可能会导致线程中出现问题,其状态与更改之前的状态相同 - 如果这是一个阻碍因素,请与我们联系,我们可以优先考虑解决方案。

Configuration
配置

When creating a graph, you can also mark that certain parts of the graph are configurable. This is commonly done to enable easily switching between models or system prompts. This allows you to create a single "cognitive architecture" (the graph) but have multiple different instance of it.
创建图表时,您还可以标记图表的某些部分是可配置的。这样做通常是为了能够在模型或系统提示之间轻松切换。这使您可以创建单个“认知架构”(图形),但具有多个不同的实例。

You can optionally specify a config_schema when creating a graph.
您可以选择在创建图表时指定config_schema

class ConfigSchema(TypedDict):
    llm: str

graph = StateGraph(State, config_schema=ConfigSchema)

You can then pass this configuration into the graph using the configurable config field.
然后,您可以使用可配置的配置字段将此配置传递到图形中。

config = {"configurable": {"llm": "anthropic"}}

graph.invoke(inputs, config=config)

You can then access and use this configuration inside a node:
然后,您可以在节点内访问和使用此配置:

def node_a(state, config):
    llm_type = config.get("configurable", {}).get("llm", "openai")
    llm = get_llm(llm_type)
    ...

See this guide for a full breakdown on configuration
有关配置的完整细分,请参阅本指南

Breakpoints
断点

It can often be useful to set breakpoints before or after certain nodes execute. This can be used to wait for human approval before continuing. These can be set when you "compile" a graph. You can set breakpoints either before a node executes (using interrupt_before) or after a node executes (using interrupt_after.)
在某些节点执行之前或之后设置断点通常很有用。这可用于等待人工批准,然后再继续。这些可以在“编译”图形时设置。您可以在节点执行之前(使用 interrupt_before)或在节点执行后(使用 interrupt_after)设置断点。

You MUST use a checkpoiner when using breakpoints. This is because your graph needs to be able to resume execution.
使用断点时,必须使用checkpoiner。这是因为您的图形需要能够恢复执行。

In order to resume execution, you can just invoke your graph with None as the input.
为了恢复执行,您只需调用将 None 作为输入的图形。

# Initial run of graph
graph.invoke(inputs, config=config)

# Let's assume it hit a breakpoint somewhere, you can then resume by passing in None
graph.invoke(None, config=config)

See this guide for a full walkthrough of how to add breakpoints.
请参阅本指南,了解如何添加断点的完整演练。

Visualization
可视化

It's often nice to be able to visualize graphs, especially as they get more complex. LangGraph comes with several built-in ways to visualize graphs. See this how-to guide for more info.
能够可视化图表通常很高兴,尤其是当它们变得更加复杂时。LangGraph 带有几种内置的可视化图表的方法。有关详细信息,请参阅此操作指南

Streaming 流式处理

LangGraph is built with first class support for streaming. There are several different streaming modes that LangGraph supports:
LangGraph 具有一流的流媒体支持。LangGraph 支持几种不同的流式处理模式:

  • "values": This streams the full value of the state after each step of the graph.
    “values”:在图形的每个步骤之后,这将流式传输状态的全部值。
  • "updates: This streams the updates to the state after each step of the graph. If multiple updates are made in the same step (e.g. multiple nodes are run) then those updates are streamed separately.
    “updates:这会在图形的每个步骤之后将更新流式传输到状态。如果在同一步骤中进行了多个更新(例如,运行多个节点),则这些更新将单独流式传输。
  • "debug": This streams as much information as possible throughout the execution of the graph.
    “debug”:在图形的整个执行过程中,这将尽可能多地传输信息。

In addition, you can use the astream_events method to stream back events that happen inside nodes. This is useful for streaming tokens of LLM calls.
此外,还可以使用 astream_events 方法流回节点内部发生的事件。这对于流LLM式传输调用令牌非常有用。

Comments 评论