Topic と BinaryOperatorAggregate:累積チャネルとリデューサー
役割
LastValue は上書きしかできませんが、実際のエージェント状態では多くのフィールドが累積を要求します。メッセージリストは append、カウンタは加算、TODO はマージ。LangGraph はこの種の要件に2つの道を提供します。Topic(Topic class:23)は pub/sub 風のリストチャネルで、BinaryOperatorAggregate(BinaryOperatorAggregate class:65)は任意の二項演算子 (reducer) を受け取る汎用チャネルです。どちらも直接または間接的に BaseChannel(BaseChannel:19)を継承します。
BinaryOperatorAggregate は、ユーザーが書く Annotated[list, add_messages] という宣言の背後にある実装です。StateGraph はコンパイル時に _is_field_binop(_is_field_binop:1890-1908)を呼び、Annotated メタデータの最後にある2引数 callable を reducer として取り出し、BinaryOperatorAggregate(typ, reducer) をインスタンス化します。update が呼ばれるたびに新しい値を operator で現在値にマージします(update:123-144)。
Topic は内部チャネルとして使われることが多いです。すべての Pregel インスタンスは __pregel_tasks という名前で Topic(Send, accumulate=False)(TASKS channel:809)を持ち、Send のファンアウト (fan-out) を収容します。特徴は accumulate=False のとき毎ステップクリアされる点で、前ステップの Send リストは prepare_next_tasks で消費されたら捨てられ、次ステップで改めて収集されます。
両者に共通するのは「update が複数の値を受け取り、ある規則で fold する」ことで、これが LastValue との根本的な違いです。後者は値を1つしか受け付けません。
設計動機
- reducer は状態マージの汎用インターフェース:フィールドごとのマージセマンティクスは千差万別です。
messagesは ID で重複排除しつつ append、counterは加算、tagsは和集合。それぞれに channel クラスを作るより、operator: (Value, Value) -> Valueの二項演算子を受け取る汎用チャネルを用意し、ユーザーにマージ規則を定義させるほうがよいです。add_messages(add_messages:60-66)はそうした reducer の一例で、メッセージ ID でマージし、新規 ID は append、既存 ID は置換します。 - 型消去後に復元:
_strip_extras(_strip_extras:22-28)はAnnotated/Required/NotRequiredなどの typing ラッパーを剥がし、typ()でデフォルト値をインスタンス化します。collections.abc.Sequenceのような抽象型は直接インスタンス化できないため、typ fallback:83-88でlist/set/dictにマップし直します(typ instantiation:82-92)。 Topic.accumulateの2形態:accumulate=Trueのときはステップをまたいで累積するリストで、updateは新しい値をextendします(update:77-85)。accumulate=Falseのときは毎ステップ古い値をクリアしてから extend し、「当ステップで生じた値だけを見る」形になります。後者がまさにTASKSチャネルに必要な性質で、Sendは1ステップ内でのみ有効、消費されたら廃棄されます。_flattenが混合書き込みをサポート:Topic.updateはValue | list[Value]の混合シーケンスを受け付けます(_flatten:15-20)。そのためノードは1回でSendを1つ書いても、[Send, Send]を書いてもよく、チャネルが一律にフラット化します。Overwriteバイパス:reducer チャネルは「直接上書き」のセマンティクスもサポートします。Overwriteデータクラス(Overwrite:980)、または辞書{"__overwrite__": value}(OVERWRITE key:95)を_get_overwriteが識別し(_get_overwrite:31-51)、update内で reducer を迂回して直接代入します。1ステップで許されるOverwriteは1回だけで、2回目はInvalidUpdateErrorを投げます(overwrite dedup:132-138)。
主要ファイル
Topic class:23-41— クラス定義。ジェネリックパラメータSequence[Value]/Value | list[Value]/list[Value]。コンストラクタはaccumulateスイッチを受け取る。_flatten:15-20— ユーティリティ関数。混合Value | list[Value]シーケンスをIterator[Value]にフラット化。Topic.copy:56-61—copyを再定義し、values リストのコピーを直接再利用。checkpoint シリアライズを経由しない。Topic.checkpoint / from_checkpoint:63-75— checkpoint はself.valuesリストをそのまま返す。from_checkpointは旧版の tuple 形式にも対応。Topic.update:77-85— コアロジック。accumulate=Falseはまずクリアしてから extend、accumulate=Trueは直接 extend。空の flat シーケンスはFalseを返す。Topic.get / is_available:87-94— 空リストはEmptyChannelErrorを投げる。is_availableはbool(self.values)に再定義。Binop class:65-92— クラス定義と__init__。operatorの二項 callable を受け取り、typ()で格納値を初期化。_get_overwrite:31-51— 3種のOverwrite表現(データクラス、sentinel-keyed dict、JSON 復元後の dict)を識別。_operators_equal:54-62— lambda 比較の特殊処理。同名 lambda は等しいとみなし、再コンパイルごとにチャネルが変更扱いになるのを防ぐ。Binop.update:123-144— reducer の本当の入口。最初の値は初期化に使い、以降の値は順にoperator(self.value, value)を呼ぶ。Overwriteバイパスは例外。_is_field_binop:1890-1908— コンパイル時にAnnotated[..., reducer]メタデータ内の2引数 callable を識別し、BinaryOperatorAggregateにインスタンス化。TASKS channel:805-809—Pregel.__init__は__pregel_tasksチャネルにTopic(Send, accumulate=False)を硬编码し、ユーザー上書きを許さない。add_messages:60-66— 典型的 reducer。メッセージ ID で2つのリストをマージし、新規 ID は append、既存 ID は置換。
データフロー
BinaryOperatorAggregate.update は reducer チャネルの心臓です(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このコードで2つの分岐に注意してください。チャネルが空の場合、最初の値は直接初期値になります(reducer は2つの入力を必要とするため、1回分の reducer 呼び出しをスキップ)。以降の値は Overwrite でなければ self.value = operator(self.value, value) となります。Overwrite が現れた場合、以降の非 overwrite 値はすべて捨てられます。これにより「上書き」のセマンティクスが後続の reducer で巻き戻されないことが保証されます。
StateGraph が Annotated[list[AnyMessage], add_messages] を BinaryOperatorAggregate にコンパイルする様子は _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 Noneinspect.signature で最後の meta が2引数 callable であることを検証し、署名が合わなければそのままエラーを投げます。だから add_messages(left, right) のような2引数関数は reducer になれますが、1引数関数は拒否されます。
Topic.update は別の経路を取り、ロジックはより短いです(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 updatedaccumulate=False のときの戻り値に注意してください。古い値が非空なら、新しい値シーケンスが空でも True を返します(クリア自体が1つの変更だからです)。Pregel は channel_versions を更新します。これは LastValue が空シーケンスで False を返す挙動と逆で、チャネルごとの「空書き込み」セマンティクスの違いを反映しています。
reducer チャネルの1回の Pregel スーパーステップ内での位置付け:
境界と失敗
Overwriteの3種の識別形式が共存:overwrite forms:44-50はOverwriteデータクラス、{"__overwrite__": value}辞書、および JSON 復元後の{"type": "__overwrite__", "value": ...}辞書を同時に受け付けます。第3の分岐は LangGraph API server の orjson シリアライズ往復後もセマンティクスが失われないようにするためです。- 1ステップで許される
Overwriteは1つだけ:overwrite dedup:132-138は2回目の出現で即座にInvalidUpdateErrorを投げます。「上書き」セマンティクスが1ステップ内で複数回現れるのは矛盾するからです。 Overwrite後の非 overwrite 値は黙って捨てられる:discard after overwrite:142-143if not seen_overwriteで reducer 呼び出しを制御し、seen_overwrite=Trueになった後の値は reducer に入らないしエラーにもなりません。これは意図的で、「上書き」は以前のマージ結果を残さず、後続の値も必要ないことを意味します。- lambda reducer の同値性比較:
_operators_equal:54-62はすべての lambda を等しいとみなします。Python では<lambda>は毎回新オブジェクトになるため、同一性で比較すると__eq__が常に不一致になり、グラフキャッシュに悪影響を与えます。 Topic(accumulate=False)は空書き込みでTrueを返す:clear returns True:79-80。古い値が非空であれば、クリア自体を変更とみなします。これによりapply_writesはchannel_versionsを更新し、このチャネルに依存するノードがトリガーされます。ユーザーがTopicを自前で使う場合、この挙動に注意してください。LastValueのように空書き込みを黙って処理しません。Topicの旧 checkpoint 互換:tuple compatibility:70-74はisinstance(checkpoint, tuple)をチェックします。旧版の checkpoint は(typ, values)タプルで格納していましたが、新版は直接listに変更されました。この分岐は旧アーカイブを壊さないためのものです。- reducer 署名は厳格に2引数:
reducer signature check:1894-1907はinspect.signatureで positional 引数を数え、2つでなければエラーにします。*argsのような可変引数関数は受け付けません。これが「reducer は厳密に(a, b) -> c」を釘付けにします。 add_messagesはファクトリ:_add_messages_wrapper:41-57は@_add_messages_wrapperで装飾され、実際の関数はadd_messages(left, right)として直接呼べるほか、add_messages(format="langchain-openai")で partial を返すこともできます。この使い方は reducer パターンの一般的な拡張点で、reducer に実行時引数を与えます。
まとめ
BinaryOperatorAggregate は「reducer を持つフィールド」の統一実装で、Topic は「リスト累積」の2形態(ステップ間累積と毎ステップクリア)を提供します。これらと /channels/last-value は LangGraph チャネルシステムの三本柱で、インターフェースはすべて /channels/base-channel に由来します。add_messages のような reducer は /graph/state-graph で Annotated[..., reducer] として宣言され、実行時に /pregel/algo の apply_writes が update に流し込みます。Topic(accumulate=False) は __pregel_tasks チャネルとして Send ファンアウトの担い手で、詳細は /subgraph/send-command を参照してください。
公式資料:LangGraph 文档 · README