用 Reducer 定义状态合并规则

状态如何合并

本章目标

理解默认覆盖、列表累加和消息合并，避免并行写入冲突。

默认行为是覆盖

class State(TypedDict):
    answer: str

后一个节点返回 {"answer": "new"} 时，新值覆盖旧值。计数器通常也采用覆盖：节点读取旧值后返回完整的新值。

普通列表使用 operator.add

覆盖还是累加

import operator
from typing import Annotated


class ReviewState(TypedDict):
    feedback: Annotated[list[str], operator.add]

这表示节点返回的是要追加的增量。错误记录、检索结果和审计标签可以使用这种方式。

不要把普通字符串错误列表声明成 add_messages。add_messages 会把字符串转换成消息对象，不再是普通错误字符串。

对话消息使用 add_messages

消息按标识合并

from langchain.messages import AnyMessage
from langgraph.graph.message import add_messages


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

add_messages 能按消息 ID 更新已有消息，并把字典输入反序列化为消息对象。只有真正的对话消息字段才应使用它。

运行带 Reducer 的质量循环：

python examples/state_and_routing.py

预期结果中 feedback 同时保留第一次失败和第二次通过记录，attempts 为 2。

自定义 Reducer 的生产约束

自定义 Reducer 应当：

  1. 不修改传入对象。
  2. 相同输入产生相同结果。
  3. 重放时不会触发外部副作用。
  4. 对并行更新顺序不敏感，或明确记录排序键。

数据库连接、HTTP 客户端和模型实例不属于 State。它们应通过构图函数闭包、运行上下文或依赖注入提供。

自定义 Reducer：按业务标识去重

简单 operator.add 会保留重复项。如果并行检索源可能返回同一文档，应使用纯函数 Reducer 按稳定标识合并：

from typing import TypedDict


class Finding(TypedDict):
    id: str
    score: float
    text: str


def merge_findings(left: list[Finding], right: list[Finding]) -> list[Finding]:
    merged = {item["id"]: item for item in left}
    for item in right:
        current = merged.get(item["id"])
        if current is None or item["score"] > current["score"]:
            merged[item["id"]] = item
    return sorted(merged.values(), key=lambda item: item["id"])

排序不是装饰。并行节点的完成顺序不稳定，如果 Reducer 按到达顺序输出，快照、测试和提示词都会产生不必要的抖动。

用性质测试 Reducer

至少验证三种性质：不修改输入、相同输入结果相同、交换左右输入后业务结果等价。

def test_merge_findings_is_order_independent():
    a = [{"id": "A", "score": 0.7, "text": "old"}]
    b = [{"id": "A", "score": 0.9, "text": "new"}]
    assert merge_findings(a, b) == merge_findings(b, a)
    assert a[0]["text"] == "old"

对日志这类天然有顺序的数据，应显式增加 timestamp 或序号，再在消费端排序；不要假设并行更新的到达顺序就是业务顺序。

Reducer 选择表

数据推荐语义常见错误
当前阶段、最终答案覆盖错误地累加出多个“最终值”
普通审计标签集合式去重合并直接 operator.add 产生重复
有序事件追加后按稳定键排序依赖并行完成顺序
对话消息add_messages把普通字符串列表当消息
外部资源句柄不进入 StateCheckpoint 序列化失败

本章验收

  • 能制造两个并行节点写同一普通字段的冲突，并解释报错原因。
  • 能为一个业务对象列表实现确定性、自身无副作用的 Reducer。
  • 能说明 Reducer 在重放和并行执行时为什么不能调用外部服务。