阶段：G1-核心心智模型 ｜ 预计耗时：60 分钟 ｜ 前置：G1-01 第一个 Agent 解剖 代码：要你自己敲成
langgraph-learn/examples/g1_state_reducer.py本篇知识点对应 知识全景 1.1、1.2——1.2 Reducer 是全框架头号大坑
学习目标
- 理解 State = 图的内存：节点之间不直接传话，全靠读写 State
- 亲眼看到覆盖 vs 合并的差异，掌握
Annotated[list, add_messages]与自定义 Reducer - 知道为什么"两个节点写同一个字段"必须想清楚合并规则
核心知识
1. State 就是 TypedDict
from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages
class State(TypedDict):
messages: Annotated[list, add_messages] # 有合并规则的字段
draft: str # 无注解 = 默认规则：覆盖
revisions: int # 覆盖
节点返回 {"draft": "新诗", "revisions": 1} 是增量（delta），LangGraph 负责按规则合进总 State。
2. Reducer：同名字段的合并规则
| 写法 | 合并行为 | 适用 |
|---|---|---|
| 无注解 | 覆盖（后写赢） | 计数器、当前草稿 |
Annotated[list, add_messages] |
追加（且处理消息 ID 去重） | 对话历史——Agent 的标配 |
自定义函数 Annotated[dict, merge_dict] |
你说了算 | 多节点合写一个字典 |
3. 头号大坑现场：没有 Reducer 会怎样
两个并行节点都返回 {"log": ["A"]} 和 {"log": ["B"]}——没有 Reducer 时只剩一个（谁后到谁赢），另一个的结果凭空消失。并发节点越多，丢得越安静。这就是官方示例里 messages 都要挂 add_messages 的原因。
实验任务
任务 1：敲出对比脚本（25 分钟）
# g1_state_reducer.py —— 覆盖 vs 追加，一图看清
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
def add_log(l: list[str], r: list[str]) -> list[str]:
return l + r # 自定义 Reducer：追加
class State(TypedDict):
log_plain: list[str] # 无 Reducer（覆盖）
log_red: Annotated[list[str], add_log] # 有 Reducer（追加）
def node_a(_s: State) -> dict:
return {"log_plain": ["A"], "log_red": ["A"]}
def node_b(_s: State) -> dict:
return {"log_plain": ["B"], "log_red": ["B"]}
g = StateGraph(State)
g.add_node("a", node_a); g.add_node("b", node_b)
g.add_edge(START, "a"); g.add_edge("a", "b"); g.add_edge("b", END)
print(g.compile().invoke({"log_plain": [], "log_red": []}))
# 预测：两个 log 字段各是什么？（先写下来再跑！）
任务 2：串行没问题、并行才要命（15 分钟）
把两条边改成并行分发（g.add_edge(START, "a"); g.add_edge(START, "b")）——Node A、B 同时写。运行后读 log_plain 还剩几个？这就是并发丢数据的现场。
任务 3：消息版 Reducer（15 分钟）
把 log_red 换成 Annotated[list, add_messages]，节点返回 {"log_red": [("a", "来自A")]}——观察 add_messages 给消息自动分配 ID 的行为（m.id）。
验收标准
衔接下一阶段
State 记忆会管了 → G1-03 手写 StateGraph 与条件边：亲手搭 create_agent 藏起来的那张循环图。