目录
一、一个非常容易踩的坑
二、为什么会有 Reducer？
（一）Reducer策略
（二）后端开发视角理解 Reducer
1. SQL 类比
2. Java 类比
三、Reducer 解决的核心问题
当前值
历史记录
四、核心原理
（一）Channel 抽象
1. LastValue
2. BinaryOperatorAggregate
（二）使用 Annotated 声明 Reducer
（三）自定义 Reducer
五、源码分析
（一）_get_channels(schema)
（二）LastValue：覆盖模式
（三）BinaryOperatorAggregate：Reducer 模式
（四）完整更新流程
六、实战代码
（一）全量代码展示
（二）重点代码说明
1. 场景一：默认覆盖模式
2. 场景二：自动累加模式
3. 场景三：自定义 Reduce合并 Dict
（三）运行结果
如何理解结果？
1. 场景一：覆盖
2. 场景二：累加
3. 场景三：自定义 Reducer
七、常见坑与排查
（一）坑 1：忘记添加 Reducer
错误写法
正确写法
（二）坑 2：Reducer 字段未初始化
直接报错信息
正确做法
（三）坑 3：Reducer 有副作用
错误示例
正确示例
八、工程化思考与生产实践
（一）工程化思考简要
1. 历史记录无限增长
2. 初始化统一管理
3. Reducer 必须保持纯函数
4. 调试难度提升
（二）生产级实践建议
1. 统一初始化
2. 给历史字段加上限
3. 对话场景优先使用 add_messages
九、总结
下一篇预告
干货分享，感谢您的阅读！
本文是「LangGraph 实战」系列第 2 篇。上一篇我们已经跑通了最小直线图，并留下了一个关键问题：
节点返回的 dict，究竟是如何合并进全局 State 的？
这一篇我们就来揭开 Channel 与 Reducer 的面纱——这是从 Demo 迈向真实 Agent 的关键一步。
一、一个非常容易踩的坑
我们照着上一篇的套路，兴冲冲地写了一个对话 Agent。
每个节点都会向 messages 列表追加一条消息，最后希望把完整对话返回出来。结果一运行，却发现：
明明经过了 3 个节点，最终 messages 里却只剩下最后一个节点写入的那条消息，前面的内容全都消失了。
开始排查：
- Node 逻辑没问题
- Edge 连接也没问题
- State 定义看起来也正常
最后才发现，问题出在一个几乎没人第一次会注意到的地方：
LangGraph 默认的 State 合并策略是“覆盖（Overwrite）”，而不是“追加（Append）”。
后一个节点返回的值，会直接替换前一个节点写入的值。这也是 LangGraph 中最常见、最隐蔽的坑之一。而解决它的关键，就是理解今天的主角：Reducer
二、为什么会有 Reducer？
（一）Reducer策略
在 Demo 01 中，我们已经掌握了 StateGraph 的基本结构。但当开始构建真实 Agent 时，很快就会遇到这样的问题：
多个节点都在更新同一个字段时，到底应该覆盖，还是累加？
例如：
messages
这个字段显然希望保留整个对话历史。
而像：
current_step
status
这类字段通常只关心最新状态。
因此：
- 有些字段应该覆盖
- 有些字段应该累加
LangGraph 并没有替你做统一决定，而是把这个权力交给开发者：
每个 State 字段都可以拥有独立的合并策略。
而这个策略，就是 Reducer。
（二）后端开发视角理解 Reducer
如果你有后端开发经验，可以这样理解：
1. SQL 类比
普通字段：
UPDATE task SET status = 'done'
覆盖旧值。
而消息历史更像：
INSERT INTO messages (…)
不断累积。
2. Java 类比
覆盖：
map.put(key, value)
自定义合并：
map.merge(
key,
value,
(oldValue, newValue) -> oldValue + newValue
)
这里的：
(oldValue, newValue) -> …
本质上就是 Reducer。
三、Reducer 解决的核心问题
两种策略的区别如下：
| 覆盖（默认） | 新值直接替换旧值 | 当前状态、步骤标记 | map.put() / SQL UPDATE |
| 累加（Reducer） | 新旧值通过函数合并 | 消息历史、执行记录 | map.merge() / SQL INSERT |
判断标准其实非常简单：
这个字段表示“当前值”，还是“历史记录”？
如果是：
当前值
只关心最新状态：
status
current_step
current_node
使用默认覆盖即可。
历史记录
希望保留全部过程：
messages
steps
tool_calls
使用 Reducer。
四、核心原理
（一）Channel 抽象
LangGraph 的状态管理底层建立在 Channel 机制之上。每一个 State 字段，都会对应一个 Channel。Channel 负责决定：
这个字段应该如何接收和更新数据。
主要有两种类型：
1. LastValue
只保留最新值。
旧值 -> 新值
直接覆盖。
2. BinaryOperatorAggregate
通过 Reducer 合并：
旧值 + 新值 -> 合并结果
（二）使用 Annotated 声明 Reducer
定义方式如下：
from typing import Annotated, TypedDict
import operator
class MyState(TypedDict):
current_step: str # → LastValue（覆盖）
messages: Annotated[list[str], operator.add] # → 累加器（追加）
其中：
Annotated[类型, Reducer]
表示：
- 第一个参数：字段类型
- 第二个参数：Reducer 函数
（三）自定义 Reducer
除了：
operator.add
你还可以使用任意函数。
要求签名统一为：
(old, new) -> merged
例如：
def merge_dicts(old: dict, new: dict) -> dict:
return {
**old,
**new
}
class State(TypedDict):
metadata: Annotated[
dict,
merge_dicts
]
这样多个节点写入的 dict 就会自动合并。
五、源码分析
（一）_get_channels(schema)
当创建 StateGraph 时：
StateGraph(…)
内部会调用：
_get_channels()
解析 TypedDict 上的类型注解。
简化后的逻辑如下：
# langgraph/graph/state.py（简化）
def _get_channels(schema):
channels = {}
for field_name, field_type in get_type_hints(schema, include_extras=True).items():
if hasattr(field_type, '__metadata__'):
# 有 Annotated → 取 metadata[0] 作为 Reducer
reducer = field_type.__metadata__[0]
channels[field_name] = BinaryOperatorAggregate(field_type.__origin__, reducer)
else:
# 无 Annotated → 默认 LastValue（覆盖）
channels[field_name] = LastValue(field_type)
return channels
逻辑非常直接：
有 Annotated
↓
BinaryOperatorAggregate
没有 Annotated
↓
LastValue
（二）LastValue：覆盖模式
核心代码几乎只有一行：
class LastValue(BaseChannel):
def update(self, values):
if values:
self.value = values[-1] # 直接覆盖
只取最后一个值。
（三）BinaryOperatorAggregate：Reducer 模式
class BinaryOperatorAggregate(BaseChannel):
def update(self, values):
for value in values:
if self.value is not None:
self.value = self.operator(self.value, value) # old, new → merged
else:
self.value = value
本质就是不断执行：
merged = reducer(old, new)
（四）完整更新流程
假设节点返回：
{
"messages": ["新消息"],
"status": "done"
}
更新过程如下：
Node 返回结果
↓
messages
↓
BinaryOperatorAggregate
↓
operator.add(
old,
["新消息"]
)
↓
旧列表 + 新列表
status
↓
LastValue
↓
直接覆盖
最终 State 更新完成
六、实战代码
（一）全量代码展示
下面通过三个场景演示不同的 State 合并策略。原始代码全量展示如下：
"""Demo 02: State Management — 状态定义与 Reducer 机制。
演示 LangGraph 的状态管理核心：
1. TypedDict 定义类型安全的 State
2. Reducer（Annotated）实现自动累加 vs 覆盖
3. 自定义 Reducer 函数
4. State 在节点间的传递和更新
运行方式：
python stages/stage1_fundamentals/02_state_management/main.py
"""
from __future__ import annotations
import operator
import sys
from pathlib import Path
from typing import Annotated, TypedDict
from langgraph.graph import END, START, StateGraph
sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent.parent))
from shared import get_logger, log_step
logger = get_logger("demo.02_state_mgmt")
# ============================================================
# 场景一：覆盖模式 vs 累加模式对比
# ============================================================
class OverwriteState(TypedDict):
"""无 Reducer 的 State — 所有字段都是覆盖模式。"""
value: str
counter: int
class AccumulateState(TypedDict):
"""带 Reducer 的 State — 列表自动累加，计数覆盖。"""
messages: Annotated[list[str], operator.add]
steps: Annotated[list[str], operator.add]
current_node: str
def demo_overwrite():
"""演示覆盖模式：每个 Node 的返回值直接替换旧值。"""
print("\\n— 场景一：覆盖模式（无 Reducer）—")
def node_a(state: OverwriteState) -> dict:
log_step(logger, "node_a", f"接收到 value='{state['value']}', counter={state['counter']}")
return {"value": "来自 node_a", "counter": 10}
def node_b(state: OverwriteState) -> dict:
log_step(logger, "node_b", f"接收到 value='{state['value']}', counter={state['counter']}")
return {"value": "来自 node_b", "counter": 20}
graph = StateGraph(OverwriteState)
graph.add_node("a", node_a)
graph.add_node("b", node_b)
graph.add_edge(START, "a")
graph.add_edge("a", "b")
graph.add_edge("b", END)
app = graph.compile()
result = app.invoke({"value": "初始值", "counter": 0})
print(f" 最终 value = '{result['value']}' ← node_b 的值覆盖了 node_a")
print(f" 最终 counter = {result['counter']} ← node_b 的值覆盖了 node_a")
return result
def demo_accumulate():
"""演示累加模式：带 Reducer 的字段自动合并。"""
print("\\n— 场景二：累加模式（Annotated + operator.add）—")
def collector_a(state: AccumulateState) -> dict:
log_step(logger, "collector_a", "添加消息和步骤记录")
return {
"messages": ["collector_a: 第一条消息"],
"steps": ["collector_a"],
"current_node": "collector_a",
}
def collector_b(state: AccumulateState) -> dict:
log_step(logger, "collector_b", f"当前已有 {len(state['messages'])} 条消息")
return {
"messages": ["collector_b: 第二条消息", "collector_b: 第三条消息"],
"steps": ["collector_b"],
"current_node": "collector_b",
}
def summarizer(state: AccumulateState) -> dict:
log_step(logger, "summarizer", f"汇总 {len(state['messages'])} 条消息")
summary = f"共收集 {len(state['messages'])} 条消息，经过节点: {' → '.join(state['steps'])}"
return {
"messages": [f"总结: {summary}"],
"steps": ["summarizer"],
"current_node": "summarizer",
}
graph = StateGraph(AccumulateState)
graph.add_node("collector_a", collector_a)
graph.add_node("collector_b", collector_b)
graph.add_node("summarizer", summarizer)
graph.add_edge(START, "collector_a")
graph.add_edge("collector_a", "collector_b")
graph.add_edge("collector_b", "summarizer")
graph.add_edge("summarizer", END)
app = graph.compile()
result = app.invoke({
"messages": [],
"steps": [],
"current_node": "",
})
print(f" messages 数量 = {len(result['messages'])}（自动累加，不是覆盖！）")
for i, msg in enumerate(result["messages"]):
print(f" [{i+1}] {msg}")
print(f" steps = {' → '.join(result['steps'])}")
print(f" current_node = '{result['current_node']}'（无 Reducer，被最后一个节点覆盖）")
return result
# ============================================================
# 场景三：自定义 Reducer 函数
# ============================================================
def merge_dicts(left: dict, right: dict) -> dict:
"""自定义 Reducer: 合并字典，保留所有 key，新值覆盖旧值。"""
return {**left, **right}
def dedup_list(left: list, right: list) -> list:
"""自定义 Reducer: 列表合并并去重。"""
seen = set()
result = []
for item in left + right:
if item not in seen:
seen.add(item)
result.append(item)
return result
class CustomReducerState(TypedDict):
"""使用自定义 Reducer 的 State。"""
metadata: Annotated[dict, merge_dicts]
tags: Annotated[list[str], dedup_list]
status: str
def demo_custom_reducer():
"""演示自定义 Reducer 函数。"""
print("\\n— 场景三：自定义 Reducer —")
def tagger_a(state: CustomReducerState) -> dict:
log_step(logger, "tagger_a", "添加标签和元数据")
return {
"metadata": {"source": "tagger_a", "priority": "high"},
"tags": ["python", "langgraph", "ai"],
"status": "tagging",
}
def tagger_b(state: CustomReducerState) -> dict:
log_step(logger, "tagger_b", "添加更多标签和元数据")
return {
"metadata": {"author": "demo", "priority": "low"},
"tags": ["langgraph", "agent", "demo"],
"status": "done",
}
graph = StateGraph(CustomReducerState)
graph.add_node("tagger_a", tagger_a)
graph.add_node("tagger_b", tagger_b)
graph.add_edge(START, "tagger_a")
graph.add_edge("tagger_a", "tagger_b")
graph.add_edge("tagger_b", END)
app = graph.compile()
result = app.invoke({"metadata": {}, "tags": [], "status": ""})
print(f" metadata = {result['metadata']}")
print(" ↑ merge_dicts: source 来自 tagger_a, author 来自 tagger_b, priority 被 tagger_b 覆盖")
print(f" tags = {result['tags']}")
print(" ↑ dedup_list: 'langgraph' 只出现一次（去重）")
print(f" status = '{result['status']}'（无 Reducer，直接覆盖）")
return result
def run_demo() -> dict:
"""运行 State Management Demo。"""
print("=" * 60)
print(" Demo 02: State Management — 状态定义与 Reducer 机制")
print("=" * 60)
result1 = demo_overwrite()
result2 = demo_accumulate()
result3 = demo_custom_reducer()
print()
print("=" * 60)
print(" 关键概念回顾")
print("=" * 60)
print(" 1. 无 Reducer : 字段直接覆盖（LastValue 策略）")
print(" 2. operator.add : 列表自动拼接、数字自动相加")
print(" 3. 自定义 Reducer : 任意合并逻辑（dict 合并、列表去重等）")
print(" 4. Annotated : 使用 typing.Annotated 声明 Reducer")
print(" 5. Channel : LangGraph 内部的数据管道抽象")
print()
return {"overwrite": result1, "accumulate": result2, "custom": result3}
if __name__ == "__main__":
run_demo()
（二）重点代码说明
1. 场景一：默认覆盖模式
# 场景一：覆盖模式（无 Reducer）
class OverwriteState(TypedDict):
value: str # 每个节点写它都会覆盖
counter: int
没有 Reducer。所有更新都会覆盖旧值。
2. 场景二：自动累加模式
# 场景二：累加模式（operator.add）
class AccumulateState(TypedDict):
messages: Annotated[list[str], operator.add] # 自动追加
steps: Annotated[list[str], operator.add]
current_node: str # 无 Reducer → 仍覆盖
其中：
messages
steps
自动追加。
而：
current_node
仍然保持覆盖模式。
3. 场景三：自定义 Reduce合并 Dict
# 场景三：自定义 Reducer
def merge_dicts(left: dict, right: dict) -> dict:
return {**left, **right}
def dedup_list(left: list, right: list) -> list:
seen, result = set(), []
for item in left + right:
if item not in seen:
seen.add(item)
result.append(item)
return result
class CustomReducerState(TypedDict):
metadata: Annotated[dict, merge_dicts] # dict 合并
tags: Annotated[list[str], dedup_list] # 追加并去重
status: str
（三）运行结果
实际执行输出如下：
Connected to server 127.0.0.1:50252
============================================================
Demo 02: State Management — 状态定义与 Reducer 机制
============================================================
— 场景一：覆盖模式（无 Reducer）—
最终 value = '来自 node_b' ← node_b 的值覆盖了 node_a
最终 counter = 20 ← node_b 的值覆盖了 node_a
— 场景二：累加模式（Annotated + operator.add）—
messages 数量 = 4（自动累加，不是覆盖！）
[1] collector_a: 第一条消息
[2] collector_b: 第二条消息
[3] collector_b: 第三条消息
[4] 总结: 共收集 3 条消息，经过节点: collector_a → collector_b
steps = collector_a → collector_b → summarizer
current_node = 'summarizer'（无 Reducer，被最后一个节点覆盖）
— 场景三：自定义 Reducer —
metadata = {'source': 'tagger_a', 'priority': 'low', 'author': 'demo'}
↑ merge_dicts: source 来自 tagger_a, author 来自 tagger_b, priority 被 tagger_b 覆盖
tags = ['python', 'langgraph', 'ai', 'agent', 'demo']
↑ dedup_list: 'langgraph' 只出现一次（去重）
status = 'done'（无 Reducer，直接覆盖）
============================================================
关键概念回顾
============================================================
1. 无 Reducer : 字段直接覆盖（LastValue 策略）
2. operator.add : 列表自动拼接、数字自动相加
3. 自定义 Reducer : 任意合并逻辑（dict 合并、列表去重等）
4. Annotated : 使用 typing.Annotated 声明 Reducer
5. Channel : LangGraph 内部的数据管道抽象
2026-06-04 23:12:06 INFO demo.02_state_mgmt | 📌 接收到 value='初始值',
counter=0
INFO demo.02_state_mgmt | 📌 接收到 value='来自
node_a', counter=10
INFO demo.02_state_mgmt | 📌 添加消息和步骤记录
INFO demo.02_state_mgmt | 📌 当前已有 1 条消息
INFO demo.02_state_mgmt | 📌 汇总 3 条消息
INFO demo.02_state_mgmt | 📌 添加标签和元数据
INFO demo.02_state_mgmt | 📌 添加更多标签和元数据
Process finished with exit code 0
如何理解结果？
1. 场景一：覆盖
value
counter
都只保留了最后一个节点的写入结果。这正是默认行为。
2. 场景二：累加
messages
经过多个节点后保留了全部内容。
而：
current_node
因为没有 Reducer，
最终只剩：
summarizer
这说明：
同一个 State 内，不同字段可以采用完全不同的更新策略。
3. 场景三：自定义 Reducer
merge_dicts
实现：
metadata
自动合并。
dedup_list
实现：
tags
自动去重。
例如：
langgraph
只会保留一份。
七、常见坑与排查
（一）坑 1：忘记添加 Reducer
最终只剩最后一条消息。
错误写法
class State(TypedDict):
messages: list[str]
正确写法
class State(TypedDict):
messages: Annotated[
list[str],
operator.add
]
（二）坑 2：Reducer 字段未初始化
直接报错信息
app.invoke({})
直接报错：
KeyError
TypeError
Reducer 需要：
old + new
但：
None + […]
这里基本就是没有初始化导致显然不成立。
正确做法
app.invoke({
"messages": [],
"steps": [],
"current_node": ""
})
（三）坑 3：Reducer 有副作用
错误示例
def bad(a, b):
a.extend(b)
return a
直接修改了旧对象。
正确示例
def good(a, b):
return a + b
返回全新对象。
💡 踩坑笔记
我在一开始学习 LangGraph 过程中真实遇到的问题例如：
“我曾经盯着 messages 为什么只剩一条消息排查了五分钟，最后发现只是少写了一个 Annotated。”
这种真实经历往往比知识点本身更有价值，大家快动手操作一下吧。
八、工程化思考与生产实践
（一）工程化思考简要
1. 历史记录无限增长
Reducer 只负责追加，不会自动清理。
因此：
messages
会越来越长。最终导致：
- Token 爆炸
- 内存上涨
需要定期裁剪。
2. 初始化统一管理
建议不要在每次：
invoke()
时手动传空列表。
统一使用工厂函数：
def make_initial_state():
…
3. Reducer 必须保持纯函数
不要：
- 写日志
- 写文件
- 修改入参
把它当成数学函数即可。
4. 调试难度提升
State 字段越来越多后，很难定位是谁修改了什么。建议结合：
stream()
观察每个节点产生的增量更新。
（二）生产级实践建议
1. 统一初始化
def make_initial_state():
return {
"messages": [],
"steps": [],
"current_node": ""
}
2. 给历史字段加上限
def add_capped(
old,
new,
cap=100
):
return (
old + new
)[-cap:]
只保留最近 100 条。
3. 对话场景优先使用 add_messages
from langgraph.graph.message import add_messages
class ChatState(TypedDict):
messages: Annotated[
list,
add_messages
]
相比：
operator.add
它支持：
- 追加
- 去重
- 更新
- 删除
是生产环境中的首选方案。
九、总结
State 管理是 LangGraph 中最容易被忽视、却又最重要的机制之一。
Node 和 Edge 决定：
Graph 怎么走。
而 Reducer 决定：
数据怎么流。
记住一句最重要的话：
每定义一个 State 字段，都先问自己：它是“当前值”，还是“历史记录”？
- 当前值 → 覆盖
- 历史记录 → Reducer
掌握这一点后，你就真正拥有了构建复杂 Agent 状态系统的能力。
下一篇预告
掌握了状态管理之后，Graph 已经能够稳定传递数据。
但现实中的 Agent 不会永远沿着一条直线执行：
- 用户问题不同
- 工具调用结果不同
- 模型决策不同
Graph 必须学会“分叉”。
下一篇：
《让 Graph 学会“分叉”：Conditional Edge 与路由函数实战》
我们将正式进入 LangGraph 的分支控制世界，学习如何根据状态动态决定下一步执行路径。
而在后续第 4 篇中，我们还会专门深入讲解：
MessagesState 与 add_messages —— LangGraph 对话场景中的智能 Reducer。
系列导航：LangGraph从零构建生产级 AI Agent 平台的递进式学习项目-CSDN博客
