LangGraph 框架指南 · 第 03 章

第 3 章状态：图的数据心脏

本章目标：理解图状态的真正身份——它不是普通字典，而是"一组通道（channel）的集合"；每个通道有自己的更新规则（reducer）。学完本章，"多节点写同一字段"这类经典 bug 将与你绝缘

约 20 分钟8,076 字10 个代码示例

本章目标：理解图状态的真正身份——它不是普通字典，而是"一组通道（channel）的集合"；每个通道有自己的更新规则（reducer）。学完本章，"多节点写同一字段"这类经典 bug 将与你绝缘。

3.1 一个必然撞上的 bug

继续演进简历筛选系统。产品要求：候选人要接受两轮评估——硬性条件（学历、年限、必备技能）和软性条件（项目质量、成长曲线），两轮各自给分，最后求和。你顺手写出这样的代码：

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

class ScreenState(TypedDict):
    resume_text: str
    hard_score: int      # 硬性条件得分
    soft_score: int      # 软性条件得分

def hard_review(state: ScreenState) -> dict:
    return {"hard_score": 70}

def soft_review(state: ScreenState) -> dict:
    return {"soft_score": 80}

builder = StateGraph(ScreenState)
builder.add_node("hard_review", hard_review)
builder.add_node("soft_review", soft_review)
builder.add_edge(START, "hard_review")
builder.add_edge("hard_review", "soft_review")
builder.add_edge("soft_review", END)
graph = builder.compile()
print(graph.invoke({"resume_text": "..."}))
# {'resume_text': '...', 'hard_score': 70, 'soft_score': 80}

这个版本没问题——两个字段各写各的。但换个需求就出事了：产品要求每个评估维度都往同一个 score 字段里加分。你把代码改成：

class ScreenState(TypedDict):
    resume_text: str
    score: int            # 所有维度向这里累加

def hard_review(state: ScreenState) -> dict:
    return {"score": 70}

def soft_review(state: ScreenState) -> dict:
    return {"score": 80}

运行结果：score 是 80，不是 150。硬性条件的 70 被无声覆盖了。更麻烦的是：这里没有任何报错——毕竟在 Python 的字典语义里，"后写覆盖先写"天经地义。

LangGraph 的默认状态更新规则就是"覆盖"。这个默认值在单写者场景下没问题，但只要出现"多个节点写同一字段"（顺序写、尤其是并行写），就必然出错。LangGraph 的解法不是禁止多写，而是把"怎么合并多次写入"变成状态的显式属性——这就是 reducer。

3.2 状态的真相：通道

现在可以揭开图状态的真正身份了。

你在 TypedDict 里声明的每个字段，在运行时对应一个通道（channel）。一个通道包含两件事：

flowchart LR subgraph CH[一个通道] V[当前值（存储）] R[更新规则 reducer] end N1[节点 A 的补丁] --> R N2[节点 B 的补丁] --> R R --> V

存储：这个字段当前存了什么
更新规则（reducer）：当有新的写入到来时，如何与当前值合并

默认 reducer 是"覆盖"（LastValue 通道）：新值直接替换旧值。第 2 章的例子一直工作良好，是因为每个字段只有一个节点在写。

而 reducer 可以换成别的语义。比如"累加"：

from typing import Annotated, TypedDict
import operator

class ScreenState(TypedDict):
    resume_text: str
    score: Annotated[int, operator.add]     # 声明：写入按加法合并

Annotated[int, operator.add] 读作："这个字段是 int，多次写入时用 operator.add 合并"。再跑一次上面的图：

print(graph.invoke({"resume_text": "..."}))
# {'resume_text': '...', 'score': 150}

70 + 80 = 150。没有改任何节点代码，只在状态定义处声明了合并规则——这就是声明式状态管理的味道：节点只管"产出增量"，怎么合并是状态自己的事。

术语说明：operator.add 这类"合并函数"在 LangGraph 文档里叫 reducer。它来源于函数式编程里的 reduce 概念：把一串值按规则归并成一个值。

3.3 用累加语义重构评分流程

把 reducer 用在更贴近真实的地方。评估维度从两个涨到三个（硬性、软性、潜力），并且允许在补丁里一次加多分：

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

class ScreenState(TypedDict):
    resume_text: str
    score: Annotated[int, operator.add]          # 总分：累加
    breakdown: Annotated[list[str], operator.add]  # 明细：追加

def hard_review(state: ScreenState) -> dict:
    return {"score": 35, "breakdown": ["硬性条件 +35（年限/技能匹配）"]}

def soft_review(state: ScreenState) -> dict:
    return {"score": 30, "breakdown": ["软性条件 +30（项目质量）"]}

def potential_review(state: ScreenState) -> dict:
    return {"score": 25, "breakdown": ["成长潜力 +25（技术栈演进）"]}

builder = StateGraph(ScreenState)
builder.add_node("hard_review", hard_review)
builder.add_node("soft_review", soft_review)
builder.add_node("potential_review", potential_review)
builder.add_edge(START, "hard_review")
builder.add_edge("hard_review", "soft_review")
builder.add_edge("soft_review", "potential_review")
builder.add_edge("potential_review", END)
graph = builder.compile()

result = graph.invoke({"resume_text": "..."})
print(result["score"])       # 90
print(result["breakdown"])   # ['硬性条件 +35...', '软性条件 +30...', '成长潜力 +25...']

Annotated[list[str], operator.add] 是另一个高频组合：列表按拼接合并。注意第一个节点执行时 breakdown 还没有值——首次写入时引擎直接赋值（不经过 reducer），所以这里不需要提供初始值；但引擎版本行为可能有差异，自定义 reducer 仍应容忍 None（见 3.5）。

3.4 add_messages：最常用的内置 reducer

有一个 reducer 特殊到值得单独一节：add_messages。它解决的是 LLM 应用里最常见的需求——消息列表的追加与合并。

先想一个问题：如果消息字段用默认"覆盖"语义，对话图就废了——第二轮节点写入消息时会把第一轮的覆盖掉。如果用 operator.add 纯拼接也不够：LLM 多轮对话里，某条消息可能需要替换而不是追加（比如模型重新生成答案、修正工具调用结果）。

add_messages 的合并规则恰好覆盖这两点：

新消息的 ID 不存在 → 追加
新消息的 ID 已存在 → 替换同 ID 的旧消息

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

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

def greet(state: ChatState) -> dict:
    from langchain_core.messages import HumanMessage, AIMessage
    return {"messages": [
        HumanMessage("你好，我叫李思远", id="m1"),
        AIMessage("你好李思远！", id="m2"),
    ]}

def correct(state: ChatState) -> dict:
    # 同一个 id：不是追加，而是替换 m2
    from langchain_core.messages import AIMessage
    return {"messages": [AIMessage("你好，李思远！很高兴认识你。", id="m2")]}

builder = StateGraph(ChatState)
builder.add_node("greet", greet)
builder.add_node("correct", correct)
builder.add_edge(START, "greet")
builder.add_edge("greet", "correct")
builder.add_edge("correct", END)
graph = builder.compile()

for m in graph.invoke({"messages": []})["messages"]:
    print(m.id, m.content)
# m1 你好，我叫李思远
# m2 你好，李思远！很高兴认识你。    <- 被替换成了新内容

顺带一提，HumanMessage("文本") 不传 id 时会自动生成随机 ID，此时永远是追加语义——日常对话场景足够；显式传 ID 则获得替换能力。此外 add_messages 还支持传 RemoveMessage(id=...) 来删除指定消息，这是第 15 章做"记忆裁剪"的基础。

因为"消息列表 + add_messages"太常用了，LangGraph 提供了一个现成的状态类作为快捷方式：

from langgraph.graph import MessagesState

class MyState(MessagesState):     # 继承得到 messages 字段
    extra_field: str              # 再补充自己的字段

MessagesState 等价于 TypedDict 里声明 messages: Annotated[list[AnyMessage], add_messages]，直接继承它再扩展字段即可，不必每次手写 Annotated。

3.5 自定义 reducer

内置语义不够用时，reducer 就是一个普通函数，随手可写。签名是 (当前值, 新写入值) -> 合并结果。

需求实例：多个评估节点都会产出"关注点"（attention points），要求合并时去重。顺手把首次调用的边界情况处理掉：

def merge_unique(existing: list[str] | None, new: list[str] | None) -> list[str]:
    """合并两个列表并去重，保持出现顺序；容忍 None（首次写入）"""
    existing = existing or []
    new = new or []
    merged = list(existing)
    for item in new:
        if item not in merged:
            merged.append(item)
    return merged

class ScreenState(TypedDict):
    resume_text: str
    concerns: Annotated[list[str], merge_unique]   # 用自定义 reducer

def hard_review(state: ScreenState) -> dict:
    return {"concerns": ["跨行经验偏多"]}

def soft_review(state: ScreenState) -> dict:
    return {"concerns": ["跨行经验偏多", "项目量化数据少"]}

builder = StateGraph(ScreenState)
builder.add_node("hard_review", hard_review)
builder.add_node("soft_review", soft_review)
builder.add_edge(START, "hard_review")
builder.add_edge("hard_review", "soft_review")
builder.add_edge("soft_review", END)
graph = builder.compile()

print(graph.invoke({"resume_text": "..."})["concerns"])
# ['跨行经验偏多', '项目量化数据少']   <- 重复项被合并

写自定义 reducer 的三条纪律：

不要原地修改入参。existing.append(...) 后返回 existing 是危险写法——持久化、时间旅行（第 12、13 章）都依赖"历史快照不被后续执行偷偷改掉"。始终构造并返回新对象。
容忍初始值缺失。首次写入时 existing 可能是 None（取决于是否提供过初始值），reducer 要能处理。
保持纯函数。reducer 里不要做 IO、不要打日志、不要依赖全局状态——它是状态合并的仲裁逻辑，会在你意想不到的时刻被重放（时间旅行的核心机制之一就是重放 reducer）。

3.6 状态定义的三种姿势

TypedDict 之外，Pydantic BaseModel 和 dataclass 也可以定义状态。三者的差别主要在运行时校验与默认值：

定义方式	运行时校验	默认值	适用场景
`TypedDict`	无	不支持	大多数场景的默认选择，零开销
Pydantic `BaseModel`	有（类型不符即报错）	支持	希望尽早暴露字段拼写/类型错误
`dataclass`	无	支持	想要默认值但不想引入校验

Pydantic 版本长这样（注意 reducer 写法一致）：

from typing import Annotated
from pydantic import BaseModel
import operator

class ScreenState(BaseModel):
    resume_text: str
    score: Annotated[int, operator.add] = 0

带校验的定义方式能把 3.1 节"字段拼错静默通过"的问题提前到写入时暴露，团队协作的项目里值得考虑。

常见坑与排错

忘了加 reducer，多写场景静默覆盖。这是本章主角，再强调一次：只要"多于一个节点"可能写同一个字段——包括并行分支（第 5 章）和循环回写（第 4 章）——就必须认真考虑该字段的合并语义。症状是"数值只有最后一次的"或"列表只剩一项"，且无任何报错。
reducer 里原地修改可变对象。operator.add 对数字、对列表都安全（[1,2] + [3] 返回新列表），但自定义 reducer 里写 existing.extend(new); return existing 就埋了雷：一旦启用持久化，历史检查点里的列表会被污染，时间旅行结果错乱。规则：永远返回新对象。
以为 Annotated 里的类型注解有运行时效果。Annotated[int, operator.add] 里的 int 不会在运行时强制类型——传个字符串进去，加法会在运行时炸掉。要运行时校验请用 Pydantic 定义状态。
add_messages 场景下"消息没追加"。通常是消息带了固定 ID 又被再次写入——触发的是替换而非追加。排查时打印 m.id 看是否撞了 ID。
在 reducer 上做复杂业务逻辑。reducer 应当廉价、纯粹、可重放。业务规则放进节点；reducer 只做合并。判断标准：把 reducer 想象成"字典的一个方法"，你会往 dict.update 里写业务吗？
并行节点 + 覆盖语义 + 相同值。有一种侥幸情况：两个并行节点写同一字段且值恰好相同，覆盖语义下看不出问题，上线后某天值不同了才爆雷。代码评审时对"并行写同一字段"要保持本能警觉。

动手练习

给 3.3 节的评分图加一个 deduct 节点：当简历里出现"频繁跳槽"关键词时总分扣 20 分（即写入 {"score": -20}），验证累加 reducer 对负数同样成立。
实现一个 max_reducer：多个节点写入候选人的"期望薪资"（万元/月），状态里始终保留最大值。思考它比覆盖语义好在哪——如果各节点写入顺序不确定，覆盖语义的结果还可靠吗？
用 MessagesState 搭一个两轮对话图：节点 A 写入用户消息和模型回复各一条，节点 B 通过替换 ID 的方式修正模型回复中的人名。运行后确认只有 2 条消息且内容为修正后的版本。

本章速查

概念 / API	一句话
通道 channel	状态的每个字段在运行时对应一个通道：存储 + 更新规则
默认 reducer	覆盖（LastValue）——多写必错，仅适用于单写者字段
`Annotated[T, reducer]`	在状态定义处声明该字段的合并规则
`operator.add`	内置累加/拼接语义，数字与列表通吃
`add_messages`	消息专用：新 ID 追加、同 ID 替换；配合 `RemoveMessage` 可删
`MessagesState`	自带 `messages: Annotated[..., add_messages]` 的快捷基类
自定义 reducer	签名 `(旧值, 新值) -> 新结果`；要纯、要容忍 None、不要原地改
状态定义三姿势	TypedDict（轻）/ Pydantic（校验）/ dataclass（默认值）