所有通道都放在 libs/langgraph/langgraph/channels/ 目录，公共父类只有一个文件：channels/base.py。我们先看它的类头和构造：

# channels/base.py:12
Value = TypeVar("Value")
Update = TypeVar("Update")
Checkpoint = TypeVar("Checkpoint")

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

    __slots__ = ("key", "typ")            # base.py:22

    def __init__(self, typ: Any, key: str = "") -> None:   # base.py:24
        self.typ = typ
        self.key = key

BaseChannel[Value, Update, Checkpoint] 这三个类型参数是理解一切通道的钥匙。它们回答三个不同的问题：

关键洞察：这三者可以不一样！看抽象属性 ValueType / UpdateType（base.py:28-36）：

    @property
    @abstractmethod
    def ValueType(self) -> Any:        # base.py:28
        """The type of the value stored in the channel."""

    @property
    @abstractmethod
    def UpdateType(self) -> Any:       # base.py:33
        """The type of the update received by the channel."""

update() 是通道最重要的方法——你写的每一个 reducer，最终都是某个通道的 update 逻辑。它是抽象方法，父类只定契约、不给实现（base.py:89-99）：

    @abstractmethod
    def update(self, values: Sequence[Update]) -> bool:   # base.py:89
        """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.

        Raises `InvalidUpdateError` if the sequence of updates is invalid.

        Returns `True` if the channel was updated, `False` otherwise."""

这段 docstring 每一句都是"硬契约"，逐句拆：

写完看读。读侧有两个方法，一个抽象（必须实现）、一个有默认实现（base.py:69-85）：

    @abstractmethod
    def get(self) -> Value:            # base.py:69
        """Return the current value of the channel.
        Raises `EmptyChannelError` if the channel is empty (never updated yet)."""

    def is_available(self) -> bool:    # base.py:75
        """Return `True` if the channel is available (not empty), `False` otherwise.
        Subclasses should override this method to provide a more efficient
        implementation than calling `get()` and catching `EmptyChannelError`.
        """
        try:
            self.get()
            return True
        except EmptyChannelError:
            return False

断点续跑（阶段 6 的核心）能实现，靠的就是通道会"拍照存档"和"照存档还原"。看这对方法（base.py:49-65）：

    def checkpoint(self) -> Checkpoint | Any:      # base.py:49
        """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

    @abstractmethod
    def from_checkpoint(self, checkpoint: Checkpoint | Any) -> Self:   # base.py:60
        """Return a new identical channel, optionally initialized from a checkpoint.
        If the checkpoint contains complex data structures, they should be copied."""

父类还提供了一个基于这对方法的 copy()（base.py:40-47）：

    def copy(self) -> Self:            # base.py:40
        """Return a copy of the channel.
        By default, delegates to `checkpoint()` and `from_checkpoint()`.
        Subclasses can override this method with a more efficient implementation."""
        return self.from_checkpoint(self.checkpoint())

最后两个方法很多人忽略，但它们是 Day 30/32 那些"发布订阅""同步屏障"通道的命门。两个都有默认实现（都默认 no-op），子类按需重写（base.py:101-121）：

    def consume(self) -> bool:         # base.py:101
        """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

    def finish(self) -> bool:          # base.py:112
        """Notify the channel that the Pregel run is finishing.
        By default, no-op.
        A channel can use this method to modify its state, preventing finish.
        Returns `True` if the channel was updated, `False` otherwise."""
        return False

把接口和引擎连起来看，理解才落地。Day 21 的 apply_writes（pregel/_algo.py）就是把上面所有方法串起来用的地方，节选核心四段：

# _algo.py:291 —— ① 被读过的通道，通知它"订阅者跑过了"
for chan in {...读过的通道...}:
    if channels[chan].consume() and next_version is not None:
        checkpoint["channel_versions"][chan] = next_version

# _algo.py:319 —— ② 有人写的通道，落写入；返回 True 才推进版本
for chan, vals in pending_writes_by_channel.items():
    if channels[chan].update(vals) and next_version is not None:
        checkpoint["channel_versions"][chan] = next_version
        if channels[chan].is_available():
            updated_channels.add(chan)

# _algo.py:329 —— ③ 这一步没人写的通道，用空序列通知它"新的一步到了"
if channels[chan].update(EMPTY_SEQ) and next_version is not None:
    ...

# _algo.py:338 —— ④ 若判定是最后一步，通知所有通道 finish
if channels[chan].finish() and next_version is not None:
    ...

✋ 10 分钟动手

# 1. 通读通道基类（今天的主角，就 122 行）
sed -n '1,122p' libs/langgraph/langgraph/channels/base.py

# 2. 看目录下都有哪些具体通道（接下来 5 天的地图）
cat libs/langgraph/langgraph/channels/__init__.py

# 3. 看引擎在哪调用 consume/update/finish
sed -n '284,340p' libs/langgraph/langgraph/pregel/_algo.py

# 4. 看编译时"字段→通道"的兜底选择
sed -n '1845,1862p' libs/langgraph/langgraph/graph/state.py

# 5. 自己写个最小通道验证接口（继承 BaseChannel，实现 update/get/from_checkpoint）