PydanticState

发表于 | 更新于 | 分类于 ， | 阅读 次
用 Pydantic v2 定义 AgentState：Field 约束、语义类型、自定义校验，以及 Annotated 同时挂 reducer 与校验元数据的功能图谱。

业务字段越来越多——user_id 必须是通用唯一识别码（Universally Unique Identifier，UUID）、$amount \ge 0$、SKU 必须匹配正则。纯 TypedDict 只有静态类型；Pydantic v2 模型作 State schema，可在 invoke 入图时做运行时校验，并继续用 注解类型（Annotated） 挂归约器（reducer）。

段末注释：Annotated = 在类型上附加元数据的写法；LangGraph 读其中的 reducer，Pydantic 读其中的 Field / Validator。两者可叠在同一字段上。

社区方案：State 用官方支持的 pydantic.BaseModel（Define state — Pydantic），约束用 Pydantic v2 Field / 语义类型，不要自研校验层。风险：校验只发生在首节点输入；create_agent 不支持 Pydantic State；递归校验比 TypedDict / dataclass 慢。

图 1　Annotated 双通道：左抽屉给 Pydantic 校验，右抽屉给 LangGraph reducer（对应 §5）

1. 定位

维度 内容
角色 带运行时校验的 State schema（入图闸门）
输入 → 输出 invoke(dict | BaseModel) → 校验通过后跑图 → 返回 dict
核心 API BaseModel、Field、Annotated[T, reducer | Field(...)]
依赖 LangChain 对话字段用 AnyMessage + add_messages；与 ChatModel 参数无关

出现背景：TypedDict 够编排、不够守边界。对外 API / 工单入参需要 UUID、金额、枚举时，用 Pydantic 把非法输入挡在第一拍之前。

2. 图拓扑

节点表

节点名 职责 读 State 写 State
validate_order 读金额与明细 amount, items status, logs, messages
confirm 确认 status status, logs

边表

源 目标 类型
START validate_order 固定
validate_order confirm 固定
confirm END 固定

图 2　Pydantic 闸门在 START 之后、首节点之前；`amount=-1` 进 ValidationError 桶，图不调度（对应 §3～§4）

3. invoke 生命周期

1
2
3
4
5
6
1. invoke 传入 dict（或 BaseModel 实例）
2. 用 AgentState 校验「首节点输入」：负 amount / 脏字段 / 坏 UUID → ValidationError，图不跑
3. validate_order 读已校验对象，返回 partial update
4. 合并时读 Annotated 上的 reducer（logs 走 operator.add，messages 走 add_messages；标量 last-write-wins）
5. confirm 再写 status / logs → END
6. 返回 dict（不是 BaseModel）；需要模型时再 AgentState(**result)

后续节点的返回值默认不再走一遍 Pydantic 全量校验——闸门只守入口。

4. 原理

4.1 一次 invoke 的数据链路

  1. StateGraph(AgentState) 把 BaseModel 子类登记为 state_schema；compile 后得到 CompiledGraph。
  2. invoke(input) 用该 schema 构造/校验首节点输入。Field(ge=0)、UUID、嵌套 LineItem、extra="forbid" 都在这一步生效。失败抛 校验错误（ValidationError），节点函数不会被调用。
  3. 节点入参是已校验对象，可用 state.amount（部分版本也兼容 state["amount"]）。返回 partial dict，不要返回全量 State。
  4. 每个字段独立合并：无 Annotated → 覆盖；Annotated[list, operator.add] → 拼接；Annotated[list[AnyMessage], add_messages] → 按消息 id 去重追加。Overwrite(value) 可绕过 reducer 整表替换。
  5. 下一拍节点看到的是合并后的 channels。官方限制：从第二拍起不再对写入做 Pydantic 运行时校验；validate_order 若返回 {"amount": -1}，ge=0 未必再拦。
  6. 到达 END 后返回 dict。嵌套模型在输出里通常是 dict；要属性访问需 AgentState(**result)。

少了第 2 步 → 与 TypedDict 无异，脏数据进图。
少了第 4 步的 reducer → logs / messages 被后写节点整表盖掉。
把第 2 步误当成「每步都校验」→ 节点写坏字段会静默进终态。

4.2 官方限制（挂在链路上）

限制 落在哪一步 后果
只校验首节点输入 步骤 2 后续节点写入不受 Field 约束
输出不是 BaseModel 步骤 6 type(result) 是 dict
报错不标节点名 步骤 2 堆栈只说 schema，不说哪个 node
递归校验偏慢 步骤 2 深嵌套 / 热路径更宜 dataclass
field_validator 改写可能不落盘 步骤 2 能 raise，return 规范化值不一定写回 channels
create_agent 不支持 Pydantic State 构图 预构建 Agent 用 TypedDict

5. 功能图谱：Annotated × Pydantic

Annotated[T, *metadata] 是同一语法、两套读者：

元数据形态 谁读 何时读 典型用途
可调用对象（operator.add、add_messages、自定义 (left, right)） LangGraph 节点返回后合并 多写字段
Field(...) / Gt / MinLen Pydantic 入图校验 边界、长度、正则
BeforeValidator / AfterValidator / WrapValidator Pydantic 入图校验 规范化、跨格式解析
UUID、HttpUrl、Literal、嵌套 BaseModel（写在 T 上，不是 metadata） Pydantic 入图校验 语义类型 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from typing import Annotated
from pydantic import Field
import operator

# 只校验
sku: Annotated[str, Field(min_length=3, pattern=r"^[A-Z0-9-]+$")]

# 只合并
logs: Annotated[list[str], operator.add] = Field(default_factory=list)

# 外层 reducer + 内层元素约束（列表本身也可再 Field(max_length=...)）
tags: Annotated[
    list[Annotated[str, Field(min_length=1, max_length=32)]],
    operator.add,
] = Field(default_factory=list, max_length=20)

T 给类型检查器看；metadata 按框架分流。reducer 必须是 callable，不能写成字符串。

5.1 Field 声明参数（非约束）

参数 类型 / 默认 作用 AgentState 用法
default 任意 缺省值 status="pending"
default_factory () -> T 每次新建，避免共享可变默认 logs / items / messages 必须用这个，禁止 =[]
description str 文档与 JSON Schema 给 API 层生成 OpenAPI
examples list Schema 示例 文档
alias / validation_alias str 入参别名 兼容外部字段名 userId
frozen bool 校验后不可改该字段 入图后不应变的 user_id
deprecated bool / str 标记弃用 过渡字段
exclude bool 序列化时丢掉 内部缓存不对外

默认值写在注解右侧：status: str = "pending" 与 Field(default="pending") 等价；可变容器只用 default_factory。

5.2 数值约束

适用 int / float / Decimal。JSON Schema 对应 minimum / exclusiveMinimum 等。

参数 语义 合法例 非法例 Agent 场景
gt $x > n$ gt=0 → 0.01 0 单价必须为正
ge $x \ge n$ ge=0 → 0 -1 金额、计数
lt $x < n$ lt=1 → 0.99 1 概率开区间
le $x \le n$ le=1 → 1 1.01 置信度闭区间
multiple_of $x$ 是 $n$ 的倍数 0.01 → 9.90 9.901 货币分位
allow_inf_nan 是否允许 inf / nan 默认对 float 常为 True 视配置 金额建议 False
max_digits / decimal_places Decimal 总位数 / 小数位 max_digits=5, decimal_places=2 → 123.45 1234.56 精确金额 
1
2
3
amount: float = Field(ge=0, le=1_000_000, allow_inf_nan=False)
confidence: float = Field(ge=0, le=1)
qty: int = Field(ge=1, le=99)

5.3 字符串与集合约束

参数 适用 语义 Agent 场景
min_length str / list / dict / set 最小长度 / 元素数 非空备注、至少 1 条明细
max_length 同上 上限 防超长 prompt / 防日志膨胀
pattern str 整串匹配正则 SKU、工单号
fail_fast 序列 遇第一个非法元素即停 大列表入参
strict 任意 关闭强制转换 "42" 不再变成 int 
1
2
3
sku: str = Field(min_length=3, max_length=32, pattern=r"^[A-Z0-9-]+$")
note: str | None = Field(default=None, max_length=200)
items: list[LineItem] = Field(default_factory=list, max_length=20)

pattern 是整值匹配；不要指望它做「包含某词」。需要子串规则用 AfterValidator 或 field_validator。

5.4 语义类型（写在 T 上）

这些类型自带解析与校验，比手写 pattern 稳。

类型 来源 入参形态 失败条件 注意
UUID uuid.UUID UUID 对象或标准字符串 非 UUID 文本 推荐 user_id
datetime / date 标准库 ISO 字符串或对象 无法解析 时区：要感知时区用 AwareDatetime
Decimal decimal 数字或数字字符串 非数字 金额精确计算
Literal["web","app"] typing 恰好这些值 "WEB"（除非先规范化） 渠道、路由键
Enum / StrEnum enum 成员名或值 未定义成员 订单状态
HttpUrl / AnyUrl pydantic URL 字符串 缺 scheme 等 回调地址
EmailStr pydantic 邮箱字符串 格式非法 需 pip install email-validator 或 pydantic[email]
IPvAnyAddress pydantic IP 字符串 非法 IP 来源地址
SecretStr pydantic 字符串 — 打印被掩码；checkpoint 仍可能落明文，密钥不要放 State
AnyMessage LangChain Message 或可反序列化 dict 缺 type 不要用 BaseMessage 当序列化字段

Pydantic 默认会做强制转换：{"number": "42", "flag": "true"} → int / bool。需要「字符串就是字符串」时设 strict=True 或 model_config = ConfigDict(strict=True)。

5.5 复合与嵌套

写法 校验行为 Agent 场景
list[LineItem] 每个元素递归成 LineItem 订单行
dict[str, int] 键必须是 str，值必须是 int 计数器表
LineItem 嵌套模型 入参 dict → 模型；缺字段失败 结构化子对象
T | None None 合法；有值则按 T 可选回调、可选备注
Annotated[int, Field(gt=0)] | None None 或正整数 可选正数
判别联合（discriminator） 按字段选子模型 多形态工具结果

嵌套模型与 list[Model] 会触发递归校验（官方点名偏慢）。热路径、超大明细用 TypedDict / dataclass，只在 input_schema 用 Pydantic。

5.6 自定义校验器

API 能做什么 在 LangGraph 里是否可靠
Field 约束 边界 / 长度 / 正则 可靠（入图时）
@field_validator(..., mode="after") 里 raise 拒绝非法值 可靠（入图时）
同一 validator 里 return 改写值（如首字母大写） 规范化 不可靠：官方实现常「只验不写回」，改写可能丢
@model_validator(mode="after") 跨字段 如 amount == sum(qty * price) raise 可用；改写同样可能不落盘
AfterValidator(fn) / BeforeValidator(fn) 挂在 Annotated 上的函数式校验 与 field_validator 同类限制

规范化（trim、大小写、单位换算）放在节点里显式返回，不要指望 validator 的 return 写进 channels。

1
2
3
4
5
6
7
8
9
10
11
from pydantic import field_validator

class AgentState(BaseModel):
    amount: float = Field(ge=0)
    items: list[LineItem] = Field(default_factory=list)

    @field_validator("items")
    @classmethod
    def items_match_amount(cls, items: list[LineItem], info):
        # 跨字段更宜 model_validator；这里只演示 raise
        return items

5.7 ConfigDict（模型级）

项 常用值 作用 指导
extra "ignore" / "forbid" / "allow" 未知键 对外入口用 "forbid" 防脏字段
strict False / True 是否强制转换 API 字符串入参保持默认；内部精确类型用 True
str_strip_whitespace False / True 入参去首尾空白 工单号、SKU 建议开
frozen False / True 整模型不可变 与节点「返回新 dict」相容；少用
validate_assignment False / True 属性赋值再校验 图内部少做 state.x = ...，无效 
1
2
class AgentState(BaseModel):
    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)

5.8 三套 schema 怎么选

方案 运行时校验 默认值 性能 适用
TypedDict + Annotated 无 无（键常需给齐） 最快 图内部、热路径
dataclass + Annotated 无 有 快 只要默认值
BaseModel + Annotated 入图有 有 慢 对外工单、金额、UUID
input_schema=Pydantic，内部 TypedDict 只校验图输入 内部自定 折中 生产常用：闸门与编排分离

StateGraph(OverallState, input_schema=OrderIn, output_schema=OrderOut)：输入用 Pydantic 守边界，内部 channels 仍可用 TypedDict + reducer；输出按 output_schema 裁剪。预构建 create_agent 只能走 TypedDict。

6. 最小可运行示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
import operator
from typing import Annotated, Literal
from uuid import UUID

from langchain_core.messages import AIMessage, AnyMessage, HumanMessage
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
from pydantic import BaseModel, ConfigDict, Field, ValidationError


class LineItem(BaseModel):
    sku: str = Field(min_length=3, max_length=32, pattern=r"^[A-Z0-9-]+$")
    qty: int = Field(ge=1, le=99)
    unit_price: float = Field(gt=0, allow_inf_nan=False)


class AgentState(BaseModel):
    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)

    user_id: UUID
    amount: float = Field(ge=0, le=1_000_000, allow_inf_nan=False)
    status: Literal["pending", "validated", "confirmed"] = "pending"
    items: list[LineItem] = Field(default_factory=list, max_length=20)
    logs: Annotated[list[str], operator.add] = Field(default_factory=list)
    messages: Annotated[list[AnyMessage], add_messages] = Field(default_factory=list)


def validate_order(state: AgentState) -> dict:
    # 拓扑对应 §2 节点 validate_order
    n = len(state.items)
    return {
        "status": "validated",
        "logs": [f"amount={state.amount} items={n}"],
        "messages": [AIMessage(content="订单已校验")],
    }


def confirm(state: AgentState) -> dict:
    # 拓扑对应 §2 节点 confirm
    return {"status": "confirmed", "logs": ["confirmed"]}


builder = StateGraph(AgentState)
builder.add_node("validate_order", validate_order)
builder.add_node("confirm", confirm)
builder.add_edge(START, "validate_order")
builder.add_edge("validate_order", "confirm")
builder.add_edge("confirm", END)
graph = builder.compile()

ok = graph.invoke(
    {
        "user_id": "550e8400-e29b-41d4-a716-446655440000",
        "amount": 99.0,
        "items": [{"sku": "SKU-01", "qty": 2, "unit_price": 49.5}],
        "messages": [HumanMessage(content="下单")],
    }
)
print(ok["status"], ok["logs"], type(ok))  # confirmed [...] <class 'dict'>

for bad in (
    {"user_id": "not-a-uuid", "amount": 1},
    {"user_id": "550e8400-e29b-41d4-a716-446655440000", "amount": -1},
    {
        "user_id": "550e8400-e29b-41d4-a716-446655440000",
        "amount": 1,
        "items": [{"sku": "bad sku", "qty": 1, "unit_price": 1}],
    },
    {
        "user_id": "550e8400-e29b-41d4-a716-446655440000",
        "amount": 1,
        "unknown": True,
    },
):
    try:
        graph.invoke(bad)
    except ValidationError as e:
        print("拦截", e.error_count(), e.errors()[0]["type"])

7. 执行追踪

合法输入 amount=99、1 条明细、1 条 HumanMessage：

步骤 status logs messages 条数
入图校验后 pending [] 1（Human）
validate_order validated ["amount=99.0 items=1"] 2（+ AI）
confirm confirmed ["amount=99.0 items=1", "confirmed"] 2

amount=-1 / 坏 UUID / 坏 SKU / 多余键：停在步骤 2，两节点都不跑。

重要配置参数

参数（API 名） 类型 / 默认值 功能说明 作用与影响 参考起点 / 常用范围 配置指导
Field(ge/gt/le/lt) 数值约束 / 无 入图时检查标量边界 过严：合法单被拒；过松：负金额进图 金额 ge=0；置信度 [0,1] 只守入口，节点内仍可能写坏
Field(min_length/max_length/pattern) 序列/字符串约束 长度与整串正则 max_length 过小截断业务；无上限则日志膨胀 SKU 正则、明细 max_length=20 pattern 不支持「包含」语义
default_factory () -> T 为可变字段提供独立默认实例 写成 =[] 会跨 invoke 共享同一 list logs/items/messages 可变默认只用 factory
Annotated[T, reducer] 类型 + callable 告诉 LangGraph 如何合并该字段 不加则后写覆盖；并行写 list 必丢数据 operator.add、add_messages reducer 必须是函数
Annotated[T, Field(...)] 类型 + 约束 把 Field 嵌进注解，可叠 reducer 与右侧 = Field(...) 等价；内层可约束 list 元素 元素级 gt=0 外层挂 reducer，内层挂约束
ConfigDict(extra=...) "ignore" 默认 未知键策略 "forbid" 多一个键就失败；"allow" 脏字段进 state 对外 "forbid" 内部调试可临时 "ignore"
ConfigDict(strict=...) False 关闭 "42"→int 等强制转换 True 时 API 字符串入参大量失败 默认 False 适配 HTTP JSON 要精确类型再开
input_schema type / 同 state StateGraph(..., input_schema=) 单独校验入参 与内部 State 解耦；输出仍按 output_schema 裁剪 对外 Pydantic、对内 TypedDict create_agent 不可用 Pydantic State
AnyMessage 联合消息类型 带 type 判别器，供序列化还原 写成 BaseMessage 过线后可能丢子类信息 messages 字段 与 add_messages 成对
ValidationError 异常 入图校验失败 error_count() / errors()[0]["type"] 给 API 映射 422 边界测试 不要在节点里吞掉

8. 易踩坑

  1. logs: list = []：类级共享可变默认。必须 Field(default_factory=list)。
  2. 以为 Field 能管住每一拍：后续节点返回负 amount 可能写进终态。关键不变量在节点里再断言，或把校验放到 input_schema + 节点显式检查。
  3. field_validator 做规范化：raise 有效，return 改写常不落盘。去空格、改大小写写在节点返回值里。
  4. 消息字段用 list[BaseMessage]：过线/checkpoint 还原可能丢具体子类。用 list[AnyMessage] + add_messages。
  5. create_agent 传入 Pydantic State：官方不支持，改 TypedDict。
  6. Pydantic v1 @validator：LangGraph 文档面向 v2，用 field_validator / Field。
  7. 把密钥放进 SecretStr State：打印被掩，checkpoint 仍可能持久化明文。

小结

  • Pydantic State = 入图闸门（类型、边界、嵌套、未知键）+ TypedDict 同款 Annotated reducer。
  • Annotated 一张图：左通道 Field / Validator / 语义类型，右通道 operator.add / add_messages / 自定义 (left, right)；可叠用。
  • 输出是 dict；校验不覆盖后续节点写入。对外严、对内快：input_schema 用 Pydantic，内部用 TypedDict。

参考链接

-------------本文结束感谢您的阅读-------------