状态

定义

State 由模式+归约器函数组成
State 的模式将作为图中所有 Node（节点）和 Edge（边）的输入模式
所有 Node 都会向 State 发出更新，然后使用指定的 归约器函数完成这些更新

模式

TypedDict

from typing import TypedDict
from typing_extensions import Annotated

def add(left, right):
  return left + right

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

Pydantic BaseModel

from pydantic import BaseModel

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

嵌套的 Pydantic 模型

class NestedModel(BaseModel):
  value: str

class State(BaseModel):
  foo: str
  nested: NestedModel

单模式

通常，所有图节点都使用单一模式进行通信，它们会读写相同的状态

from typing import TypedDict
from langgraph.graph import StateGraph, START

class State(TypedDict):
  foo: str

def node_1(state: State) -> State:
  return { "foo": "1" }

def node_2(state: State) -> State:
  return { "foo": "2" }

builder = StateGraph(State).add_sequence([node_1, node_2]).add_edge(START, "node_1")

多模式

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:
    # 写入 OverallState
    return {"foo": state["user_input"] + " name"}

def node_2(state: OverallState) -> PrivateState:
    # 从 OverallState 读取，写入 PrivateState
    return {"bar": state["foo"] + " is"}

def node_3(state: PrivateState) -> OutputState:
    # 从 PrivateState 读取，写入 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'}

归约器（reduce）

默认归约器（覆盖）

from typing_extensions import TypedDict

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

图的输入是 {"foo": 1, "bar": ["hi"]}。
第一个 Node 返回 {"foo": 2}。应用此更新后，State 将变为 {"foo": 2, "bar": ["hi"]}。
第二个节点返回 {"bar": ["bye"]}，则 State 将变为 {"foo": 2, "bar": ["bye"]}。

自定义归约器

from typing import TypedDict
from typing_extensions import Annotated

def add(left, right):
  return left + right

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

图的输入是 {"foo": 1, "bar": ["hi"]}。
第一个 Node 返回 {"foo": 2}。应用此更新后，State 将变为 {"foo": 2, "bar": ["hi"]}。
第二个节点返回 {"bar": ["bye"]}，则 State 将变为 {"foo": 2, "bar": ["hi", "bye"]}。