avatar yuguan bitou
发布日期

第 03 讲 · Channel 抽象：BaseChannel 接口契约

深入 BaseChannel 接口契约：update/get/checkpoint/consume/finish 语义解析

langgraphchannelstate

学习目标

  • 吃透 BaseChannel 的 5 个核心方法语义：update / get / checkpoint / consume / finish。
  • 理解通道的"读/写/序列化/触发"四类职责如何分工。
  • 知道这些方法分别在超步的哪个阶段被调用。

通道是什么

回顾第 1 讲：节点不直接通信，全部通过 通道（channel） 读写。 一个通道就是一个带版本、可序列化、定义了"多值如何合并"的状态格子。

源码：libs/langgraph/langgraph/channels/base.py，核心类 BaseChannel。

26:libs/langgraph/langgraph/channels/base.py
class BaseChannel(Generic[Value, Update, Checkpoint], ABC):
    """Base class for all channels."""

    __slots__ = ("key", "typ")

    def __init__(self, typ: Any, key: str = "") -> None:

三个泛型参数点明了通道的三种"形态"：

  • Value：get() 读出来的值类型。
  • Update：update() 接收的单条更新类型。
  • Checkpoint：checkpoint() 序列化出去、from_checkpoint() 还原回来的类型。

很多通道三者相同（如 LastValue），但有的不同（如 Topic 读出 Sequence、更新可为单值或列表）。

五个核心方法

1. update(values) — 写（Update 阶段调用）

99:libs/langgraph/langgraph/channels/base.py
    @abstractmethod
    def update(self, values: Sequence[Update]) -> bool:
        """Update the channel's value with the given sequence of updates.
        The order of the updates in the sequence is arbitrary.
        This method is called by Pregel for all channels at the end of each step.

        If there are no updates, it is called with an empty sequence.

关键点：

  • 批量：参数是本超步内所有写该通道的更新的序列，顺序不保证。
  • 每步都调：即使没有更新，也会用空序列调用一次（用于"新超步开始"的通知）。
  • 返回 bool：True 表示通道真的变了（需要 bump 版本），False 表示没变。
  • 这正是 reducer 的落点：LastValue 只许一个值，BinaryOperatorAggregate 把多个值 fold 起来。

这就是第 1 讲"步内不可变、步末批量合并"的代码体现：update 由 apply_writes 在 after_tick 阶段统一调用（第 11 讲细讲）。

2. get() — 读（Plan / 执行阶段调用）

73:libs/langgraph/langgraph/channels/base.py
    @abstractmethod
    def get(self) -> Value:
        """Return the current value of the channel.

        Raises `EmptyChannelError` if the channel is empty (never updated yet)."""
  • 通道从没被写过时 get() 抛 EmptyChannelError，这是 LangGraph 内部到处用的"空"信号。
  • 配套 is_available() 用来"问而不取"，子类常重写它以避免异常开销。

3. checkpoint() / from_checkpoint() — 序列化（保存/恢复）

58:libs/langgraph/langgraph/channels/base.py
    def checkpoint(self) -> Checkpoint | Any:
        """Return a serializable representation of the channel's current state.

        Raises `EmptyChannelError` if the channel is empty (never updated yet),
        or doesn't support checkpoints.
        """
        try:
            return self.get()
        except EmptyChannelError:
            return MISSING
  • checkpoint() 把通道当前状态变成可序列化对象，存进 Checkpoint.channel_values。
  • from_checkpoint(blob) 反向：从存档重建一个等价通道。
  • 默认 copy() = from_checkpoint(checkpoint())，子类可重写更高效的拷贝。
  • 注意 MISSING 哨兵：空通道不写进 checkpoint。

4. consume() — 消费（被触发后调用，默认 no-op）

110:libs/langgraph/langgraph/channels/base.py
    def consume(self) -> bool:
        """Notify the channel that a subscribed task ran.

        By default, no-op.

        A channel can use this method to modify its state, preventing the value from being consumed again.

        Returns `True` if the channel was updated, `False` otherwise.
        """
        return False
  • 当一个订阅了该通道的节点被触发执行后，框架调 consume()。
  • 用途：让通道"用过即焚"，避免同一份值反复触发同一节点。
  • 典型实现：Topic(accumulate=False)、NamedBarrierValue（凑齐后清空 seen）。

5. finish() — 收尾（run 结束前调用，默认 no-op）

121:libs/langgraph/langgraph/channels/base.py
    def finish(self) -> bool:
        """Notify the channel that the Pregel run is finishing.
  • 整个 run 即将结束时调用。
  • 用途：...AfterFinish 系列通道靠它把"暂存值"在最后一刻才对外暴露（第 5 讲）。

方法 × 超步阶段 对照表

方法调用时机由谁调
from_checkpointrun 开始、加载存档channels_from_checkpoint（第 17 讲）
get / is_availablePlan 组装节点输入、执行时读_proc_input / local_read
updateUpdate 阶段（超步末）apply_writes（第 11 讲）
consume节点被触发后apply_writes
finishrun 结束前apply_writes(... is_finish=True)
checkpoint超步末存档create_checkpoint（第 15/17 讲）

把这张表记牢，后面读执行引擎就不会迷路——所有通道方法的调用都集中在 apply_writes 和加载/存档两处。

使用场景：什么时候要自定义 Channel

99% 的情况你用内置通道 + reducer 就够了。但以下场景可能要自定义 BaseChannel：

  • 特殊合并语义：比如"只保留 top-K 分数"、"按时间窗口滑动"。
  • 副作用感知：在 consume() 里做"已读标记"、在 finish() 里做"收尾刷盘"。
  • 自定义序列化：通道内部是不可直接 JSON 化的对象，需要在 checkpoint/from_checkpoint 里转换。

自定义时务必遵守契约：update 返回是否变化、空通道 get 抛 EmptyChannelError、 checkpoint 产出可被 serde 处理的对象（第 16 讲）。

动手实验

实现一个"只保留最大值"的通道，验证你对契约的理解：

from collections.abc import Sequence
from langgraph.channels.base import BaseChannel
from langgraph._internal._typing import MISSING
from langgraph.errors import EmptyChannelError

class MaxValue(BaseChannel):
    def __init__(self, typ, key=""):
        super().__init__(typ, key)
        self.value = MISSING

    @property
    def ValueType(self): return self.typ
    @property
    def UpdateType(self): return self.typ

    def from_checkpoint(self, checkpoint):
        c = self.__class__(self.typ, self.key)
        if checkpoint is not MISSING:
            c.value = checkpoint
        return c

    def update(self, values: Sequence) -> bool:
        if not values:
            return False
        new = max(values) if self.value is MISSING else max(self.value, *values)
        changed = new != self.value
        self.value = new
        return changed

    def get(self):
        if self.value is MISSING:
            raise EmptyChannelError()
        return self.value

用 Annotated[int, MaxValue] 接到 StateGraph 状态字段上（第 7 讲讲如何映射），多次写入观察是否只留最大值。

阅读作业

  • 精读 channels/base.py 全文，对每个方法标注它在"读/写/序列化/触发"哪一类。
  • 对照 channels/last_value.py，看 LastValue 如何实现这 5 个方法（下一讲展开）。

小结

  • 通道 = 带版本、可序列化、定义了合并语义的状态格子。
  • 五方法：update（批量写）、get（读）、checkpoint/from_checkpoint（存档）、consume（用过即焚）、finish（收尾）。
  • 所有调用都集中在 apply_writes 与加载/存档，记住这点读引擎不迷路。

下一讲：内置通道家族——LastValue / Topic / BinaryOperatorAggregate / EphemeralValue。