从零开始学LangGraph（9）：详解State的定义、更新与管理 - 文章 - 开发者社区

关注我~第一时间学习如何更好地使用AI。

重要的不是我们是否会被AI替代，

而是我们要比被替代的人更懂AI。

前期导览：

从零开始学LangGraph（1）：Chat Model --- 如何通过代码与模型对话（上）

从零开始学LangGraph（2）：Message和Template —— 如何通过代码与模型对话（下）

从零开始学LangGraph（3）：Tools --- 如何实现大模型与外部系统对接

从零开始学LangGraph（4）：用Excel表格来理解LangGraph的基本工作原理

从零开始学LangGraph（5）：手把手教你搭建简易Graph

从零开始学LangGraph（6）：轻松玩转Conditional Edges

从零开始学LangGraph（7）：手把手教你手搓带Memory的Chat Bot

从零开始学LangGraph（8）：Tools + Subgraph实战，手把手教你构建ReAct Agent

大家好，前期我已经陆续给大家分享了一些利用LangGraph的学习笔记，相信大家已经开始大开脑洞搭建自己的AI Agent了。本期我想放缓一下脚步，再回到一些比较重要的概念上来加深一下学习，以提升对LangGraph这个工具的掌控感。

要说LangGraph中重要的概念，那我第一个想到的就是State了。无论我们定义什么样的节点、边、图，核心目的都是希望通过对State的操作，以达到我们想要的效果。

本期我将带着大家结合官方文档，从state的定义、更新与管理三个维度 ，来温故知新。

关于State的定义

LangGraph中的State，可以用TypedDict，Pydantic model，或者dataclass来定义，其中前两种比较常用。

TypedDict：轻量级的选择

TypedDict主打一个轻量级 和简单直接 。它本质上就是Python标准库typing模块中的一个类型提示工具，不需要安装额外的依赖，使用起来非常方便。

正如我在前期文章中给大家展示的，用TypedDict定义State的典型写法如下：


 
 
 
 
   
from typing import TypedDict  
  
class AppState(TypedDict):  
    Key\_1: str  
    Key\_2: int

需要注意的是，由于typing本质上是个类型提示工具，它只在开发阶段（通过IDE或类型检查工具）发挥提示作用，而并不会在应用实际运行时对其中的参数、键值进行强制的类型验证。举个例子，虽然我们已经设定Key_1的数据类型为str，但如果我们向它传入一个int，程序不会报错（但可能会导致后续逻辑错误）。因此。如果你需要更严格的运行时验证，那就需要考虑使用Pydantic model了。

Pydantic Model：更强大的选择

Pydantic是一个强大的数据验证库，它可以在项目运行时对数据进行验证，确保数据的类型和格式符合预期。这对于构建健壮的AI Agent来说是非常重要的。

下面我们来看一个使用Pydantic model定义State的示例：


 
 
 
 
   
from pydantic import BaseModel, Field  
  
class AppState(BaseModel):  
    Key\_1: str  
    Key\_2: int = Field(ge=0, le=100)  # 限制Key\_2的值必须在0到100之间  
    Key\_3: str = "默认值"  # 如果没传这个字段，就自动使用"默认值"

可以看到，与TypedDict相比，Pydantic model提供了几个非常实用的功能：

限制取值范围 ：通过 Field 可以给数值类型的字段设置范围限制。比如 ge=0 表示"大于等于0"（ge是greater or equal的缩写）， le=100 表示"小于等于100"（le是less or equal的缩写）。这样就能确保数据在合理的范围内，避免出现负数等级、超过上限的血量这种不合理的情况。

自动验证数据 ：当你传入的数据不符合要求时，Pydantic会在程序运行时立即报错。比如上面例子中，如果你传入的 Key\_2 是负数或者大于100，程序就会抛出异常，告诉你数据有问题。

设置默认值 ：如果某个字段是可选的，你可以给它设置一个默认值。比如上面的 Key\_3 ，如果调用时没有传这个字段，它就会自动使用"默认值"。这样就不需要每次都手动检查字段是否存在了。

当然Pydantic还提供了其他一些更复杂的功能，比如：

• 自定义验证器 ：可以写自己的验证函数，对字段值进行更复杂的检查，比如验证邮箱格式、密码强度等
• 字段间的依赖关系 ：可以让一个字段的值依赖于另一个字段，比如总价 = 单价 × 数量
• 数据转换 ：可以在验证时自动转换数据类型，比如把字符串"123"自动转换成整数123
• 嵌套模型 ：可以在一个模型中包含另一个模型，实现复杂的数据结构

这些功能在构建复杂应用时非常有用。如果你想深入了解，可以查看Pydantic的官方文档：https://docs.pydantic.dev/

那么在实际开发中，我们应该如何选择呢？通常的建议是：

• TypedDict ：如果你的项目比较简单，不需要复杂的验证逻辑，或者你希望保持代码的轻量级，那么 TypedDict 是个不错的选择
• Pydantic model ：如果你需要严格的数据验证，或者你的State结构比较复杂，需要字段间的依赖关系，那么 Pydantic model 会更合适

关于State的更新

在LangGraph中，State的更新其实是一个两阶段 的过程：

第一阶段：节点函数处理

Node函数接收当前的state，根据业务逻辑进行处理，计算出需要更新的新值。

第二阶段：Reducer决定更新方式

Reducer函数决定如何将节点返回的新值更新到state中，是覆盖（overwrite）还是拼接（append）等

理解这两个阶段的工作机制，对于构建功能完善的Agent至关重要。

第一阶段：节点函数处理

节点函数是State更新的第一步。它接收当前的state作为输入，根据业务逻辑进行处理，然后返回一个结果，包含需要更新的state字段和新值。

回顾一下我们在从零开始学LangGraph（5）：手把手教你搭建简易Graph中的例子：


 
 
 
 
   
def dodge\_check\_node(state: AppState) -> AppState:  
    """检查闪避结果并处理HP变化"""  
    dice\_roll = random.randint(1, 6)  
      
    if dice\_roll > 3:  
        print("你成功闪避了伤害")  
        return {}  # 没有变化，返回空字典  
    else:  
        new\_HP = state['char\_HP'] - dice\_roll  
        print(f"闪避失败！受到{dice\_roll}点伤害")  
        return {"char\_HP": new\_HP}  # 返回需要更新的字段和新值

这里，节点函数计算出了新的HP值，并返回{"char\_HP": new\_HP}。但这时候，我们还无法判断这个新值将如何对state产生影响，因为这还得由定义state时设置的Reducer来决定。

第二阶段：Reducer决定更新方式

Reducer函数决定了节点返回的新值如何更新到state中，每一个键都可以有其独立的Reducer。这是State更新的关键环节，不同的Reducer会产生不同的更新效果。

1.Reducer的基本用法

在从零开始学LangGraph（7）：手把手教你手搓带Memory的Chat Bot一文中，我向大家介绍了一种常用的Reducer，add\_messages：


 
 
 
 
   
from typing import Annotated, List  
from langchain\_core.messages import AnyMessage  
from langgraph.graph.message import add\_messages  
  
class ChatState(TypedDict):  
    messages: Annotated[List[AnyMessage], add\_messages]

这里的关键是Annotated的使用。Annotated是Python的类型注解工具，它允许我们为类型添加额外的元数据。在LangGraph中，我们用它来指定Reducer函数。

add\_messages的作用是：

• 将新消息追加到现有消息列表的末尾
• 如果新消息的ID与旧消息相同，则更新旧消息而不是追加

除了add\_messages，LangGraph中还有其他一些内置的Reducer，比如：

operator.add ：用于数值累加，将新值加到旧值上。比如统计总分、计数等场景：


 
 
 
 
   
from typing import Annotated, TypedDict  
from operator import add  
  
class ScoreState(TypedDict):  
    total\_score: Annotated[int, add]  # 每次返回的分数会累加到总分上

operator.extend ：用于列表扩展，将新列表的元素追加到当前列表中。比如收集多个节点的结果：


 
 
 
 
   
from typing import Annotated, List, TypedDict  
from operator import extend  
  
class ResultState(TypedDict):  
    results: Annotated[List[str], extend]  # 新列表的元素会追加到现有列表中

除了这些内置Reducer，LangGraph也允许我们自定义Reducer来实现想要的效果。

2.自定义Reducer

当内置的Reducer无法满足需求时，我们可以自定义Reducer函数实现对state更新过程的控制。自定义Reducer的过程非常简单，只需要定义一个函数即可。

自定义Reducer的基本要求：

参数：Reducer必须接收两个参数，依次从左到右分别为：

• old\_value ：state中该字段的当前值
• new\_value ：节点函数返回的新值

返回值 ：返回合并后的结果，类型应该与字段类型一致

下面我给大家一个简单的例子。我们定义一个作用是取新旧值中的较大者的Reducer，并将它设定在 state中。


 
 
 
 
   
from typing import Annotated, TypedDict  
  
# 定义一个自定义Reducer函数  
def take\_max(old\_value: int, new\_value: int) -> int:  
    """自定义Reducer：取新旧值中的较大者"""  
    return max(old\_value, new\_value)  
  
# 在State定义中使用自定义Reducer  
class AppState(TypedDict):  
    max\_score: Annotated[int, take\_max]  # 使用自定义的take\_max作为Reducer

然后，我们假设一个更新分数的节点：


 
 
 
 
   
def update\_score\_node(state: AppState) -> AppState:  
    """更新分数"""  
    return {"max\_score": 85}  # 返回新的分数值

上述代码意味着，根据这个节点函数的计算操作（被简化），输出max_score的值为85，但由于我们设定了Reducer，所以这个85不会直接对state进行更新，而是需要与经过这个节点之前的state中保存的值（这才是当前的state！）进行比较：

如果当前max_score是80，节点返回85，Reducer会取max(80, 85) = 85
如果当前max_score是90，节点返回85，Reducer会取max(90, 85) = 90

是不是很酷炫？

3.Overwrite：强制覆盖机制

有时候，即使我们已经为某个字段设定了Reducer（比如add\_messages），但在某些特殊场景下，我们可能希望忽略这个Reducer，直接用新值覆盖state中的旧值。这时候，我们就可以使用Overwrite机制。

Overwrite允许我们在节点返回时，强制忽略已设定的Reducer，直接覆盖state中的值。

LangGraph提供了两种使用Overwrite的方式：

使用Overwrite类型包装值 ：


 
 
 
 
   
from langgraph.graph import Overwrite  
  
def reset\_node(state: ChatState) -> ChatState:  
    """重置消息列表，忽略add\_messages的Reducer"""  
    return {"messages": Overwrite([])}  # 直接覆盖为空列表，忽略add\_messages的拼接逻辑

使用\_\_overwrite\_\_键 ：


 
 
 
 
   
def reset\_node(state: ChatState) -> ChatState:  
    """重置消息列表，忽略add\_messages的Reducer"""  
    return {"messages": {"\_\_overwrite\_\_": []}}  # 效果与上面相同

Overwrite在以下场景特别有用：

• 重置状态 ：需要清空累积的数据（比如重置消息历史）
• 强制替换 ：需要完全替换某个字段的值，而不考虑之前的累积结果
• 特殊处理 ：某些节点需要绕过Reducer的常规逻辑

需要注意的是，如果不指定Reducer，LangGraph默认就会直接覆盖（相当于自动使用overwrite），所以Overwrite主要用于在已设定Reducer的情况下强制覆盖 的场景。

关于State的管理

在实际应用中，我们有时候需要更精细地管理State的可见性和访问权限。LangGraph提供了几种机制来实现这个目标，包括input schema、output schema 和private schema 。

input schema和output schema

在某些场景下，我们可能希望Graph对外暴露的接口与内部使用的State结构有所不同。比如，我们可能希望用户只需要传入简单的参数，而不需要了解Graph内部复杂的State结构。

这时候，我们就可以使用input schema 和output schema 来定义Graph的输入和输出格式。

1.基本用法

假设我们有一个内部State，结构比较复杂：


 
 
 
 
   
from typing import TypedDict, List  
  
class InternalState(TypedDict):  
    user\_name: str  
    user\_age: int  
    messages: List[str]  
    result: str  
    status: str  
    internal\_counter: int  
    debug\_info: dict

但我们希望用户调用Graph时，只需要传入用户名和年龄，而不需要关心内部的messages、internal\_counter、debug\_info等字段。同时，我们也希望Graph返回给用户的只是处理结果，而不是整个内部State。

这时候，我们可以分别定义input和output schema：


 
 
 
 
   
from typing import TypedDict  
  
class InputSchema(TypedDict):  
    """用户输入格式"""  
    user\_name: str  
    user\_age: int  
  
class OutputSchema(TypedDict):  
    """用户输出格式"""  
    result: str  
    status: str

然后在创建Graph时指定这些schema：


 
 
 
 
   
graph = StateGraph(InternalState,input\_schema=InputSchema,  
    output\_schema=OutputSchema)

2.效果说明

使用input/output schema后，Graph的行为会发生以下变化：

（1）输入转换：

用户调用时，只需要传入InputSchema格式的数据


 
 
 
 
   
result = app.invoke({"user\_name": "张三", "user\_age": 25})

LangGraph会自动将InputSchema转换为InternalState，未输入的地方会自动初始化为空值。转换后的内部state如下：


 
 
 
 
   
{  
"user\_name": "张三",  
"user\_age": 25,  
"messages": [],  # 自动初始化为空列表  
"internal\_counter": 0,  # 自动初始化为0  
"debug\_info": {}  # 自动初始化为空字典  
}

（2）输出转换：

Graph执行完成后，内部state可能是：


 
 
 
 
   
{  
    "user\_name": "张三",  
    "user\_age": 25,  
    "messages": ["欢迎张三！"],  
    "result": "处理完成",  
    "status": "success"  
    "internal\_counter": 1,  
    "debug\_info": {"processed": True}  
}

这时，OutputSchema就像一个筛选器一样，只会向用户返回在其中定义的键值。在我们的示例中，实际返回给用户的sate可能是：


 
 
 
 
   
{  
    "result": "处理完成",  
    "status": "success"  
}

关键点：

• input schema定义了用户调用 invoke() 时需要传入的字段格式
• output schema定义了Graph返回给用户的字段格式
• 内部State的完整结构仍然存在，只是在输入输出时进行了转换
• 如果output schema中的字段在内部State中不存在，需要在节点中返回这些字段，或者使用转换函数

private schema

有时候，我们希望在Graph的执行过程中使用一些中间状态，这些状态只在节点之间传递，但不应该出现在Graph的最终输出中。比如：

中间计算结果 ：某些节点需要计算中间值，这些值会被后续节点使用，但不需要对外暴露；

临时数据存储 ：在Graph执行过程中需要临时存储一些数据，用于节点间的信息传递；

内部状态管理 ：需要跟踪Graph内部的执行状态，但这些状态对用户来说是不必要的。

LangGraph提供了private state 机制来实现这个需求。

1.基本用法

private state的使用方法非常简单，就是是定义一个独立的PrivateState类型，然后在节点函数中使用它。关键点在于：

• 定义独立的 PrivateState 类型，包含需要在节点间传递但不对外暴露的字段
• 节点函数可以接收 PrivateState 作为输入，也可以返回 PrivateState 作为输出
• PrivateState 中的字段会在节点间正常传递，但不会出现在Graph的最终输出中

下面我们通过一个示例来理解private state的使用：


 
 
 
 
   
from typing import TypedDict  
from langgraph.graph import StateGraph, START, END  
  
# 定义Graph的整体状态：包含公有字段和私有字段  
classOverallState(TypedDict):  
    user\_input: str      # 公有字段，会出现在最终输出中  
    graph\_output: str    # 公有字段，会出现在最终输出中  
    foo: str            # 公有字段，会出现在最终输出中  
  
# 定义私有状态：只在节点间传递，不会出现在最终输出中  
classPrivateState(TypedDict):  
    bar: str            # 私有字段，只在节点间传递  
  
# 节点1：处理用户输入，写入OverallState  
defnode\_1(state: OverallState) -> OverallState:  
    # 处理用户输入，生成中间结果  
    return {"foo": state["user\_input"] + " name"}  
  
# 节点2：读取OverallState，写入PrivateState  
defnode\_2(state: OverallState) -> PrivateState:  
    # 从OverallState读取foo，计算中间值并存入PrivateState  
    return {"bar": state["foo"] + " is"}  
  
# 节点3：读取PrivateState，写入OverallState  
defnode\_3(state: PrivateState) -> OverallState:  
    # 从PrivateState读取bar，生成最终输出  
    return {"graph\_output": state["bar"] + " Lance"}  
  
# 创建Graph，只使用OverallState作为主状态  
builder = StateGraph(OverallState)  
builder.add\_node("node\_1", node\_1)  
builder.add\_node("node\_2", node\_2)  
builder.add\_node("node\_3", node\_3)  
builder.add\_edge(START, "node\_1")  
builder.add\_edge("node\_1", "node\_2")  
builder.add\_edge("node\_2", "node\_3")  
builder.add\_edge("node\_3", END)  
  
graph = builder.compile()  
  
# 调用Graph  
result = graph.invoke({"user\_input": "My"})  
print(result)

这段代码的运行结果将是：{'user\_input': 'My', 'graph\_output': 'My name is Lance', 'foo': 'My name'}。

2.效果说明

（1）私有状态在节点间正常传递：

在上面的示例中，PrivateState中的bar字段在node\_2中被设置，然后在node\_3中被读取使用。这说明private state在Graph内部的节点之间是完全可用的，可以正常传递数据。

（2）私有状态不会出现在最终输出中：

虽然PrivateState中的bar字段在Graph执行过程中被使用（在node\_2中写入，在node\_3中读取），但最终的输出结果中只包含OverallState中定义的字段（user\_input、graph\_output、foo），bar字段被自动过滤掉了。

结语

本期我们从三个维度深入了解了LangGraph中的State机制：

State的定义 ：可以使用TypedDict或Pydantic model来定义State。TypedDict轻量级但只有类型提示，Pydantic model提供运行时验证、默认值设置等更强大的功能。

State的更新 ：State更新分为两个阶段——节点函数处理业务逻辑并返回新值，Reducer决定如何将新值合并到State中。LangGraph提供了add\_messages、operator.add等内置Reducer，也支持自定义Reducer。当需要强制覆盖时，可以使用Overwrite机制。

State的管理 ：通过input\_schema和output\_schema可以控制Graph的输入输出格式，简化对外接口。通过private state可以定义只在节点间传递但不对外暴露的中间状态，保持内部实现细节的封装性。

理解这些机制有助于我们更好地设计和管理Agent的状态流转，构建更清晰、更易维护的Graph结构。

好了，以上就是本期的主要内容，希望对大家有帮助，喜欢的朋友别忘了点赞、收藏、转发，祝大家玩的开心~

—— END——

往期精华：

1.COZE教程

零基础搞定！萌新的 Coze 开源版保姆级本地部署指南

AI工作流编排手把手指南之一：Coze智能体的创建与基本设置

AI工作流编排手把手指南之二：Coze智能体的插件添加与调用

AI工作流编排手把手指南之三：Coze智能体的工作流

Agent | 工作流编排指南4：萌新友好的Coze选择器节点原理及配置教程

Agent | 工作流编排指南5：长文扩写自由 — Coze循环节点用法详解