LangGraph实战教程:一文搞懂图状态管理

作者:技术拾光时间:2026-07-27 06:32:13浏览:478

LangGraph状态管理采用TypedDict、dataclass和Pydantic三种Schema定义。TypedDict语法简洁,节点返回字典实现部分更新,配合add归约器可自动合并列表;Pydantic提供严格类型校验,dataclass适合面向对象编程风格。推荐优先使用TypedDict,因其灵活高效。

前情回顾中,我们梳理了LangGraph的入门基础、Reducer机制以及核心概念。现在,聚焦整个图的“灵魂”——状态(State)。在LangGraph计算图中,状态如同贯穿始终的动脉,每个节点均可从中读取数据,并向其写入更新,实现信息流转与共享。本文将系统拆解三种状态Schema的定义方式,配合代码示例,助你快速上手。

LangGraph实战教程:一文搞懂图状态管理

一、什么是状态(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,原因很实在:

  1. 语法简洁:无需额外装饰器,直接定义字典结构。
  2. 类型提示友好:IDE 能正确推断字段类型,自动补全顺畅。
  3. 贴近更新逻辑:节点返回的是字典形式的“部分更新”,与 TypedDict 天然契合。
  4. 轻量无负担:不像 Pydantic 有校验开销,性能更好。
  5. 官方首选:绝大多数官方示例都用 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。如果你需要在不同阶段使用不同的数据结构,有以下几种方案:

  1. 状态嵌套:在状态中定义一个嵌套字段,例如 sub_state: dict,然后在不同节点中操作这个子字典。
  2. 多个子图:将不同的逻辑拆分成独立的子图(Subgraph),每个子图有自己的状态 Schema,然后通过父图协调它们。
  3. 动态字段:使用 TypedDict 的 total=False 选项定义可选字段,或者使用 dict 作为兜底字段类型。

推荐优先考虑第一种方案,因为它保持了图的整体性和可追踪性。

希望这篇教程能帮你彻底掌握 LangGraph 的状态管理。如果你有任何疑问或想深入了解某个知识点,欢迎留言讨论。