Lang Graphでの差分エンジンの実装

# はじめに
「差分エンジン」はアレン・ニューエル、クリフォード・ショー、ハーバート・サイモンが1957年に開発した「[一般問題解決プログラム（GPS）](https://ja.wikipedia.org/wiki/General_Problem_Solver)」の概念です。これはAI研究の黎明期に問題解決の汎用的アプローチとして革新的でした。

### 「差分エンジン」の現代的意義
現代のAIシステムやロボット工学、自己学習アルゴリズムはこの基本概念を洗練させています。差分エンジンの原理は、特に強化学習や自律エージェントの設計において影響は顕著です。GPSの方法論は後に [Soar](https://ja.wikipedia.org/wiki/Soar_(%E8%AA%8D%E7%9F%A5%E3%82%A2%E3%83%BC%E3%82%AD%E3%83%86%E3%82%AF%E3%83%81%E3%83%A3)) へと発展しました。


## 差分エンジンの仕組み

差分エンジンは以下のプロセスで機能します。

1. **目標状態と現在状態の比較**：システムは現状と達成したい目標との間の「差分」を特定
2. **差分の優先順位付け**：最も重要または解決すべき差分を識別
3. **適切な技術の適用**：その差分を減少させる特定の方法の実行

## 3つの重要特性

1. **目標（方向性）**
   - 明確な目標状態の設定
   - 現状との差分を定量的に認識する能力
   - 差分の重要度を評価する能力

2. **リソースフルさ（方法）**
   - 様々な差分タイプに対応する多様な解決策の保有
   - 適切な手段を状況に応じて選択する能力
   - 解決手段の効果予測能力

3. **持続性（継続的適用）**
   - 目標達成まで方法を繰り返し適用する能力
   - 進捗を監視し方法を調整する能力
   - 失敗からの回復と代替手段への切り替え能力





## 差分エンジンの実装シーケンス

```mermaid

sequenceDiagram
    participant User
    participant DE as DifferenceEngine
    participant Ops as Operators Library
    participant State as State Manager

    rect rgba(255, 230, 230, 0.5)
    note over User, State: 初期化と状態設定
    User->>+DE: 初期化
    DE->>+Ops: 演算子ライブラリの初期化
    Ops-->>-DE: 初期化完了
    User->>DE: 現在状態と目標状態の設定
    DE->>+State: 状態の保存
    State-->>-DE: 状態保存完了
    end

    rect rgba(230, 255, 230, 0.5)
    note over DE, State: 差分検出と優先順位付け
    DE->>+DE: 差分検出(detect_differences)
    DE->>+State: 現在状態の取得
    State-->>-DE: 現在状態
    DE->>+State: 目標状態の取得
    State-->>-DE: 目標状態
    DE->>DE: 差分計算
    DE->>DE: 差分の大きさ計算(calculate_magnitude)
    DE->>DE: 差分の優先順位付け(prioritize_differences)
    DE-->>-DE: 優先順位付き差分リスト
    end

    rect rgba(230, 230, 255, 0.5)
    note over DE, Ops: 演算子の選択
    DE->>+DE: 最優先差分の取得
    DE->>+Ops: 適切な演算子の選択(select_operator)
    Ops->>Ops: 差分に対応可能な演算子の特定
    Ops->>Ops: 演算子の適合性評価
    Ops-->>-DE: 最適な演算子
    end

    rect rgba(255, 255, 230, 0.5)
    note over DE, State: 演算子の適用と状態更新
    DE->>+Ops: 演算子の取得
    Ops-->>-DE: 選択された演算子
    DE->>+State: 現在状態の取得
    State-->>-DE: 現在状態
    DE->>DE: 演算子の適用(apply_operator)
    DE->>+State: 状態の更新
    State-->>-DE: 状態更新完了
    end

    rect rgba(255, 230, 255, 0.5)
    note over DE, State: 目標達成の確認
    DE->>+DE: 目標達成チェック(is_goal_reached)
    DE->>+State: 現在状態の取得
    State-->>-DE: 現在状態
    DE->>+State: 目標状態の取得
    State-->>-DE: 目標状態
    DE->>DE: 状態比較
    alt 目標達成
        DE-->>User: 解決成功
    else 目標未達成かつイテレーション残存
        DE->>DE: 次のイテレーションへ
    else 目標未達成かつイテレーション終了
        DE-->>User: 解決失敗
    end
    DE-->>-DE: プロセス完了
    end
```
### 

# 差分エンジンの実装要素一覧表

差分エンジンの実装の主要要素を機能と目的に基づいて整理しました。

| 要素 | 機能 | 目的 |
|------|------|------|
| `detect_differences` | 差分検出 | 現在状態と目標状態の差異を特定 | 高 |
| `calculate_magnitude` | 差分評価 | 差分の大きさ・重要度を数値化 | 中 |
| `prioritize_differences` | 優先順位付け | 最も重要な差分を特定 | 高 |
| `register_operator` | 演算子登録 | 差分を減らすための方法を登録 | 中 |
| `select_operator` | 演算子選択 | 特定の差分に最適な演算子を選定 | 高 |
| `solve` | メインループ | 差分を減らす反復的なプロセスを管理 | 最高 |
| `apply_operator` | 状態更新 | 選択された演算子を適用して状態を変更 | 高 |
| `is_goal_reached` | 目標達成確認 | 現在状態が目標状態に到達したか検証 | 中 |
| `can_reduce_difference` | 演算子適合性確認 | 特定の演算子が差分を減らせるか評価 | 中 |
| `evaluate_operator_suitability` | 演算子評価 | 演算子の効果性・効率性を評価 | 中 |

## 要素間の相互関係

2. **問題解決サイクル**
   - `solve` → `detect_differences` → `prioritize_differences` → `select_operator` → `apply_operator` → `is_goal_reached` → （繰り返しまたは終了）

3. **演算子管理**
   - `register_operator` → `can_reduce_difference` → `evaluate_operator_suitability` → `select_operator`

4. **差分処理**
   - `detect_differences` → `calculate_magnitude` → `prioritize_differences`

この構造により、状態間の差分を識別し、それを減らすための適切な方法を選択・適用する一連のプロセスが明確に定義されています。

## 差分エンジンの変数一覧

差分エンジンの実装において使用される主要な変数をまとめました。

| 変数名 | データ型 | 用途 | スコープ | 説明 |
|--------|----------|------|----------|------|
| `self.operators` | 辞書 | 演算子保存 | クラス | 差分を減らすための演算子（方法）を名前をキーとして格納 |
| `self.current_state` | 辞書 | 状態保持 | クラス | 現在の状態を特徴量と値のペアで保持 |
| `self.goal_state` | 辞書 | 状態保持 | クラス | 目標とする状態を特徴量と値のペアで保持 |
| `differences` | リスト | 計算結果 | メソッド | 検出された差分のリスト |
| `diff` | 辞書 | 一時保存 | メソッド | 個別の差分情報（特徴、現在値、目標値、大きさ） |
| `feature` | 文字列 | ループ変数 | メソッド | 状態内の特徴量の名前 |
| `suitable_operators` | リスト | 計算結果 | メソッド | 特定の差分に適用可能な演算子のリスト |
| `name` | 文字列 | ループ変数 | メソッド | 演算子の名前 |
| `operator` | 辞書 | 参照 | メソッド | 個別の演算子の詳細（前提条件、効果、アクション） |
| `suitability` | 数値 | 計算結果 | メソッド | 演算子の適合度スコア |
| `iteration` | 整数 | カウンタ | メソッド | 解決プロセスの反復回数 |
| `max_iterations` | 整数 | パラメータ | メソッド | 最大反復回数の制限 |
| `prioritized_diffs` | リスト | 計算結果 | メソッド | 優先順位付けされた差分のリスト |
| `current_diff` | 辞書 | 参照 | メソッド | 現在処理中の最優先差分 |
| `operator_name` | 文字列 | 計算結果 | メソッド | 選択された演算子の名前 |
| `preconditions` | 関数/条件 | パラメータ | メソッド | 演算子の適用条件 |
| `effects` | 関数/記述 | パラメータ | メソッド | 演算子の期待される効果 |
| `action` | 関数 | パラメータ | メソッド | 演算子が実行する具体的な処理 |

## 変数間の関連性

1. **状態管理変数**：
   - `self.current_state`と`self.goal_state`は差分検出の基盤
   - `feature`を通じて両状態の特徴量が比較される

2. **差分関連変数**：
   - `differences`は`diff`オブジェクトのコレクション
   - `prioritized_diffs`は`differences`を優先順位付けしたもの
   - `current_diff`は`prioritized_diffs`から選択された最優先項目

3. **演算子関連変数**：
   - `self.operators`は`name`をキーとした`operator`のコレクション
   - 各`operator`は`preconditions`、`effects`、`action`を持つ
   - `suitable_operators`は特定の差分に対する候補演算子
   - `operator_name`は最終的に選択された演算子の識別子

この変数構造により、差分エンジンは現在状態と目標状態の差分を体系的に分析し、最適な演算子を選択・適用して目標への到達を目指します。

# Lang Graph での差分エンジン実装検討

LangGraphはLLMを使用したグラフベースのワークフローを構築するためのフレームワークです。これを使って差分エンジンを実装する方法を検討します。

## 基本アーキテクチャ

Lang Graphを使用した差分エンジンの実装は、以下のようなノードで構成されるグラフとして設計できます。

```python
from langgraph.graph import StateGraph
import operator
from typing import TypedDict, List, Dict, Any, Optional

# 状態の型定義
class DifferenceEngineState(TypedDict):
    current_state: Dict[str, Any]  # 現在の状態
    goal_state: Dict[str, Any]     # 目標状態
    differences: List[Dict]        # 検出された差分
    current_diff: Optional[Dict]   # 現在処理中の差分
    operators: Dict[str, Dict]     # 利用可能な演算子
    selected_operator: Optional[str]  # 選択された演算子
    iteration: int                 # 実行イテレーション数
    success: Optional[bool]        # 解決の成功/失敗状態
```

## グラフの構築

```python
# グラフの構築
def build_difference_engine_graph():
    graph = StateGraph(DifferenceEngineState)
    
    # ノードの定義
    graph.add_node("detect_differences", detect_differences)
    graph.add_node("prioritize_differences", prioritize_differences)
    graph.add_node("select_operator", select_operator)
    graph.add_node("apply_operator", apply_operator)
    graph.add_node("check_goal", check_goal)
    
    # エッジとフロー制御の定義
    graph.set_entry_point("detect_differences")
    
    graph.add_edge("detect_differences", "prioritize_differences")
    graph.add_edge("prioritize_differences", "select_operator")
    
    # 条件付きエッジ
    graph.add_conditional_edges(
        "select_operator",
        lambda state: "apply_operator" if state["selected_operator"] else "check_goal"
    )
    
    graph.add_edge("apply_operator", "check_goal")
    
    # 終了条件または繰り返し
    graph.add_conditional_edges(
        "check_goal",
        lambda state: "END" if state["success"] is not None else "detect_differences"
    )
    
    return graph.compile()
```

## 各ノードの実装

```python
# 差分検出ノード
def detect_differences(state: DifferenceEngineState) -> DifferenceEngineState:
    differences = []
    
    for feature, goal_value in state["goal_state"].items():
        if feature in state["current_state"]:
            if state["current_state"][feature] != goal_value:
                diff = {
                    'feature': feature,
                    'current': state["current_state"][feature],
                    'goal': goal_value,
                    'magnitude': calculate_magnitude(feature, state)
                }
                differences.append(diff)
        else:
            diff = {
                'feature': feature,
                'current': None,
                'goal': goal_value,
                'magnitude': calculate_magnitude(feature, state)
            }
            differences.append(diff)
    
    return {
        **state,
        "differences": differences,
        "iteration": state["iteration"] + 1
    }

# 差分の優先順位付けノード
def prioritize_differences(state: DifferenceEngineState) -> DifferenceEngineState:
    # 差分がない場合は成功として終了
    if not state["differences"]:
        return {**state, "success": True}
    
    # 差分の優先順位付け
    prioritized_diffs = sorted(state["differences"], key=lambda x: x['magnitude'], reverse=True)
    
    return {
        **state,
        "current_diff": prioritized_diffs[0]
    }

# 演算子選択ノード
def select_operator(state: DifferenceEngineState) -> DifferenceEngineState:
    if not state["current_diff"]:
        return {**state, "selected_operator": None}
    
    suitable_operators = []
    
    for name, operator in state["operators"].items():
        if can_reduce_difference(operator, state["current_diff"], state):
            suitability = evaluate_operator_suitability(operator, state["current_diff"], state)
            suitable_operators.append((name, suitability))
    
    if not suitable_operators:
        # 適用可能な演算子がない場合は失敗
        return {**state, "selected_operator": None, "success": False}
    
    # 最も適切な演算子を選択
    selected = max(suitable_operators, key=lambda x: x[1])[0]
    
    return {
        **state,
        "selected_operator": selected
    }

# 演算子適用ノード
def apply_operator(state: DifferenceEngineState) -> DifferenceEngineState:
    operator = state["operators"][state["selected_operator"]]
    
    # 新しい状態のコピーを作成
    new_current_state = state["current_state"].copy()
    
    # 演算子のアクション関数を適用
    operator["action"](new_current_state, state["current_diff"])
    
    return {
        **state,
        "current_state": new_current_state,
        "selected_operator": None
    }

# 目標達成確認ノード
def check_goal(state: DifferenceEngineState) -> DifferenceEngineState:
    # すでに成功/失敗が決定している場合はそのまま
    if state["success"] is not None:
        return state
    
    # 最大イテレーション数に達した場合
    if state["iteration"] >= MAX_ITERATIONS:
        return {**state, "success": False}
    
    # 目標達成チェック
    is_reached = True
    for feature, value in state["goal_state"].items():
        if feature not in state["current_state"] or state["current_state"][feature] != value:
            is_reached = False
            break
    
    if is_reached:
        return {**state, "success": True}
    
    return state
```

## 補助関数

```python
# 差分の大きさを計算
def calculate_magnitude(feature, state):
    # 特徴の種類によって異なる計算方法
    # ここでは単純化のため常に1を返す
    return 1

# 演算子が差分を減らせるかチェック
def can_reduce_difference(operator, difference, state):
    return operator.get("preconditions", lambda s, d: True)(state, difference)

# 演算子の適合性を評価
def evaluate_operator_suitability(operator, difference, state):
    # ここでは単純化のため常に1を返す
    return 1
```

## LLMを活用した拡張

LangGraphの特性を活かし、差分検出や演算子生成にLLMを組み込みます。

```python
from langchain.llms import OpenAI

# LLMを使用して演算子を動的に生成
def generate_operator_with_llm(state: DifferenceEngineState) -> DifferenceEngineState:
    llm = OpenAI()
    
    current_diff = state["current_diff"]
    prompt = f"""
    現在状態: {state["current_state"]}
    目標状態: {state["goal_state"]}
    
    現在の差分:
    特徴: {current_diff['feature']}
    現在値: {current_diff['current']}
    目標値: {current_diff['goal']}
    
    この差分を減らすための操作（演算子）を生成してください。
    操作は以下の形式で記述してください：
    1. 操作の名前
    2. 適用条件
    3. 期待される効果
    4. 具体的なアクション（Python関数形式）
    """
    
    response = llm.predict(prompt)
    # レスポンスからオペレータを解析して追加
    # ...
    
    return state
```

## 実行方法

```python
# 初期状態の設定
initial_state = {
    "current_state": {"位置": "A", "持ち物": []},
    "goal_state": {"位置": "B", "持ち物": ["鍵"]},
    "differences": [],
    "current_diff": None,
    "operators": {
        "移動": {
            "preconditions": lambda s, d: d["feature"] == "位置",
            "effects": "位置の変更",
            "action": lambda state, diff: state.update({"位置": diff["goal"]})
        },
        "拾う": {
            "preconditions": lambda s, d: d["feature"] == "持ち物" and s["current_state"]["位置"] == "A",
            "effects": "持ち物の追加",
            "action": lambda state, diff: state["持ち物"].extend(["鍵"])
        }
    },
    "selected_operator": None,
    "iteration": 0,
    "success": None
}

# グラフの作成と実行
difference_engine = build_difference_engine_graph()
result = difference_engine.invoke(initial_state)

print(f"解決成功: {result['success']}")
print(f"最終状態: {result['current_state']}")
print(f"実行イテレーション: {result['iteration']}")
```

## LangGraphの適合性

LangGraphは差分エンジンのような段階的な問題解決アプローチと非常に相性が良く、状態変化とフロー制御を明確に表現できる点が大きな強みです。

1. **宣言的なフロー定義**: 差分エンジンの処理フローをグラフとして視覚的に定義できる
2. **状態管理の簡素化**: 不変な状態パターンにより副作用の管理が容易
3. **条件分岐の明示化**: 条件付きエッジにより処理フローの分岐が明確
4. **LLMとの統合**: 差分の評価や演算子生成にLLM APIコールを簡単に組み込める
5. **拡張性**: 新しいノードの追加によるシステム拡張が容易

```mermaid
flowchart TD
    subgraph StateManagement["状態管理 (DifferenceEngineState)"]
        direction TB
        InitialState[初期状態] --> CurrentState[現在状態<BR>ncurrent_state]
        InitialState --> GoalState[目標状態<BR>goal_state]
        CurrentState --> DiffDetection[差分検出処理]
        GoalState --> DiffDetection
        DiffDetection --> Differences[差分リスト<BR>ndifferences]
        Differences --> PriorityProcessing[優先順位付け処理]
        PriorityProcessing --> CurrentDiff[現在の差分<BR>ncurrent_diff]
        OperatorLib[演算子ライブラリ<BR>operators] --> OperatorSelection[演算子選択処理]
        CurrentDiff --> OperatorSelection
        OperatorSelection --> SelectedOp[選択された演算子<BR>selected_operator]
        SelectedOp --> ApplyOperation[演算子適用処理]
        ApplyOperation --> NewCurrentState[新しい現在状態]
        NewCurrentState -->|状態更新| CurrentState

    end
    
    style StateManagement fill:#f0f8ff,stroke:#4682b4,stroke-width:2px
    style InitialState fill:#e6ffe6,stroke:#006400
    style CurrentState fill:#ffe6e6,stroke:#8b0000
    style GoalState fill:#e6e6ff,stroke:#00008b
    style Differences fill:#fff0e6,stroke:#d2691e
    style CurrentDiff fill:#ffe6f0,stroke:#8b008b
    style SelectedOp fill:#e6fff0,stroke:#2e8b57
    style NewCurrentState fill:#ffe6e6,stroke:#8b0000,stroke-dasharray: 5 5
```

各処理ノードは状態を不変（イミュータブル）として扱い、新しい状態オブジェクトを生成することで状態の一貫性を保ちます。

## 条件分岐図

以下の条件分岐図では、LangGraphにおける条件付きエッジの定義と実際の処理フローを示しています。
LangGraphでは、これらの条件分岐はラムダ関数として定義され、状態オブジェクトに基づいて次のステップを決定します。このアプローチにより、差分エンジンの複雑な状態遷移ロジックが明確かつ管理しやすくなります。

```mermaid

stateDiagram-v2
    
    state "検出ノード<BR>detect_differences" as Detect
    state "優先順位付けノード<BR>prioritize_differences" as Prioritize
    state "演算子選択ノード<BR>select_operator" as Select
    state "演算子適用ノード<BR>apply_operator" as Apply
    state "目標達成確認ノード<BR>check_goal" as Check
    
    [*] --> Detect
    
    Detect --> Prioritize
    
    state if_differences <<choice>>
    Prioritize --> if_differences
    
    if_differences --> Select : 差分あり
    if_differences --> Check : 差分なし<BR>success = True
    
    state if_operator <<choice>>
    Select --> if_operator
    
    if_operator --> Apply : 適用可能な演算子あり<BR>selected_operator != None
    if_operator --> Check : 適用可能な演算子なし<BR>success = False
    
    Apply --> Check
    
    state if_goal <<choice>>
    Check --> if_goal
    
    if_goal --> [*] : 目標達成または失敗<BR>success != None
    if_goal --> Detect : 処理継続<BR>success = None
    
```

## 「演算子」の意味について

このアーキテクチャにおける「演算子」（オペレーター）は、差分エンジンのコンテキストで特別な意味を持っています。

### 差分エンジンにおける「演算子」とは：

**1. 問題解決の方法や行動を表す機能的単位**
現在の状態と目標状態の間の差分（ギャップ）を減らすために適用される具体的な操作や手順を指します。

**2. 演算子の構成要素**
   - **前提条件（preconditions）**: 演算子を適用するための条件
   - **効果（effects）**: 演算子の適用による期待される結果や変化
   - **アクション（action）**: 実際に状態を変更する具体的な処理

例として「移動」演算子は、
   - **前提条件**: 位置の変更が必要な差分が存在する
   - **効果**: ある位置から別の位置への変更
   - **アクション**: 状態の「位置」属性を目標値に更新する関数

**3. 演算子は問題領域に応じて定義される**
システムが目標状態に到達するために選択・適用する「ツール」のセットとして機能します。

:::note info
要するに、演算子とは「差分を減らすための方法」であり、差分エンジンのコアとなる問題解決メカニズムを構成する要素です。
:::

## まとめ

「差分エンジン」は1957年に開発された問題解決の概念で、現在のAI研究にも影響を与えている基本的なアプローチです。このエンジンは現在状態と目標状態の差分を特定し、その差分を減らすための演算子（操作方法）を選択・適用することで、段階的に目標達成を目指すシステムです。

差分エンジンの3つの重要特性は「目標（方向性）」「リソースフルさ（方法）」「持続性（継続的適用）」であり、これらが組み合わさることで柔軟な問題解決能力を発揮します。LangGraphを用いた実装では、グラフベースのワークフローとして差分エンジンを構築し、状態管理や処理フローを明確に定義できます。

この差分エンジンの考え方は、特に強化学習や自律エージェントの設計など、現代のAIシステムの基盤となる重要な概念です。差分を検出し、それを減らすための適切な演算子を選択・適用するという単純ながら強力なアプローチは、様々な問題解決アルゴリズムの基礎となります。