1. 从“健忘”到“博闻强记”为什么 LangGraph 的状态管理是核心如果你刚开始接触 LangGraph可能会觉得它和 LangChain 有点像都是用来构建 LLM 应用的工作流框架。但当你真正上手尤其是尝试构建一个多轮对话、需要记住上下文的应用时你可能会遇到一个尴尬的局面你的 Agent 像个金鱼说完上句就忘了下句。你明明在代码里传了历史消息但 Agent 在处理新问题时似乎总是“失忆”了。这背后的关键就在于 LangGraph 的状态State管理机制。LangGraph 的核心设计哲学是将应用建模为一个有状态Stateful的图Graph。这里的“图”指的是由节点Node和边Edge构成的计算流而“状态”则是贯穿整个计算流、可以被所有节点读写和更新的共享内存。这和我们熟悉的 LangChain 的链Chain或代理Agent有本质区别。在 LangChain 中信息通常是通过一个固定的输入输出管道传递状态管理相对隐式或者需要开发者自己通过回调、内存等机制来维护。而 LangGraph 把状态管理提到了明面上成为一等公民。为什么这很重要想象一下你要构建一个客服机器人。用户第一轮问“我的订单 12345 状态如何” 系统查询后回答“已发货。” 第二轮用户又问“预计什么时候到” 一个合格的机器人必须记得“订单 12345”和“已发货”这两个关键信息才能给出准确的物流预估。在 LangGraph 中这个“记忆”就存储在状态对象里。每个处理节点比如“查询订单”、“查询物流”都可以读取状态中的历史信息并将自己的处理结果写回状态供后续节点使用。所以理解并掌握 LangGraph 的状态是解锁其强大能力——构建复杂、有记忆、可回溯的智能工作流——的第一把钥匙。默认的状态定义可能很简单但当你需要处理复杂的数据结构比如同时维护对话历史、工具调用记录、用户画像、会话元数据时自定义状态就变得至关重要。而归约器Reducer则是你精细化控制状态如何更新的“手术刀”。2. 解剖 LangGraph 的状态对象TypedDict 与 Annotated 的共舞在 LangGraph 中状态不是一个黑盒子。它被明确定义为一个 Python 的TypedDict。这是一种类型提示Type Hints用来规定状态字典里每个字段的名字和类型。这样做的好处是代码清晰、易于维护并且能获得 IDE 的自动补全和类型检查支持。2.1 基础状态定义一个简单的聊天记录器让我们从一个最简单的例子开始定义一个只记录聊天消息的状态。from typing import TypedDict, List from langgraph.graph import StateGraph, END from langchain_core.messages import BaseMessage class BasicState(TypedDict): # 定义一个字段 messages其类型是 BaseMessage 的列表 messages: List[BaseMessage] # 初始化图 graph_builder StateGraph(BasicState)这里我们创建了一个BasicState类型它只有一个字段messages用来存放所有的聊天消息。BaseMessage是 LangChain 中表示消息的基类可以是HumanMessage用户输入、AIMessageAI 回复、ToolMessage工具调用结果等。接下来我们需要定义节点函数来操作这个状态。在 LangGraph 中节点函数接收两个参数当前状态state和一个可选的配置config。它必须返回一个字典这个字典的键对应状态字段名值是对该字段的更新操作。def call_model(state: BasicState): 模拟调用大模型并生成回复 # 从状态中读取历史消息 history state[“messages”] # 这里简单模拟总是回复“这是AI的回复” from langchain_core.messages import AIMessage new_message AIMessage(content“这是AI的回复”) # 返回更新指令将新消息追加到 messages 列表 return {“messages”: [new_message]} def human_input(state: BasicState): 模拟用户输入 user_input input(“用户: “) from langchain_core.messages import HumanMessage new_message HumanMessage(contentuser_input) return {“messages”: [new_message]} # 添加节点 graph_builder.add_node(“call_model”, call_model) graph_builder.add_node(“human_input”, human_input) # 设置入口点和边 graph_builder.set_entry_point(“human_input”) graph_builder.add_edge(“human_input”, “call_model”) graph_builder.add_edge(“call_model”, END) # 编译图 graph graph_builder.compile()运行这个图你会发现一个问题每次调用节点返回{“messages”: [new_message]}它都会用一个新的单元素列表[new_message]替换掉状态中原来的整个messages列表而不是追加。我们的对话历史永远只有最后一条消息。这显然不是我们想要的。2.2 引入归约器控制状态更新的逻辑这就是Annotated和归约器登场的时候。Annotated是 Python 的一个类型注解允许我们为类型附加额外的“元数据”。在 LangGraph 中我们用Annotated[FieldType, reducer_function]来声明一个状态字段并指定如何更新它。这里的reducer_function就是归约器。归约器是一个函数它接收两个参数当前字段的旧值current和节点返回的更新值update然后返回合并后的新值。最常见的归约器是add_messages它专门用于处理List[BaseMessage]的追加。让我们修正上面的状态定义from typing import TypedDict, List, Annotated from langgraph.graph import StateGraph, END from langgraph.graph import add_messages # 导入内置的归约器 from langchain_core.messages import BaseMessage class AdvancedState(TypedDict): # 使用 Annotated 注解字段类型是 List[BaseMessage]更新逻辑由 add_messages 函数控制 messages: Annotated[List[BaseMessage], add_messages] # 重新构建图 graph_builder StateGraph(AdvancedState) # ... 节点函数保持不变 ... graph_builder.add_node(“call_model”, call_model) graph_builder.add_node(“human_input”, human_input) graph_builder.set_entry_point(“human_input”) graph_builder.add_edge(“human_input”, “call_model”) graph_builder.add_edge(“call_model”, END) graph graph_builder.compile()现在当我们运行这个图时节点函数返回{“messages”: [new_message]}LangGraph 不会直接用[new_message]替换旧列表而是会调用add_messages(old_messages_list, [new_message])。这个内置归约器的逻辑很简单将更新列表中的所有消息追加到旧列表的末尾。这样messages字段就变成了一个不断增长的对话历史记录。注意add_messages是 LangGraph 为消息列表提供的“特供”归约器它处理了一些底层细节比如消息的合并优化。对于普通列表你可以使用另一个内置归约器add_messages不适用应该用更通用的方法。3. 超越消息列表自定义归约器处理复杂状态现实中的应用状态远比一个消息列表复杂。你可能需要跟踪会话 ID、用户偏好、已执行的操作列表、临时计算结果等等。add_messages只解决了消息追加的问题对于其他类型的字段你需要自己编写归约器。3.1 场景一个支持“撤销”功能的计算器状态假设我们正在构建一个智能计算助手它不仅能计算还能记住计算历史并允许用户撤销上一步。我们的状态可能需要包含current_value: 当前的计算结果浮点数。history: 计算历史记录是一个列表每一项是一个字典包含操作和操作前的值。messages: 和模型的对话记录。首先我们定义状态类型和自定义归约器。from typing import TypedDict, List, Dict, Any, Annotated from langgraph.graph import StateGraph, END from langgraph.graph import add_messages class CalculatorState(TypedDict): # 当前值更新规则直接用新值覆盖旧值 current_value: float # 历史记录更新规则将新的历史条目追加到列表末尾 history: Annotated[List[Dict[str, Any]], append_history] # 对话消息使用内置归约器 messages: Annotated[List[BaseMessage], add_messages] # 自定义归约器用于 history 字段 def append_history(current: List[Dict[str, Any]], update: List[Dict[str, Any]]): 将 update 列表中的元素追加到 current 列表末尾。 # 重要归约器必须返回一个新的对象而不是修改原对象。 # 对于列表通常返回 current update if update is None: return current return current update # 自定义归约器用于 current_value 字段覆盖式更新 def replace_value(current: float, update: float): 直接用 update 值替换 current 值。 # 虽然覆盖逻辑简单但显式定义归约器能让意图更清晰也便于未来扩展比如添加验证逻辑。 return update现在我们需要更新状态定义为current_value也加上归约器注解。但这里有个问题Annotated只能附加一个元数据。我们需要一个能表达“覆盖”逻辑的归约器。实际上对于简单的覆盖LangGraph 有一个约定如果节点返回的更新字典中某个字段的值不是None且该字段没有用Annotated指定归约器则默认执行覆盖操作。但为了代码清晰和一致性尤其是未来可能增加逻辑比如更新前检查值是否有效定义一个显式的归约器是更好的实践。我们可以定义一个通用的replace归约器或者直接为float类型定义一个。这里我们采用更清晰的方式修改状态定义class CalculatorState(TypedDict): # 使用自定义归约器 replace_value current_value: Annotated[float, replace_value] history: Annotated[List[Dict[str, Any]], append_history] messages: Annotated[List[BaseMessage], add_messages]3.2 实现计算器节点接下来我们实现两个节点一个执行计算一个处理撤销。def calculate_node(state: CalculatorState): 解析用户消息中的计算指令并更新状态 # 1. 获取最新的用户消息 last_message state[“messages”][-1] if last_message.type ! “human”: return {} # 如果不是用户消息不做任何更新 user_input last_message.content # 2. 简单解析指令例如 “add 5” 或 “multiply 3” parts user_input.strip().lower().split() if len(parts) ! 2: # 如果指令格式不对让AI节点回复错误信息 return {“messages”: [AIMessage(content“指令格式错误请使用‘操作 数字’如‘add 5’。”)]} op, num_str parts[0], parts[1] try: num float(num_str) except ValueError: return {“messages”: [AIMessage(contentf“无法将‘{num_str}’解析为数字。”)]} old_value state.get(“current_value”, 0.0) # 默认从0开始 new_value old_value op_symbol “” if op “add”: new_value old_value num op_symbol “” elif op “subtract”: new_value old_value - num op_symbol “-” elif op “multiply”: new_value old_value * num op_symbol “*” elif op “divide”: if num 0: return {“messages”: [AIMessage(content“除数不能为零。”)]} new_value old_value / num op_symbol “/” else: return {“messages”: [AIMessage(contentf“不支持的操作: {op}”)]} # 3. 准备状态更新 # 更新当前值 value_update {“current_value”: new_value} # 更新历史记录记录操作前的值和执行的操作 history_entry {“old_value”: old_value, “operation”: f“{op_symbol}{num}“, “new_value”: new_value} history_update {“history”: [history_entry]} # 生成AI回复消息 ai_reply AIMessage(contentf“计算: {old_value} {op_symbol} {num} {new_value}。当前结果为: {new_value}”) message_update {“messages”: [ai_reply]} # 4. 合并所有更新并返回 # LangGraph 会自动将不同字段的更新分发给对应的归约器处理 return {**value_update, **history_update, **message_update} def undo_node(state: CalculatorState): 撤销上一步计算 history state.get(“history”, []) if not history: # 没有历史可撤销 return {“messages”: [AIMessage(content“没有可撤销的操作。”)]} # 取出最后一条历史记录 last_op history[-1] # 当前值回退到操作前的值 old_value last_op[“old_value”] # 从历史记录中移除最后一条 remaining_history history[:-1] # 准备更新 value_update {“current_value”: old_value} history_update {“history”: remaining_history} # 注意这里我们传入了新的完整历史列表 ai_reply AIMessage(contentf“已撤销操作 ‘{last_op[‘operation’]}’。当前结果回退到: {old_value}”) message_update {“messages”: [ai_reply]} return {**value_update, **history_update, **message_update}这里有一个关键点需要注意在undo_node中我们更新history字段时返回的是{“history”: remaining_history}。append_history归约器会把remaining_history这个列表追加到旧的history列表后面这显然不对。我们希望的是替换整个历史列表。这暴露了append_history归约器的一个局限性它只能处理追加操作。对于替换操作我们需要另一种方式。在 LangGraph 中如果你需要替换整个列表或其他可变结构一个常见的模式是在归约器内部判断更新值的类型或内容或者使用一个更智能的归约器。让我们修改append_history使其能处理“替换”语义。我们可以约定如果update参数不是一个列表或者我们通过某种标志比如update是一个字典且包含特定键则执行替换操作。但更简单清晰的做法是为替换操作单独定义一个归约器并在状态定义中根据字段的更新需求来选择使用哪个。然而一个字段只能绑定一个归约器。因此更实用的方案是让一个归约器能处理多种更新意图。我们可以通过检查update的值来实现def smart_history_reducer(current: List[Dict[str, Any]], update: List[Dict[str, Any]]): 智能历史记录归约器。 如果 update 是列表则追加。 如果 update 是 None则返回 current无操作。 注无法通过此归约器实现‘替换’因为替换需要传入新列表这与追加传入的类型相同 if update is None: return current # 默认行为追加 return current update对于“替换”需求在 LangGraph 的范式下更好的做法是不通过归约器实现而是通过节点的返回值直接覆盖。还记得之前的约定吗如果一个字段没有用Annotated指定归约器节点返回的值会直接覆盖旧值。所以我们可以将history字段的归约器注解去掉当需要追加时节点返回{“history”: [new_entry]}系统会直接覆盖这不对。当需要替换时节点返回{“history”: new_list}这可以实现替换。但这又带来了新问题我们如何区分“追加一条”和“替换整个列表”节点在返回时必须知道完整的、新的历史列表是什么这增加了节点的负担破坏了“节点只关心自己产生的增量”这一原则。这是一个经典的权衡。在实践中对于history这种结构我推荐两种方案方案A使用两个独立的字段。history_entriesAnnotated[List, append]用于追加新条目history_snapshot无归约器或一个替换归约器用于存储完整的、可能被修改过的列表快照。undo_node可以修改history_snapshot并清空或调整history_entries。这稍显复杂。方案B让节点负责构建完整的新状态。即calculate_node读取旧的history构建一个包含新条目的全新列表new_history old_history [new_entry]然后返回{“history”: new_history}。undo_node同样构建一个移除了最后一项的新列表。此时history字段可以不使用归约器或使用一个简单的替换归约器。这种方案下节点逻辑更重但状态更新语义非常简单直接——总是替换。对于初学者和大多数场景方案B更简单直观。我们调整一下class CalculatorState(TypedDict): current_value: Annotated[float, replace_value] # 覆盖更新 history: List[Dict[str, Any]] # 不使用归约器总是替换 messages: Annotated[List[BaseMessage], add_messages] def calculate_node(state: CalculatorState): # ... 前面的解析和计算逻辑不变 ... old_history state.get(“history”, []) new_history old_history [history_entry] # 构建全新的历史列表 history_update {“history”: new_history} # 返回全新列表用于替换 # ... 其他更新 ... return {**value_update, **history_update, **message_update} def undo_node(state: CalculatorState): history state.get(“history”, []) if not history: return {“messages”: [AIMessage(content“没有可撤销的操作。”)]} last_op history[-1] old_value last_op[“old_value”] new_history history[:-1] # 构建移除了最后一项的新列表 history_update {“history”: new_history} # 返回新列表用于替换 # ... 其他更新 ... return {**value_update, **history_update, **message_update}这个例子清晰地展示了如何根据业务逻辑选择状态更新策略。对于简单的追加使用归约器如add_messages非常方便。对于需要复杂逻辑更新如条件替换、合并、去重的字段可能需要自定义归约器。而对于“整个替换”这种操作有时不使用归约器让节点返回完整的新值反而是更清晰的选择。4. 实战中的模式、陷阱与最佳实践通过前面的例子我们已经掌握了自定义状态和归约器的基本用法。但在实际项目中你会遇到更复杂的场景和一些容易踩的坑。下面分享一些实战经验和模式。4.1 模式一分层状态与模块化当应用变得复杂时状态可能包含多个逻辑模块。例如一个客服 Agent 的状态可能包含conversation: 对话历史 (Annotated[List[BaseMessage], add_messages])user_profile: 用户信息 (Dict[str, Any]使用自定义合并归约器)session_data: 临时会话数据如当前查询的订单号 (str 直接覆盖)tool_calls: 本回合已调用的工具列表 (Annotated[List, append_tool_call])一个好的实践是使用TypedDict的继承或组合来组织状态使其更清晰。from typing import TypedDict, List, Dict, Any, Annotated from pydantic import BaseModel class UserProfile(BaseModel): user_id: str tier: str “standard” preferences: Dict[str, Any] {} class ConversationState(TypedDict): messages: Annotated[List[BaseMessage], add_messages] pending_tool_calls: List[Dict] # 等待执行的工具调用 class AppState(TypedDict): # 使用嵌套的 TypedDict 或 Pydantic 模型来组织 conversation: ConversationState user: UserProfile # Pydantic 模型需要自定义归约器处理 session_id: str metadata: Dict[str, Any]对于UserProfile这样的 Pydantic 模型你需要一个自定义归约器来合并更新。通常可以使用 Pydantic 模型的dict()和update方法或者更精细地处理。def update_user_profile(current: UserProfile, update: Dict[str, Any]): 更新用户档案。update 是一个字典包含要更新的字段。 if not update: return current # 创建当前配置的副本并用 update 字典更新它 # 注意这里为了简单直接使用了字典更新。对于复杂嵌套可能需要递归合并。 updated_data current.dict() updated_data.update(update) # 返回一个新的 Pydantic 模型实例 return UserProfile(**updated_data) # 在状态定义中使用 class AppState(TypedDict): user: Annotated[UserProfile, update_user_profile] # ... 其他字段 ...4.2 陷阱一归约器的副作用与不可变性归约器必须是纯函数。它不应该修改传入的current或update参数也不应该产生其他副作用如读写文件、网络请求。它应该根据输入计算并返回一个新的值。错误的例子def bad_reducer(current: List, update: List): current.extend(update) # 错误直接修改了 current return current正确的例子def good_reducer(current: List, update: List): # 返回一个新的列表对象 return current update这一点至关重要因为 LangGraph 可能出于性能或调试原因对状态进行缓存或快照。修改输入参数会导致不可预测的行为。4.3 陷阱二None值的处理在归约器中update参数有可能为None如果节点没有返回该字段的更新。你的归约器应该能优雅地处理这种情况通常的做法是直接返回current。def safe_reducer(current: Any, update: Any): if update is None: return current # ... 正常的合并逻辑 ...4.4 最佳实践为复杂归约器编写单元测试自定义归约器是业务逻辑的一部分而且其正确性至关重要。务必为它们编写单元测试覆盖各种边界情况update为Nonecurrent为空值如[],{},None正常合并场景冲突处理场景如字典键冲突def test_append_history(): reducer append_history assert reducer([], [{“op”: “add”}]) [{“op”: “add”}] assert reducer([{“op”: “add”}], []) [{“op”: “add”}] assert reducer([{“op”: “add”}], [{“op”: “subtract”}]) [{“op”: “add”}, {“op”: “subtract”}] assert reducer([{“op”: “add”}], None) [{“op”: “add”}] # 处理 None4.5 模式二使用“标记”字段控制流程状态不仅可以存储数据还可以存储控制流程的标记。例如一个字段need_human_input: bool可以决定图的下一个节点是调用模型还是等待用户输入。边Edge的条件判断可以基于这些状态字段。class StateWithFlag(TypedDict): messages: Annotated[List[BaseMessage], add_messages] need_human_input: bool def some_node(state: StateWithFlag): # ... 处理逻辑 ... if some_condition: return {“messages”: [AIMessage(content“Done”)], “need_human_input”: False} else: return {“messages”: [AIMessage(content“I need more info.”)], “need_human_input”: True} # 在图中定义条件边 graph_builder.add_conditional_edges( “some_node”, # 这个函数根据状态决定下一个节点 lambda state: “human_input_node” if state[“need_human_input”] else “next_auto_node”, )这种模式使得工作流能够根据运行时状态动态改变路径极大地增强了灵活性。掌握自定义状态和归约器你就掌握了 LangGraph 记忆系统的钥匙。从简单的消息列表到包含用户画像、会话上下文、操作历史的复杂状态你可以通过精心设计的TypedDict和归约器来精确描述和控制应用的数据流。记住状态设计是 LangGraph 应用架构的核心值得在项目初期投入时间仔细考量。一个好的状态设计能让后续的节点开发和流程编排事半功倍。