Topic and BinaryOperatorAggregate: accumulating channels and reducers

源码版本1.2.9

Responsibilities

LastValue only overwrites, but real agent state often needs to accumulate: message lists append, counters add, todos merge. LangGraph offers two paths for this — Topic (Topic class:23) is a pub/sub-style list channel, and BinaryOperatorAggregate (BinaryOperatorAggregate class:65) is a general channel that accepts any binary operator (reducer). Both inherit directly or indirectly from BaseChannel (BaseChannel:19).

BinaryOperatorAggregate is the implementation behind declarations like Annotated[list, add_messages]. At compile time StateGraph calls _is_field_binop (_is_field_binop:1890-1908), treats the last two-argument callable in the Annotated metadata as a reducer, and instantiates BinaryOperatorAggregate(typ, reducer). Each update folds new values into the current value via operator (update:123-144).

Topic is more commonly used for internal channels: every Pregel instance ships with a Topic(Send, accumulate=False) channel named __pregel_tasks (TASKS channel:809) to hold Send fan-out. When accumulate=False it clears every step — the previous step's Send list is dropped once prepare_next_tasks consumes it, and the next step starts collecting fresh.

The shared trait is "update accepts multiple values and folds them by some rule" — that is the fundamental difference from LastValue, which accepts only one value.

Design motivation

• Reducer is the universal interface for state merging: different fields have wildly different merge semantics — messages dedupes by ID and appends, counter adds, tags unions. Rather than writing a channel class for each, provide a general channel that takes a binary operator operator: (Value, Value) -> Value and let the user define the merge rule. add_messages (add_messages:60-66) is one such reducer: merge by message ID, append on new ID, replace on existing ID.
• Rehydrate after type erasure: _strip_extras (_strip_extras:22-28) peels off typing wrappers like Annotated / Required / NotRequired, then uses typ() to instantiate a default value. Abstract types like collections.abc.Sequence cannot be instantiated directly, so typ fallback:83-88 maps them back to list / set / dict (typ instantiation:82-92).
• Topic.accumulate dual mode: when accumulate=True it is a cross-step accumulating list — update extends new values in (update:77-85); when accumulate=False it clears the old value first, then extends, becoming "only values produced this step". The latter is exactly what the TASKS channel needs: a Send is valid for one step and discarded once consumed.
• _flatten supports mixed writes: Topic.update accepts a mixed sequence of Value | list[Value] (_flatten:15-20), so a node can write one Send or [Send, Send] in one go, and the channel flattens it uniformly.
• Overwrite bypass: reducer channels also support a "direct overwrite" semantics: the Overwrite dataclass (Overwrite:980) or the dict {"__overwrite__": value} (OVERWRITE key:95) is recognized by _get_overwrite (_get_overwrite:31-51) and bypasses the reducer to assign directly in update. At most one Overwrite is allowed per step; a second one raises InvalidUpdateError (overwrite dedup:132-138).

Key files

• Topic class:23-41 — class definition, generic parameters Sequence[Value] / Value | list[Value] / list[Value]; the constructor takes an accumulate toggle.
• _flatten:15-20 — utility: flattens a mixed Value | list[Value] sequence into an Iterator[Value].
• Topic.copy:56-61 — overrides copy to reuse a copy of the values list directly, skipping checkpoint serialization.
• Topic.checkpoint / from_checkpoint:63-75 — checkpoint returns self.values directly; from_checkpoint is also compatible with the legacy tuple form.
• Topic.update:77-85 — core logic: when accumulate=False, clear first then extend; when accumulate=True, just extend; an empty flat sequence returns False.
• Topic.get / is_available:87-94 — empty list raises EmptyChannelError; is_available is overridden as bool(self.values).
• Binop class:65-92 — class definition + __init__: takes a binary operator callable, initializes the stored value with typ().
• _get_overwrite:31-51 — recognizes three Overwrite representations (dataclass, sentinel-keyed dict, dict restored from JSON).
• _operators_equal:54-62 — special lambda comparison: same-named lambdas are treated as equal, so recompiling the graph does not flag channels as changed.
• Binop.update:123-144 — the real reducer entry point: the first value seeds, subsequent values each call operator(self.value, value); the Overwrite bypass is the exception.
• _is_field_binop:1890-1908 — at compile time, recognizes the two-argument callable in Annotated[..., reducer] metadata and instantiates a BinaryOperatorAggregate.
• TASKS channel:805-809 — Pregel.__init__ hardcodes Topic(Send, accumulate=False) for the __pregel_tasks channel; user overrides are not allowed.
• add_messages:60-66 — typical reducer: merges two lists by message ID, appending on new ID, replacing on existing ID.

Data flow

BinaryOperatorAggregate.update is the heart of the reducer channel (update:123-144):

def update(self, values: Sequence[Value]) -> bool:
    if not values:
        return False
    if self.value is MISSING:
        self.value = values[0]
        values = values[1:]
    seen_overwrite: bool = False
    for value in values:
        is_overwrite, overwrite_value = _get_overwrite(value)
        if is_overwrite:
            if seen_overwrite:
                msg = create_error_message(
                    message="Can receive only one Overwrite value per super-step.",
                    error_code=ErrorCode.INVALID_CONCURRENT_GRAPH_UPDATE,
                )
                raise InvalidUpdateError(msg)
            self.value = overwrite_value
            seen_overwrite = True
            continue
        if not seen_overwrite:
            self.value = self.operator(self.value, value)
    return True

Two branches to note: if the channel is empty, the first value seeds directly (skipping one reducer call, since the reducer needs two inputs); afterwards, each value that is not an Overwrite becomes self.value = operator(self.value, value). Once an Overwrite appears, every subsequent non-overwrite value is discarded — this guarantees that the "overwrite" semantics will not be reverted by a later reducer.

How StateGraph compiles Annotated[list[AnyMessage], add_messages] into BinaryOperatorAggregate is in _is_field_binop:1890-1908:

def _is_field_binop(typ: type[Any]) -> BinaryOperatorAggregate | None:
    if hasattr(typ, "__metadata__"):
        meta = typ.__metadata__
        if len(meta) >= 1 and callable(meta[-1]):
            sig = signature(meta[-1])
            params = list(sig.parameters.values())
            if (
                sum(
                    p.kind in (p.POSITIONAL_ONLY, p.POSITIONAL_OR_KEYWORD)
                    for p in params
                )
                == 2
            ):
                return BinaryOperatorAggregate(typ, meta[-1])
            else:
                raise ValueError(
                    f"Invalid reducer signature. Expected (a, b) -> c. Got {sig}"
                )
    return None

It uses inspect.signature to verify that the last metadata entry is a two-argument callable; if the signature does not match, it raises. That is why a two-argument function like add_messages(left, right) can serve as a reducer while a single-argument function is rejected.

Topic.update takes a different, shorter path (update:77-85):

def update(self, values: Sequence[Value | list[Value]]) -> bool:
    updated = False
    if not self.accumulate:
        updated = bool(self.values)
        self.values = list[Value]()
    if flat_values := tuple(_flatten(values)):
        updated = True
        self.values.extend(flat_values)
    return updated

Note the return value when accumulate=False: if the old value was non-empty, it returns True even when the new sequence is empty (because clearing is itself a change), and Pregel will bump channel_versions. This is the opposite of LastValue, which returns False on an empty sequence — reflecting the different semantics these channels attach to "empty write".

The reducer channel within one Pregel superstep:

Boundaries and failure modes

• Three Overwrite forms coexist: overwrite forms:44-50 accepts the Overwrite dataclass, the {"__overwrite__": value} dict, and the {"type": "__overwrite__", "value": ...} dict restored from JSON. The third branch is there so that a round-trip through the LangGraph API server's orjson serialization does not lose semantics.
• At most one Overwrite per step: overwrite dedup:132-138 raises InvalidUpdateError on the second occurrence, because "overwrite" appearing more than once in a single step is contradictory.
• Non-overwrite values after an Overwrite are silently discarded: discard after overwrite:142-143 if not seen_overwrite decides whether to call the reducer; once seen_overwrite=True, subsequent values neither enter the reducer nor raise. This is intentional — "overwrite" means we do not want to keep the previous merge result, and we do not need later values either.
• Lambda reducer equality: _operators_equal:54-62 treats all lambdas as equal, because in Python each <lambda> is a new object; comparing by identity would make __eq__ never true and break graph caching.
• Topic(accumulate=False) returns True on empty write: clear returns True:79-80 — as long as the old value was non-empty, clearing counts as a change. This causes apply_writes to bump channel_versions and trigger any node depending on this channel. If you use Topic directly, be aware that it does not handle empty writes as silently as LastValue.
• Topic legacy checkpoint compatibility: tuple compatibility:70-74 checks isinstance(checkpoint, tuple); old versions stored checkpoints as (typ, values) tuples, the new version stores list directly. This block is there to avoid breaking old archives.
• Reducer signature requires two arguments: reducer signature check:1894-1907 uses inspect.signature to count positional parameters and raises if the count is not two. Functions with variadic *args are rejected — this nails down "reducer must be strictly (a, b) -> c".
• add_messages is a factory: _add_messages_wrapper:41-57 wraps with @_add_messages_wrapper; the resulting function can be called directly as add_messages(left, right) or as add_messages(format="langchain-openai") to return a partial. This is a common extension pattern for reducers — adding runtime parameters to a reducer.

Summary

BinaryOperatorAggregate is the unified implementation for "fields with a reducer"; Topic is the two forms of "list accumulation": cross-step accumulating or per-step clearing. Together with /channels/last-value they make up the three pillars of the LangGraph channel system, all built on the interface defined in /channels/base-channel. Reducers like add_messages are declared via Annotated[..., reducer] in /graph/state-graph and called from apply_writes in /pregel/algo into update. Topic(accumulate=False) as the __pregel_tasks channel is the carrier of Send fan-out, see /subgraph/send-command.

See official docs: LangGraph 文档 · README.