langgraph

LangGraph ワークフローパターン

本番環境向けの LangGraph ワークフローを構築するための包括的なパターンです。**LangGraph 1.x は LTS（長期サポート）**であり、最初の安定メジャーリリースとして、Uber、LinkedIn、Klarna のエージェントを支えています。各カテゴリには個別のルールファイルがあり、必要に応じて rules/ から読み込まれます。

LangGraph 1.2（2026-05-12 リリース）— フォールトトレランスのリリースです。 特に記載がない限り、以下はすべて StateGraph.add_node(...) で利用できます。

• ノードごとのタイムアウト — timeout= は float | timedelta | TimeoutPolicy を受け入れます。 TimeoutPolicy(run_timeout=, idle_timeout=, refresh_on="auto"|"heartbeat") は、進捗に応じて更新されるアイドル上限と、固定の 実時間上限を分けて設定します。期限切れになると、LangGraph は NodeTimeoutError（kind="idle"|"run" と elapsed を含む）を発生させ、その試行の書き込みを破棄して、 リトライポリシーに処理を委ねます。協調的な仕組みです。asyncio のキャンセルを利用するため、GIL をブロックするノードは中断 されません。rules/resilience-node-timeouts.md を参照してください。
• ノードのエラーハンドラー — error_handler= は、リトライ予算を使い切ったときに実行される復旧ノードを登録します。 NodeError 型のパラメーターを宣言すると、失敗コンテキスト（フィールド node、error）を受け取り、 状態を更新してルーティングを変更する Command を返します。 rules/resilience-error-handlers.md を参照してください。
• RunControl（langgraph.runtime）— 協調的なグレースフルシャットダウンです。任意のスレッドから request_drain(reason) を呼び出します。ノードは runtime.drain_requested を確認し、チェックポイントの境界で停止することで、 一部だけ適用されたスーパーステップではなく、再開可能なスレッドを残します。rules/resilience-graceful-drain.md を参照してください。
• DeltaChannel（langgraph.channels.delta、ベータ版）— チェックポイントには差分の書き込みのみを保存し、バッチリデューサーで再生します。snapshot_frequency の更新ごとにスナップショットを作成します。スレッドの長さに伴ってチェックポイントのコストが増大する問題を解消します。リデューサーは バッチを受け取るため、バッチ化の方法に依存しない必要があります。rules/state-delta-channel.md を参照してください。
• runtime.heartbeat() — 明示的な進捗シグナルです。refresh_on="heartbeat" の場合、アイドルタイムアウトを更新する唯一の方法です。

これらは以前の 1.1 で追加されたもので、1.2 の機能ではありません（現在もサポートされている機能であり、過去のバージョンのこのスキルで誤っていたのはリリースの 帰属のみです）：遅延ノード（defer=True）、ノードレベルのキャッシュ（CachePolicy と graph.compile(cache=...)）、および create_agent のモデルミドルウェア （before_model / after_model）。

クイックリファレンス

カテゴリ	ルール	影響度	使用する場面
状態管理	5	CRITICAL	ワークフローの状態スキーマ、アキュムレーター、リデューサー、差分チャネルの設計
レジリエンス	3	CRITICAL	ノードのタイムアウト、エラーハンドラー、グレースフルドレイン（1.2 以降）
ルーティングと分岐	4	HIGH	動的ルーティング、リトライループ、セマンティックルーティング、グラフ間の連携
並列実行	3	HIGH	ファンアウト／ファンイン、マップリデュース、並行エージェント
スーパーバイザーパターン	3	HIGH	中央コーディネーター、ラウンドロビン、優先度に基づくディスパッチ
ツール呼び出し	4	CRITICAL	ツールのバインド、ToolNode、動的な選択、承認
チェックポイント	3	HIGH	永続化、復旧、スレッド間で共有する Store メモリ
ヒューマン・イン・ザ・ループ	3	MEDIUM	承認ゲート、フィードバックループ、中断／再開
ストリーミング	3	MEDIUM	リアルタイム更新、トークンストリーミング、カスタムイベント
サブグラフ	3	MEDIUM	モジュール構成、ネストされたグラフ、状態のマッピング
関数型 API	3	MEDIUM	@entrypoint/@task デコレーター、StateGraph からの移行
プラットフォーム	3	HIGH	デプロイ、RemoteGraph、ダブルテキスト送信の戦略

合計：12 カテゴリにわたる 41 個のルール

状態管理

状態スキーマによって、ノード間のデータの流れが決まります。スキーマを誤ると、気付かないうちにデータが失われます。

ルール	ファイル	主なパターン
TypedDict State	rules/state-typeddict.md	アキュムレーターには TypedDict + Annotated[list, add]
Pydantic Validation	rules/state-pydantic.md	境界では BaseModel、内部では TypedDict
MessagesState	rules/state-messages.md	MessagesState または add_messages リデューサー
Custom Reducers	rules/state-reducers.md	マージ／上書きには Annotated[T, reducer_fn]
Delta Channels (1.2, beta)	rules/state-delta-channel.md	大規模なアキュムレーターには DeltaChannel(reducer, snapshot_frequency=)

レジリエンス

外部と通信するノードのフォールトトレランスです。1.2 で追加されました。それ以前は、ノードが失敗せず、処理も返さない場合には役に立たない retry_policy しか手段がありませんでした。

ルール	ファイル	主なパターン
Node Timeouts	rules/resilience-node-timeouts.md	add_node(..., timeout=TimeoutPolicy(run_timeout=, idle_timeout=))
Error Handlers	rules/resilience-error-handlers.md	add_node(..., error_handler=) + NodeError 型のパラメーター → Command
Graceful Drain	rules/resilience-graceful-drain.md	RunControl().request_drain() + runtime.drain_requested

from langgraph.types import RetryPolicy, TimeoutPolicy
from langgraph.errors import NodeError

builder.add_node(
    "call_vendor",
    call_vendor,
    timeout=TimeoutPolicy(run_timeout=300, idle_timeout=30),
    retry_policy=RetryPolicy(max_attempts=3),
    error_handler=lambda state, error: Command(
        update={"failure": f"{error.node}: {error.error}"}, goto="degraded_path"
    ),
)

ルーティングと分岐

ノード間の制御フローです。処理が停止しないよう、必ず END のフォールバックを含めてください。

ルール	ファイル	主なパターン
Conditional Edges	rules/routing-conditional.md	明示的なマッピングを指定した add_conditional_edges
Retry Loops	rules/routing-retry-loops.md	最大再試行回数カウンターを使ったループバックエッジ
Semantic Routing	rules/routing-semantic.md	埋め込み類似度または Command API によるルーティング
Cross-Graph Navigation	rules/routing-cross-graph.md	親／兄弟へのルーティングには Command(graph=Command.PARENT)

並列実行

独立したノードを並行して実行します。結果を蓄積するには Annotated[list, add] を使用してください。

ルール	ファイル	主なパターン
Fan-Out/Fan-In	rules/parallel-fanout-fanin.md	動的な並列分岐には Send API
Map-Reduce	rules/parallel-map-reduce.md	asyncio.gather + 結果の集約
Error Isolation	rules/parallel-error-isolation.md	return_exceptions=True + 分岐ごとのタイムアウト

スーパーバイザーパターン

中央のコーディネーターが、専門のワーカーに処理を振り分けます。ワーカーはスーパーバイザーに戻ります。

ルール	ファイル	主なパターン
Basic Supervisor	rules/supervisor-basic.md	状態の更新とルーティングには Command API
Priority Routing	rules/supervisor-priority.md	優先度辞書でエージェントの実行順を決定
Round-Robin	rules/supervisor-round-robin.md	agents_completed で完了状況を追跡

ツール呼び出し

LangGraph エージェントに関数呼び出しを統合します。エージェントあたりのツール数は 10 個未満にしてください。

ルール	ファイル	主なパターン
Tool Binding	rules/tools-bind.md	model.bind_tools(tools) + tool_choice
ToolNode Execution	rules/tools-toolnode.md	並列実行用の組み込み機能 ToolNode(tools)
Dynamic Selection	rules/tools-dynamic.md	埋め込みベースのツール関連性フィルタリング
Tool Interrupts	rules/tools-interrupts.md	ツールの承認ゲートには interrupt()

チェックポイント

復旧とデバッグのためにワークフローの状態を永続化します。

ルール	ファイル	主なパターン
Checkpointer Setup	rules/checkpoints-setup.md	開発環境では MemorySaver、本番環境では PostgresSaver
State Recovery	rules/checkpoints-recovery.md	thread_id による再開 + get_state_history
Cross-Thread Store	rules/checkpoints-store.md	スレッドをまたぐ長期記憶には Store

ノードレベルのキャッシュ（1.2+）

チェックポイントとは独立しています。個々のノード出力をキャッシュすることで、同じ入力で再実行した場合に実行を完全にスキップします。

from langgraph.graph import StateGraph
from langgraph.types import CachePolicy
from langgraph.cache.sqlite import SqliteCache

graph = StateGraph(State)
graph.add_node(
    "expensive_fetch",
    fetch_fn,
    cache_policy=CachePolicy(ttl=3600, key_func=lambda s: s["query"]),
)
# RedisCache(url=...) for distributed workers
compiled = graph.compile(cache=SqliteCache("cache.db"))

ノードが冪等で、実行コストが高い場合（埋め込み、外部 API）に使用します。key_func がその変動を捉えない限り、出力が実時間や変更可能な外部状態に依存するノードには使用しないでください。

遅延ノードとモデルミドルウェア（1.2+）

# defer=True — node execution is deferred until the run is about to end,
# i.e. after every other upstream node has completed
graph.add_node("aggregate", aggregate_fn, defer=True)

# Model middleware — no subclassing required.
# create_react_agent is @deprecated since v1.0; use create_agent from langchain.agents.
# The legacy pre_model_hook/post_model_hook are now before_model/after_model middleware.
from langchain.agents import create_agent

agent = create_agent(
    model=model,
    tools=tools,
    middleware=[compress_history, redact_pii],  # before_model / after_model hooks
    system_prompt="...",                          # prompt= renamed to system_prompt
)

人間参加型ワークフロー

人間の介入のためにワークフローを一時停止します。状態を永続化するにはチェックポインターが必要です。

ルール	ファイル	主なパターン
中断／再開	rules/human-in-loop-interrupt.md	interrupt() 関数 + Command(resume=)
承認ゲート	rules/human-in-loop-approval.md	interrupt_before + 状態更新 + 再開
フィードバックループ	rules/human-in-loop-feedback.md	承認されるまで繰り返し中断

ストリーミング

ワークフローのリアルタイム更新と進捗追跡を行います。LangGraph 1.2 は version="v2" をサポートしています（1.1 で導入）。これはオプトイン形式のストリーミング形式で、stream()、astream()、invoke()、ainvoke() の完全な型安全性を提供します。

ルール	ファイル	主なパターン
ストリームモード	rules/streaming-modes.md	5 つのモード：values、updates、messages、custom、debug
トークンストリーミング	rules/streaming-tokens.md	ノード／タグのフィルタリングを使用した messages モード
カスタムイベント	rules/streaming-custom-events.md	進捗イベント用の get_stream_writer()
ストリーミング v2	rules/streaming-v2-format.md	型付きストリーミングに version="v2" を使用（LG 1.1+）

サブグラフ

ネストされたグラフを使用して、モジュール式で再利用可能なワークフローコンポーネントを構成します。

ルール	ファイル	主なパターン
ノードから呼び出す	rules/subgraphs-invoke.md	異なるスキーマ、明示的な状態マッピング
ノードとして追加	rules/subgraphs-add-as-node.md	共有状態、add_node(name, compiled_graph)
状態マッピング	rules/subgraphs-state-mapping.md	親／子間の境界変換

関数型 API

明示的なグラフ構築の代わりに、@entrypoint と @task デコレーターを使用してワークフローを構築します。

ルール	ファイル	主なパターン
@entrypoint	rules/functional-entrypoint.md	オプションのチェックポインターを指定できるワークフローのエントリーポイント
@task	rules/functional-task.md	future を返し、.result() でブロック
移行	rules/functional-migration.md	StateGraph から関数型 API への変換

プラットフォーム

永続化、ストリーミング、マルチテナンシーに対応したマネージド API としてグラフをデプロイします。

ルール	ファイル	主なパターン
デプロイ	rules/platform-deployment.md	langgraph.json + CLI + Assistants API
RemoteGraph	rules/platform-remote-graph.md	デプロイ済みグラフの呼び出しに RemoteGraph を使用
ダブルテキスト送信	rules/platform-double-texting.md	4 つの戦略：reject、rollback、enqueue、interrupt

クイックスタートの例

from langgraph.graph import StateGraph, START, END
from langgraph.types import Command
from typing import TypedDict, Annotated, Literal
from operator import add

class State(TypedDict):
    input: str
    results: Annotated[list[str], add]

def supervisor(state) -> Command[Literal["worker", END]]:
    if not state.get("results"):
        return Command(update={"input": state["input"]}, goto="worker")
    return Command(goto=END)

def worker(state) -> dict:
    return {"results": [f"Processed: {state['input']}"]}

graph = StateGraph(State)
graph.add_node("supervisor", supervisor)
graph.add_node("worker", worker)
graph.add_edge(START, "supervisor")
graph.add_edge("worker", "supervisor")
app = graph.compile()

2026 年の主なパターン

• ストリーミング v2（LG 1.1）：型安全なストリーミングには version="v2" を使用します。stream() と astream() の戻り値は完全に型付けされます。後方互換性のため、デフォルトは引き続き "v1" です。
• Command API：状態の更新とルーティングをまとめて行う場合は、Command(update=..., goto=...) を使用します
• context_schema：状態を汚染せずに、実行時設定（temperature、provider）を渡します
• CachePolicy：SqliteCache（本番環境）または langgraph.cache.memory の InMemoryCache（開発環境）を使用し、TTL を設定して実行コストの高いノード結果をキャッシュします
• RemainingSteps：再帰制限に先回りして対処します
• Store：チェックポインターとは別のスレッド間メモリです（スレッドスコープ）
• interrupt()：ノードのロジック内で動的に中断します（条件付きの場合は interrupt_before の代わりに使用）
• add_edge(START, node)：set_entry_point() は使用しません（非推奨）
• LTS リリース：LangGraph 1.x は LTS であり、v2.0 まで ACTIVE のままです

重要な決定事項

決定事項	推奨事項
状態の型	内部では TypedDict、境界では Pydantic
エントリーポイント	set_entry_point() ではなく add_edge(START, node)
ルーティングと状態の更新	Command API
ルーティングのみ	条件付きエッジ
アキュムレーター	常に Annotated[list[T], add]
開発用チェックポインター	MemorySaver
本番用チェックポインター	PostgresSaver
短期メモリ	チェックポインター（スレッド単位）
長期メモリ	Store（スレッド間で共有、名前空間付き）
並列分岐の最大数	同時実行で 5～10
エージェントあたりのツール数	最大 5～10（それ以上の場合は動的に選択）
承認ゲート	リスクの高い操作には interrupt()
ストリームモード	ほとんどの UI では ["updates", "custom"]
サブグラフのパターン	分離には Invoke、状態共有には Add-as-Node
Functional と Graph	単純なフローには Functional、複雑なトポロジーには Graph

よくある間違い

1. add リデューサーを忘れる（累積されず、上書きされる）
2. 状態をその場で変更する（チェックポインティングが壊れる）
3. ルーティングに END のフォールバックがない（ワークフローが停止しなくなる）
4. 無限の再試行ループ（最大回数カウンターがない）
5. ルーター関数で副作用を発生させる
6. エージェントあたりのツールが多すぎる（コンテキストがあふれる）
7. ツールで例外を発生させる（エージェントのループがクラッシュする）
8. 本番環境にチェックポインターがない（クラッシュ時に進捗が失われる）
9. interrupt() を try/except で囲む（メカニズムが壊れる）
10. サブグラフの境界で状態を変換しない
11. Functional API のタスクで .result() を呼び忘れる
12. set_entry_point() を使う（非推奨。add_edge(START, ...) を使用）

評価

すべてのカテゴリのテストケースを統合した test-cases.json を参照してください。

関連スキル

• ork:agent-orchestration - より高度なマルチエージェント連携、ReAct ループのパターン、フレームワークの比較
• temporal-io - 永続的実行の代替手段
• ork:llm-integration - 一般的な LLM 関数呼び出し
• type-safety-validation - Pydantic モデルのパターン