在 LangGraph 中，StateGraph 以 State 贯穿整个 Graph 的执行过程。每个 Node 读取当前 State 完成自己的任务后，通常只返回本次执行产生的部分更新，而不是重新构造完整 State。

LangGraph 会根据这些更新修改当前 State，再将新的 State 传递给后续 Node，推动 Graph 继续执行。

对于普通字段，这种更新方式很好理解。但如果 State 中保存的是消息、日志、搜索结果这类列表数据，新的值应该直接覆盖原来的内容，还是追加到已有数据之后？如果多个 Node 都更新同一个字段，LangGraph 又该如何合并这些结果？

这些问题，本质上都指向同一个机制：State 的更新与合并规则。而在 LangGraph 中，负责定义这套规则的核心机制就是 Reducer。

一、State 的更新规则

先看一个简单的 State：

代码片段Python
from typing_extensions import TypedDict
 
class TaskState(TypedDict):
    topic: str
    status: str
    logs: list[str]

一个 Node 可以只返回自己修改的字段：

代码片段Python
def start_task(state: TaskState):
    return {
        "status": "running"
    }

假设执行前的 State 是：

代码片段Python
{
    "topic": "LangGraph",
    "status": "pending",
    "logs": ["task created"]
}

执行以后得到：

代码片段Python
{
    "topic": "LangGraph",
    "status": "running",
    "logs": ["task created"]
}

Node 返回的并不是新的完整 State，而是一次 Partial State Update。

对于 status，新值直接替换旧值即可：

代码片段Text
pending
   ↓
running

但如果 Node 返回：

代码片段Python
{
    "logs": ["task started"]
}

就会出现两种可能：

代码片段Text
["task created"]
        ↓
["task started"]

或者：

代码片段Text
["task created"]
        +
["task started"]
        ↓
["task created", "task started"]

字段类型是 list[str]，并不能决定采用哪一种更新方式。

LangGraph 需要知道：这个字段收到新值时，旧值和新值应该怎样合并。

这就是 Reducer。

二、Reducer 的合并机制

如果 State 字段没有配置 Reducer，LangGraph 默认采用覆盖方式更新，正如刚才我们已经看到的。

对于消息、日志、搜索结果集合等需要持续累积的数据，则往往需要显式定义 Reducer。

Reducer 可以理解为字段级别的合并函数。

一次 State Update 到来时，它接收：

代码片段Text
当前值
  +
本次更新值

并计算：

代码片段Text
下一步的值

例如希望日志不断追加，可以这样定义：

代码片段Python
import operator
from typing import Annotated
from typing_extensions import TypedDict
 
 
class TaskState(TypedDict):
    logs: Annotated[list[str], operator.add]

其中 Annotated[list[str], operator.add]，它告诉 LangGraph：

代码片段Text
logs 的类型是 list[str]

logs 收到更新时
使用 operator.add 合并

因此：

代码片段Text
["task created"]
+
["step a"]

得到：

代码片段Text
["task created", "step a"]

把前面的 Graph 改成 Reducer 版本：

代码片段Python
import operator
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
 
 
class TaskState(TypedDict):
    logs: Annotated[list[str], operator.add]
 
 
def step_a(state: TaskState):
    return {
        "logs": ["step a"]
    }
 
 
def step_b(state: TaskState):
    return {
        "logs": ["step b"]
    }
 
 
builder = StateGraph(TaskState)
 
builder.add_node("step_a", step_a)
builder.add_node("step_b", step_b)
 
builder.add_edge(START, "step_a")
builder.add_edge("step_a", "step_b")
builder.add_edge("step_b", END)
 
graph = builder.compile()
 
result = graph.invoke(
    {
        "logs": ["task created"]
    }
)
 
print(result)

其运行结果如下：

代码片段Text
{'logs': ['task created', 'step a', 'step b']}

Node 本身仍然只返回：

代码片段Python
{
    "logs": ["step a"]
}

Node 并不知道旧值是什么，也没有自己执行：

代码片段Python
state["logs"] + ["step a"]

合并逻辑由 Reducer 负责。

因此 Node 和 Reducer 职责分别如下：

代码片段Text
Node：产生 State Update
Reducer：决定 Update 怎样合并进 State

四、自定义 Reducer

Reducer 不一定使用 operator.add。

只要能够接收当前值和新值，并返回合并后的结果，就可以作为 Reducer。

例如一个研究 Agent 会从多个步骤产生标签：

代码片段Text
analyze_topic
    ↓
["agent", "langgraph"]

analyze_content
    ↓
["langgraph", "state"]

如果直接追加，会得到：

代码片段Text
["agent", "langgraph", "langgraph", "state"]

如果希望自动去重，可以定义自己的 Reducer：

代码片段Python
from typing import Annotated
from typing_extensions import TypedDict
 
 
def merge_tags(
    current: list[str],
    new: list[str]
) -> list[str]:
    return list(dict.fromkeys(current + new))
 
 
class ResearchState(TypedDict):
    tags: Annotated[list[str], merge_tags]

在 Graph 中使用：

代码片段Python
from langgraph.graph import StateGraph, START, END
 
 
def analyze_topic(state: ResearchState):
    return {
        "tags": ["agent", "langgraph"]
    }
 
 
def analyze_content(state: ResearchState):
    return {
        "tags": ["langgraph", "state"]
    }
 
 
builder = StateGraph(ResearchState)
 
builder.add_node("analyze_topic", analyze_topic)
builder.add_node("analyze_content", analyze_content)
 
builder.add_edge(START, "analyze_topic")
builder.add_edge("analyze_topic", "analyze_content")
builder.add_edge("analyze_content", END)
 
graph = builder.compile()
 
result = graph.invoke(
    {
        "tags": []
    }
)
 
print(result)

Reducer 对两个列表进行合并和去重，运行结果如下：

代码片段Text
{'tags': ['agent', 'langgraph', 'state']}

四、MessagesState 的消息更新

Agent 中最常见的累积型 State 就是消息历史。

例如：

代码片段Text
HumanMessage
AIMessage
ToolMessage
HumanMessage
AIMessage

直觉上，可以使用：

代码片段Python
messages: Annotated[
    list,
    operator.add
]

这样每次返回新的消息：

代码片段Python
return {
    "messages": [new_message]
}

都可以追加到旧列表后面。

但消息和普通日志列表存在一个区别：已有消息有时需要被修改。

假设 State 中已经存在：

代码片段Text
id = msg-1
content = "helo"

后来需要改成：

代码片段Text
id = msg-1
content = "hello"

如果只是列表追加，最终会同时存在：

代码片段Text
msg-1  helo
msg-1  hello

这显然不适合消息状态。

因此 LangGraph 提供了专门的 Reducer：

代码片段Python
from langgraph.graph.message import add_messages

可以这样定义：

代码片段Python
from typing import Annotated
from typing_extensions import TypedDict
from langchain.messages import AnyMessage
from langgraph.graph.message import add_messages
 
 
class ChatState(TypedDict):
    messages: Annotated[
        list[AnyMessage],
        add_messages
    ]

它和普通 Reducer 的使用方式完全一样，只是合并规则针对 Message 做了专门处理。

add_messages 会根据消息 ID 合并消息：

代码片段Text
新的 ID
→ 追加

已有 ID
→ 更新对应消息

因此，它比：

代码片段Python
operator.add

更适合 Agent 的消息历史。

六、MessagesState 的使用

由于消息 State 非常常见，LangGraph 已经提供了预定义的：

代码片段Python
from langgraph.graph import MessagesState

可以直接用来创建 Graph：

代码片段Python
builder = StateGraph(MessagesState)

可以把它理解成已经预先配置好了：

代码片段Python
messages: Annotated[
    list[AnyMessage],
    add_messages
]

来看一个简单例子：

代码片段Python
from langchain.messages import AIMessage
from langgraph.graph import (
    MessagesState,
    StateGraph,
    START,
    END,
)
 
 
def reply(state: MessagesState):
    last_message = state["messages"][-1]
 
    return {
        "messages": [
            AIMessage(
                content=f"收到：{last_message.content}"
            )
        ]
    }
 
 
builder = StateGraph(MessagesState)
 
builder.add_node("reply", reply)
 
builder.add_edge(START, "reply")
builder.add_edge("reply", END)
 
graph = builder.compile()
 
result = graph.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "介绍一下 Reducer",
            }
        ]
    }
)
 
for message in result["messages"]:
    print(
        type(message).__name__,
        message.content
    )

运行结果如下：

代码片段Text
HumanMessage 介绍一下 Reducer
AIMessage 收到：介绍一下 Reducer

reply Node 只返回了一条新的 AIMessage，最终 State 中却保留了用户消息和 AI 消息。

这就是 add_messages 在处理消息 State。

实际 Agent 往往还需要其他字段。

可以直接扩展 MessagesState：

代码片段Python
from langgraph.graph import MessagesState
 
 
class AgentState(MessagesState):
    topic: str
    retry_count: int

如果其他字段也需要累积，可以继续单独配置 Reducer。

同一个 State 中，不同字段可以使用完全不同的更新规则。

七、State Schema 的边界

理解 Reducer 以后，设计 State 时就不能只考虑：

代码片段Text
需要哪些字段

还应该同时考虑：

代码片段Text
这个字段由谁更新
是否会多次更新
是否会被并行 Node 更新
新值应该覆盖还是合并
如果合并应该采用什么规则

例如：

代码片段Python
class AgentState(MessagesState):
    query: str
    documents: list[str]
    summary: str
    retry_count: int

几个字段的更新语义可能完全不同。

query 通常代表当前查询，新值可以直接覆盖。

summary 通常只需要保存最新结果，也适合默认覆盖。

retry_count 如果 Node 每次返回完整计数，也可以继续覆盖。

documents 则取决于 Graph 结构。

如果只有一个搜索 Node：

代码片段Text
search
  ↓
documents

覆盖可能就足够。

如果多个搜索 Node 会同时产生文档：

代码片段Text
search_web
search_database
search_files

那么 documents 很可能需要 Reducer。

messages 则通常使用 add_messages。

可以用一个简单的判断方式：

代码片段Text
新值代表最新状态
        ↓
默认覆盖

新值需要和旧值同时保留
        ↓
使用 Reducer

存在特殊合并规则
        ↓
自定义 Reducer

State 本身也不应该包含所有中间变量。

只有需要跨 Node 使用、参与路由、持久化或者任务恢复的数据，才更适合进入 State。

Node 内部的一次性临时变量，继续留在函数内部即可。

这样 State schema 会更清楚，后面的 Persistence、Interrupt 和 Subgraph 设计也会更容易。

总结

本文介绍了 LangGraph 的 State 更新机制。Node 返回的是部分状态更新，字段默认采用覆盖方式；需要累积、去重或合并数据时，可以通过 Reducer 定义更新规则。

add_messages 专门处理消息状态，MessagesState 则提供了常用封装。设计 State 时，还应提前考虑并发更新以及不同字段的合并语义。