17.1 AgentState 是 LangGraph 中所有节点间流转的公共数据容器

AgentState 是 LangGraph 中所有节点间流转的公共数据容器它的设计和更新机制是高效 Agent 与复杂工作流的关键。 1. AgentState 概念速记角色与作用在 LangGraph 的StateGraph中State特指AgentState类是贯穿所有节点的核心共享数据对象充当一块公共黑板。每个节点Node都接收当前 State 作为输入并返回对该 State 的更新。它为多步、多节点工作流提供了结构化的数据传递和共享方式。关系辨析AgentState是 Agent当前正在使用的可变容器存储着对话历史和中间计算结果。Checkpoint则是该 State 在某一时刻的不可变快照用于实现故障恢复与时间旅行。你可以把它们的关系理解为AgentState像一张草稿纸而Checkpoint是草稿纸的留影照片。 2. AgentState 的三种主要形态 形态一最简单的StateTypedDict这是最基础的自定义方式只定义一个结构而不指定归约器。比如你的AgentState中就只定义了需要包含哪些字段。如果不用Annotated指定归约器节点返回的字段更新默认会直接覆盖旧值。这种方式适用于只需固定值的简单场景。# 简单状态每次更新都会覆盖原有值fromtypingimportTypedDictclassSimpleState(TypedDict):counter:intuser_name:str 形态二预置的MessagesState(带add_messages)LangGraph 为你预置了MessagesState它是最常见、最基础的状态设计之一。它的核心是messages字段并配置了add_messages函数作为归约器Reducer。这个归约器确保节点返回的新消息会被追加到现有消息列表中而不是覆盖从而自然地构建对话历史。这对于构建聊天机器人或 Agent 是至关重要的。fromlanggraph.graphimportMessagesStateclassMyAgentState(MessagesState):pass# 也可以扩展其他字段但继承了 messages 的追加行为 形态三自定义 归约器用于复杂场景在你的实际业务中Agent 往往需要记住更多业务数据比如用户的首选项温度单位、搜索偏好、流程状态是否已搜索、重试次数、业务字段订单号、购物车等。为此你会继承TypedDict并通过Annotated对特定字段指定归约器Reducer。归约器定义了节点返回的状态更新如何与现有状态合并。这是官方推荐的最佳实践因为它能让你明确控制每个字段的更新逻辑。⚙️ 3. AgentState 状态更新机制详解这是掌握 AgentState 最核心的部分。当你的节点函数返回一个字典时LangGraph 不是简单地替换旧状态而是会根据每个字段的归约规则将节点的更新合并到现有状态中。(1). 默认归约器Default Reducer如果一个字段没有使用Annotated指定归约器LangGraph 会使用默认行为用新值完全覆盖旧值。要覆盖一个字段节点函数返回该字段的新值即可。# 状态定义fromtypingimportTypedDictclassMyState(TypedDict):user_input:str# 默认会覆盖step_count:int# 默认会覆盖# 节点函数会这样更新状态defsome_node(state:MyState)-dict:return{user_input:新的输入,step_count:state[step_count]1}(2). 预置归约器add_messages这是专门用于messages列表的归约器。它的逻辑是将节点返回的新消息追加到现有消息列表的末尾。fromtypingimportAnnotated,TypedDictfromlanggraph.graph.messageimportadd_messagesfromlangchain_core.messagesimportAnyMessageclassMyState(TypedDict):messages:Annotated[list[AnyMessage],add_messages]# 使用 add_messages 归约器other_field:str# 默认覆盖当一个节点return {messages: [AIMessage(content...)]}时该 AIMessage 会被追加而非替换原有列表。(3). 自定义归约器Annotated语法官方推荐使用Annotated[FieldType, ReducerFunction]语法来为字段绑定归约器这也是类型注解的最佳实践。fromtypingimportAnnotated,TypedDictdefmy_reducer(current:int|None,new:int)-int:一个自定义归约器返回当前值和新增值的和ifcurrentisNone:returnnewreturncurrentnewclassMyCustomReducerState(TypedDict):accumulated:Annotated[int,my_reducer](4). 自定义归约器旧版operator.add方式对于简单的列表追加也可以使用operator.add但可读性不如add_messages。当前主流最佳实践是add_messages。importoperatorfromtypingimportAnnotated,TypedDictclassMyCustomState(TypedDict):results:Annotated[list[str],operator.add]# 使用 operator.add 实现追加(5). 更底层的StateSchema类补充认知作为知识补充LangGraph 提供了更底层的StateSchema类来定义状态它提供了统一的 API可与 Zod 等标准 schema 库集成。但在日常使用中TypedDict Annotated的语法更常见。(6). 归约器函数签名总结归约器函数的标准签名是reducer(current: Value, new_value: UpdateValue) - Value你编写的逻辑需要基于这个模式来合并新旧数据。 4. 最佳实践与面试高频点为 List 字段使用归约器如果想让节点能追加数据到列表字段务必为其绑定归约器。否则默认是覆盖可能会丢失信息。为新增字段绑定归约器在自定义状态时根据业务逻辑决定归约策略是覆盖还是累积。合理拆分状态将业务数据、流程状态、对话消息分层存放结构更清晰。节点设计原则节点职责应聚焦于核心业务逻辑状态管理则交给框架和归约器处理这符合单一职责原则。类型与安全推荐使用TypedDict配合Annotated进行强类型约束这能提升代码健壮性和可维护性。文档与可视化为条件边的路由函数提供Literal类型注解能让代码自文档化也便于 LangGraph Studio 可视化。✨ 5. 完整示例代码以下是一个综合案例演示如何在一个 LangGraph 工作流中定义和使用包含多种归约器的自定义 State。importosfromtypingimportTypedDict,Annotated,Literalfromdotenvimportload_dotenvfromlangchain_openaiimportChatOpenAIfromlanggraph.graphimportStateGraph,END,STARTfromlanggraph.graph.messageimportadd_messagesfromlangchain_core.messagesimportHumanMessage,AIMessage# 1. 加载 .envload_dotenv()# 2. 创建模型实例llmChatOpenAI(modelqwen-plus,temperature0.7,api_keyos.getenv(DASHSCOPE_API_KEY),base_urlos.getenv(DASHSCOPE_BASE_URL,https://dashscope.aliyuncs.com/compatible-mode/v1),model_kwargs{extra_body:{enable_thinking:False}})# 3. 定义自定义 AgentState展示多种归约器classAgentState(TypedDict):# 使用 LangGraph 内置的 add_messages用于对话历史追加messages:Annotated[list,add_messages]# 默认覆盖记录当前用户的温度单位偏好temperature_unit:str# 自定义归约器记录一段对话中的模型调用次数model_call_count:Annotated[int,lambdax,y:xyifxisnotNoneelsey]# 4. 节点函数 1处理用户请求更新偏好defprocess_request(state:AgentState)-dict:user_msgs[msgformsginstate[messages]ifisinstance(msg,HumanMessage)]ifuser_msgs:contentuser_msgs[-1].contentif华氏度incontent:return{temperature_unit:华氏度}elif摄氏度incontent:return{temperature_unit:摄氏度}return{}# 5. 节点函数 2调用 LLM 并记录调用次数展示多种归约器的作用defcall_model(state:AgentState)-dict:current_countstate.get(model_call_count,0)# 这个返回体现了 add_messages 和 model_call_count 归约器的同时生效responsellm.invoke(state[messages])return{messages:[response],model_call_count:1}# 6. 路由函数决定后续流程defshould_continue(state:AgentState)-Literal[continue,end]:# 连续调用次数不超过 2 次才继续ifstate.get(model_call_count,0)2:returncontinuereturnend# 7. 构建图workflowStateGraph(AgentState)workflow.add_node(process_preference,process_request)workflow.add_node(process_model,call_model)workflow.set_entry_point(process_preference)workflow.add_edge(process_preference,process_model)# 添加条件循环workflow.add_conditional_edges(process_model,should_continue,{continue:process_model,# 循环回自身继续end:END})appworkflow.compile()# 8. 运行测试if__name____main__:# 模拟多轮对话initial_state{messages:[HumanMessage(content帮我查一下上海现在的温度我用摄氏度。)],temperature_unit:摄氏度,model_call_count:0,}print(初始状态:,initial_state)final_stateapp.invoke(initial_state)print(\n 最终状态 )print(fTemperature Unit:{final_state[temperature_unit]})print(fModel Call Count:{final_state[model_call_count]})print(fMessages:{final_state[messages]}) 总结AgentState是 LangGraph 工作流的记忆载体其通过归约器Reducer机制实现了灵活可控的状态更新。我们可以使用TypedDict定义状态结构用Annotated为特定字段绑定归约器。在编写 Agent 时合理设计和运用AgentState包括消息与业务字段是保证任务连贯性、正确性和可调试性的基石。