Day 12 / 共 60 天 · 阶段 2 状态与数据流

Pydantic State:带校验的状态,代价是什么

前面状态都用 TypedDict——轻,但没有运行时校验(塞进去一个类型不对的值它不管)。把状态换成 Pydantic BaseModel,就能在每次进节点前自动校验、类型强转、跑 field_validator。今天从官方压测文件 bench/pydantic_state.py 看真实的 Pydantic 状态怎么写,再钻进 graph/state.py_internal/_pydantic.py,看它是怎么在每个超步把 dict 强转成模型实例(每次都 validate),以及一个隐蔽但重要的坑——字段名撞上 pydantic 保留字时的自动重映射。

📍 阶段 2「状态与数据流」(D07-12) · 你在第 12 天(收官)
D07 reducers D08 add_messages D09 累加通道 D10 并行写 D11 输入/输出 D12 Pydantic
💡 一句话锚点 用 Pydantic 当状态,等于给状态加了一道"安检门":每次节点读状态时,LangGraph 会把内部那份状态 dict 用 Schema(**dict) 重新构造成模型实例——这一步自动触发 pydantic 的类型校验和你写的 field_validator。好处是脏数据进不来、错误早暴露;代价是每个超步、每个节点都要 validate 一遍,字段多时是实打实的开销(所以官方专门有 bench 压它)。
L01

痛点:TypedDict 不校验,脏数据静悄悄溜进来

🤔 痛点TypedDict 的类型标注只是"给人和 IDE 看的注释",运行时不强制。你的字段声明 count: int,某个节点因为 bug 返回了 {"count": "3"}(字符串),TypedDict 状态照单全收,直到几步之后 count + 1TypeError——离出错点已经很远,极难定位。你想要的是"脏值一进状态就报错"。
💡 本质:把"数据契约"变成运行时强制Pydantic 模型在构造实例的那一刻校验每个字段:类型不符尝试强转("3"3)、强转不了就抛 ValidationError、还会跑你定义的 @field_validator 自定义规则。LangGraph 的做法是:内部照旧用通道存 dict(省内存、好合并),但每次把状态交给节点前,临时构造一次模型实例,借这次构造完成校验。
大白话TypedDict 是"口头约定,违约不管";Pydantic 是"合同+门卫,进门就查证件"。查证件的时机就是"节点拿到 state 之前"。
L02

真实的 Pydantic 状态:bench/pydantic_state.py

官方拿来做性能压测的状态,正是一个"字段多、每个都带校验"的 Pydantic 模型。节选前几个字段:

libs/langgraph/bench/pydantic_state.py:14-56
class State(BaseModel):
    messages: Annotated[list, operator.add] = Field(default_factory=list)

    @field_validator("messages", mode="after")
    @classmethod
    def validate_messages(cls, v):
        if not isinstance(v, list):
            raise TypeError("messages must be a list")
        for msg in v:
            if not isinstance(msg, dict):
                raise TypeError("messages must be a list of dicts")
        return v

    trigger_events: Annotated[list, operator.add] = Field(default_factory=list)
    ...
    primary_issue_medium: Annotated[str, lambda x, y: y or x] = Field(default="email")
    autoresponse: Annotated[dict | None, lambda _, y: y] = Field(default=None)  # 总是覆盖
Annotated[list, operator.add]★注意:reducer 照样写在 Annotated 里(D05 的嗅探对 Pydantic 字段一样有效)。= Field(default_factory=list) 是 pydantic 的默认值语法。两套机制叠加,互不冲突。
@field_validator(mode="after")pydantic 的自定义校验:after 表示在类型转换之后再跑你的检查(这里确保 messages 是"list of dicts")。这些校验会在每次构造实例时执行。
lambda x, y: y or xreducer 写成 lambda,恰好 2 个位置参数——满足 D05 的"数参数"规则。y or x = 有新值用新值、否则保留旧值。
为什么拿它做 bench因为它有十几个字段、每个都带 validator——是"Pydantic 状态开销"的放大镜。压测就是要量"每步都 validate 一遍"到底多贵。

它的建图部分和 TypedDict 完全一样(Pydantic 只是换了 state 的"壳"):

libs/langgraph/bench/pydantic_state.py:249-297(节选)
    builder = StateGraph(State)                 # 直接把 BaseModel 当 state
    builder.add_edge(START, "one")
    builder.add_node("one", partial(read_write, "messages", ["trigger_events", ...]))
    ...
    builder.add_conditional_edges(
        "six", lambda state: END if len(state.messages) > n else "one")   # 注意 state.messages 属性访问
看这行 state.messages:TypedDict 状态是 state["messages"](下标),Pydantic 状态是 state.messages(属性)。因为节点拿到的是模型实例不是 dict——这正是 L03 那次"coerce"的产物。
数据结构:同一份状态,两种"壳" TypedDict 状态 通道存 dict 节点直接拿到 dict state["count"](下标) 无 mapper · 无校验 最快,脏值不拦 Pydantic 状态 通道仍存 dict 进节点前 State(**dict) state.count(属性) 每步 coerce · 每步校验 较慢,脏值挡门外 底层都是 dict,区别在"交给你函数前要不要构造实例+校验"
图注:Pydantic 只是在节点入口多套了一层"构造实例"的壳,通道底层存储没变。
L03

核心:每次读状态都 coerce 成实例(=validate)

D04 讲 attach_node 时见过一个 mapper,注释写着"coerce state dict to schema class (eg. pydantic model)"。它由 _pick_mapper 挑出:

libs/langgraph/langgraph/graph/state.py:1718-1732
def _pick_mapper(state_keys, schema):
    if state_keys == ["__root__"]:
        return None
    if isclass(schema) and (issubclass(schema, BaseModel) or is_dataclass(schema)):
        return partial(_coerce_state, schema)      # ★是 Pydantic/dataclass → 用 coerce
    return None                                    # 是 TypedDict → 不转,直接给 dict

_S = TypeVar("_S")

def _coerce_state(schema, input: dict) -> _S:
    return schema(**input)                         # ★关键:用 dict 构造模型实例
schema 是 BaseModel只有状态是 Pydantic(或 dataclass)时才返回 mapper;TypedDict 返回 None(不转换,节点直接拿 dict,无校验开销)。
schema(**input)★整个 Pydantic 校验的"引爆点"就这一行。把内部状态 dict 展开成关键字参数构造模型——pydantic 在这一步做全部校验 + 强转 + 跑 field_validator。脏数据在这里就会抛 ValidationError。
每个节点读状态都调D04 学过:节点被触发时,引擎用它的 channels 读出状态 dict,再过 mapper。所以每次进节点 = 构造一次实例 = validate 一次

它挂载在 PregelNode 的 mapper 上(D04 见过的那段):

libs/langgraph/langgraph/graph/state.py:1506-1523(节选)
            if input_schema in self.schema_to_mapper:
                mapper = self.schema_to_mapper[input_schema]      # 缓存:同 schema 复用
            else:
                mapper = _pick_mapper(input_channels, input_schema)
                self.schema_to_mapper[input_schema] = mapper
            self.nodes[key] = PregelNode(
                ...
                mapper=mapper,     # 读出 dict 后,用它 coerce 成模型实例交给你的函数
                bound=node.runnable)
控制流:节点读状态时的 coerce/validate 关卡 通道里的状态{"count":"3", ...} mapperState(**dict)校验+强转+validator 你的节点函数拿到 State 实例 "3"→3 强转成功;类型不可转→抛 ValidationError(脏数据挡在门外) TypedDict 状态:没有 mapper,dict 直达节点(无此关卡、无开销)
图注:Pydantic 的安全性来自这道每步都过的关卡;TypedDict 省掉了它,换来速度、失去校验。
L04

隐蔽坑:字段名撞 pydantic 保留字会被重映射

LangGraph 内部经常需要动态造一个 pydantic 模型(比如为 JSON schema、输入校验)。造模型时,如果你的字段名以 _ 开头、或撞上 pydantic 的保留名(如 schemamodel_*),会被自动改名:

libs/langgraph/langgraph/_internal/_pydantic.py:152-178
def _remap_field_definitions(field_definitions):
    remapped = {}
    for key, value in field_definitions.items():
        if key.startswith("_") or key in _RESERVED_NAMES:      # 撞保留字/下划线开头
            if isinstance(value, FieldInfo):
                raise NotImplementedError("Remapping ... not supported if ... Field instance ...")
            type_, default_ = value
            remapped[f"private_{key}"] = (                     # ★改名成 private_xxx
                type_,
                Field(default=default_, alias=key,             # 但保留原名作为 alias
                      serialization_alias=key,
                      title=key.lstrip("_").replace("_", " ").title()))
        else:
            remapped[key] = value
    return remapped
_RESERVED_NAMES{key for key in dir(BaseModel) if not key.startswith("_")} 动态算出(_pydantic.py:149)——即 BaseModel 的所有公共方法名(schema/dict/json/copy/model_dump…)。你的字段不能叫这些。
改名 private_key + alias如果撞了,内部字段名改成 private_xxx,但用 alias=key 保住原名对外可见。这样序列化/反序列化时你仍看到原名。
Field 实例不给改边界:若这个撞名字段的值是一个显式 Field(...) 实例,无法安全重映射,直接 NotImplementedError——宁可报错也不猜。
⚠️ 边界:别把状态字段命名成 schema / model_name / copy 这类这些是 pydantic BaseModel 的保留公共名。用它们当字段名,轻则触发上面的重映射(行为变得难以预测)、重则和 pydantic 内部机制冲突。create_model 还专门对 model 开头的名字临时静音警告_pydantic.py:233-240),侧面说明这类命名是"已知雷区"。实践建议:状态字段名保持普通业务词(question/answer/retry_count),避开 schema/json/dict/copy/model_*
L05

谁算"pydantic 能处理的类型"

LangGraph 用一个函数判断某个复杂类型能不能交给 pydantic 处理(决定要不要走校验路径):

libs/langgraph/langgraph/_internal/_pydantic.py:252-275
def is_supported_by_pydantic(type_) -> bool:
    if is_dataclass(type_):
        return True
    if isinstance(type_, type) and issubclass(type_, BaseModel):
        return True
    if hasattr(type_, "__orig_bases__"):
        for base in type_.__orig_bases__:
            if base is TypedDict:                    # typing_extensions.TypedDict
                return True
            elif base is typing.TypedDict:           # 标准库 typing.TypedDict
                if sys.version_info >= (3, 12):      # 仅 3.12+ pydantic 才支持
                    return True
    return False
dataclass / BaseModel直接支持——它们能被 pydantic 校验/内省。
TypedDict 分两种★细节:typing_extensions.TypedDict 全版本支持;标准库 typing.TypedDict 只有 Python 3.12+ 的 pydantic 才认。这就是为什么 LangGraph 文档一直推荐 from typing_extensions import TypedDict——兼容老 Python。
返回 False 的类型裸 int/str 这种原始类型返回 False(注释明说"对原始类型返回 False")——它们走 __root__ 路径,不需要 pydantic 建模。
🍼 一句话这个函数就是"要不要给这个 schema 上 pydantic 那套机器"的开关。理解它,你就知道为什么"用 typing_extensions.TypedDict"是一句有技术含义的建议,不是随便说说。
L06

两处设计取舍 + 一处边界

🎨 设计取舍①:为什么内部存 dict、只在"进节点"时才 coerce 成模型? 朴素做法:状态从头到尾都以模型实例存在。LangGraph 选择内部通道存 dict、只在交给节点前临时构造实例好处:通道的合并(reducer)在 dict/原始值层面做最省事、最好序列化落盘(阶段 6 checkpoint 要存的是 dict 不是对象);模型实例只是"给用户函数用的临时视图"。代价:每进一次节点就构造一次实例、validate 一次——字段多、超步多时开销可观(正是 bench/pydantic_state.py 要量化的)。这是"持久化友好 + 用户友好"与"运行开销"的权衡。
🎨 设计取舍②:撞保留字为什么"自动重映射"而不直接报错? _remap_field_definitions 遇到撞名字段自动改成 private_xxx + alias。好处:LangGraph 内部大量动态造模型(用你的字段名当 pydantic 字段名),若某个业务字段恰好叫 schema 就直接崩,用户体验差;自动重映射让绝大多数情况"照常工作"。代价:重映射后字段的内部名和你写的不一致,出问题时调试更绕;且对 Field 实例这种没法安全重映射的情况仍要 NotImplementedError。所以它是"尽量兜住,兜不住才报错"的折中。
⚠️ 边界:Pydantic 状态的校验是"每步都跑",不是"只在入口跑一次"很多人以为校验只发生在 invoke 输入那一下。实际上因为 coerce 挂在每个节点的 mapper 上,状态在图里流转 N 个超步,就会被 validate N 次。这意味着:① 你的 field_validator 若很重(比如做正则、查表),会被反复执行,成为热点;② 如果某个内部节点写回了一个暂时不满足 validator 的中间态,下一个节点读时就会抛 ValidationError——即使这个中间态本打算马上被修正。排查性能或"莫名 ValidationError"时,记住这条"每步都验"的事实。
L07

今日小结 + 动手 + 明日预告

🧠 今天你应该能回答

  • Pydantic 状态相比 TypedDict 多了什么?(运行时校验 + 类型强转 + field_validator)
  • 校验在哪一步、哪一行触发?(节点读状态时 mapper 调 _coerce_stateschema(**dict)
  • 校验跑几次?(每个节点每次被触发都跑一次,不是只入口一次)
  • 为什么内部还是存 dict?(便于 reducer 合并 + checkpoint 序列化,实例只是临时视图)
  • Pydantic 状态里 reducer 怎么写?(照样 Annotated[类型, reducer],和 Field 默认值并存)
  • 为什么推荐 typing_extensions.TypedDict?(标准库 typing.TypedDict 仅 3.12+ 被 pydantic 支持)
  • 字段名不能叫什么?(pydantic 保留名 schema/dict/json/copy/model_* 等,会被重映射或冲突)

✋ 10 分钟动手

# 1. 读真实压测状态(字段多+每个带 validator)
sed -n '14,70p' libs/langgraph/bench/pydantic_state.py

# 2. 看 coerce 引爆点:schema(**input)
sed -n '1718,1732p' libs/langgraph/langgraph/graph/state.py

# 3. 看保留字重映射 + 支持判定
sed -n '152,178p' libs/langgraph/langgraph/_internal/_pydantic.py
sed -n '252,275p' libs/langgraph/langgraph/_internal/_pydantic.py

# 4. 亲手感受"脏数据被挡在门外"
python3 -c "
from langgraph.graph import StateGraph, START, END
from pydantic import BaseModel
class S(BaseModel):
    count: int
def bad(s): return {'count': 'not-an-int'}   # 故意写脏值
g=StateGraph(S); g.add_node('n', bad); g.add_edge(START,'n'); g.add_edge('n',END)
try:
    g.compile().invoke({'count': 1})
except Exception as e:
    print(type(e).__name__, '被拦住了')   # 下一个节点读到脏值时 ValidationError
"
💡 明日预告 · Day 13(进入阶段 3)阶段 2 到此结束——你已吃透"状态怎么定义、怎么合并、怎么校验、对外怎么收发"。明天进入阶段 3「控制流」:D13 讲普通边和条件边的区别,钻进 graph/_branch.pyadd_conditional_edges,看那个决定"下一步去哪"的路由函数是怎么被解析和执行的。
← Day 11 · 输入/输出 schema Day 13 · 普通边 vs 条件边 →