State 与 Reducer

State（状态）是 LangGraph 的核心概念。每次图执行时，当前的状态会传入每个节点，节点的返回值被合并回状态中。理解 State 和 Reducer 的机制是掌握 LangGraph 的第一步。

State 的作用

State 本质上是一个"上下文对象"，它在工作流中扮演以下角色：

1. 数据传递：节点之间通过 state 交换数据
2. 持久化：配合 Checkpointer，state 可以被保存和恢复
3. 追踪：记录工作流的执行进度和中间结果
4. 调试：每次 checkpoint 都保存了完整的 state 快照

定义 State

TypedDict 方式（推荐）

from typing import TypedDict, List, Annotated
import operator

class AgentState(TypedDict):
    # 简单字段：后写入的覆盖先写入的
    question: str
    answer: str

    # 带 reducer 的字段：多个节点的写入会合并
    messages: Annotated[List[dict], operator.add]
    documents: Annotated[List[str], operator.add]

dataclass 方式

from dataclasses import dataclass, field
from typing import List

@dataclass
class AgentState:
    question: str = ""
    answer: str = ""
    messages: List[dict] = field(default_factory=list)
    documents: List[str] = field(default_factory=list)

Reducer（归约器）详解

Reducer 决定了当多个节点写入同一个 state 字段时，如何合并这些值。

默认行为：覆盖

如果没有显式指定 reducer，后写入的值会覆盖先写入的值：

class State(TypedDict):
    count: int  # 后写入的覆盖之前的值

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

def node_b(state):
    return {"count": 2}  # 覆盖 node_a 写入的值

# 最终 count = 2

operator.add：追加列表

class State(TypedDict):
    items: Annotated[List[str], operator.add]

def node_a(state):
    return {"items": ["a"]}

def node_b(state):
    return {"items": ["b"]}

# 最终 items = ["a", "b"]

自定义 Reducer

你可以编写自己的 reducer 函数：

def custom_reducer(old_value: List[str], new_value: List[str]) -> List[str]:
    """只保留最新的 10 条消息"""
    combined = old_value + new_value
    return combined[-10:]

class State(TypedDict):
    messages: Annotated[List[str], custom_reducer]

add_messages：内置消息 Reducer

LangGraph 提供了一个内置的 add_messages reducer，专门用于处理消息列表：

from langgraph.graph import add_messages
from typing import Annotated

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

add_messages 比 operator.add 更智能：

• 自动去重（根据消息 ID）
• 支持更新已有消息（用相同 ID 的新消息替换旧消息）
• 正确处理 AI 消息和工具消息的关系

常见模式

读写分离

class State(TypedDict):
    # 输入字段
    user_input: str

    # 内部状态
    intermediate_result: str

    # 输出字段
    final_output: str

累计追踪

class State(TypedDict):
    # 累积 LLM 调用次数
    llm_calls: int
    # 累积工具调用
    tool_calls_made: Annotated[List[str], operator.add]

注意事项

1. Reducer 只适用于 list、dict 等复杂类型，简单类型（str、int）不需要 reducer
2. 使用 operator.add 时节点必须返回 list，不能返回 None
3. Reducer 的输入是已有值和新增值，第一个参数是旧值，第二个参数是新值
4. 不要修改 state 对象的属性，始终返回新的更新字典

参考

• 官方文档：Persistence
• API 参考：add_messages