第 4 章 State 与 Channel

4.1 State 的定义

State 是图执行期间所有节点共享的全局数据容器。节点函数的第一个参数是当前 State 的只读快照，所有对 State 的修改必须通过返回一个增量字典来提交——运行时在超步末的 Update 阶段按字段规则合并到真实状态。State 本身不要求是特定基类的实例，只要能被编译器解析字段类型即可。

4.2 State 的三种定义方式

TypedDict（推荐）：标准库 typing.TypedDict，零运行时开销、类型提示友好，是绝大多数场景的首选。

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages

class ResearchState(TypedDict):
    messages: Annotated[list, add_messages]  # 带 reducer
    query: str                               # 普通字段（覆盖写）
    papers: list[str]                        # 普通字段（注意：多节点写需 reducer）

Pydantic BaseModel（运行时校验）：适合 State 结构严格、需要在节点写入时做类型校验或 JSON 序列化的场景。

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

class ResearchState(BaseModel):
    messages: Annotated[list, add_messages] = Field(default_factory=list)
    query: str = ""
    papers: list[str] = Field(default_factory=list)

注意：Pydantic 模型作为 State Schema 时，字段仍需用 Annotated[..., reducer] 声明合并规则，否则同超步多节点写入同字段会报错。

dataclass（不推荐）：标准库 dataclass 可被识别，但在泛型解析与 Pydantic v2 兼容性上存在历史问题，新代码不建议使用。

4.3 字段分类与 Channel 概念

State 字段分两类：普通字段（默认覆盖写）和带 reducer 的字段（合并写）。在编译阶段，StateGraph 会把每一个字段包装成一个 Channel（通道） 对象。Channel 封装了两件事："当前值"与"合并规则"。运行时在 Update 阶段对每个 Channel 独立处理待提交写入。

LangGraph 内置的核心 Channel 类型如下：

• LastValue：默认通道。保存最后一次写入的值。同一超步内若有多个节点写同一个 LastValue 通道，运行时会抛出 InvalidUpdateError: Can receive only one value per step。适用于"当前阶段"、"最终答案"等在每轮只有一个节点负责写入的字段。
• BinaryOperatorAggregate：带 reducer 的聚合通道。接收一个二元函数 (old, new) -> merged，每次写入时用该函数把新值与当前值合并。常见 reducer 包括 operator.add、max、min，以及 LangGraph 专用的 add_messages。
• Topic：发布-订阅通道，支持 accumulate（是否累积所有消息，默认 False 即每轮清空）与 unique（是否按 key 去重）。用于广播消息队列、事件总线等场景，用户代码一般通过高级 API 间接接触。
• EphemeralValue：运行时内部使用的临时通道，用于保存 __start__ 输入、分支路由结果 branch:to:x 等，用户代码不直接读写。
• DeltaChannel / ContextChannel：面向高级场景的增量通道与上下文通道，前者用于传递 JSON Patch 等增量更新，后者用于跨子图的上下文透传，普通业务流程一般不必直接使用。

4.4 reducer 的语义与陷阱

reducer 决定了一个字段在多节点、多超步写入时的合并行为，使用不当会直接导致数据被覆盖或合并报错。

operator.add 语义：对 list 为拼接（old + new），对 int/float/str 为算术/字符串相加。若字段声明为 Annotated[list, operator.add]，多个节点在同一超步返回的 list 会被按节点完成顺序依次拼接到旧列表后。

add_messages（langgraph.graph.message.add_messages）：专为聊天消息列表设计的 reducer。消息以 BaseMessage（或 dict 形式）为单位，按 id 字段做覆盖更新：若新消息 id 与已有消息相同，则替换旧消息；若不同则追加；配合 RemoveMessage 可实现消息删减。MessagesState 预定义了 messages: Annotated[list[BaseMessage], add_messages]，快速构建对话型 Agent 时可直接继承：

from langgraph.graph import MessagesState

class MyState(MessagesState):
    # 自动拥有 messages 字段
    user_id: str

自定义 reducer 签名为 (old_value, new_value) -> merged_value，必须是纯函数（相同输入必相同输出、无副作用）且可序列化（不能是 lambda 或闭包，否则跨进程/跨平台部署会失败）：

def merge_papers(old: list[str], new: list[str]) -> list[str]:
    """去重合并论文链接。"""
    seen = set(old)
    for p in new:
        if p not in seen:
            old.append(p)
            seen.add(p)
    return old

class State(TypedDict):
    papers: Annotated[list[str], merge_papers]

字段语义选择指南：

业务语义	字段类型	推荐 Channel/Reducer
单值、每轮只有一个节点写（如当前轮次、最终答案）	任意	LastValue（默认，不写 Annotated）
追加列表（如日志、检索结果汇集）	list	operator.add 或自定义去重 reducer
聊天消息	list[BaseMessage]	add_messages
数值累加/取最大	int/float	operator.add、max、min
覆盖更新（如当前路由目标）	str/enum	LastValue
广播事件	自定义	Topic（高级）

4.5 私有状态与输入输出 Schema

State 中可能包含仅内部节点使用、不应暴露给外部调用者的字段（如内部重试计数、中间草稿）。通过 input_schema 与 output_schema 参数可以裁剪外部可见字段：

builder = StateGraph(
    StateSchema=InternalState,
    input_schema=InputSchema,   # 外部 invoke 时只需提供这部分字段
    output_schema=OutputSchema, # 外部 invoke 返回时只暴露这部分字段
)

未在 input_schema 中声明的字段必须在图内部被赋值或在初始 state 中给出默认值，否则首个执行节点读取时会得到 KeyError。

4.6 状态更新的可见性与大 State 问题

可见性规则：节点返回的增量字典在当前超步内只写入待提交缓冲区，同轮其他节点不可见；下一个超步开始时，节点入参 state 才反映上一轮合并后的结果。这与第 3 章 BSP 模型一致。state 对象应视为只读，对其中的 list/dict 做 in-place 修改（如 state["papers"].append(x)）不会被运行时捕获，等于无效写入。

大 State 序列化问题：Checkpointer 在每个超步末会序列化整个 State 写入后端。若字段中塞入大对象（完整文档文本、base64 图片、大型 embedding 矩阵），会导致：快照体积膨胀、数据库 IO 飙升、resume 时反序列化耗时。此类数据应放入 Store（LangGraph 的跨线程键值存储，第 9 章详述）或外部对象存储（S3、Redis 等），State 中只保存引用标识（如文档 ID、URL、对象 key）。

本章小结

State 是节点共享的类型化数据容器，字段在编译期映射为 Channel，Channel 定义合并规则。TypedDict 是推荐定义方式；LastValue 与 BinaryOperatorAggregate（含 operator.add、add_messages、自定义 reducer）是最常用的两类通道；Topic 等高级通道服务于特定场景。reducer 必须是纯函数且可序列化；input/output schema 用于裁剪外部可见字段；大对象不应直接放入 State。

常见错误

1. 忘记给多写入字段加 reducer。现象：两个节点同超步写同一字段时抛 InvalidUpdateError。原因：字段默认是 LastValue 通道，不允许多写入。修复：按合并语义加 Annotated[..., reducer]，或把写入合并到同一个节点。
2. 在节点内直接修改 state 对象。现象：返回后发现变更没生效或只偶发生效。原因：state 是只读快照，in-place 修改不会被提交。修复：返回包含所需字段的增量字典。
3. reducer 中包含副作用或不可序列化对象。现象：使用 Postgres Checkpointer 或多 worker 部署时报 pickle/序列化错误，或 reducer 被多次调用结果不一致。原因：reducer 内写了文件、调用了随机函数、使用了 lambda。修复：将 reducer 改为顶层纯函数，副作用放到节点内。
4. 把大对象直接塞进 State。现象：Checkpoint 表行数不变但单行体积巨大，LangSmith trace 卡顿。原因：大字段每次超步都被序列化一次。修复：将大字段迁移到 Store 或外部存储，State 只存引用。

生产建议

• 所有 State Schema 统一放在 graph/state.py 中定义，跨文件复用，避免节点本地声明匿名 dict；
• 对聊天消息始终使用 add_messages 与 RemoveMessage，不要用 operator.add 追加消息（否则编辑旧消息无法生效）；
• reducer 模块单独存放并写单元测试，覆盖"空 old 值""重复 id 写入""跨超步累积"等边界；
• 在 State 中为关键字段设计版本号或时间戳字段，便于做幂等与排障；
• 定期巡检 Checkpointer 中快照大小，对超过 100KB 的 State 做字段拆分或外置化。

---