LastValue:上書き書き込みのデフォルトチャネル

源码版本1.2.9

役割

LastValue は BaseChannel の最もシンプルなサブクラスです。1つの値を保持し、書き込むたびに古い値を上書きします(LastValue class:20-21)。これは「state フィールドに Annotated 注釈を付けないときにデフォルトで使われるチャネル (channel)」で、StateGraph で class State(TypedDict): count: int と書いたとき、コンパイル時に見える count フィールドの背後には LastValue(int) があります(fallback LastValue:1857-1859)。

重要な制約は「1ステップで書き込める値は1つだけ」です。update は長さが 1 より大きいシーケンスを受け取ると、マージも上書きもせず、そのまま InvalidUpdateError を投げます(update:56-67)。ユーザーには Annotated[int, reducer] で「このステップで複数ノードが同時に書いたときにどう fold するか」を表現するよう促します。この制約は技術的な制限ではなく、セマンティクス上の選択です。スカラー状態には妥当なマージ方法がもともと存在せず、黙って値を捨てるより明示的にエラーにするほうがよいからです。

また、このファイルには変種の LastValueAfterFinish も含まれます(LastValueAfterFinish:81-152)。挙動は LastValue に似ますが、書き込んだ値は Pregel の finish() が呼ばれた後でないと外部に見えず、1回読まれるとクリアされます。「中断シグナル」や「終了シグナル」など、実行の終わりでのみ有効になる内部チャネル用です。

設計動機

• 上書きはスカラー状態で唯一妥当なデフォルト:整数、文字列、ブール値といった値には「マージ」のセマンティクスがありません。count: int のようなフィールドでは、新しい値が来たら古い値を上書きするのが直感に最も合います。そのため StateGraph は fallback:1857 で reducer 注釈のないフィールドを LastValue にコンパイルします。
• 黙って値を捨てない:同じスーパーステップ (superstep) で2つのノードが同時に count に書き込んでも、LastValue はどちらかを勝手に選びません。INVALID_CONCURRENT_GRAPH_UPDATE を投げます(InvalidUpdateError:60-64)。エラーメッセージは「Use an Annotated key to handle multiple values」と明示し、reducer を追加するという正しい解法にユーザーを導きます。/channels/topic-binop を参照してください。
• None ではなく MISSING:初期値には langgraph._internal._typing.MISSING を使い(value = MISSING:29)、None は正当な格納値として扱います。get() は MISSING のとき EmptyChannelError を投げます(get:69-72)。これは BaseChannel のデフォルト実装と整合しますが、is_available はここでは self.value is not MISSING の1行に再定義されており(is_available:74-75)、毎回 try/except を通すのを避けています。
• copy は from_checkpoint より安い:親クラスのデフォルトは copy = from_checkpoint(checkpoint()) ですが、ここでは copy を再定義し(copy:44-48)、同じ value の参照をそのまま共有します。スカラーはそもそも深コピー不要だからです。from_checkpoint(from_checkpoint:50-54)は MISSING かどうかの分岐で値を代入するか決めます。
• LastValueAfterFinish の遅延可視性:一部の制御シグナル(Command の goto など)は、書き込まれた瞬間に prepare_next_tasks に見えると、同じステップ内で自己トリガーしてループを形成してしまいます。LastValueAfterFinish は finished フラグ(finished flag:93-95)を持ち、get() は finished=True のときだけ値を返します(get gated:145-148)。finish() は Pregel が終了処理で呼び(finish:138-143)、consume() は1回消費された後にクリアします(consume:130-136)。

主要ファイル

• LastValue class:20-25 — クラス定義。ジェネリック BaseChannel[Value, Value, Value]、つまり格納値型・書き込み値型・checkpoint 型がすべて同じ Value。
• __init__:27-29 — 構築時は value = MISSING。書き込まれた初回だけ MISSING ではなくなります。
• ValueType / UpdateType:34-42 — どちらも self.typ を返します。宣言で与えられた型そのものです。
• copy:44-48 — value の参照を直接再利用。スカラーは深コピー不要。
• from_checkpoint:50-54 — checkpoint 値からインスタンスを再構築。MISSING は空からの開始を意味します。
• update:56-67 — コア制約。空シーケンスは False を返し、長さ 1 より大で InvalidUpdateError を投げ、それ以外は最後の値を格納して True を返します。
• get / is_available:69-75 — 読み取りインターフェース。MISSING は EmptyChannelError を投げます。is_available は1行の判定に再定義。
• LastValueAfterFinish:81-95 — 変種。finished フラグを追加し、finish() 後まで可視性を遅延。
• LastValueAfterFinish.update:122-128 — 書き込み時に finished をリセット。新たな書き込みが前回の可視性を取り消すことを意味します。
• consume / finish / get:130-148 — 3つのフックが連携して「1回見たらクリア」のセマンティクスを実現。
• fallback LastValue:1857-1859 — StateGraph はコンパイル時に reducer 注釈のないフィールドを LastValue(annotation) にフォールバック。

データフロー

LastValue.update のロジックはこれだけです(update:56-67):

def update(self, values: Sequence[Value]) -> bool:
    if len(values) == 0:
        return False
    if len(values) != 1:
        msg = create_error_message(
            message=f"At key '{self.key}': Can receive only one value per step. Use an Annotated key to handle multiple values.",
            error_code=ErrorCode.INVALID_CONCURRENT_GRAPH_UPDATE,
        )
        raise InvalidUpdateError(msg)

    self.value = values[-1]
    return True

3つ目の分岐で values[0] ではなく values[-1] を書いていることに注意してください。直前で len != 1 を弾いているため、この行に達するのは len == 1 の場合だけで、[-1] と [0] は同値です。空シーケンスが False を返すのが重要で、apply_writes が未変更チャネルに対して update(EMPTY_SEQ) を呼ぶ処理(EMPTY_SEQ update:329)と組み合わさり、LastValue は空呼び出しで「変更なし」を正しく返します。

StateGraph が未注釈フィールドを LastValue にマップする方法は field_to_channel:1840-1859 にあります:

if manager := _is_field_managed_value(name, annotation):
    if allow_managed:
        return manager
    else:
        raise ValueError(f"This {annotation} not allowed in this position")
elif channel := _is_field_channel(annotation):
    channel.key = name
    return channel
elif channel := _is_field_binop(annotation):
    channel.key = name
    return channel

fallback: LastValue = LastValue(annotation)
fallback.key = name
return fallback

順に試行します:managed value(MessagesState の特殊管理)→ 明示的な BaseChannel インスタンス → Annotated[..., reducer](BinaryOperatorAggregate、/channels/topic-binop 参照)→ どれにもヒットしなければ LastValue にフォールバック。

読み書き全体の時系列は次のとおりです:

境界と失敗

• 多値の並行書き込みは即エラー:multi-value error:59-64 は InvalidUpdateError を投げ、エラーコードは INVALID_CONCURRENT_GRAPH_UPDATE。初心者が最もよく陥る罠で、2つのノードが同じスカラーフィールドに同時に書き込むケースです。修正は Annotated[int, lambda a, b: a + b] のように reducer を持つフィールドに変えること。
• 空書き込みは False:empty seq:57-58 は False を返します。BaseChannel 契約と整合し、apply_writes はこれに従って channel_versions の更新をスキップします。
• __eq__ は値ではなく型のみを見る:__eq__:31-32 は isinstance(value, LastValue) を返します。つまり2つの LastValue インスタンスは何を格納していても等しいとみなされます。これはグラフコンパイル時の重複排除用で、値比較に使ってはいけません。
• MISSING と None:None は正当な格納値です。MISSING だけが「一度も書き込まれていない」ことを意味します(get / is_available:70-75)。業務フィールドが None を許すなら、安心して書き込んでください。チャネルはそれを空とは扱いません。
• LastValueAfterFinish の新規書き込みは可視性を取り消す:LAF.update:122-128 の冒頭で self.finished = False に設定されます。つまり finish() 後に再び書き込みが来ると、チャネルは「非公開」状態に戻り、次の finish() まで読めなくなります。
• LastValueAfterFinish.consume は finished が偽のとき no-op:LAF.consume:130-136 を参照。「書き込まれたがまだ finish されていない」値はクリアされず、1回消費された後でのみクリアされます。これが「終了シグナル」のセマンティクスに必要な挙動です。

まとめ

LastValue は LangGraph 状態のデフォルトの形です。1つのスカラーを格納し、上書きで更新し、多値書き込み時には黙って捨てるのではなくエラーにします。これを理解すれば、reducer 注釈のないすべてのフィールドが実行時にどう変化するかが分かります。並行書き込みの正しいマージ方法は /channels/topic-binop に、チャネルインターフェースの定義は /channels/base-channel に、update が呼ばれる具体的タイミングは /pregel/algo に続きます。

公式資料:LangGraph 文档 · README