Agent 状态 Schema 设计:TypedDict + Annotated Reducers
Agent 状态 Schema 设计:TypedDict + Annotated Reducers
学习目标
本章要解决什么问题:多个中间件和节点如何安全地读写 Agent 状态,而不会互相覆盖或引入不一致?
读者将学到:
AgentState的 TypedDict 定义Annotated注解与 reducer 函数的组合OmitFromSchema如何实现输入/输出 Schema 分离- 中间件状态字段的自动合并机制
前置知识
本章涉及状态 Schema 的工程实现,建议先阅读:
- StateGraph Agent 循环 — 状态 Schema 概念
项目实践
AgentState 的核心定义
# 伪代码class AgentState(TypedDict): messages: Required[Annotated[list[AnyMessage], add_messages]] jump_to: NotRequired[Annotated[JumpTo | None, EphemeralValue, PrivateStateAttr]] structured_response: NotRequired[Annotated[ResponseT, OmitFromInput]]关键字段:
| 字段 | 类型 | Reducer | 用途 |
|---|---|---|---|
messages | list[AnyMessage] | add_messages | 消息累加器,多个节点的更新合并为追加 |
jump_to | `Literal[“tools”, “model”, “end”] | None` | EphemeralValue:每次消费后自动清除 |
structured_response | ResponseT | 覆盖 | OmitFromInput:不作为输入接收 |
Annotated 注解的组合
LangChain 使用 Python 的 typing.Annotated 将 reducer 函数作为元数据附加到类型注解上:
Annotated[list[AnyMessage], add_messages]# 含义:当多个节点同时更新 messages 字段时,使用 add_messages 函数合并add_messages 是 LangGraph 提供的 reducer,将新消息追加到列表而非覆盖。这保证了消息历史的一致性。
OmitFromSchema 标注
中间件可以定义自己的状态字段,并通过 OmitFromSchema 控制这些字段在输入/输出 Schema 中的可见性:
# 伪代码OmitFromInput = OmitFromSchema(input=True, output=False) # 仅输出可见OmitFromOutput = OmitFromSchema(input=False, output=True) # 仅输入可见PrivateStateAttr = OmitFromSchema(input=True, output=True) # 完全内部使用在 AgentState 中的应用:
jump_to使用PrivateStateAttr:仅中间件内部使用,不作为 Agent 的输入或输出structured_response使用OmitFromInput:模型输出时设置,但用户不需要在输入中提供
中间件状态 Schema 的自动合并
当多个中间件都定义了自己的 state_schema 时,create_agent() 自动合并:
# 伪代码state_schemas = [m.state_schema for m in middleware] + [base_state]resolved_state_schema = _resolve_schemas(state_schemas)合并规则:
- 后面的 Schema 覆盖前面的同名字段
base_state(AgentState)最后合并,确保核心字段不被覆盖OmitFromSchema注解在输入/输出 Schema 解析时被尊重
问题与规避
字段名称冲突
问题:两个中间件定义了同名字段但类型不同。
对策:合并时后声明的覆盖先声明的。中间件应使用前缀命名(如 retry_attempt_count、cache_hit_count)避免冲突。
Reducer 的副作用
问题:自定义 reducer 函数如果有副作用(如修改外部状态),可能在每次状态更新时触发意外行为。
对策:Reducer 必须是纯函数——只接收输入并返回合并后的值,不修改外部状态。
设计取舍
TypedDict vs Pydantic BaseModel
| 方案 | 优势 | 代价 |
|---|---|---|
| TypedDict | 轻量、无运行时开销、纯类型检查 | 无运行时校验、需要类型检查工具 |
| Pydantic BaseModel | 运行时校验、序列化方便 | 额外依赖、实例化开销 |
LangChain 选择 TypedDict,因为状态 Schema 主要用于编译时类型检查和 LangGraph 的状态归约逻辑,不需要运行时校验。
消息累加 vs 覆盖
| 方案 | 优势 | 代价 |
|---|---|---|
| add_messages(累加) | 保留完整对话历史、多节点更新不冲突 | 状态可能变大 |
| 覆盖 | 状态大小可控 | 多节点更新时后者覆盖前者 |
LangChain 选择累加,因为 Agent 的对话历史是核心状态,丢失消息会导致上下文不一致。