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 + 1 报 TypeError——离出错点已经很远,极难定位。你想要的是"脏值一进状态就报错"。💡 本质:把"数据契约"变成运行时强制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"的产物。图注: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)
图注:Pydantic 的安全性来自这道每步都过的关卡;TypedDict 省掉了它,换来速度、失去校验。
L04
隐蔽坑:字段名撞 pydantic 保留字会被重映射
LangGraph 内部经常需要动态造一个 pydantic 模型(比如为 JSON schema、输入校验)。造模型时,如果你的字段名以 _ 开头、或撞上 pydantic 的保留名(如 schema、model_*),会被自动改名:
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_state→schema(**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.py 的 add_conditional_edges,看那个决定"下一步去哪"的路由函数是怎么被解析和执行的。