跳转到内容

Agent 状态 Schema 设计:TypedDict + Annotated Reducers

Agent 状态 Schema 设计:TypedDict + Annotated Reducers

学习目标

本章要解决什么问题:多个中间件和节点如何安全地读写 Agent 状态,而不会互相覆盖或引入不一致?

读者将学到:

  • AgentState 的 TypedDict 定义
  • Annotated 注解与 reducer 函数的组合
  • OmitFromSchema 如何实现输入/输出 Schema 分离
  • 中间件状态字段的自动合并机制

前置知识

本章涉及状态 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用途
messageslist[AnyMessage]add_messages消息累加器,多个节点的更新合并为追加
jump_to`Literal[“tools”, “model”, “end”]None`EphemeralValue:每次消费后自动清除
structured_responseResponseT覆盖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_stateAgentState)最后合并,确保核心字段不被覆盖
  • OmitFromSchema 注解在输入/输出 Schema 解析时被尊重

问题与规避

字段名称冲突

问题:两个中间件定义了同名字段但类型不同。

对策:合并时后声明的覆盖先声明的。中间件应使用前缀命名(如 retry_attempt_countcache_hit_count)避免冲突。

Reducer 的副作用

问题:自定义 reducer 函数如果有副作用(如修改外部状态),可能在每次状态更新时触发意外行为。

对策:Reducer 必须是纯函数——只接收输入并返回合并后的值,不修改外部状态。

设计取舍

TypedDict vs Pydantic BaseModel

方案优势代价
TypedDict轻量、无运行时开销、纯类型检查无运行时校验、需要类型检查工具
Pydantic BaseModel运行时校验、序列化方便额外依赖、实例化开销

LangChain 选择 TypedDict,因为状态 Schema 主要用于编译时类型检查和 LangGraph 的状态归约逻辑,不需要运行时校验。

消息累加 vs 覆盖

方案优势代价
add_messages(累加)保留完整对话历史、多节点更新不冲突状态可能变大
覆盖状态大小可控多节点更新时后者覆盖前者

LangChain 选择累加,因为 Agent 的对话历史是核心状态,丢失消息会导致上下文不一致。

参考来源