AI Agent 开发（07）：State 设计详解

第 6 篇用 StateGraph 把"模型循环"变成了图：节点是函数，边是路由。但决定一个 Agent 能长多大的，往往是那个容易被一笔带过的 State——所有节点共享它、读写它，图每走一步它就变一次模样。State 设计得草率，循环计数会丢、消息会重复、并行分支会直接报错；设计得讲究，中断恢复、时间旅行、多 Agent 协作才有落点。

本篇是《AI Agent 开发》系列第 7 篇，代码基于 langgraph 1.2.x（langchain-core 1.6.x），环境沿用第 2 篇的 uv 项目。全文按"心智模型 → 覆盖陷阱 → Reducer 机制 → 定义方式选型 → 私有 State → 设计原则"展开，看完你能对自己 Agent 的每个字段做出有依据的设计决策。

1、State 是共享内存

先立结论：State 是随图流转的共享内存，不是数据库。它本质上是一份结构由你定义的 Python 字典：图内所有节点都能读全量、写局部，数据沿边整体传递；一次 invoke 结束，它的生命周期随之结束。没有查询语句、没有索引、也没有默认的持久化——持久化是 checkpointer 的职责，不是 State 的。

节点函数的约定是"收全量、返增量"：入参是当前完整 State，返回值只包含本次要修改的字段，未提及的字段原样传给下游。最小示例如下：

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class QAState(TypedDict):
    question: str    # 用户问题
    answer: str      # 最终回答
    docs: list[str]  # 检索到的资料

def retrieve(state: QAState) -> dict:
    docs = [f"关于 {state['question']} 的片段 A", "片段 B"]
    return {"docs": docs}            # 只写要改的字段

def generate(state: QAState) -> dict:
    ctx = "\n".join(state["docs"])   # 能读到全量 State
    return {"answer": f"综合资料作答：{ctx}"}

builder = StateGraph(QAState)
builder.add_node("retrieve", retrieve)
builder.add_node("generate", generate)
builder.add_edge(START, "retrieve")
builder.add_edge("retrieve", "generate")
builder.add_edge("generate", END)
print(builder.compile().invoke({"question": "LangGraph 是什么？"}))

retrieve 只返回 docs，question 不必重复携带；generate 却能同时读到两者——这就是"整体传递、局部更新"。再记一条性质：compile 时接上 checkpointer（本地调试用 InMemorySaver，来自 langgraph.checkpoint.memory），每个超步（superstep）结束都会生成一份不可变快照，后续的中断恢复与时间旅行全部踩在这些快照上，第 12 篇展开。

心智模型：State 是块黑板，节点轮流上前擦写，checkpointer 负责给每一步拍照。

2、同名字段覆盖陷阱

默认更新策略是 LastValue：后写覆盖先写。串行图里它通常正合心意——下游节点"改写"上游结果，覆盖就是语义。但只要两个字段的更新意图是"各自贡献一份"，这个默认值就从便利变成陷阱。

看一个对照实验，两份 State 只差一行 Annotated，其余完全相同：

import operator
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class Overwrite(TypedDict):
    count: int                          # 默认：覆盖

class Accumulate(TypedDict):
    count: Annotated[int, operator.add] # 声明 reducer：累加

def node_a(state) -> dict:
    return {"count": 1}

def node_b(state) -> dict:
    return {"count": 1}

for schema in (Overwrite, Accumulate):
    builder = StateGraph(schema)
    builder.add_node("a", node_a)
    builder.add_node("b", node_b)
    builder.add_edge(START, "a")
    builder.add_edge("a", "b")
    builder.add_edge("b", END)
    print(schema.__name__, builder.compile().invoke({"count": 0}))

输出如下：

Overwrite {'count': 1}
Accumulate {'count': 2}

两个节点各 +1 的意图，在默认覆盖下 node_b 的 1 直接顶掉了 node_a 的 1，最终计数停在 1——没有报错、没有日志，是最难排查的那类静默 bug。挂上 operator.add 后，LangGraph 在每次写入时执行 count = count + 1，结果才是 2。

覆盖还有个更凶的变体：如果 a、b 是从同一节点并行分出的两支（第 8 篇），两个分支在同一超步写同一个无 reducer 字段，LangGraph 无法裁决谁覆盖谁，直接抛 InvalidUpdateError。所以 reducer 不只是"想要累加时才加的糖"，它是并行写入同一字段的前提。

3、Reducer 合并机制

3.1、Annotated 与合并函数

Annotated[T, reducer] 的第二个参数是一个普通二元函数：节点返回 {"count": 1} 时，LangGraph 对每个被更新的字段执行 新值 = reducer(旧值, 1)，再用新值合成下一份 State；没有 Annotated 的字段等价于走 reducer = lambda 旧, 新: 新，即覆盖。reducer 只在字段被写入时触发，节点没提的字段原封不动。LangGraph 内部把每个字段实现为一个通道（channel）：默认是 LastValue 通道，挂了 reducer 就变成二元聚合通道，机制详见 Graph API 文档。

Annotated[int, operator.add] 之外，更常见的是列表累加：Annotated[list[str], operator.add] 用 + 拼接列表。多分支各自检索、在汇合点自动合并结果，第 8 篇的分支合并模式就建在它上面。类型与函数可以自由组合，只要满足"接收两个同型参数、返回同型结果"的签名。

3.2、add_messages 消息通道

messages 是 Agent State 里最特殊的字段：对话要追加，但模型重写、人工修订时又要能替换旧消息——追加与替换并存，没有任何内置函数能兼任，于是 LangGraph 为它预置了专门的 reducer add_messages。完整演示如下：

from typing import Annotated
from typing_extensions import TypedDict
from langchain_core.messages import AIMessage, HumanMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

class ChatState(TypedDict):
    messages: Annotated[list, add_messages]

def draft(state: ChatState) -> dict:
    return {"messages": [AIMessage(content="草稿回复", id="ai-1")]}

def revise(state: ChatState) -> dict:
    return {"messages": [AIMessage(content="修订回复", id="ai-1")]}

builder = StateGraph(ChatState)
builder.add_node("draft", draft)
builder.add_node("revise", revise)
builder.add_edge(START, "draft")
builder.add_edge("draft", "revise")
builder.add_edge("revise", END)
result = builder.compile().invoke(
    {"messages": [HumanMessage(content="在吗？", id="human-1")]}
)
for msg in result["messages"]:
    print(msg.id, msg.content)

输出如下：

human-1 在吗？
ai-1 修订回复

draft 与 revise 写的是同一个 id="ai-1"，最终列表里只有一条 AI 消息且内容是修订版。add_messages 的规则共三条：无 id 或新 id 的消息追加到尾部；id 与已有消息相同的原位替换；RemoveMessage 按 id 删除，id=REMOVE_ALL_MESSAGES 一次清空（这几个名字都从 langgraph.graph.message 导入）。

两个使用要点。其一，不显式给 id 时 LangChain 会自动生成随机 id，等于永远走追加分支——想替换就必须自己持有并复用 id。其二，这套"按 id 去重更新"正是流式重写上一条消息、人机协同（HITL）人工修订（第 11 篇）、上下文裁剪（第 23 篇）的地基。图省事可以直接 from langgraph.graph import MessagesState 用官方预置定义。细节见 add_messages API 参考。

4、TypedDict 与 Pydantic

定义 State 的容器有两派：前面一直用的 TypedDict，以及 Pydantic BaseModel。换成 Pydantic 的写法如下：

from typing import Annotated
from pydantic import BaseModel, Field
from langgraph.graph.message import add_messages

class AgentState(BaseModel):
    messages: Annotated[list, add_messages] = Field(
        default_factory=list, description="对话消息")
    retries: int = Field(default=0, ge=0, description="重试计数")
    max_retries: int = Field(default=3, ge=1)

结构几乎没变，reducer 照样用 Annotated 挂，差异全在运行时：节点返回 {"retries": "三次"} 这类脏值时，LangGraph 按模型校验写入值，直接抛 ValidationError，错误数据进不了 State；Field(default=...) 让字段天生带默认值，invoke 不传也能跑。代价是每次更新多一层校验开销。两派对比：

维度	TypedDict	Pydantic BaseModel
运行时校验	无，注解仅静态检查	写入即校验，脏值抛错
字段默认值	需调用方显式传入	Field(default=...) 声明
IDE 提示	基础补全	字段、描述、类型全量提示
运行时开销	零	每次更新多一层校验

选型建议：一次性脚本与教程 demo 用 TypedDict，零负担；生产项目与多人协作的 Agent 用 Pydantic——校验把脏数据挡在入口，默认值让 State 平滑演进（第 6 节回收这个伏笔）。字段约束语法见 Pydantic 字段文档。

5、私有 State 隐藏内部字段

State 一旦混杂"对外要暴露的"与"纯内部的"两类字段，麻烦就来了：调用方依赖了 scratchpad，以后想重构内部实现就得小心翼翼。LangGraph 的解法是给 StateGraph 加两张进出口清单——input 与 output schema：

import operator
from typing import Annotated
from typing_extensions import TypedDict
from langchain_core.messages import AIMessage, HumanMessage
from langgraph.graph.message import add_messages

class ChatIO(TypedDict):
    """对外接口：进出只认消息列表。"""
    messages: Annotated[list, add_messages]

class AgentState(ChatIO):
    """内部全量 State：多出的字段图外不可见。"""
    scratchpad: str                          # 中间推理笔记
    tool_calls: Annotated[int, operator.add] # 工具调用计数

def think(state: AgentState) -> dict:
    return {"scratchpad": "先拆解问题，再决定检索策略",
            "tool_calls": 1}

def reply(state: AgentState) -> dict:
    return {"messages": [AIMessage(
        content=f"（基于笔记作答：{state['scratchpad']}）")]}

ChatIO 是对外契约，AgentState 继承它并追加两个私有字段；节点一律按 AgentState 读写。组装与验证如下：

from langgraph.graph import StateGraph, START, END

builder = StateGraph(AgentState, input=ChatIO, output=ChatIO)
builder.add_node("think", think)
builder.add_node("reply", reply)
builder.add_edge(START, "think")
builder.add_edge("think", "reply")
builder.add_edge("reply", END)
graph = builder.compile()

result = graph.invoke({"messages": [
    HumanMessage(content="查一下 LangGraph 最新版本", id="q-1")
]})
print(result)  # 只有 messages 键，scratchpad 与 tool_calls 不在其中

invoke 的返回值只剩 messages——scratchpad 与 tool_calls 依旧在图内正常流转，但出了图就不可见。要说清一个细节：input/output schema 收窄的是图的进出口，不是节点的可见性，节点拿到的始终是全量 AgentState，别指望用它做字段级权限隔离。它的真实收益是对外接口稳定：内部加多少临时字段、计数器，调用方与下游测试都不受打扰；子图作为节点嵌入父图时（第 24 篇），窄接口也让两边的 State 互不污染。更多用法见 Use the Graph API 指南。

6、State 设计三原则

原则一，字段最小化。只放"跨节点共享"的数据；只在单个节点里用的中间变量，留在函数局部，别上 State。字段每多一个，读代码的人就要多问一句"谁写它、谁读它"。

原则二，一个字段一种合并语义。累加语义配 operator.add，消息语义配 add_messages，"最新值生效"的快照语义用默认覆盖。反模式是一个字段时而想累加、时而想覆盖——那说明它其实是两个字段，拆开。

原则三，演进兼容。新增字段务必给默认值（Pydantic 的 Field(default=...) 或入口节点兜底写入），保证接了 checkpointer 的旧线程在新代码上仍能加载；删字段则要确认旧快照里的数据可以一起丢。配套还有一条体积纪律：checkpointer 对每一步全量快照，把大 JSON 塞进 State 等于把快照成本乘以步数。

三条原则收成一句话：State 的每个字段，都要说清"谁写、谁读、怎么合"。想不清楚的字段，先别加。

7、总结

回望全文：State 是随图流转的共享内存，节点"收全量、返增量"，checkpointer 为每个超步拍照；默认覆盖既制造静默丢数也触发并行报错，Annotated 挂 reducer 才能声明合并语义；add_messages 以"追加 + 按 id 替换"统治消息通道；Pydantic 带来校验与默认值；input/output schema 把内部字段关在图里。

留一个练习：为"研究任务 Agent"设计 State，要求三个字段——messages 存对话与工具结果，todos 存待办子任务，done_count 记已完成数量。

• messages 用 add_messages，done_count 用 Annotated[int, operator.add]
• 思考题：todos 配 operator.add 还是默认覆盖？提示——完成一项要移出列表，"只进不出"打不了勾
• 验收：两个节点先后写 done_count，终值是两次之和
• 加分项：用 Pydantic 定义并给全部字段默认值；再用 output schema 藏起内部字段

对于新起的 Agent，建议从最小 State 起步：先 TypedDict 跑通主链路，出现共享字段后逐个补 reducer，生产化时换 Pydantic 并用 input/output schema 一次定好对外边界。下一篇进入控制流：条件边、循环与递归上限，把"决策 → 行动 → 回到决策"的环真正画进图里，届时第 2 节的并行陷阱会再次登场，不妨先想好对策。