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（用户输入）、AIMessage（AI 回复）、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(content=user_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(content=f“无法将‘{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(content=f“不支持的操作: {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(content=f“计算: {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(content=f“已撤销操作 ‘{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_entries（Annotated[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 应用架构的核心，值得在项目初期投入时间仔细考量。一个好的状态设计，能让后续的节点开发和流程编排事半功倍。
