LangGraph のノードで env を読むとき、「いつ読むか」を設計判断として扱っていますか。ノード実行時に毎回読む実装は、バッチ処理や並列テストで env が書き換わると、ノードごとに別の挙動を取ります。本記事では、factory 関数の呼び出し時点で env を 1 回だけ読み、closure に固定する設計を扱います。NodeFn 型エイリアスと依存注入を組み合わせると、責務分離とテスト容易性も同時に得られます。

対象バージョンと前提

Python 3.11+ — 型注釈の遅延評価に from __future__ import annotations を使用。typing.Literal と collections.abc.Callable を利用
LangGraph のノード関数規約 (state を受け取り、更新する state の部分 dict を返す)
closure と factory の基礎は前提とします。「env をいつ読むか」を設計判断として扱う部分に焦点を当てます

コード例は HR マッチンググラフの検証ゲートノード (ConditionVerificationGate) から引用します。

env はノード実行時ではなく、ノードを組み立てるときに確定する

env をノード実行時に毎回読む実装は、グラフ構築後に env が書き換わると、同一グラフ実行内でもノードごとに別の値で動きます。env を factory 呼び出し時に 1 回だけ読んで closure に固定する実装は、グラフ構築時点の設定が、その後のノード実行を通じて固定されます。本記事は後者を選びます。理由は、グラフは「構築 → 実行」が分離した使い方をするからです。構築時点で決まった設定が実行中に揺れない方が、挙動を予測しやすく、テストも安定します。

env をノード実行時に読むとずれる場面

問題は、env がプロセス稼働中に書き換わる場面で起きます。代表例は 2 つです。

pytest の monkeypatch.setenv: 並列実行するテストが互いの env を上書きし合う
バッチ処理: 1 プロセスで複数グラフを順に流す間、別のコードが env を更新する

ノード実行のたびに env を読むと、同一グラフ実行の途中で読んだ値が変わります。次の図は、グラフ実行中に env が shadow から enforce に書き換わり、後段ノードだけが別モードで動くずれを示します。

同一グラフ実行内で前段と後段が別モードで動くと、原因の切り分けが難しくなります。ログには 1 回の実行として現れるのに、内部では 2 つのモードが混在しているためです。

factory + closure による env スナップショット

make_condition_verification_gate_node は factory です。factory の先頭で get_gate_mode() を 1 回呼び、結果を mode というローカル変数に束縛します。返すノード関数は、この mode を closure 経由で参照するだけで、自分では env を読み直しません。ここでの GateMode は一般的な用語ではなく、この実装の中で使っている「off / shadow / enforce の 3 値を表す型名」です。

次のコードは、factory が env を 1 回読んで mode に固定し、ノード関数がそれを参照する構造です。

NodeFn = Callable[[MatchingState], Awaitable[dict[str, Any]]]


def make_condition_verification_gate_node(
    synthesis_log_repo: ISynthesisDecisionLogRepository,
) -> NodeFn:
    """ConditionVerificationGate ノードの factory。graph.py から呼び出す。

    trial_result_logger の make_trial_result_logger_node(repo) と同型。
    ゲートモードは factory 呼び出し時に 1 回だけ確定する (バッチ内での env 書き換えを防ぐ)。
    """
    mode: GateMode = get_gate_mode()
    if mode == "enforce":
        logger.warning(
            "ConditionVerificationGate: enforce モードで動作します ..."
        )

    async def condition_verification_gate_node(state: MatchingState) -> dict[str, Any]:
        # D2: off モードは完全 no-op。state を一切読まない・変更しない・ログを書かない。
        if mode == "off":
            return {}
        # ... mode を参照して shadow / enforce の処理に分岐する ...

    return condition_verification_gate_node

mode は factory のローカル変数です。内側の condition_verification_gate_node がこれを参照することで、mode は closure に捕捉されます。ノード関数が何度呼ばれても、参照する mode は factory 呼び出し時の値のまま固定されます。

読み取りタイミングの違いを整理します。

観点	実行時に毎回読む	生成時に 1 回読む (本記事)
env を読む場所	ノード関数の本体	factory の先頭
env を読む回数	ノード実行のたび	グラフ構築時に 1 回
実行中の env 書き換えの影響	受ける (途中で値が変わる)	受けない (固定済み)
同一実行内のモード一貫性	保証されない	保証される
再読み込みの方法	自動 (毎回読む)	factory を作り直す

get_gate_mode はキャッシュしない — タイミングの責務を呼び出し側に置く

env を読む get_gate_mode は、呼び出しのたびに env を読む純粋な変換に保ちます。lru_cache などでキャッシュしません。

def get_gate_mode() -> GateMode:
    """env CONDITION_VERIFICATION_GATE_MODE を off/shadow/enforce に正規化する。

    不正値・未設定は off (完全 no-op)。
    AppSettings() を呼び出しごとにインスタンス化するため lru_cache 相当のキャッシュなし。
    テスト時の monkeypatch.setenv と実行時 env 書き換えに対応する。
    モード読み取りはノード初期化時に 1 回呼び出してローカル変数に固定することを推奨する
    (バッチ内での env 書き換えを防ぐ)。
    """
    raw = AppSettings().condition_verification_gate_mode.strip().lower()
    if raw in _VALID_GATE_MODES:
        return cast(GateMode, raw)
    if raw:
        _warn_gate_env_once(_ENV_NAME, raw, "off")
    return "off"

ここで「固定する責務」を reader 側に持たせない判断が要点です。reader が lru_cache を持つと、プロセス内で最初に読んだ値が以降の全呼び出しに固定されます。これはテストにとって扱いにくい状態です。monkeypatch.setenv で env を変えても、キャッシュ済みの値が返り続けるからです。

そこで責務を次のように分けます。

reader (get_gate_mode): 呼ばれた時点の env を都度読んで正規化する。固定しない
factory (make_..._node): reader を 1 回だけ呼び、結果を closure に固定する

「いつ固定するか」を factory が決めるため、reader は素直な変換のままで済みます。テストは factory を呼び直すか、factory 呼び出し前に env を設定するだけで、任意の値を固定できます。

NodeFn 型エイリアスで factory とノードの契約を 1 箇所に置く

NodeFn はグラフノードの実行関数シグネチャを表す型エイリアスです。

NodeFn = Callable[[MatchingState], Awaitable[dict[str, Any]]]

factory の戻り値型を NodeFn と宣言すると、factory が返す関数は「MatchingState を受け取り、dict[str, Any] を返す非同期関数」であることが型で保証されます。ノード本体のシグネチャがこの契約から外れると、型検査で検出されます。

同じグラフの make_trial_result_logger_node(repo) も同型の factory です。両者が NodeFn という 1 つの型を共有することで、グラフに追加するノードの契約が 1 箇所に集約されます。グラフ構築側 (workflow.add_node) は、どの factory の戻り値も同じ型として扱えます。

factory とノード関数の責務分離

factory パターンを使うと、初期化時の処理と 1 件ごとの処理を自然に分けられます。

factory が担う (構築時に 1 回)	ノード関数が担う (実行のたび)
env を読んで `mode` に確定する	`state` から offer / candidate / job を読む
依存 (`synthesis_log_repo`) を束縛する	1 件ごとに `evaluate_gate` で判定する
`enforce` 時の警告を 1 回だけ出す	`SynthesisDecisionLog` を構築してログ追加する
ノード関数を生成して返す	`mode` に応じて返す state を組み立てる

enforce 時の警告を factory 側に置く理由は、警告を出す回数を 1 回に抑えるためです。警告をノード関数に置くと、ノードが呼ばれるたびに同じ警告がログに出ます。factory は構築時に 1 回だけ実行されるため、初期化時の警告は factory に置くのが適切です。

依存注入との組み合わせ — テスト時に env も repo も外から固定する

factory は依存も外から受け取ります。synthesis_log_repo を factory 引数で注入することで、テストは fake repository を渡せます。env は factory 呼び出しの前に設定すれば closure に固定されます。

import pytest
from orchestrator.app.graph.nodes.condition_verification_gate import (
    make_condition_verification_gate_node,
)

@pytest.mark.asyncio
async def test_gate_enforce_rejects_offer(monkeypatch):
    monkeypatch.setenv("CONDITION_VERIFICATION_GATE_MODE", "enforce")
    fake_repo = FakeSynthesisDecisionLogRepository()

    # factory 呼び出し時点の env (enforce) が closure に固定される
    node = make_condition_verification_gate_node(fake_repo)

    # この後に env を書き換えても、node の mode は enforce のまま
    monkeypatch.setenv("CONDITION_VERIFICATION_GATE_MODE", "off")

    state = {
        "request_id": "req-test-001",
        "candidate_profiles": [{"candidate_id": "c1", "domain_deal_breakers": []}],
        "job_requirements": [{"job_id": "j1", "salary_max": 10_000_000.0}],
        "synthesized_offers": [{"candidate_id": "c1", "job_id": "j1"}],
        "fit_scores_by_pair": {},
    }
    result = await node(state)

    # enforce が closure に固定されているため、off への書き換え後も synthesized_offers キーが返る
    assert "synthesized_offers" in result

closure に固定しているため、テストの並列実行で別テストが env を書き換えても、すでに生成済みのノードは干渉を受けません。env (動作モード) と repo (ログ書き込み先) の両方を外から渡せるため、テストごとに独立した状態を作れます。

off モードの early return が非破壊性の境界になる

mode を closure に固定しておくと、ノード冒頭で mode == "off" を一度判定するだけで、以降の処理をすべて飛ばせます。

async def condition_verification_gate_node(state: MatchingState) -> dict[str, Any]:
    # off モードは完全 no-op。state を一切読まない・変更しない・ログを書かない。
    if mode == "off":
        return {}
    # off 以外のときだけ、ここから先の評価・ログ構築・state 更新を行う

off のとき、このノードは state を読まず、変更せず、ログも残しません。空 dict を返すだけです。既定を off にしておくことで、このノードをグラフに追加しても既存バッチの挙動を一切変えません。新しいゲートを shadow / enforce で有効化するまで、グラフは従来通り動きます。

この early return は、mode が実行中に揺れないことを前提にしています。env を毎回読む実装では、off で始めた実行が途中で enforce に変わり、後段ノードだけがゲートを効かせる事態が起こり得ます。factory で固定しておけば、実行の最初から最後まで off のままで、非破壊性が保たれます。

適用範囲と注意点

この設計が機能するのは、「グラフ構築 → 実行」が分離している場合です。構築時点の設定で実行が完結するなら、env を構築時に固定するのは自然です。

逆に長時間動くプロセスで途中から設定を変えたい要件には向きません。closure に固定した値は実行中に変わらないため、env を書き換えても反映されません。この場合は、factory を作り直して新しいノード (新しいグラフ) を構築します。「設定を変える = グラフを作り直す」という対応にすることで、いつ設定が変わったかがグラフの再構築という明示的な操作に対応します。

参考情報