第 06 讲 · DeltaChannel 与增量快照（beta）

学习目标

理解 DeltaChannel 解决的问题：大状态/长历史的存储膨胀。
掌握"只存增量 + 按需 replay 重建"的机制。
理解 reducer 的 batching-invariant（折叠不变性） 约束为什么必须成立。
知道它的 beta 风险与适用边界。

注意：DeltaChannel 当前标记为 beta，磁盘表示与周边契约可能变化。生产使用前先评估。

问题背景：reducer 通道的存储膨胀

像 BinaryOperatorAggregate 或 add_messages 这类累积型通道，状态会越来越大（消息列表越来越长）。而 LangGraph 默认每个超步都把通道完整值快照进 checkpoint。于是：一个 1000 轮的对话，可能存了 1000 份"几乎一样、只多一条消息"的完整列表—— 存储和 IO 都浪费。

DeltaChannel 的思路：大多数 checkpoint 不存完整值，只存一个哨兵；要读时把祖先的"增量写" 重新 replay 一遍，重建出当前值。 偶尔（按频率）才存一份完整快照，限制 replay 深度。

源码：libs/langgraph/langgraph/channels/delta.py。

核心契约：reducer 必须 batching-invariant

48:libs/langgraph/langgraph/channels/delta.py

        reducer(reducer(state, xs), ys) == reducer(state, xs + ys)

    This lets LangGraph replay checkpointed writes in larger batches than they
    were originally produced without changing reconstructed state.

含义：分两批折叠 和 合并成一批折叠 必须得到相同结果。为什么必须？因为 replay 时 LangGraph 会把历史上分散在多个超步的写合并成更大的批 一次性喂给 reducer，以减少调用次数。若 reducer 不满足这条，重建出的值就会和原值不同——数据损坏。

operator.add（list 拼接）天然满足；但带"去重/限长/依赖顺序敏感"的 reducer 要特别小心。

它还要求 reducer 形态是 reducer(state, [w1, w2, ...]) -> new_state（一次收一批），而不是逐个二元折叠：

39:libs/langgraph/langgraph/channels/delta.py

    The reducer receives the current accumulated value and a batch of writes
    in one call: `reducer(state, [write1, write2, ...]) -> new_state`.

update：在内存里就是普通 reducer

185:libs/langgraph/langgraph/channels/delta.py

        base = self.typ() if self.value is MISSING else self.value
        self.value = self.reducer(base, list(values))
        return True

运行期它和普通累积通道没区别：把本步写折叠进当前值。差异全在序列化（checkpoint）上。

checkpoint：默认只存哨兵 MISSING

204:libs/langgraph/langgraph/channels/delta.py

    def checkpoint(self) -> Any:
        """Return stored representation: always `MISSING`.

        Snapshot decisions live in `create_checkpoint` (which has the channel
        version) and write `_DeltaSnapshot(ch.get())` directly into
        `channel_values`. For non-snapshot steps the channel does not appear
        in `channel_values`; reconstruction walks ancestor writes via the
        saver's `get_delta_channel_history`.

通道自己的 checkpoint() 永远返回 MISSING——意思是"非快照步不要存我的完整值"。
真正的"要不要存完整快照"的决策在 create_checkpoint（它有 channel 版本号信息），快照时往 channel_values 写一个 _DeltaSnapshot(value)。

快照节奏（snapshot cadence）：两个计数器

55:libs/langgraph/langgraph/channels/delta.py

    Snapshot cadence is driven by two counters: per-channel update count and
    total supersteps since last snapshot. `create_checkpoint` writes a full
    `_DeltaSnapshot` blob when EITHER the update count reaches
    `snapshot_frequency` OR the supersteps count reaches the system-wide
    `DELTA_MAX_SUPERSTEPS_SINCE_SNAPSHOT` bound (default 5000), bounding
    replay depth even for channels that stop receiving writes.

per-channel 更新计数达到 snapshot_frequency（默认 1000）→ 存快照。
距上次快照的超步数达到系统上界 DELTA_MAX_SUPERSTEPS_SINCE_SNAPSHOT（默认 5000）→ 存快照。
两个条件任一满足即快照。第二个条件保证"即使某通道很久不更新，replay 深度也有上界"。

重建：from_checkpoint + replay_writes

137:libs/langgraph/langgraph/channels/delta.py

    def from_checkpoint(self, checkpoint: Any) -> Self:
        """Initialize from a stored blob.

        Blob types:
          * `MISSING`: start empty; caller replays writes.
          * `_DeltaSnapshot(value)`: restore value directly from snapshot.
          * plain value (migration from old `BinaryOperatorAggregate` blobs):
            use directly.
        """

157:libs/langgraph/langgraph/channels/delta.py

    def replay_writes(self, writes: Sequence[PendingWrite]) -> None:
        """Apply ancestor writes oldest-to-newest via a single reducer call.

        If any write is an Overwrite, the last one in the sequence acts as
        the reset point: its value becomes the new base and only writes
        after it are passed to the reducer.
        """

重建逻辑：

找到最近的快照（_DeltaSnapshot）作为基底；没有则从空开始。
把快照之后的所有祖先写按时间序拼成一批，一次 reducer(base, writes) 重放（这里就用到了 batching-invariant）。
若中途有 Overwrite，最后一个 Overwrite 作为"重置点"，只重放它之后的写。

这套"祖先写 walk + 重放"由 checkpoint saver 的 get_delta_channel_history 提供（第 17 讲会看到 Postgres/SQLite 各自的实现）。

机制全景图

flowchart TD
    subgraph 写入路径
        W[每步 update] --> M[内存里 reducer 折叠]
        M --> CK{到快照频率?}
        CK -->|是| SNAP[channel_values 存 _DeltaSnapshot]
        CK -->|否| SENT[只留哨兵, 不存完整值]
    end
    subgraph 读取路径
        L[from_checkpoint] --> F{有快照?}
        F -->|有| BASE[以快照为基底]
        F -->|无| EMPTY[从空开始]
        BASE --> RP[replay 祖先写]
        EMPTY --> RP
        RP --> V[重建当前值]
    end

使用场景

超长对话/Agent 记忆：消息历史动辄上千条，用 DeltaChannel 大幅降低每步 checkpoint 体积。
大型累积状态：如不断追加的检索结果、日志、事件流。
何时别用：状态本身很小（省不了多少）、reducer 不满足 batching-invariant、或你需要稳定的磁盘格式（beta 不保证）——这些情况继续用 BinaryOperatorAggregate / add_messages。

动手实验

用一个 Annotated[list, DeltaChannel(operator.add 风格的批量 reducer)] 字段（注意 reducer 形态是 (state, list_of_writes) -> state），跑很多步，观察 checkpoint 大小相比普通累积通道的变化。
故意写一个非 batching-invariant 的 reducer（如对写入去重且依赖到达顺序），触发 replay 后对比重建值与运行期值，理解契约为何必须成立。

阅读作业

精读 channels/delta.py 全文，重点是 update / checkpoint / from_checkpoint / replay_writes。
预读 libs/langgraph/langgraph/pregel/_checkpoint.py 中 delta_channels_to_snapshot 与 create_checkpoint，看快照决策代码（第 15/17 讲细讲）。

小结

DeltaChannel = 增量存储 + 按需 replay 重建，解决累积状态的 checkpoint 膨胀。
核心约束：reducer 必须 batching-invariant，否则重建会损坏数据。
快照按"更新次数 OR 超步数"双计数器触发，限制 replay 深度。
beta 阶段：评估磁盘格式稳定性后再上生产。

模块二完结。下一模块进入"图构建与编译"——看 StateGraph 如何变成会跑的 Pregel。