状态 State

状态（State）是 LangGraph 一切的核心。节点之间的协作、循环的执行、记忆的保持——全都靠它。本篇彻底讲清楚 State 是什么、怎么定义、怎么进化。

一、状态是什么

State 是图执行过程中在节点之间共享的数据容器——整个图的"工作内存"。

graph LR
    A[节点A] -->|读/写| S[(State)]
    B[节点B] -->|读/写| S
    C[节点C] -->|读/写| S
    S -->|快照+合并| A
    S -->|快照+合并| B
    S -->|快照+合并| C

执行模型：

1. 图启动时，传入初始 state。
2. 调度到某个节点时，把当前 state 的快照传给它。
3. 节点返回一个 dict（要更新的字段子集）。
4. 框架用 reducer 把这个 dict 合并进 state，得到新的 state。
5. 继续调度下一个节点，循环直到 END。

二、为什么需要状态

如果不共享状态，节点之间怎么传数据？最朴素的想法是"节点 A 调节点 B 时直接传参"。但这样：

• 节点之间强耦合（A 必须知道 B 的存在和签名）。
• 没法做循环（A 调 B，B 又调 A？参数怎么传？）。
• 没法做持久化（中间状态保存在哪？）。

引入共享 state 后：

• 节点解耦：节点只读写 state，不关心"上一个是谁、下一个是谁"。
• 支持循环：状态在循环中被反复读写，自然累积。
• 支持记忆：状态可被 checkpointer 持久化，断点续跑。

对比 Java 思路：这类似一个"会话上下文（Context）"对象，所有处理器（Processor）都读写它，而不是互相直接调用。

三、两种定义方式

方式一：TypedDict（推荐入门用）

TypedDict 是 Python typing 模块提供的，本质是"带类型提示的字典"，运行时仍是普通 dict，没有运行时校验。

from typing import TypedDict, Annotated
from operator import add

class State(TypedDict):
    user_query: str                # 用户原始问题
    retrieved_docs: list           # 检索到的文档
    answer: str                    # 最终回答
    step: Annotated[int, add]      # 步数计数，累加
    messages: Annotated[list, add] # 消息历史，追加

优点：轻量、写法简洁、与 LangGraph 内部 dict 模型完美契合。缺点：没有运行时校验。

方式二：Pydantic v2（需要校验时用）

Pydantic 提供运行时类型校验和数据验证，适合对数据严格性要求高的场景。

from pydantic import BaseModel, Field
from typing import Annotated
from operator import add

class State(BaseModel):
    user_query: str = Field(..., description="用户原始问题")
    retrieved_docs: list[str] = Field(default_factory=list)
    answer: str = ""
    step: Annotated[int, add] = 0
    messages: Annotated[list, add] = Field(default_factory=list)

注意：Pydantic v2 用 BaseModel，不是 v1 的写法。装包：pip install pydantic。

维度	TypedDict	Pydantic v2
运行时校验	❌	✅
数据转换	❌	✅（如 str→int）
默认值	字段没默认值时要手动传	支持 Field(default=...)
序列化	dict 即结果	.model_dump()
性能	最快	略慢
学习成本	低	中
适合	简单状态、快速迭代	严格业务、对外接口

本教程默认用 TypedDict，除非特别说明。

四、字段定义与默认更新策略

State 里每个字段有"默认更新策略"：

• 没有 reducer：节点返回的新值覆盖旧值。
• 有 reducer（Annotated[T, reducer_fn]）：用 reducer_fn(old, new) 合并。

class State(TypedDict):
    # 覆盖型：最后写入者获胜
    answer: str
    # 累加型：每次返回的值与旧值相加
    counter: Annotated[int, add]
    # 追加型：每次返回的列表追加到旧列表
    history: Annotated[list, add]

# 一个节点
def some_node(state: State) -> dict:
    return {
        "answer": "新答案",        # 覆盖
        "counter": 1,              # 旧 counter + 1
        "history": ["新事件"],      # 旧 history + ["新事件"]
    }

reducer 详见 Reducer 归约器。

五、Annotated 语义详解

Annotated[T, metadata] 是 PEP 593 引入的，给类型加"元数据"而不改变类型本身。LangGraph 用它来挂 reducer：

from typing import Annotated
from operator import add

# 解释：counter 字段类型是 int，更新时用 add 函数合并
counter: Annotated[int, add] = 0

Annotated[int, add] 在类型检查器眼里仍是 int，但 LangGraph 在编译时会读取第二个元素 add 作为 reducer。

可以挂多个元数据（LangGraph 只看第一个可调用的）：

counter: Annotated[int, add, "一些说明文字"]

六、内置 MessagesState

写对话/智能体时几乎都要存消息历史。LangGraph 提供了内置 MessagesState，已经定义好 messages 字段并配了 add_messages reducer：

from langgraph.graph import MessagesState

# 等价于
class MessagesState(TypedDict):
    messages: Annotated[list, add_messages]

你可以直接用它，也可以继承扩展：

from langgraph.graph import MessagesState

class MyState(MessagesState):
    # 在 messages 之外加自己的字段
    user_id: str
    turn: int

add_messages 不是简单追加——它会按消息 id 去重和更新（同 id 的新消息覆盖旧消息）。这点对工具调用流式更新很关键。详见 Reducer 归约器。

七、状态在节点间流转的机制

sequenceDiagram
    participant U as 用户
    participant G as Graph 引擎
    participant S as State 存储
    participant N as Node

    U->>G: invoke(initial_state)
    G->>S: 写入 initial_state
    G->>N: 传 state 快照
    N->>N: 执行逻辑
    N->>G: 返回 update dict
    G->>S: 用 reducer 合并 update
    G->>N: 传下一个节点的 state 快照
    Note over G,S: ...循环...
    G->>U: 返回最终 state

关键点：

1. 节点收到的是快照：节点函数里读到的 state 是当时的快照，节点内对 state 的修改不会影响其它节点（除非通过返回值）。
2. 返回值是"差量"：节点返回的 dict 只包含要更新的字段，没列出的字段保持不变。
3. reducer 决定合并：reducer 函数 (old_value, new_value) -> merged_value，把旧值和新值合并。
4. 不可变快照 + 合并：这是 LangGraph 状态管理的核心机制，类似 Redux 的 reducer 模式。

八、完整可运行示例

# state_demo.py
from typing import TypedDict, Annotated
from operator import add
from langgraph.graph import StateGraph, START, END


class State(TypedDict):
    topic: str
    summary: str               # 覆盖型
    notes: Annotated[list, add]  # 追加型
    counter: Annotated[int, add] # 累加型


def step1(state: State) -> dict:
    print(f"[step1] 收到 topic={state['topic']}")
    return {
        "summary": f"关于{state['topic']}的初步总结",
        "notes": ["step1 笔记"],
        "counter": 1,
    }


def step2(state: State) -> dict:
    print(f"[step2] 当前 summary={state['summary']}")
    print(f"[step2] 当前 notes={state['notes']}")
    return {
        "summary": "step2 升级版总结",   # 覆盖
        "notes": ["step2 笔记"],         # 追加
        "counter": 10,                    # 加 10
    }


def step3(state: State) -> dict:
    print(f"[step3] 最终 summary={state['summary']}")
    print(f"[step3] notes 长度={len(state['notes'])}, counter={state['counter']}")
    return {"counter": 1}


g = StateGraph(State)
g.add_node("step1", step1)
g.add_node("step2", step2)
g.add_node("step3", step3)
g.add_edge(START, "step1")
g.add_edge("step1", "step2")
g.add_edge("step2", "step3")
g.add_edge("step3", END)
app = g.compile()

result = app.invoke({
    "topic": "学 LangGraph",
    "summary": "",
    "notes": [],
    "counter": 0,
})
print("\n最终 state：")
print(result)

输出：

[step1] 收到 topic=学 LangGraph
[step2] 当前 summary=关于学 LangGraph的初步总结
[step2] 当前 notes=['step1 笔记']
[step3] 最终 summary=step2 升级版总结
[step3] notes 长度=2, counter=11

最终 state：
{'topic': '学 LangGraph', 'summary': 'step2 升级版总结', 'notes': ['step1 笔记', 'step2 笔记'], 'counter': 12}

观察 counter：step1 +1 → 1；step2 +10 → 11；step3 +1 → 12。这就是 reducer 累加的威力。 观察 summary：每次新值覆盖旧值，最终是 step2 的版本（step3 没返回 summary 所以不变）。 观察 notes：两节点都返回了列表，被追加合并。

九、与 LangGraph4j 的 State 对比

维度	LangGraph (Python)	LangGraph4j (Java)
类型载体	TypedDict / Pydantic	类或 Map<String,Object>
reducer 标注	Annotated[T, fn]	@Reducer 注解 + 类继承
默认策略	覆盖	覆盖
内置消息状态	MessagesState	MessagesState
运行时校验	Pydantic 时有	严格类校验

Java 因静态类型系统，State 通常定义成具体类，reducer 用注解；Python 因灵活，State 多用 TypedDict 轻量表达。

十、常见踩坑

1. 忘了给需要累加的字段配 reducer：消息历史只留最后一条，counter 永远是 1——都是因为默认覆盖。
2. 节点返回字段不在 State 里：被静默忽略。先检查 State 定义有没有这个字段。
3. TypedDict 没法设默认值：TypedDict 不支持字段默认值。需要默认值要么用 total=False（所有字段可选）+ 节点里判空，要么改用 Pydantic。
4. Pydantic v1 写法：用 BaseModel 但写成 v1 的 class Config: / .dict()。LangGraph 0.2+ 要求 Pydantic v2，写法是 model_config = ConfigDict(...) / .model_dump()。
5. 节点函数里直接修改入参 state：不该这样——状态是快照，应该通过返回值更新。直接改入参在并发场景下不可靠。
6. Pydantic 与 TypedDict 混用：同一 State 不要一半字段 TypedDict 一半 BaseModel，行为不一致。整选一种。
7. messages 字段没用 add_messages：用了普通 add，结果工具消息的更新/去重失效，对话历史会重复。
8. 初始化时漏传字段：可选字段可以漏，但被节点读取的字段必须有初始值或节点里用 .get() 取默认。

十一、小结

• State 是图执行过程中在节点间共享的数据容器。
• 两种定义方式：TypedDict（轻量、推荐）和 Pydantic v2（带校验）。
• 默认更新策略是覆盖；用 Annotated[T, reducer] 指定合并策略。
• 内置 MessagesState 适合对话/智能体场景。
• 状态流转机制：节点收到快照 → 返回差量 dict → 框架用 reducer 合并。
• 与 LangGraph4j 概念对应，Java 用类 + 注解，Python 用 TypedDict + Annotated。

下一篇讲 节点 Node——读写 State 的执行单元。