langgraph-architecture

LangGraph アーキテクチャの設計判断

LangGraph を使用する場合

次のようなニーズがある場合は LangGraph を使用します：

• 状態を保持する会話 - メモリを使った複数ターンのやり取り
• Human-in-the-loop - 承認ゲート、修正、介入
• 複雑な制御フロー - ループ、分岐、条件付きルーティング
• マルチエージェントの連携 - 複数の LLM が連携して動作
• 永続化 - チェックポイントからの再開、タイムトラベルデバッグ
• ストリーミング - リアルタイムのトークンストリーミング、進捗の更新
• 信頼性 - 再試行、エラー復旧、耐久性の保証

次のような場合は代替手段を検討します：

シナリオ	代替手段	理由
単一の LLM 呼び出し	直接 API を呼び出す	オーバーヘッドに見合わない
線形パイプライン	LangChain LCEL	抽象化がよりシンプル
ステートレスなツール使用	関数呼び出し	永続化が不要
シンプルな RAG	LangChain retrievers	組み込みのパターン
バッチ処理	非同期タスク	実行モデルが異なる

状態スキーマの設計判断

TypedDict と Pydantic の比較

TypedDict	Pydantic
軽量で高速	実行時の検証
辞書形式のアクセス	属性アクセス
検証のオーバーヘッドなし	型の強制変換
シンプルなシリアライズ	複雑なネスト構造のモデル

推奨：ほとんどの場合は TypedDict を使用します。検証や複雑なネスト構造が必要な場合は Pydantic を使用します。

Reducer の選択

用途	Reducer	例
チャットメッセージ	add_messages	ID と RemoveMessage を処理
単純な追加	operator.add	Annotated[list, operator.add]
最新の値を保持	なし（LastValue）	field: str
カスタムマージ	Lambda	Annotated[list, lambda a, b: ...]
リストを上書き	Overwrite	Reducer をバイパス

状態サイズに関する考慮事項

# SMALL STATE (< 1MB) - Put in state
class State(TypedDict):
    messages: Annotated[list, add_messages]
    context: str

# LARGE DATA - Use Store
class State(TypedDict):
    messages: Annotated[list, add_messages]
    document_ref: str  # Reference to store

def node(state, *, store: BaseStore):
    doc = store.get(namespace, state["document_ref"])
    # Process without bloating checkpoints

グラフ構造の設計判断

単一グラフとサブグラフの比較

単一グラフを使用する場合：

• すべてのノードで同じ状態スキーマを共有する
• シンプルな線形フローまたは分岐フローである
• ノード数が 10 未満である

サブグラフを使用する場合：

• 異なる状態スキーマが必要である
• グラフ間で再利用可能なコンポーネントがある
• チーム間で関心事を分離する
• 複雑な階層型ワークフローである

条件付きエッジと Command の比較

条件付きエッジ	Command
状態に基づくルーティング	ルーティングと状態の更新
独立したルーター関数	ノード内での判断
より明確な可視化	より柔軟
標準的なパターン	動的な遷移先

# Conditional Edge - when routing is the focus
def router(state) -> Literal["a", "b"]:
    return "a" if condition else "b"
builder.add_conditional_edges("node", router)

# Command - when combining routing with updates
def node(state) -> Command:
    return Command(goto="next", update={"step": state["step"] + 1})

静的ルーティングと動的ルーティング

静的エッジ（add_edge）：

• ビルド時にフローが確定している
• グラフをより明確に可視化できる
• 動作を把握しやすい

動的ルーティング（add_conditional_edges、Command、Send）：

• 実行時に状態に基づいて判断する
• エージェント主導のナビゲーション
• ファンアウトパターン

永続化戦略

Checkpointer の選択

Checkpointer	用途	特徴
InMemorySaver	テストのみ	再起動時に失われる
SqliteSaver	開発	単一ファイル、ローカル
PostgresSaver	本番環境	スケーラブル、並行処理に対応
カスタム	特別な要件	BaseCheckpointSaver を実装

チェックポイントの範囲

# Full persistence (default)
graph = builder.compile(checkpointer=checkpointer)

# Subgraph options
subgraph = sub_builder.compile(
    checkpointer=None,   # Inherit from parent
    checkpointer=True,   # Independent checkpointing
    checkpointer=False,  # No checkpointing (runs atomically)
)

チェックポイントを無効にする場合

• アトミックに処理すべき短命なサブグラフ
• 互換性のない状態スキーマを持つサブグラフ
• 再開が不要なパフォーマンス重視の処理経路

マルチエージェントアーキテクチャ

Supervisor パターン

次のような場合に適しています：

• 明確な階層がある
• 意思決定を一元化する
• エージェントごとに専門分野が異なる

          ┌─────────────┐
          │  Supervisor │
          └──────┬──────┘
    ┌────────┬───┴───┬────────┐
    ▼        ▼       ▼        ▼
┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐
│Agent1│ │Agent2│ │Agent3│ │Agent4│
└──────┘ └──────┘ └──────┘ └──────┘

ピアツーピアパターン

次のような場合に適しています：

• エージェントが協力して動作する
• 明確な階層がない
• 柔軟なコミュニケーションが必要

┌──────┐     ┌──────┐
│Agent1│◄───►│Agent2│
└──┬───┘     └───┬──┘
   │             │
   ▼             ▼
┌──────┐     ┌──────┐
│Agent3│◄───►│Agent4│
└──────┘     └──────┘

ハンドオフパターン

次のような場合に適しています：

• 専門分野を順番に引き継ぐ
• 各ステージの移行が明確である
• ステージごとに能力が異なる

┌────────┐    ┌────────┐    ┌────────┐
│Research│───►│Planning│───►│Execute │
└────────┘    └────────┘    └────────┘

ストリーミング戦略

ストリームモードの選択

モード	ユースケース	データ
updates	UI の更新	ノードの出力のみ
values	状態の検査	各ステップで完全な状態
messages	チャット UX	LLM トークン
custom	進捗／ログ	StreamWriter 経由のデータ
debug	デバッグ	タスク＋チェックポイント

サブグラフのストリーミング

# Stream from subgraphs
async for chunk in graph.astream(
    input,
    stream_mode="updates",
    subgraphs=True  # Include subgraph events
):
    namespace, data = chunk  # namespace indicates depth

Human-in-the-Loop の設計

割り込みの配置

戦略	ユースケース
interrupt_before	アクション前の承認
interrupt_after	完了後のレビュー
ノード内の interrupt()	動的でコンテキストに応じた一時停止

再開パターン

# Simple resume (same thread)
graph.invoke(None, config)

# Resume with value
graph.invoke(Command(resume="approved"), config)

# Resume specific interrupt
graph.invoke(Command(resume={interrupt_id: value}), config)

# Modify state and resume
graph.update_state(config, {"field": "new_value"})
graph.invoke(None, config)

ゲート（順番に実施）

LangGraph の設計を確定する前に、順番に完了してください。各ステップには客観的な合格条件（成果物または明示的な「該当なし」）があり、「検討した」という自己申告だけでは不十分です。

1. 代替案 — 合格条件： 対象のワークロードについて、(a) 代替案を検討する場合 に該当する項目を少なくとも 1 つ評価し、却下した理由を 1 行で示すか、(b) ユースケースが明らかに LangGraph を使用する場合 に該当し、「代替案を検討する」項目には当てはまらないこと。
2. 状態の契約 — 合格条件： すべての状態フィールドにリデューサー（またはデフォルト／LastValue）が割り当てられ、スキーマと同じ場所に記載されていること。大きなペイロードはインラインの塊ではなく、参照または Store によるものにすること（状態サイズに関する考慮事項を参照）。
3. チェックポインター — 合格条件： チェックポインターの選択に従い、対象環境に適したセーバーの種類を選択していること（例：本番環境では、テスト専用と明示されていない限り InMemorySaver を使用しない）。
4. ループと不安定なノード — 合格条件： 循環する可能性のあるグラフでは recursion_limit（または同等の設定）が指定されていること。外部呼び出しでは、ノードごとの RetryPolicy または「リトライなし」という選択が文書化されていること（リトライの設定を参照）。

エラー処理戦略

リトライの設定

# Per-node retry
RetryPolicy(
    initial_interval=0.5,
    backoff_factor=2.0,
    max_interval=60.0,
    max_attempts=3,
    retry_on=lambda e: isinstance(e, (APIError, TimeoutError))
)

# Multiple policies (first match wins)
builder.add_node("node", fn, retry_policy=[
    RetryPolicy(retry_on=RateLimitError, max_attempts=5),
    RetryPolicy(retry_on=Exception, max_attempts=2),
])

フォールバックパターン

def node_with_fallback(state):
    try:
        return primary_operation(state)
    except PrimaryError:
        return fallback_operation(state)

# Or use conditional edges for complex fallback routing
def route_on_error(state) -> Literal["retry", "fallback", "__end__"]:
    if state.get("error") and state["attempts"] < 3:
        return "retry"
    elif state.get("error"):
        return "fallback"
    return END

スケーリングに関する考慮事項

水平スケーリング

• 共有状態には PostgresSaver を使用します
• マネージドインフラストラクチャには LangGraph Platform を検討します
• チェックポイントの外にある大きなデータにはストアを使用します

パフォーマンスの最適化

1. 状態サイズを最小限にする - 大きなデータには参照を使用します
2. 並列ノード - 可能な場合はファンアウトします
3. コストの高い操作をキャッシュする - CachePolicy を使用します
4. すべてを非同期にする - ainvoke、astream を使用します

リソース制限

# Set recursion limit
config = {"recursion_limit": 50}
graph.invoke(input, config)

# Track remaining steps in state
class State(TypedDict):
    remaining_steps: RemainingSteps

def check_budget(state):
    if state["remaining_steps"] < 5:
        return "wrap_up"
    return "continue"

判断チェックリスト

ゲート（順番に実施）の後、実装の前に確認してください。

1. LangGraph は適切なツールですか？（よりシンプルな代替案と比較）
2. 適切なリデューサーを使った状態スキーマが定義されていますか？
3. 永続化戦略を選択しましたか？（開発環境と本番環境のチェックポインター）
4. ストリーミングの要件を特定しましたか？
5. Human-in-the-Loop の箇所を定義しましたか？
6. エラー処理とリトライ戦略を決めましたか？
7. マルチエージェントの連携パターンを決めましたか？（該当する場合）
8. リソース制限を設定しましたか？