3.1. 状态定义状态的定义实际上是在声明状态的Schema后者是状态字段的完整描述。官方推荐了三种定义Schema的方式TypedDict、dataclass、Pydantic3.1.1. TypedDictfrom typing import TypedDict,Annotated from operator import add from langgraph.graph import StateGraph,START,END #定义状态 class OverAllStatus(TypedDict): #日志的类型是list[str]最新的方式不是覆盖是add 追加 logs: Annotated[list[str], add] cur_id:str # 定义节点 def node_1(state: OverAllStatus) - OverAllStatus: pre_idstate[cur_id] return { logs:[node_1 执行完], cur_id:pre_id,node_1 } # 定义节点 def node_2(state: OverAllStatus) - OverAllStatus: pre_idstate[cur_id] return { logs:[node_2 执行完], cur_id:pre_id,node_2 } # 定义边 #3.1 创建图 获取建造者 builderStateGraph(state_schemaOverAllStatus) #3.2 添加节点 builder.add_node(node_1) builder.add_node(node_2) #3.3 添加边 builder.add_edge(START,node_1) builder.add_edge(node_1,node_2) builder.add_edge(node_2,END) #4. 编译图 graphbuilder.compile() #5. 执行图 resultgraph.invoke({logs:[],cur_id:start}) print(result)输出{logs: [node_1 执行完, node_2 执行完], cur_id: start,node_1,node_2}3.1.2. dataclass属性调用方式由[字段名]变为.字段名。from typing import Annotated from dataclasses import dataclass from operator import add from langgraph.graph import StateGraph,START,END #定义状态 dataclass class OverAllStatus: #日志的类型是list[str]最新的方式不是覆盖是add 追加 logs: Annotated[list[str], add] cur_id:str # 定义节点 def node_1(state: OverAllStatus) - OverAllStatus: pre_idstate.cur_id return OverAllStatus( [node_1 执行完], pre_id,node_1 ) # 定义节点 def node_2(state: OverAllStatus) - OverAllStatus: pre_idstate.cur_id return OverAllStatus( [node_2 执行完], pre_id,node_2 ) # 定义边 #3.1 创建图 获取建造者 builderStateGraph(state_schemaOverAllStatus) #3.2 添加节点 builder.add_node(node_1) builder.add_node(node_2) #3.3 添加边 builder.add_edge(START,node_1) builder.add_edge(node_1,node_2) builder.add_edge(node_2,END) #4. 编译图 graphbuilder.compile() #5. 执行图 resultgraph.invoke({logs:[],cur_id:start}) print(result)输出{logs: [node_1 执行完, node_2 执行完], cur_id: start,node_1,node_2}3.1.3. PydanticPydantic模型的字段访问方式和dataclass相同。from typing import Annotated from operator import add from pydantic import BaseModel from langgraph.graph import StateGraph, START, END # 定义状态 class OverAllStatus(BaseModel): # 使用 add 作为归约函数将每个节点返回的日志追加到原列表 logs: Annotated[list[str], add] cur_id: str # 定义节点 def node_1(state: OverAllStatus) - dict: return { logs: [node_1 执行完], cur_id: f{state.cur_id},node_1, } def node_2(state: OverAllStatus) - dict: return { logs: [node_2 执行完], cur_id: f{state.cur_id},node_2, } # 创建图 builder StateGraph(state_schemaOverAllStatus) # 添加节点并明确指定节点名称 builder.add_node(node_1, node_1) builder.add_node(node_2, node_2) # 添加边 builder.add_edge(START, node_1) builder.add_edge(node_1, node_2) builder.add_edge(node_2, END) # 编译图 graph builder.compile() # 执行图 result graph.invoke({ logs: [], cur_id: start, }) print(result)输出{logs: [node_1 执行完, node_2 执行完], cur_id: start,node_1,node_2}3.1.4. 校验行为学习LangChain的结构化输出时我们提到Pydantic对格式要求最严格如果模型返回的内容不符合结构化Schema的要求则抛出ValidationError。而其余方式都不会对模型的返回结果进行校验即便模型返回的内容不符合结构化要求也会原样返回给用户。而作为LangGraph计算图的状态时这三种方式都要求字段名称完全一致。只是处理方式不同。具体规则如下3.1.4.1. 输入字段不匹配TypedDict将输入字段视为字典的Key不匹配时抛出KeyError异常。dataclass将输入字段视为类的属性不匹配时抛出TypeError类型错误异常Pydantic对输入字段进行校验不匹配时抛出ValidationError异常。3.1.4.2. 节点返回字段不匹配图节点返回的是对于状态的更新如果返回字段和状态字段不匹配上述三种Schema定义方式的行为是统一的状态更新会被忽略3.1.5. 推荐用法在实际使用中推荐优先使用TypedDict定义 LangGraph 状态图的 State Schema。大多数官方案例也采用TypedDict方式定义状态Schema。这种方式写法简洁、结构清晰能够直接描述状态中包含哪些字段以及每个字段对应的数据类型非常适合用于定义图运行过程中的共享状态。相比普通dictTypedDict可以提供更明确的字段约束和类型提示相比dataclassTypedDict更贴近 LangGraph 中状态的更新方式因为节点通常返回的是表示“部分状态更新”的字典而不是完整对象相比Pydantic BaseModel它又更加轻量不会引入额外的数据校验开销。因此在没有复杂校验需求的情况下TypedDict是定义 LangGraph State Schema 的首选方式。3.2. State Reducer3.2.1. 什么是State ReducerState Reducer 是 LangGraph 中用于合并状态更新的核心机制。在 LangGraph 的StateGraph中每个节点可以读取和写入共享状态而 Reducer 定义了如何将多个节点对同一状态键的更新合并Reducer 的核心特征函数签名(Value, Value) - Value接收当前值和更新值返回合并后的新值注解定义通过Annotated[Type, reducer_function]为状态键指定 Reducer默认行为未指定 Reducer 的状态键使用覆盖策略Last-Write-Wins3.2.2. 如何定义Reducer3.2.2.1. 定义Reducer函数Reducer 本质上是一个二元合并函数用于定义当同一个字段产生多个更新值时LangGraph 应该如何将这些值合并为一个最终结果。函数签名(Value, Value) - Value示例代码如下# left 从最开始的位置合并到当前节点的值 # right 当前节点的值 # 返回值 合并后的值 def my_reducer(left: list[str], right: list[str]) - list[str]: return leftright # 1.node1 运行之后的值 left_state[start,node_1 执行完] right_state[node_2 执行完] mergedmy_reducer(left_state,right_state) print(merged)其中my_reducer用于处理list[str]类型的数据。它接收两个列表参数left当前已累计的状态值right本次待合并的新值。函数内部通过left right将两个列表合并并返回合并后的结果。因此该 Reducer 的作用是当某个状态字段存在多次列表更新时将这些列表内容追加合并而不是直接覆盖原值。运行结果如下[start, node_1 执行完, node_2 执行完]3.2.2.2. 将Reducer和状态字段关联在 LangGraph 中Reducer 通常通过 Python 的typing.Annotated与状态字段进行关联。Annotated[]是 Python 提供的一种类型注解扩展机制用于在原始类型之外附加额外的元数据信息。需要注意的是Annotated[]本身并不规定这些元数据的具体含义它只负责在类型注解中保留这些信息。严格来说Annotated的第一个参数是被注解的原始类型后续参数是附加的元数据。至于这些元数据表示什么、如何解析则由使用它的框架或工具自行决定。在 LangGraph 中框架利用这一机制将状态字段的类型和 Reducer 规则同时声明在字段定义中。其基本形式如下Annotated[Type, reducer_function]其中Type表示状态字段的数据类型reducer_function表示该字段对应的 Reducer 函数。示例代码如下from typing import TypedDict, Annotated class OverAllState(TypedDict): logs: Annotated[list[str], my_reducer] cur_id: str在上述代码中logs字段的类型是list[str]my_reducer是与logs字段关联的 Reducer 函数当多个节点同时更新logs字段时LangGraph 会使用my_reducer将多个列表合并cur_id字段没有指定 Reducer因此采用默认更新规则。3.2.2.3. 常用内置Reducer函数1. operator.addoperator.add是 Python 内置的加法操作函数底层由 C 实现它接收两个参数等价于a第一个参数b第二个参数代码如下from operator import add print(f{add(1,2) }) print(f{add([1,2], [3,4]) }) print(f{add([a,b], [c]) })输出如下add(1,2) 3 add([1,2], [3,4]) [1, 2, 3, 4] add([a,b], [c]) [a, b, c]2. langgraph.graph.message.add_messagesadd_messages是 LangGraph 中专用于合并消息列表的 Reducer 函数常用于维护对话历史类的状态字段。其函数签名如下add_messages是 LangGraph 中专用于合并消息列表的 Reducer 函数常用于维护对话历史类的状态字段。其函数签名如下def add_messages( left: Messages, right: Messages, *, format: Literal[langchain-openai] | None None, ) - Messages: ... return merged参数说明left状态中已有的消息列表right当前节点返回的消息更新值format可选参数用于指定返回消息的格式通常无需手动设置。left与right的类型均为Messages。Messages可以理解为 LangChain 消息对象的列表其中每个元素都是BaseMessage或其子类的实例常见子类包括HumanMessage用户的输入消息AIMessageAI 的回复消息SystemMessage系统提示消息ToolMessage工具调用的结果消息。add_messages处理的是对话消息序列而非普通的字符串列表。BaseMessage包含一个可选的id属性用于唯一标识一条消息。add_messages在合并left与right时不是简单地执行列表拼接而是依据消息的id进行合并若right中的某条消息的id在left中不存在则将该消息追加到结果列表末尾若right中的某条消息的id与left中已有消息的id相同则使用right中的新消息替换left中的旧消息。因此add_messages的作用可以概括为在保留历史消息的基础上追加新消息并允许通过相同的消息id覆盖已有消息。需要特别说明add_messages并非简单地对left与right求“并集”。更准确地说它是一个基于消息id的消息列表合并函数既支持追加新消息也支持更新已有消息。可以理解为merged left right但若right中存在与left相同id的消息则最终结果中不会出现重复消息而是用right中的消息覆盖left中对应的旧消息。示例代码如下from langchain_core.messages import HumanMessage, AIMessage, SystemMessage from langgraph.graph import add_messages left_state[ SystemMessage(content你是一个专业的翻译,id1), HumanMessage(content你好,id2), AIMessage(content你好我是AI助手,id3), ] right_state[ HumanMessage(content我是老王你是小王,id2), AIMessage(content好的我记住了,id3), HumanMessage(content你是谁,id4), AIMessage(content我是小王,id5), ] mergedadd_messages(left_state,right_state) print(merged)输出如下[SystemMessage(content你是一个专业的翻译, additional_kwargs{}, response_metadata{}, id1), HumanMessage(content我是老王你是小王, additional_kwargs{}, response_metadata{}, id2), AIMessage(content好的我记住了, additional_kwargs{}, response_metadata{}, id3, tool_calls[], invalid_tool_calls[]), HumanMessage(content你是谁, additional_kwargs{}, response_metadata{}, id4), AIMessage(content我是小王, additional_kwargs{}, response_metadata{}, id5, tool_calls[], invalid_tool_calls[])]3.2.3. 默认行为如果某个 State 字段没有显式定义ReducerLangGraph 会使用默认的状态更新行为后一次更新值会覆盖该字段原有的状态值。换句话说当节点返回的更新结果中包含某个字段时如果该字段没有配置 ReducerLangGraph 不会对新旧值进行追加、合并或累加而是直接使用本次返回的新值替换原来的旧值。示例代码如下from langgraph.graph import StateGraph, START, END from typing import TypedDict class OverAllState(TypedDict): logs: list[str] id: str def node_a(state: OverAllState): return { logs: [node_a], id: node_a } def node_b(state: OverAllState): return { logs: [node_b], id: node_b } builder StateGraph(state_schemaOverAllState) builder.add_node(node_a, node_a) builder.add_node(node_b, node_b) builder.add_edge(START, node_a) builder.add_edge(node_a, node_b) builder.add_edge(node_b, END) graph builder.compile() result graph.invoke({logs: [START], id: start}) print( * 30, - result -, * 30) print(result)输出如下 - result - {logs: [node_b], id: node_b}可以看到logs字段和id字段都没有定义Reducer因此节点返回的新值会覆盖初始状态中的旧值图运行结果中的状态值和最后一次更新保持一致。