第二十五节的主题是State和Reducer函数这一节详细讲解了 LangGraph 状态管理的核心机制。 第25课核心知识点梳理1. State状态—— 图的“公共黑板”State 是贯穿 LangGraph 整个工作流的核心数据结构负责存储和传递任务执行过程中的关键信息它反映了应用在某一时刻的完整快照如用户输入、中间结果、工具调用日志等。State 的核心价值在于消除数据孤岛通过共享状态池打破节点间的数据壁垒保持执行上下文完整记录智能体从启动到终止的全生命周期数据保障确定性执行通过状态快照实现执行流程的可回溯与可重现2. Reducer归约器—— 决定“如何合并”Reducer 是 LangGraph 中用于定义状态字段更新逻辑的核心机制通常与Annotated类型注解结合使用。它的核心思想是将字段的更新逻辑从业务代码中解耦通过类型注解绑定到字段上。Reducer 函数的标准签名为reducer(current: Value, new: Value) - Value接收当前状态值和新值返回合并后的结果。⚙️ State 更新机制不可变性与增量更新节点读节点接收完整的当前 State节点计算节点执行业务逻辑确定需要修改哪些字段节点返回节点只返回一个包含变更字段的字典部分更新而不是整个 State系统合并LangGraph 自动将返回的更新浅合并到全局 State 中defnode(state:AgentState)-dict:# 只返回需要更新的字段LangGraph 会自动合并return{temperature_unit:new_unit} Reducer 的核心价值LangGraph默认的状态合并策略是覆盖Overwrite“而不是追加Append”。后一个节点返回的值会直接替换前一个节点写入的值——这是 LangGraph 中最常见、最隐蔽的坑之一。因此需要为不同字段指定不同的合并策略messages、logs等需要保留历史信息的字段 → 使用 Reducer 进行追加合并current_step、status等只关心当前状态的字段 → 使用默认覆盖即可 Reducer 的核心用法Annotated 语法Reducer 通过Annotated[Type, ReducerFunc]语法与字段绑定Reducer 逻辑是声明式地绑定到字段上的fromtypingimportAnnotated,TypedDictfromlanggraph.graphimportadd_messagesfromoperatorimportaddclassMyState(TypedDict):# 追加模式新消息自动追加到列表末尾messages:Annotated[list,add_messages]# 累加模式新值自动累加到当前值counter:Annotated[int,add]# 无 Reducer默认覆盖模式current_step:str 内置 Reducer 与自定义 Reducer1.add_messages—— 消息追加归约器专为消息字段设计处理对话历史的合并fromtypingimportAnnotated,TypedDictfromlanggraph.graphimportadd_messagesfromlangchain_core.messagesimportAnyMessageclassChatState(TypedDict):messages:Annotated[list[AnyMessage],add_messages]核心行为如果新消息有id替换同 ID 的旧消息如果无 ID 或 ID 不存在追加新消息保留消息顺序2.operator.add—— 数值累加与列表扩展数值累加场景如对话轮次、分数统计fromoperatorimportaddfromtypingimportAnnotated,TypedDictclassCounterState(TypedDict):count:Annotated[int,add]# 整数累加total:Annotated[float,add]# 浮点数累加列表扩展场景fromoperatorimportaddfromtypingimportAnnotated,List,TypedDictclassSearchState(TypedDict):results:Annotated[List[str],add]# 将新元素追加到列表注意由于operator.add也支持列表的拼接这在并行分发Fan-Out场景中非常有用。多个节点并发返回结果时operator.add会将所有结果安全地合并成一个完整列表。3.operator.extend—— 列表扩展标准化版fromoperatorimportextendfromtypingimportAnnotated,List,TypedDictclassItemState(TypedDict):items:Annotated[List[str],extend]# list.extend() 行为与operator.add类似用于批量扩展列表。4. 自定义 Reducer —— 实现复杂业务逻辑自定义 Reducer 通常配合Annotated使用所有节点返回的对应字段更新都会经过这个函数处理fromtypingimportAnnotated,TypedDictdefcustom_reducer(current:dict|None,new:dict)-dict:自定义归约器深度合并字典ifcurrentisNone:returnnew result{**current}forkey,valueinnew.items():ifisinstance(value,dict)andkeyinresultandisinstance(result[key],dict):result[key]{**result[key],**value}# 字典递归合并else:result[key]valuereturnresultclassComplexState(TypedDict):metadata:Annotated[dict,custom_reducer] 完整星系案例银河探索 Agent 状态管理我们将构建一个能够记住用户偏好的星系探索助手利用 State 和 Reducer 让 Agent 具备真正的记忆。案例目标Agent 能够记住用户的名字和温度单位偏好无需重复解释构建一个能理解偏好并给出个性化响应的多轮对话 Agent演示三种 Reducer 策略的实际应用场景完整代码importosimportsysimportasynciofromtypingimportTypedDict,Annotated,Literalfromdotenvimportload_dotenvfromlangchain_openaiimportChatOpenAIfromlanggraph.graphimportStateGraph,END,add_messagesfromlanggraph.checkpoint.memoryimportMemorySaverfromlangchain_core.messagesimportHumanMessage,AIMessage,BaseMessagefromoperatorimportadd load_dotenv()ifsys.platformwin32:asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())llmChatOpenAI(modelqwen-plus,temperature0.7,api_keyos.getenv(DASHSCOPE_API_KEY),base_urlos.getenv(DASHSCOPE_BASE_URL),)# 1. 定义 State展示三种 Reducer 策略classGalaxyState(TypedDict):# ① add_messages对话历史自动追加messages:Annotated[list[BaseMessage],add_messages]# ② operator.add整数累加统计对话轮次turn_count:Annotated[int,add]# ③ 自定义 operator.add列表追加收集已访问星系与 add_messages 效果类似visited_galaxies:Annotated[list[str],add]# ④ 无 reducer默认覆盖只保留最新状态current_galaxy:struser_name:strtemperature_unit:str# celsius 或 fahrenheitmission_step:str# 2. 节点定义 defanalyze_input(state:GalaxyState):分析用户输入提取用户偏好和意图last_msgstate[messages][-1].content updates{}if我叫inlast_msg:namelast_msg.split(我叫)[-1][:2]updates[user_name]nameprint(f 记录用户姓名:{name})if摄氏度inlast_msg:updates[temperature_unit]celsiusprint(️ 记录温度单位偏好: 摄氏度)elif华氏度inlast_msg:updates[temperature_unit]fahrenheitprint(️ 记录温度单位偏好: 华氏度)if探索inlast_msg:forgalaxyin[仙女座,猎户座,银河系]:ifgalaxyinlast_msg:updates[current_galaxy]galaxy# visited_galaxies 使用 add reducer会自动追加而非覆盖updates[visited_galaxies][galaxy]print(f 记录访问星系:{galaxy})breakreturnupdatesdefgenerate_response(state:GalaxyState):根据用户偏好生成个性化回复user_namestate.get(user_name,探险者)temp_unitstate.get(temperature_unit,celsius)galaxystate.get(current_galaxy,)# 根据偏好的温度单位生成不同内容ifgalaxy仙女座:temp_celsius-110temp_fahrenheit-166description仙女座星系是我们最近的邻居包含约一万亿颗恒星elifgalaxy猎户座:temp_celsius-218temp_fahrenheit-360description猎户座是夜空中最著名的星座之一包含参宿四和参宿七else:temp_celsius数据收集中temp_fahrenheit数据收集中description正在获取星系信息temp_infof平均温度{temp_celsius}°C /{temp_fahrenheit}°Fiftemp_unitcelsius:temp_displayf️{temp_celsius}°Celse:temp_displayf️{temp_fahrenheit}°Fresponse_textf你好{user_name}{description}。{temp_display}return{messages:[AIMessage(contentresponse_text)],turn_count:1}defroute_decision(state:GalaxyState)-Literal[generate,__end__]:条件路由根据当前星系决定是否生成回复ifstate.get(current_galaxy):returngeneratereturn__end__# 3. 构建图挂载 CheckpointercheckpointerMemorySaver()builderStateGraph(GalaxyState)builder.add_node(analyze,analyze_input)builder.add_node(generate,generate_response)builder.set_entry_point(analyze)builder.add_conditional_edges(analyze,route_decision,{generate:generate,__end__:END})builder.add_edge(generate,END)graphbuilder.compile(checkpointercheckpointer)# 4. 交互式对话测试 asyncdefmain():print( 欢迎来到银河探索助手)print(*60)thread_idinput(请输入会话 ID: ).strip()orgalaxy_001config{configurable:{thread_id:thread_id}}whileTrue:user_inputinput(\n请输入您的消息: )ifuser_input.lower()in[quit,exit]:break# 使用相同的 thread_id 调用Checkpointer 自动保存状态resultgraph.invoke({messages:[HumanMessage(contentuser_input)]},configconfig)print(f助手:{result[messages][-1].content})print(f 当前状态 | 对话轮次:{result.get(turn_count,0)}| f已访问星系:{result.get(visited_galaxies,[])})if__name____main__:asyncio.run(main()) 案例执行流程图有 current_galaxy无目标State 公共黑板messages: 对话历史add_messages 追加turn_count: 对话轮次operator.add 累加visited_galaxies: 访问记录operator.add 列表追加current_galaxy: 当前星系默认覆盖user_name: 用户名默认覆盖temperature_unit: 温度单位默认覆盖用户输入analyze_inputroute_decisiongenerate_responseENDCheckpointer 自动保存 StateEND 高频面试题基础概念题面试问题核心回答要点State 在 LangGraph 中扮演什么角色State 是节点间的公共黑板作为共享数据容器实现多节点通信。核心价值是消除数据孤岛并保证信息一致性。Reducer 函数是什么决定 State 中某个字段的更新规则接收(current, new) - merged并使用Annotated绑定到字段。LangGraph 的 State 更新机制是怎样的① 每个节点接收完整的当前 State② 节点返回一个包含变更字段的部分更新③ LangGraph 自动调用 Reducer 将更新合并到全局 State。进阶题面试问题核心回答要点为什么有了operator.add还要单独提供add_messagesadd_messages专为消息序列设计支持按id智能替换已有消息而operator.add只简单拼接列表。什么情况下需要自定义 Reducer当业务逻辑超出内置 Reducer 能力时如深度合并字典、带条件的列表去重、累加器验证等。INVALID_CONCURRENT_GRAPH_UPDATE错误如何解决图并行执行多节点时若同一字段未配置 Reducer会出现此错误。为相关字段配置合适的 Reducer如add_messages或operator.add即可解决。面试题如何实现计数器 ReducerfromtypingimportAnnotated,TypedDictfromoperatorimportaddclassCounterState(TypedDict):count:Annotated[int,add]# 整数累加total:Annotated[float,add]# 浮点数累加# 节点函数中这样使用defincrement(state:CounterState):return{count:1}# 返回 1累加到 count 字段工程实践题面试问题核心回答要点如何设计生产环境的 State Schema① 为需要保留历史的字段配置 Reducer如messages② 元数据和临时状态使用默认覆盖③ 敏感信息如 API Key应放在configurable中④ 保持 Reducer 幂等。 总结State Reducer LangGraph 的状态管理心脏State是图的数据中心定义 Agent 能记住什么Reducer是数据管家定义如何记住多轮对话的自动追加能力来自add_messages而不是 LangGraph 的默认行为Checkpointer 的持久化是纵向的按时间线保存Reducer 的归约是横向的合并多节点并行更新两者正交互补