LangGraph 源码深潜：Reducer 与并行状态合并机制解剖

1
def __init__(
2
    self,
3
    state_schema: type[StateT],
4
    context_schema: type[ContextT] | None = None,
5
    *,
6
    input_schema: type[InputT] | None = None,
7
    output_schema: type[OutputT] | None = None,
8
    **kwargs: Unpack[DeprecatedKwargs],
9
) -> None:
10
    ...

1
用户代码 StateGraph(TravelState)
2
  ↓
3
StateGraph.__init__
4
  ↓
5
StateGraph._add_schema(self.state_schema)
6
  ↓
7
_get_channels(schema)

1
def __init__(self, state_schema, context_schema=None, *, input_schema=None, output_schema=None, **kwargs):
2
    # 1. 兼容旧参数：config_schema / input / output
3
    context_schema = resolve_context_schema(context_schema, kwargs)
4
    input_schema = resolve_input_schema(input_schema, kwargs)
5
    output_schema = resolve_output_schema(output_schema, kwargs)
6

7
    # 2. 初始化 builder 容器
8
    self.nodes = {}
9
    self.edges = set()
10
    self.branches = defaultdict(dict)
11
    self.schemas = {}
12
    self.channels = {}
13
    self.managed = {}
14
    self.waiting_edges = set()
15
    self.compiled = False
16

17
    # 3. 保存 schema 类型
18
    self.state_schema = state_schema
19
    self.input_schema = input_schema or state_schema
20
    self.output_schema = output_schema or state_schema
21
    self.context_schema = context_schema
22

23
    # 4. 解析 schema
24
    self._add_schema(self.state_schema)
25
    self._add_schema(self.input_schema, allow_managed=False)
26
    self._add_schema(self.output_schema, allow_managed=False)

1
self.channels

1
state key 的更新规则在构建期就已经确定，
2
不是运行到节点返回 update 时才临时猜测。

1
TravelState
2
  ↓
3
StateGraph.__init__
4
  ↓
5
self._add_schema(TravelState)
6
  ↓
7
self.channels 得到 user_request 与 research_notes 两个 channel

1
运行时不需要每次 node 返回 update 时再反射类型；
2
并行冲突可以由 channel 语义统一判断；
3
compile 后的 Pregel 图可以直接使用 channels。

1
修改 State schema 后必须重新 compile graph；
2
compile 后再改 builder 不会影响已编译图；
3
reducer 是图结构的一部分，不是临时执行参数。

1
channels: dict[str, BaseChannel]
2
managed: dict[str, ManagedValueSpec]

1
def _add_schema(
2
    self,
3
    schema: type[Any],
4
    /,
5
    allow_managed: bool = True,
6
) -> None:
7
    ...

1
StateGraph.__init__
2
  ↓
3
StateGraph._add_schema
4
  ↓
5
_get_channels(schema)

1
def _add_schema(self, schema, allow_managed=True):
2
    # 1. 同一个 schema 只解析一次
3
    if schema in self.schemas:
4
        return
5

6
    # 2. 如果 schema 形态可疑，发出 warning
7
    _warn_invalid_state_schema(schema)
8

9
    # 3. 核心：把 schema 字段解析成 channels 与 managed values
10
    channels, managed, type_hints = _get_channels(schema)
11

12
    # 4. input_schema / output_schema 不允许 managed values
13
    if managed and not allow_managed:
14
        raise ValueError("Managed channels are not permitted in Input/Output schema")
15

16
    # 5. 保存 schema 视图
17
    self.schemas[schema] = {**channels, **managed}
18

19
    # 6. 合并普通 channels
20
    for key, channel in channels.items():
21
        if key in self.channels:
22
            if self.channels[key] != channel:
23
                if isinstance(channel, LastValue):
24
                    # 允许默认 channel 不覆盖已有 reducer channel
25
                    pass
26
                else:
27
                    raise ValueError("Channel already exists with a different type")
28
        else:
29
            self.channels[key] = channel
30

31
    # 7. 合并 managed values
32
    for key, managed_value in managed.items():
33
        if key in self.managed and self.managed[key] != managed_value:
34
            raise ValueError("Managed value already exists with a different type")
35
        self.managed[key] = managed_value

1
_get_channels(schema)

1
TravelState
2
  ↓
3
_get_channels
4
  ↓
5
user_request -> LastValue(str)
6
research_notes -> BinaryOperatorAggregate(list[str], add)
7
  ↓
8
self.channels 保存两个 channel

1
不要在 state_schema、input_schema、output_schema 里给同名字段写不同 reducer。
2
如果只想隐藏输出字段，应使用 input_schema/output_schema 控制可见字段，
3
而不是改变同名字段的 reducer 语义。

1
普通 channel
2
managed value

1
def _get_channels(
2
    schema: type[dict],
3
) -> tuple[
4
    dict[str, BaseChannel],
5
    dict[str, ManagedValueSpec],
6
    dict[str, Any],
7
]:
8
    ...

1
StateGraph._add_schema
2
  ↓
3
_get_channels
4
  ↓
5
_get_channel(name, annotation)
6
  ↓
7
_is_field_channel / _is_field_binop / _is_field_managed_value

1
def _get_channels(schema):
2
    # 1. 读取类型注解，保留 Annotated metadata
3
    type_hints = get_type_hints(schema, include_extras=True)
4

5
    channels = {}
6
    managed = {}
7

8
    # 2. 遍历每个字段
9
    for name, annotation in type_hints.items():
10
        # 3. 先判断是否是 managed value
11
        if managed_spec := _is_field_managed_value(name, annotation):
12
            managed[name] = managed_spec
13
            continue
14

15
        # 4. 再解析普通 channel
16
        channel = _get_channel(name, annotation)
17
        channels[name] = channel
18

19
    return channels, managed, type_hints

1
research_notes: Annotated[list[str], add]

1
research_notes: list[str]

1
TravelState
2
  ↓
3
get_type_hints(include_extras=True)
4
  ↓
5
{
6
  "user_request": str,
7
  "research_notes": Annotated[list[str], add]
8
}
9
  ↓
10
_get_channel("user_request", str)
11
_get_channel("research_notes", Annotated[list[str], add])

1
字段声明
2
  +
3
字段更新协议声明

1
字段类型决定值的形态；
2
Annotated metadata 决定更新语义。

1
research_notes: Annotated[list[str], add]

1
BinaryOperatorAggregate(list[str], add)

1
def _get_channel(
2
    name: str,
3
    annotation: Any,
4
    *,
5
    allow_managed: bool = True,
6
) -> BaseChannel | ManagedValueSpec:
7
    ...

1
_get_channels
2
  ↓
3
_get_channel
4
  ↓
5
_is_field_channel
6
  ↓
7
_is_field_binop
8
  ↓
9
LastValue fallback

1
def _get_channel(name, annotation):
2
    # 1. 如果 Annotated metadata 直接声明了 channel，则使用该 channel
3
    if channel := _is_field_channel(annotation):
4
        channel.key = name
5
        return channel
6

7
    # 2. 如果 Annotated metadata 声明了二元 reducer，则构造 BinaryOperatorAggregate
8
    if binop_channel := _is_field_binop(annotation):
9
        binop_channel.key = name
10
        return binop_channel
11

12
    # 3. 如果是 managed value，返回 managed spec
13
    if managed := _is_field_managed_value(name, annotation):
14
        return managed
15

16
    # 4. 默认：LastValue
17
    return LastValue(annotation, key=name)

1
Annotated[list[str], add]

1
LastValue

1
user_request: str
2
  ↓
3
无 channel metadata
4
无 reducer metadata
5
  ↓
6
LastValue(str)
7

8
research_notes: Annotated[list[str], add]
9
  ↓
10
识别 add 是二元 reducer
11
  ↓
12
BinaryOperatorAggregate(list[str], add)

1
value type：字段值是什么类型
2
update type：字段如何被更新

1
如果某个字段可能被并行节点同时写入，必须显式声明 reducer。
2
否则框架不会帮你猜“追加”还是“覆盖”。

1
def __init__(self, typ: type[Value], operator: Callable[[Value, Value], Value]):
2
    ...

1
_get_channel
2
  ↓
3
_is_field_binop
4
  ↓
5
BinaryOperatorAggregate(typ, operator)

1
def __init__(self, typ, operator):
2
    super().__init__(typ)
3

4
    # 1. 保存 reducer 函数
5
    self.operator = operator
6

7
    # 2. 处理 typing 抽象类型，转成可实例化类型
8
    typ = strip_required_notrequired_annotated(typ)
9
    if typ is Sequence or MutableSequence:
10
        typ = list
11
    if typ is Set or MutableSet:
12
        typ = set
13
    if typ is Mapping or MutableMapping:
14
        typ = dict
15

16
    # 3. 尝试初始化空值
17
    try:
18
        self.value = typ()
19
    except Exception:
20
        self.value = MISSING

1
self.operator(self.value, value)

1
Sequence[str]

1
list() -> []

1
Annotated[list[str], add]
2
  ↓
3
BinaryOperatorAggregate(list[str], add)
4
  ↓
5
self.operator = add
6
self.value = []

1
[] + ["food"] + ["attraction"]

1
{"research_notes": []}

1
graph.invoke(input: InputT, config: RunnableConfig | None = None, **kwargs) -> OutputT

1
Caller
2
  ↓
3
CompiledStateGraph.invoke
4
  ↓
5
Pregel.invoke / stream
6
  ↓
7
initialize input channels

1
def invoke(input_state, config=None):
2
    # 1. 准备运行配置
3
    config = ensure_config(config)
4

5
    # 2. 基于 compiled graph 的 channels 创建本次运行的 channel 副本
6
    channels = clone_compiled_channels()
7

8
    # 3. 把 initial state 写入输入 channels
9
    for key, value in input_state.items():
10
        if key not in channels:
11
            handle_unknown_key(key)
12
            continue
13

14
        # 初始输入也走 channel update 语义
15
        channels[key].update([value])
16

17
    # 4. 启动 Pregel loop
18
    return run_pregel_loop(channels, config)

1
channels["research_notes"].update([[]])

1
input_state["research_notes"] = []
2
  ↓
3
research_notes channel.update([[]])
4
  ↓
5
channel value = []

1
不要假设 state 是直接拷贝的 dict；
2
最终 state 是 channels 当前值的投影。

1
def run_node(node, channels, config, runtime):
2
    # 1. 从 channels 读取当前 state 快照
3
    state = read_state_snapshot(channels, node.input_schema)
4

5
    # 2. 调用节点
6
    update = node.invoke(state, config=config, runtime=runtime)
7

8
    # 3. 节点只返回 partial update
9
    if not isinstance(update, dict):
10
        raise InvalidUpdateError("Expected dict update")
11

12
    # 4. 把 update 转换成 channel writes
13
    writes = []
14
    for key, value in update.items():
15
        writes.append(ChannelWrite(key, value))
16

17
    return writes

1
search_food(state)
2
  ↓
3
{"research_notes": ["推荐美食：..."]}
4
  ↓
5
ChannelWrite(channel="research_notes", value=["推荐美食：..."])

1
1. 所有更新都能被 Pregel 统一排序、合并和记录。
2
2. 并行节点之间不会直接共享可变对象，降低竞态风险。

1
node 内部不要原地修改 state["research_notes"].append(...)
2
应该 return {"research_notes": [new_note]}
3
让 reducer 合并。

1
当前 super-step 的所有已激活节点执行完成之后，
2
下一轮节点被激活之前。

1
def apply_writes(
2
    channels: dict[str, BaseChannel],
3
    pending_writes: list[Write],
4
) -> None:
5
    ...

1
Pregel loop
2
  ↓
3
execute selected tasks
4
  ↓
5
collect pending writes
6
  ↓
7
apply_writes
8
  ↓
9
channel.update(values)

1
def apply_writes(channels, pending_writes):
2
    # 1. 按 channel 名分组
3
    writes_by_channel = defaultdict(list)
4

5
    for write in pending_writes:
6
        channel_name = write.channel
7
        value = write.value
8
        writes_by_channel[channel_name].append(value)
9

10
    # 2. 对每个被写入的 channel 应用更新
11
    for channel_name, values in writes_by_channel.items():
12
        channel = channels[channel_name]
13

14
        # 3. 这里才真正触发 reducer 或覆盖逻辑
15
        changed = channel.update(values)
16

17
        # 4. 记录哪些 channel 发生了变化，用于触发下游节点、checkpoint、stream
18
        if changed:
19
            mark_channel_changed(channel_name)
20

21
    # 5. 未收到更新的 channel 不变

1
{"research_notes": [...]}

1
writes_by_channel["research_notes"] = [food_notes, attraction_notes]

1
channel.update(values)

1
pending_writes:
2
  search_food -> research_notes = [food]
3
  search_attractions -> research_notes = [attraction]
4

5
writes_by_channel:
6
  research_notes -> [[food], [attraction]]
7

8
channel:
9
  BinaryOperatorAggregate(list, add)
10

11
channel.update:
12
  [] + [food] + [attraction]

1
只有 Pregel 运行时知道同一 super-step 里有哪些节点并行完成，
2
也只有它能把所有 writes 聚合到同一个 channel 后统一处理。

1
Reducer 是并行语义的一部分。
2
只在单链路顺序执行时看不出它的重要性；
3
一旦 fan-out 并行写同一个 key，reducer 就决定系统是否能运行。

1
def update(self, values: Sequence[Value]) -> bool:
2
    ...

1
apply_writes
2
  ↓
3
channels[key].update(values)
4
  ↓
5
LastValue.update(values)

1
def update(self, values):
2
    # 1. 没有更新，不改变 channel
3
    if len(values) == 0:
4
        return False
5

6
    # 2. 默认 channel 一个 step 只能收到一个值
7
    if len(values) != 1:
8
        raise InvalidUpdateError(
9
            "Can receive only one value per step. Use an Annotated key to handle multiple values."
10
        )
11

12
    # 3. 保存唯一值
13
    self.value = values[-1]
14
    return True

1
取第一个？
2
取最后一个？
3
拼接？
4
合并 dict？
5
取最大？
6
去重？

1
search_food 返回 {destination: "东京"}
2
  ↓
3
destination channel values = ["东京"]
4
  ↓
5
LastValue.update
6
  ↓
7
state.destination = "东京"

1
destination
2
current_step
3
final_plan
4
risk_level
5
status

1
messages
2
search_results
3
research_notes
4
tool_observations
5
parallel_agent_outputs

1
如果字段可能被多个并行节点写入，不能使用默认 LastValue。

1
def update(self, values: Sequence[Value]) -> bool:
2
    ...

1
apply_writes
2
  ↓
3
channels["research_notes"].update(values)
4
  ↓
5
BinaryOperatorAggregate.update(values)
6
  ↓
7
self.operator(self.value, value)

1
def update(self, values):
2
    # 1. 没有更新，不改变 channel
3
    if not values:
4
        return False
5

6
    # 2. 如果当前没有值，用第一个 update 初始化
7
    if self.value is MISSING:
8
        self.value = values[0]
9
        values = values[1:]
10

11
    seen_overwrite = False
12

13
    # 3. 逐个处理剩余 update
14
    for value in values:
15
        is_overwrite, overwrite_value = get_overwrite(value)
16

17
        # 4. Overwrite 分支：绕过 reducer 直接覆盖
18
        if is_overwrite:
19
            if seen_overwrite:
20
                raise InvalidUpdateError("Only one Overwrite per super-step")
21
            self.value = overwrite_value
22
            seen_overwrite = True
23
            continue
24

25
        # 5. 普通 reducer 分支
26
        if not seen_overwrite:
27
            self.value = self.operator(self.value, value)
28

29
    return True

1
self.value = self.operator(self.value, value)

1
["food"] + ["attraction"]

1
["food", "attraction"]

1
初始 research_notes = []
2

3
values = [
4
  ["推荐美食：寿司、拉面、居酒屋"],
5
  ["推荐景点：浅草寺、涩谷、上野公园"]
6
]
7

8
operator.add([], [food])
9
  ↓
10
[food]
11

12
operator.add([food], [attraction])
13
  ↓
14
[food, attraction]