State 定义与 Reducer：Annotated 与 add_messages

记住一句话：节点的返回值是「这一步产生了什么」，不是「State 现在长什么样」。 LangGraph 拿到返回的 partial update 后，对每个 key 查它的 reducer：有 reducer 就调用 reducer(existing_value, update_value)，没有就直接 existing = update（覆盖）。所以「字段怎么合并」这件事，写在 State 的类型定义里，而不是写在节点逻辑里。

默认覆盖 vs Annotated 自定义合并

先安装并看最小对照。同一个图里放两个字段：一个普通字段走覆盖，一个 Annotated 字段走 operator.add 累加。两个节点各返回一次，结果一目了然。

# LangGraph 主包；本页示例只用到核心 StateGraph 与内置 reducer
pip install -U langgraph
# 若要跑后面 add_messages + 真实消息对象的示例，再装 langchain-core
pip install -U langchain-core

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


class State(TypedDict):
    # 普通字段：默认 "覆盖" 语义，后写的值替换先写的值
    last_step: str
    # Annotated 字段：挂上 operator.add 这个 reducer -> list 拼接(累加)
    # 签名上等价于 merged = operator.add(existing, update) 即 existing + update
    log: Annotated[list[str], operator.add]


def node_a(state: State):
    # 节点返回 partial update：只写自己关心的 key
    return {"last_step": "a", "log": ["enter a"]}


def node_b(state: State):
    return {"last_step": "b", "log": ["enter b"]}


builder = StateGraph(State)
builder.add_node("a", node_a)
builder.add_node("b", node_b)
builder.add_edge(START, "a")
builder.add_edge("a", "b")
builder.add_edge("b", END)
graph = builder.compile()

result = graph.invoke({"last_step": "", "log": []})
print(result)
# {'last_step': 'b', 'log': ['enter a', 'enter b']}
#   last_step 被覆盖：a 写的 'a' 被 b 写的 'b' 顶掉
#   log 被累加：operator.add 把 ['enter a'] + ['enter b'] 拼成两条

add_messages：消息列表的专用 reducer

对话 / Agent 场景里，State 几乎都有一个消息列表字段。直接用 operator.add 拼接会有两个问题：无法按 id 更新已有消息（流式时同一条消息会被多次写入），也无法删除消息。LangGraph 为此内置了 add_messages，它做三件事：追加新消息、按 message.id 去重并更新、配合 RemoveMessage 删除消息。

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import (
    HumanMessage, AIMessage, RemoveMessage,
)


class ChatState(TypedDict):
    # 这一行就是 langgraph 内置 MessagesState 的核心写法
    messages: Annotated[list, add_messages]


def append_node(state: ChatState):
    # 1) 追加：返回新消息，add_messages 把它接到列表末尾
    return {"messages": [AIMessage(content="hi there", id="ai-1")]}


def update_node(state: ChatState):
    # 2) 按 id 更新：返回一条 id 已存在的消息 -> 原地替换，而不是再追加一条
    return {"messages": [AIMessage(content="hi there (edited)", id="ai-1")]}


def delete_node(state: ChatState):
    # 3) 删除：返回 RemoveMessage(id=...)，add_messages 会把该 id 的消息移除
    first_id = state["messages"][0].id
    return {"messages": [RemoveMessage(id=first_id)]}


builder = StateGraph(ChatState)
builder.add_node("append", append_node)
builder.add_node("update", update_node)
builder.add_node("delete", delete_node)
builder.add_edge(START, "append")
builder.add_edge("append", "update")
builder.add_edge("update", "delete")
builder.add_edge("delete", END)
graph = builder.compile()

# 初始放一条用户消息（带显式 id，方便观察删除）
out = graph.invoke({"messages": [HumanMessage(content="hello", id="human-1")]})
for m in out["messages"]:
    print(type(m).__name__, repr(m.content), m.id)
# 过程：
#   start:  [Human 'hello' human-1]
#   append: [Human 'hello' human-1, AI 'hi there' ai-1]
#   update: [Human 'hello' human-1, AI 'hi there (edited)' ai-1]  # 按 ai-1 原地更新
#   delete: [AI 'hi there (edited)' ai-1]                          # human-1 被移除
# 最终输出只剩一条 AI 消息

自定义 reducer：签名与字典合并实战

reducer 就是一个普通函数，签名固定为 (existing, update) -> merged。existing 是当前 State 里该字段的值（首次可能是 None 或初始值），update 是节点这次返回的值，返回的 merged 成为字段新值。约束：必须是纯函数、不可有副作用、对相同输入可重复调用（LangGraph 在并发 / 重试 / 持久化恢复时可能多次调用）。

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


def merge_dicts(existing: dict | None, update: dict | None) -> dict:
    """自定义 reducer：浅合并两个字典，update 的同名 key 覆盖 existing。
    纯函数：不修改入参，返回新 dict。"""
    existing = existing or {}
    update = update or {}
    return {**existing, **update}


class State(TypedDict):
    # scores 字段走自定义的字典合并 reducer
    scores: Annotated[dict, merge_dicts]


def grader_a(state: State):
    return {"scores": {"relevance": 0.8}}


def grader_b(state: State):
    # 不同 key -> 合并；若是相同 key 则后者覆盖
    return {"scores": {"fluency": 0.9}}


builder = StateGraph(State)
builder.add_node("a", grader_a)
builder.add_node("b", grader_b)
builder.add_edge(START, "a")
builder.add_edge("a", "b")
builder.add_edge("b", END)
graph = builder.compile()

print(graph.invoke({"scores": {}}))
# {'scores': {'relevance': 0.8, 'fluency': 0.9}}
# 两个节点各写一个 key，merge_dicts 把它们合到同一个 dict 里，互不覆盖

并发写同一字段：reducer 是唯一合法出口

当多个节点从同一个点 fan-out 并行执行、且都写同一个字段时，LangGraph 在同一个 super-step 内会收到多份对该字段的更新。如果该字段没有 reducer，LangGraph 无法决定保留谁，直接抛 InvalidUpdateError。给字段挂 reducer，并发的多份更新就会被 reducer 依次合并——这正是 reducer 设计的核心用途。

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


class State(TypedDict):
    # 有 reducer：并发分支各写一段，operator.add 负责合并
    results: Annotated[list[str], operator.add]
    # 没有 reducer 的 plain 字段在并发写时会触发 InvalidUpdateError，见下方注释
    # winner: str


def branch_1(state: State):
    return {"results": ["from-branch-1"]}


def branch_2(state: State):
    return {"results": ["from-branch-2"]}


builder = StateGraph(State)
builder.add_node("b1", branch_1)
builder.add_node("b2", branch_2)
# 从 START 同时 fan-out 到 b1 和 b2 -> 同一 super-step 内并行执行
builder.add_edge(START, "b1")
builder.add_edge(START, "b2")
builder.add_edge("b1", END)
builder.add_edge("b2", END)
graph = builder.compile()

print(graph.invoke({"results": []}))
# {'results': ['from-branch-1', 'from-branch-2']}
# 两个分支并发写 results，operator.add 把两份更新合并；顺序由运行时决定
#
# 反例：若把 winner: str（无 reducer）让 b1、b2 同时写，会得到
#   langgraph.errors.InvalidUpdateError:
#   Can receive only one value per step. Use an Annotated key to handle multiple values.