LangGraph实战教程:一文搞懂图状态管理
LangGraph状态管理采用TypedDict、dataclass和Pydantic三种Schema定义。TypedDict语法简洁,节点返回字典实现部分更新,配合add归约器可自动合并列表;Pydantic提供严格类型校验,dataclass适合面向对象编程风格。推荐优先使用TypedDict,因其灵活高效。
前情回顾中,我们梳理了LangGraph的入门基础、Reducer机制以及核心概念。现在,聚焦整个图的“灵魂”——状态(State)。在LangGraph计算图中,状态如同贯穿始终的动脉,每个节点均可从中读取数据,并向其写入更新,实现信息流转与共享。本文将系统拆解三种状态Schema的定义方式,配合代码示例,助你快速上手。

一、什么是状态(State)?
本质上,状态是一个共享的数据容器,承载图在运行过程中需要传递和维护的全部信息。每个节点都能从状态中读取,也能向状态中写入更新。定义状态即声明其Schema——字段的完整描述。官方推荐三种方式:TypedDict、dataclass 和 Pydantic。
二、三种状态Schema定义方式
2.1 TypedDict —— 最推荐的方式
TypedDict 是 Python 标准库 typing 模块提供的类型注解工具。你可以像定义普通字典一样定义状态结构,同时为每个字段指定类型。直接看代码:
from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
from operator import add
class OverAllState(TypedDict):
logs: Annotated[list[str], add] # 使用add归约器实现日志追加
cur_id: str # 当前ID字符串
def node_1(state: OverAllState) -> OverAllState:
pre_id = state["cur_id"]
return {"logs": ["node_1 运行完毕"], "cur_id": pre_id + ", node_1"}
def node_2(state: OverAllState) -> OverAllState:
pre_id = state["cur_id"]
return {"logs": ["node_2 运行完毕"], "cur_id": pre_id + ", node_2"}
builder = StateGraph(state_schema=OverAllState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
builder.add_edge("node_2", END)
graph = builder.compile()
print(graph.invoke({"cur_id": "start"}))
输出结果:
{'logs': ['node_1 运行完毕', 'node_2 运行完毕'], 'cur_id': 'start, node_1, node_2'}
关键点:
- 通过
state["字段名"]访问字段。 - 节点函数返回字典,表示对状态的部分更新。
-
Annotated[list[str], add]中的add是一个归约器(Reducer),多个节点返回的logs列表会自动合并,而非直接覆盖。
2.2 dataclass —— 面向对象风格
如果你偏爱面向对象的写法,dataclass 会更顺手,最大变化是字段访问从 ['字段名'] 变为 .字段名。
from langgraph.graph import StateGraph, START, END
from typing import Annotated
from dataclasses import dataclass
from operator import add
@dataclass
class OverAllState:
logs: Annotated[list[str], add]
cur_id: str
def node_1(state: OverAllState) -> OverAllState:
pre_id = state.cur_id # 注意:使用 . 访问
return {"logs": ["node_1 运行完毕"], "cur_id": pre_id + ", node_1"}
def node_2(state: OverAllState) -> OverAllState:
pre_id = state.cur_id
return {"logs": ["node_2 运行完毕"], "cur_id": pre_id + ", node_2"}
# 后续构建流程与TypedDict完全相同
builder = StateGraph(state_schema=OverAllState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
builder.add_edge("node_2", END)
graph = builder.compile()
print(graph.invoke({"cur_id": "start"}))
输出结果与 TypedDict 版本完全一致。需留意:节点函数的返回值仍是字典,而非 OverAllState 对象——这是 LangGraph 的设计约定:节点返回的是“状态更新”,而非完整状态。
2.3 Pydantic —— 严格的校验派
Pydantic 是数据验证方面最强大的工具,定义状态 Schema 的同时自带校验能力。字段访问方式与 dataclass 一样,使用 .字段名。
from langgraph.graph import StateGraph, START, END
from typing import Annotated
from pydantic import BaseModel
from operator import add
class OverAllState(BaseModel):
logs: Annotated[list[str], add]
cur_id: str
def node_1(state: OverAllState) -> OverAllState:
pre_id = state.cur_id
return {"logs": ["node_1 运行完毕"], "cur_id": pre_id + ", node_1"}
def node_2(state: OverAllState) -> OverAllState:
pre_id = state.cur_id
return {"logs": ["node_2 运行完毕"], "cur_id": pre_id + ", node_2"}
builder = StateGraph(state_schema=OverAllState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
builder.add_edge("node_2", END)
graph = builder.compile()
print(graph.invoke({"cur_id": "start"}))
输出结果同样一致。
三、三种方式的校验行为对比
实际开发中,字段不匹配的情况时有发生。三种方式处理异常时的表现截然不同。
3.1 输入字段不匹配
当我们调用 graph.invoke() 传入初始状态时,如果传入的字段名与 Schema 定义不一致:
| 方式 | 异常类型 | 说明 |
|---|---|---|
| TypedDict | KeyError |
将输入视为字典,找不到键时报错 |
| dataclass | TypeError |
将输入映射到类属性,参数不匹配报类型错误 |
| Pydantic | ValidationError |
最严格的校验,给出详细的错误信息 |
3.2 节点返回字段不匹配
当节点返回的字典中包含 Schema 中没有定义的字段时,三种方式的行为统一:该字段会被静默忽略,不会报错。这意味着你需要特别小心拼写错误,否则某个节点的更新可能永远不会生效。
四、最佳实践:为什么推荐TypedDict?
在实际项目中,官方和社区普遍推荐使用 TypedDict 来定义状态 Schema,原因很实在:
- 语法简洁:无需额外装饰器,直接定义字典结构。
- 类型提示友好:IDE 能正确推断字段类型,自动补全顺畅。
- 贴近更新逻辑:节点返回的是字典形式的“部分更新”,与 TypedDict 天然契合。
- 轻量无负担:不像 Pydantic 有校验开销,性能更好。
- 官方首选:绝大多数官方示例都用 TypedDict。
什么时候用其他方式?
- 如果你习惯面向对象风格,且团队统一使用
dataclass,可以继续沿用。 - 如果你需要对输入数据进行严格的类型校验(例如来自外部 API 的数据),选择 Pydantic。
五、总结
| 维度 | TypedDict | dataclass | Pydantic |
|---|---|---|---|
| 字段访问 | ["key"] |
.attr |
.attr |
| 输入校验 | KeyError | TypeError | ValidationError |
| 返回多余字段 | 静默忽略 | 静默忽略 | 静默忽略 |
| 推荐度 | |||
| 适用场景 | 通用首选 | OOP风格团队 | 需要严格校验 |
掌握了状态管理,你就抓住了 LangGraph 的核心。接下来,就可以在此基础上构建更复杂的图结构,比如条件分支、循环等高级特性。
六、相关面试题
面试题1:LangGraph中为什么要使用 Annotated[list[str], add]这样的写法?不加 add会怎样?
参考答案:Annotated[list[str], add] 中的 add 是一个归约器(Reducer),它定义了当多个节点都对同一个字段进行更新时,如何处理这些更新的策略。add 表示将多个节点返回的列表进行拼接(+ 操作)。如果不加 add,默认行为是覆盖,后执行的节点返回的 logs 会完全替换前面节点设置的值,导致历史日志丢失。这在很多场景下是不符合预期的。
面试题2:节点函数的返回值必须是完整的State对象吗?能否只返回部分字段?
参考答案:不需要返回完整的 State 对象。节点函数返回的是一个字典,表示对状态的部分更新。LangGraph 会自动将这个字典与当前状态合并。如果返回的字典中只包含部分字段,其他字段保持不变。这也是为什么推荐使用 TypedDict 的原因之一——节点返回字典,状态也是字典,二者天然匹配。
面试题3:如果我在节点中不小心返回了一个状态Schema中没有定义的字段,会发生什么?
参考答案:这个多余的字段会被静默忽略。无论是 TypedDict、dataclass 还是 Pydantic,LangGraph 都不会因为节点返回了未定义的字段而抛出异常。这可能会导致一些难以调试的问题——比如你期望更新某个字段,但因为拼写错误,更新始终没有生效。建议在开发阶段仔细检查节点返回的字段名是否与 Schema 定义一致,或者编写单元测试来验证状态更新的正确性。
面试题4:Pydantic方式定义状态有什么独特的优势?
参考答案:Pydantic 的主要优势在于数据校验。当你从外部来源(如用户输入、API 响应、数据库查询)获取数据并注入状态时,Pydantic 会自动校验数据类型是否符合 Schema 定义。如果类型不匹配,会抛出 ValidationError 并提供详细的错误信息。此外,Pydantic 还支持字段默认值、自定义验证器、JSON Schema 导出、序列化/反序列化。如果你的应用需要处理不可信的输入数据,Pydantic 是一个很好的选择。
面试题5:如何在一个图中使用多种不同的状态Schema?
参考答案:通常情况下,一个 StateGraph 只能绑定一个 state_schema。如果你需要在不同阶段使用不同的数据结构,有以下几种方案:
- 状态嵌套:在状态中定义一个嵌套字段,例如
sub_state: dict,然后在不同节点中操作这个子字典。 - 多个子图:将不同的逻辑拆分成独立的子图(Subgraph),每个子图有自己的状态 Schema,然后通过父图协调它们。
- 动态字段:使用
TypedDict的total=False选项定义可选字段,或者使用dict作为兜底字段类型。
推荐优先考虑第一种方案,因为它保持了图的整体性和可追踪性。
希望这篇教程能帮你彻底掌握 LangGraph 的状态管理。如果你有任何疑问或想深入了解某个知识点,欢迎留言讨论。
