BaseChannel abstraction: channels as state primitives

源码版本1.2.9

Responsibilities

In the LangGraph runtime, a "channel" is the smallest unit of state. A Pregel instance compiled from a StateGraph does not directly hold the user's dict; it holds a set of BaseChannel instances — one channel per key — and each channel is responsible for storing values, merging writes, exposing reads, and serializing its current state into a checkpoint. BaseChannel is the abstract base class shared by every channel type (BaseChannel:19).

It does not store data itself (__slots__ only declares key and typ, see __slots__:22-26). It only mandates four abstract members that subclasses must implement: the ValueType and UpdateType type attributes, and the update, get, and from_checkpoint actions (abstract methods:60-99). It also provides checkpoint, is_available, consume, finish, and copy with default implementations that either fall back to get() or return False as a no-op.

In other words, BaseChannel pins down "how state is stored, how it changes, and how it leaves a trace" into one interface. The Pregel engine only needs to call update / get / checkpoint — it does not need to know whether a LastValue, Topic, or BinaryOperatorAggregate is doing the merging behind the scenes. That is how the engine and the shape of state stay fully decoupled.

Design motivation

Why model state as "channel" objects instead of passing a dict between nodes?

• Unified write-merge semantics: within the same superstep, multiple nodes may concurrently write to the same key. A dict cannot express "list appends, scalars overwrite, counters add" — these are different merge rules. Channels funnel every strategy through a single update(values: Sequence[Update]) -> bool; subclasses decide how to fold.
• Immutable intra-step view: a node reads a snapshot taken at step start; values written by other nodes during the same step only become visible next step. get is a read of the current value; update is only invoked at step boundaries by apply_writes (apply_writes:317-323), which rules out races by construction.
• Serializable: every channel can checkpoint() to a serializable value (checkpoint:49-58) and rebuild itself via from_checkpoint (from_checkpoint:60-65). The BaseCheckpointSaver family persists an entire thread's state to SQLite/Postgres through these two methods.
• Lifecycle hooks: consume and finish (consume / finish:101-121) give special channels (such as Topic(accumulate=False) or LastValueAfterFinish) a chance to mutate their state at step boundaries or run end. The default no-op signals that most channels do not need this layer of semantics.
• Self-describing types: the ValueType / UpdateType attributes (ValueType / UpdateType:28-36) let both compile time and runtime reflect on "what this channel stores, what writes it accepts". StateGraph uses them in _is_field_channel:1862-1887 to recognize Annotated[..., SomeChannel] declarations.

Key files

• BaseChannel class:19-26 — abstract base class definition, generic parameters Value / Update / Checkpoint, __slots__ only declares key and typ.
• ValueType / UpdateType:28-36 — two abstract properties subclasses use to declare the stored and written value types.
• checkpoint:49-58 — default implementation: returns self.get(), or MISSING for an empty channel.
• from_checkpoint:60-65 — abstract method: build an equivalent new channel from a checkpoint value; subclasses must implement it.
• get:69-73 — abstract read interface; an empty channel raises EmptyChannelError.
• is_available:75-85 — default implementation calls get() and catches EmptyChannelError; subclasses usually override with a cheaper check.
• update:89-99 — abstract write interface, invoked by Pregel once at the end of each superstep, in arbitrary order; returns True if the channel was mutated.
• consume / finish:101-121 — lifecycle hooks, default no-op returning False.
• apply_writes:315-345 — the single entry point through which Pregel merges all of a step's task writes back into channels; update / consume / finish are all scheduled here.
• _is_field_channel:1862-1887 — at compile time StateGraph uses isinstance(item, BaseChannel) to turn Annotated[..., channel] into a channel instance.

Data flow

Channel writes happen inside apply_writes: Pregel groups the step's writes per channel and then calls update (apply_writes:317-323):

# Apply writes to channels
updated_channels: set[str] = set()
for chan, vals in pending_writes_by_channel.items():
    if chan in channels:
        if channels[chan].update(vals) and next_version is not None:
            checkpoint["channel_versions"][chan] = next_version
            # unavailable channels can't trigger tasks, so don't add them
            if channels[chan].is_available():
                updated_channels.add(chan)

# Channels that weren't updated in this step are notified of a new step
if bump_step:
    for chan in channels:
        if channels[chan].is_available() and chan not in updated_channels:
            if channels[chan].update(EMPTY_SEQ) and next_version is not None:
                checkpoint["channel_versions"][chan] = next_version
                if channels[chan].is_available():
                    updated_channels.add(chan)

Note the second block: channels not written by any task still receive update(EMPTY_SEQ). This is an implicit contract of the BaseChannel interface — a subclass's update must accept an empty sequence, typically returning False to signal "no change". The "clear every step" semantics of Topic(accumulate=False) rely on exactly this empty call: an empty update lets it drop the previous step's value so it becomes invisible next step.

The full channel lifecycle within a single Pregel tick:

Boundaries and failure modes

• EmptyChannelError: raised by get() when the channel has never been written (get:70-73). The default is_available implementation catches this exception to decide availability; subclasses that can decide more cheaply should override it.
• MISSING sentinel: when get() raises EmptyChannelError, checkpoint() returns langgraph._internal._typing.MISSING rather than None (checkpoint fallback:55-58). This way None can be a legitimate stored value and is not confused with "empty".
• update returns False: a False return means the channel did not change, so Pregel does not bump channel_versions and subsequent prepare_next_tasks will not trigger any node because of this channel (update call:319); this is the foundation of Pregel's convergence check.
• Empty-sequence invocation: apply_writes also calls update(EMPTY_SEQ) on unchanged channels (EMPTY_SEQ update:329). The BaseChannel.update contract is that empty input returns False, but Topic(accumulate=False) will clear itself in this case and may return True — so subclasses cannot assume "empty input = no-op".
• consume and finish default to no-op (consume / finish:101-121): returning False means most channels do not participate in lifecycle management. apply_writes calls finish at the end of bump_step (finish call:338); only special channels like LastValueAfterFinish use it to implement "only visible after the run ends" semantics.
• copy defaults to going through checkpoint: BaseChannel.copy defaults to self.from_checkpoint(self.checkpoint()) (copy:40-47). If checkpointing is expensive for a subclass (say, deep-copying a large list), it should override copy with a cheaper implementation — Topic.copy does exactly that.

Summary

BaseChannel pins "how state is stored, merged, and serialized" into one interface. The Pregel engine only needs to call it in the fixed order update → is_available → get → checkpoint → finish and it will run against any channel implementation. For concrete implementations see /channels/last-value and /channels/topic-binop; how Pregel reads and writes channels is covered in /pregel/algo; how the checkpoint layer persists the return value of checkpoint() is covered in /checkpoint/base-saver.

See official docs: LangGraph 文档 · README.