TypedDict与Reducer

两个节点都要往 messages 里追加内容——若不用 reducer，后一个节点会把前一个的列表整个盖掉。Agent 对话场景里这是最常见 bug；Annotated[…, add_messages] 在每次 superstep 合并时按消息 ID 去重追加。

1. 定位

维度	内容
角色	定义共享 State 结构与字段合并规则
输入 → 输出	各节点 partial update → reducer 合并 → 新 state
核心 API	`TypedDict`、`Annotated`、`add_messages`
依赖 LangChain	`BaseMessage`、`HumanMessage`、`AIMessage`

2. 图拓扑

节点表

节点名	职责	读 State	写 State
`user_echo`	模拟用户侧	`messages`	`messages`
`ai_reply`	模拟助手	`messages`	`messages`

边表

源	目标	类型
START	user_echo	固定
user_echo	ai_reply	固定
ai_reply	END	固定

3. invoke 生命周期与 superstep

Superstep 1: user_echo 返回 {"messages": [HumanMessage(...)]}
             → add_messages 与初始 [] 合并
Superstep 2: ai_reply 返回 {"messages": [AIMessage(...)]}
             → add_messages 追加，不覆盖 HumanMessage
Superstep 3: END

Reducer 触发时机：同一 superstep 内多节点写同一字段时先各自产出 update，再统一 reducer；跨 superstep 同样适用。

4. 原理

4.1 TypedDict 作 schema

LangGraph 用 schema 校验 invoke 输入、推断节点返回值键。字段缺失时 invoke 可能报错，开发阶段建议给全键。

4.2 add_messages

专为 list[BaseMessage] 设计：按 id 更新已有消息，否则 append。适合多轮对话与 tool 消息回写。

4.3 无 Annotated 的字段

默认 last write wins：后写节点的值覆盖先写。

5. 最小可运行示例

from typing import Annotated, TypedDict

from langchain_core.messages import AIMessage, BaseMessage, HumanMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages


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


def user_echo(state: State) -> dict:
    last = state["messages"][-1].content if state["messages"] else "hi"
    return {"messages": [HumanMessage(content=f"用户说:{last}")]}


def ai_reply(state: State) -> dict:
    return {"messages": [AIMessage(content="收到")]}


builder = StateGraph(State)
builder.add_node("user_echo", user_echo)
builder.add_node("ai_reply", ai_reply)
builder.add_edge(START, "user_echo")
builder.add_edge("user_echo", "ai_reply")
builder.add_edge("ai_reply", END)

graph = builder.compile()
out = graph.invoke({"messages": [HumanMessage(content="hello")]})
print(len(out["messages"]))  # 3：初始1 + user1 + ai1

6. 执行追踪

步骤	`messages` 条数	最后一条
初始	1	Human: hello
user_echo 后	2	Human: 用户说:hello
ai_reply 后	3	AI: 收到

若无 reducer：ai_reply 后只剩 1 条 AIMessage。

重要配置参数

参数	类型 / 默认	作用与影响	参考起点	配置指导
`Annotated[T, reducer]`	必填（列表字段）	定义合并	`add_messages`	多写必加
`add_messages`	内置 reducer	消息去重追加	messages 字段	Agent 标配
`TypedDict` 字段	—	schema 约束	少而清晰	避免巨型 state
无 reducer 列表	默认覆盖	单写 OK	仅单节点写字段	并行禁用
`messages` 带 id	Message.id	更新 vs 追加	流式替换	可选 UUID
invoke 初始 messages	list	历史带入	checkpoint 续跑	与 thread 配合

7. 易踩坑

多节点写 messages 不加 add_messages：对话历史丢失。
把 reducer 写在普通变量而非 Annotated：LangGraph 不识别，仍覆盖。
节点返回全量 messages 列表：应只返回新增消息，让 reducer 合并。

小结

TypedDict 定义 State；Annotated + reducer 定义合并。
对话类 Agent 的 messages 几乎总是 add_messages。
无 reducer 字段 = 最后写入者胜出。