LangGraph状态模式定义了整个图的结构它告诉每个节点可以访问那些数据可以更新的字段。我们这里用TypedDict来定义状态模TypedDict是Python的标准库提供轻量级类型检查。基本定义fromtyping_extensionsimportTypedDictclassState(TypedDict):text:strcount:int这个State定义了两个字段text和count所有节点都会接收这个状态对象并且可以返回一个字段更新里面字段。2.1、Reducer 函数状态更新Reducer是LangGraph状态管理的核心概念它决定了多个节点更新同一个状态字段时候这些字段是替换、更新、追加等如何合并。默认行为覆盖更新默认情况下状态字段采用覆盖更新。后面的节点替换前面的节点内容。fromtyping_extensionsimportTypedDictclassState(TypedDict):foo:intbar:list[str]节点1{“foo”:1 , “bar”:[你好”],“demo”:“demo1”}节点2{“foo”:2 , “bar”:[你好2”],“test”:“测试”}最终状态 **{“foo”:2 , “bar”:[你好2”],“test”:“测试”,“demo”:“demo1”}**第二个节点就会覆盖相同字段的值。追加效果可以通过Annotated 指定 Reducer来定义更新行为可以使用 typing.Annotated 来指定 Reducer 函数**add_messages**将两个列表连接起来实现了追加效果。这种模式在处理消息列表、日志记录等场景时特别有用。fromtypingimportAnnotated,Sequencefromlangchain_core.messagesimportBaseMessage,AIMessageclassTestState2(TypedDict):message:Annotated[Sequence[BaseMessage],add_messages]节点1{ “message”“123” }节点2{ “message”“456” }最终状态[{“message”:123”},{“message”:456”}]错误案例下面是一个同时执行的节点因为征用同一个字段所以会出现错误fromtyping_extensionsimportTypedDictclassState(TypedDict):some_key:str# 没有 Reducerdefnode_a(state:State):return{some_key:value_a}defnode_b(state:State):return{some_key:value_b}builderStateGraph(State)builder.add_node(node_a)builder.add_node(node_b)builder.add_edge(START,node_a)builder.add_edge(START,node_b)如果 node_a 和 node_b 同时执行并返回 some_key 的更新LangGraph 会抛出 INVALID_CONCURRENT_GRAPH_UPDATE 错误。因为它不知道应该保留哪个值。正确案例fromtypingimportAnnotatedfromtyping_extensionsimportTypedDictimportoperatorclassState(TypedDict):some_key:Annotated[list,operator.add]defnode_a(state:State):return{some_key:[value_a]}defnode_b(state:State):return{some_key:[value_b]}2.2、自定义Reducer函数除了内置的operator还可以自定义Reducer函数、Reducer接收两个参数当前值和更新值返回合并后的值。如下defunique_add(current:list,update:list):combinedcurrentupdate seenset()result[]foritemincombined:ifitemnotinseen:seen.add(item)result.append(item)returnresultclassState(TypedDict):items:Annotated[list,unique_add]2.3、MessagesState 消息管理在LLM应用中消息历史是最常见的状态类型。LangGraph为此提供了专门的处理机制。主要解决以下问题追加新消息而不是覆盖支持消息 ID以便更新已存在的消息接受多种消息格式LangChain Message 对象、OpenAI 格式等自动反序列化2.3.1、add_messagesLangGraph提供内置的add_messages他解决了上面的所有问题fromlangchain_core.messagesimportAnyMessagefromlanggraph.graph.messageimportadd_messagesfromtypingimportAnnotatedfromtyping_extensionsimportTypedDictclassState(TypedDict):messages:Annotated[list[AnyMessage],add_messages]使用add_messages后可以多种格式传递消息# 使用 LangChain Message 对象{messages:[HumanMessage(contenthello)]}# 使用字典格式{messages:[{type:human,content:hello}]}# OpenAI 格式{messages:[{role:user,content:hello}]}add_messages 会自动处理这些格式统一转换为 LangChain Message 对象。这意味着在节点内部可以始终使用一致的 API 访问消息内容defprocess_messages(state:State):last_messagestate[messages][-1]contentlast_message.content2.3.2、MessagesState 预定义状态LangGraph 提供了预定义的 MessagesStateMessagesState 已经包含了 messages 字段和 add_messages Reducer可以直接继承并添加其他字段。这让代码更加简洁意图也更清晰。fromlanggraph.graphimportMessagesStateclassState(MessagesState):documents:list[str]user_info:dict