分散脳シミュレーションにおけるLLM指定なし時の動作検証

Copyright: 2026 Moonlight Technologies Inc. All Rights Reserved.

Author: Masahiro Aoki

実装ノート（アーティファクト）: トレーニングスクリプトが出力する artifact_manifest.json と推奨CLIフラグについては docs/implementation/ARTIFACT_MANIFESTS.md を参照してください。

検証日

2026年1月8日

1. 概要

分散脳シミュレーションシステムにおいて、ノードにLLMモデルが明示的に指定されなかった場合の動作フローを検証した。

補足: 2026年2月より導入されたノードタイプベースのLLMトレーニングパイプラインにより、学習時に --node-type を指定すると生成されるアーティファクトに node_type メタが付与され、AutoModelSelector はこれを参照して各ノードに適切なモデルを割り当てるようになっている。本ドキュメント内の検証事例はこの新仕様にも対応している。

現在の実装では、以下の層の処理により、各ノードが正常に機能することが確認された： 1. Frontend層 - モデル指定デフォルト処理 2. ZenohBrainNode層 - 自動モデルセレクター 3. AutoModelSelector層 - フォールバック機構

このドキュメントの目的と使い方

• 目的: LLM未指定時の動作を検証し、デフォルト/フォールバック挙動を共有する。
• 対象読者: 分散脳ノード実装者、QA、運用担当。
• まず読む順: 概要 → アーキテクチャ検証 → 障害/フォールバック観察 → 対応策。
• 関連リンク: 分散脳実行スクリプトは examples/run_zenoh_distributed_brain.py、PFC/Zenoh/Executive詳細は implementation/PFC_ZENOH_EXECUTIVE.md。

---

2. アーキテクチャ検証

2.1 Frontend → Backend パイプライン

ファイル: frontend/pages/distributed_brain.py (行 960-990)

# モデル設定の構築
model_config = {
    str(n.get('rank')): n.get('model_artifact_id')
    for n in flat_node_list
    if n.get('model_artifact_id')  # LLMが指定されたノードのみ
}
model_config_json = json.dumps(model_config)

特性: - ✅ モデル指定がないノードは model_config に含まれない - ✅ 各ノードはコマンドライン引数 --node-id と --module-type のみ受け取る - ✅ model_artifact_id は明示的に渡されない

2.2 ノード起動フロー

ファイル: frontend/pages/distributed_brain.py (行 1010-1050)

command_list = [
    'python', '-u', script_path,
    '--node-id', node_id,          # e.g., "lang-main-0"
    '--module-type', node_type_lower  # e.g., "lang-main"
]

起動時に渡される情報: - node_id: ノード識別子 - module_type: ノードの機能タイプ - モデルパラメータは渡されない ← 重要

---

3. ZenohBrainNode でのモデル読込

3.1 初期化シーケンス

ファイル: examples/run_zenoh_distributed_brain.py (行 1150)

config = {"d_model": 128}  # デフォルト設定
node = ZenohBrainNode(args.node_id, args.module_type, config)
node.start()

動作: 1. module_type をベースにノードを作成 2. config にはハイパーパラメータのみ（d_model等） 3. モデルは _create_model() で自動生成

3.2 モデル生成メソッド

ファイル: examples/run_zenoh_distributed_brain.py (行 245-275)

def _create_model(self) -> SNNModel:
    """Create neural model for this node using AutoModelSelector."""
    session_id = self._get_latest_model_session()

    try:
        # AutoModelSelector による自動モデル選択
        model = AutoModelSelector.get_model(
            task_type=self.module_type,     # "lang-main", "visual", etc.
            session_id=session_id,          # DBから取得
            api_client=self.client,
            d_model=self.config.get("d_model", 128)
        )

        # tokenizer の読込（lang-main の場合）
        if self.module_type == "lang-main" and session_id:
            self._load_tokenizer_from_session(session_id)

        return model

    except Exception as e:
        self.logger.error(f"AutoModelSelector failed: {e}. Falling back...")
        # フォールバック: SpikingEvoTextLM をデフォルト初期化
        vocab_size = 30522
        return SpikingEvoTextLM(vocab_size=vocab_size, d_model=128)

処理フロー: 1. API から最新セッション取得: _get_latest_model_session() - DB に保存されている最新のモデルセッション ID を検索 - 失敗時は session_id = None

1. AutoModelSelector で自動選択:
2. session_id が有効 → DB からモデルをダウンロード

3. session_id が None → デフォルトパラメータで初期化

4. Tokenizer の読込（lang-main ノードのみ）

5. フォールバック:

6. モデル読込失敗 → デフォルトモデルで初期化
7. SpikingEvoTextLM(vocab_size=30522, d_model=128)

---

4. AutoModelSelector の詳細動作

4.1 モデルクラスのマッピング

ファイル: evospikenet/model_selector.py (行 87-99)

@staticmethod
def _get_model_class(task_type: str):
    if task_type == 'text' or task_type == 'lang-main':
        return SpikingEvoTextLM
    elif task_type == 'vision' or task_type == 'visual':
        return SpikingEvoVisionEncoder
    elif task_type == 'audio' or task_type == 'auditory':
        return SpikingEvoAudioEncoder
    elif task_type == 'multimodal':
        return SpikingEvoMultiModalLM
    return None

特性: - ✅ module_type から対応するモデルクラスを決定 - ✅ 複数の別名をサポート（"lang-main" = "text"） - ✅ 不明なタイプの場合は None を返す

4.2 デフォルトパラメータ

ファイル: evospikenet/model_selector.py (行 102-131)

@staticmethod
def _get_default_params(task_type: str):
    """Returns robust default parameters for each model type."""
    common = {'time_steps': 10}

    if task_type in ['text', 'lang-main']:
        return {
            'vocab_size': 30522,
            'd_model': 128,
            'n_heads': 4,
            'num_transformer_blocks': 2,
            **common
        }
    elif task_type in ['vision', 'visual']:
        return {
            'input_channels': 1,
            'output_dim': 128,
            'image_size': (28, 28),
            **common
        }
    elif task_type in ['audio', 'auditory']:
        return {
            'input_features': 13,  # MFCC
            'output_neurons': 128,
            **common
        }
    # ...

特性: - ✅ 各ノードタイプごとに堅牢なデフォルト値を定義 - ✅ LLM指定がない場合、これらのパラメータで自動初期化 - ✅ ノード機能に最適化された設定値

4.3 モデル読込フロー（API からの取得）

ファイル: evospikenet/model_selector.py (行 56-75)

@staticmethod
def get_model(task_type: str, session_id: str = None,
              api_client=None, device=None, **kwargs):
    """
    Factory method to get an initialized model instance.
    """
    device = device or AutoModelSelector.get_device()

    # 1. モデルクラスを決定
    model_class = AutoModelSelector._get_model_class(task_type)
    if not model_class:
        raise ValueError(f"Unknown task_type: {task_type}")

    # 2. API からのロードを試みる（session_id がある場合）
    if session_id and api_client:
        try:
            return AutoModelSelector._load_from_api(
                model_class, task_type, session_id,
                api_client, device
            )
        except Exception as e:
            logger.error(f"Failed to load from API: {e}. "
                        f"Falling back to default initialization.")

    # 3. フォールバック: デフォルト初期化
    logger.info(f"Initializing {model_class.__name__} "
               f"with default/provided parameters.")
    params = AutoModelSelector._get_default_params(task_type)
    params.update(kwargs)  # CLI 引数で上書き可能

    model = model_class(**params).to(device)
    return model

処理フロー図:

get_model(task_type, session_id=None/有効, api_client)
  │
  ├─ task_type から モデルクラス決定
  │
  ├─ session_id が有効か？
  │  ├─ YES: API から artifact ダウンロード
  │  │  ├─ config.json, weights 取得
  │  │  └─ モデル復元
  │  │
  │  └─ NO または 取得失敗
  │     └─ フォールバック
  │
  └─ デフォルトパラメータで初期化
     └─ model_class(**params).to(device)

---

5. 実装上のフォールバック機構

5.1 段階的なフォールバック

段階 1: DB からモデルダウンロード

if session_id and api_client:
    return AutoModelSelector._load_from_api(...)

段階 2: API 接続失敗時

except Exception as e:
    logger.error(f"Failed to load from API: {e}")

段階 3: デフォルト初期化

params = AutoModelSelector._get_default_params(task_type)
model = model_class(**params).to(device)

段階 4: ZenohBrainNode のフォールバック

except Exception as e:
    self.logger.error(f"AutoModelSelector failed: {e}")
    return SpikingEvoTextLM(vocab_size=30522, d_model=128)

5.2 パラメータ上書き機構

ファイル: evospikenet/model_selector.py (行 75)

params.update(kwargs)  # CLI 引数や config で上書き可能

使用例:

AutoModelSelector.get_model(
    task_type="lang-main",
    session_id=None,
    api_client=None,
    d_model=256,  # kwargs で上書き
    n_heads=8
)

---

6. データベース統合検証

6.1 セッション ID 取得

ファイル: examples/run_zenoh_distributed_brain.py (行 192-210)

def _get_latest_model_session(self):
    """Find the session ID of the latest model artifact."""
    try:
        response = requests.get(
            f"{self.api_base_url}/api/artifacts",
            params={"artifact_type": "model"},
            timeout=5
        )
        if response.status_code == 200:
            artifacts = response.json()
            # Filter for weights file
            model_artifacts = [
                a for a in artifacts
                if a['name'] == 'spiking_lm.pth'
            ]
            if model_artifacts:
                # Sort by creation time desc
                model_artifacts.sort(
                    key=lambda x: x['created_at'],
                    reverse=True
                )
                # Return latest session_id
                return model_artifacts[0]['session_id']
    except requests.exceptions.RequestException as e:
        self.logger.warning(f"Failed to fetch artifacts: {e}")

    return None  # No session found

動作: - ✅ API から全モデルアーティファクト取得 - ✅ 最新のセッション ID を抽出 - ✅ 失敗時は None を返す

6.2 Artifact ダウンロード

ファイル: evospikenet/model_selector.py (行 134-170)

@staticmethod
def _load_from_api(model_class, task_type, session_id,
                   api_client, device):
    """Helper to download config and weights, then load the model."""

    # Determine artifact names based on task type
    config_name = "config.json"

    if task_type in ['text', 'lang-main']:
        weights_name = "spiking_lm.pth"
    elif task_type in ['vision', 'visual']:
        weights_name = "vision_encoder.pth"
    elif task_type in ['audio', 'auditory']:
        weights_name = "audio_encoder.pth"
    elif task_type == 'multimodal':
        weights_name = "multi_modal_lm.pth"
    else:
        # Fallback to generic weights
        weights_name = "model.pth"

    # Download config and weights (the helper uses a simple cache directory
    # under `/tmp/evospikenet_cache/{session_id}` so repeated invocations do not
    # re-fetch already retrieved files).
    config_path = AutoModelSelector._download_artifact(
        api_client, session_id, config_name
    )
    weights_path = AutoModelSelector._download_artifact(
        api_client, session_id, weights_name
    )

    # Load model from saved weights
    # (Implementation details omitted for brevity)

特性: - ✅ task_type に基づいて適切な artifact 名を決定 - ✅ API から config と weights をダウンロード - ✅ ダウンロード失敗時は例外を発生

---

7. ノードタイプ別の動作確認

7.1 Lang-Main ノード

項目	動作	確認状況
モデル指定あり	DB から SpikingEvoTextLM をロード	✅ 実装済
モデル指定なし	デフォルト SpikingEvoTextLM(vocab_size=30522) で初期化	✅ 実装済
Tokenizer	DB から BERT tokenizer ロード、失敗時は bert-base-uncased を使用	✅ フォールバック機構あり
API 接続失敗	メモリ内でデフォルト初期化	✅ 実装済

7.2 Visual ノード

項目	動作	確認状況
モデル指定あり	DB から SpikingEvoVisionEncoder をロード	✅ 実装済
モデル指定なし	デフォルト SpikingEvoVisionEncoder(input_channels=1, image_size=(28,28)) で初期化	✅ 実装済
API 接続失敗	メモリ内でデフォルト初期化	✅ 実装済

7.3 Audio ノード

項目	動作	確認状況
モデル指定あり	DB から SpikingEvoAudioEncoder をロード	✅ 実装済
モデル指定なし	デフォルト SpikingEvoAudioEncoder(input_features=13, output_neurons=128) で初期化	✅ 実装済
API 接続失敗	メモリ内でデフォルト初期化	✅ 実装済

7.4 PFC ノード

項目	動作	確認状況
推論モデル	PFCDecisionEngine または AdvancedPFCEngine で初期化（LLM 不要）	✅ LLM非依存
モデル指定の影響	受けない（ルーティング用なので専用モデル）	✅ 設計通り

7.5 Motor ノード

項目	動作	確認状況
モデル指定あり	DB から MotorControlLM をロード（実装予定）	🔄 フェーズ2
モデル指定なし	SimpleLIFNode または AutonomousMotorNode で動作	✅ 実装済

---

8. 環境変数と設定

8.1 API URL 設定

ファイル: examples/run_zenoh_distributed_brain.py (行 1150-1157)

self.api_base_url = os.environ.get("API_URL", "http://api:8000")

デフォルト: http://api:8000

設定方法:

export API_URL="http://custom-api-server:8000"
python examples/run_zenoh_distributed_brain.py \
    --node-id lang-main-0 \
    --module-type lang-main

8.2 デバイス自動選択

ファイル: evospikenet/model_selector.py (行 31-37)

@staticmethod
def get_device():
    """Auto-detects the best available device."""
    if torch.cuda.is_available():
        return 'cuda'
    elif torch.backends.mps.is_available():
        return 'mps'
    else:
        return 'cpu'

優先順序: CUDA > MPS (Metal) > CPU

---

9. エラーハンドリングと復旧戦略

9.1 API 接続失敗時の動作

シナリオ	実装状況	動作
DB が起動していない	✅	デフォルト初期化
セッション ID 取得失敗	✅	session_id = None → フォールバック
Artifact ダウンロード失敗	✅	例外キャッチ → デフォルト初期化
API タイムアウト	✅	5秒 timeout で失敗を切り上げ

9.2 ログ出力

ファイル: evospikenet/model_selector.py (行 55, 65)

logger.info(f"AutoModelSelector: Selected device '{device}'...")
logger.error(f"Failed to load from API: {e}...")
logger.info(f"Initializing {model_class.__name__}...")

ログレベル: - INFO: 正常な処理フロー（デバイス選択、初期化） - ERROR: API 接続失敗 - WARNING: DB 接続失敗（フォールバック実行）

---

10. 実装の安全性評価

10.1 Robustness チェックリスト

• ✅ LLM 指定なし時の初期化: 全ノードタイプで実装済
• ✅ API 接続失敗時の動作: デフォルト初期化で保証
• ✅ パラメータ検証: デフォルト値で安全な設定を保証
• ✅ デバイス互換性: CUDA/MPS/CPU を自動選択
• ✅ エラーハンドリング: 段階的なフォールバック機構

10.2 パフォーマンス考慮

処理	推定時間	影響
API からセッション ID 取得	100-500ms	起動時 1 回のみ
Artifact ダウンロード	1-5s	API 通信依存
モデル初期化	100-500ms	起動時 1 回のみ
デフォルト初期化	10-50ms	高速フォールバック

結論: LLM 指定なしでの起動は高速（デフォルト初期化で 50ms 以内）

---

11. 推奨事項

11.1 本番環境での設定

# API が必ず起動している状態を確認
docker-compose up -d api

# ノードを起動
API_URL="http://api:8000" python examples/run_zenoh_distributed_brain.py \
    --node-id lang-main-0 \
    --module-type lang-main

11.2 開発・テスト環境での使用

# API なしでもノードが起動可能
python examples/run_zenoh_distributed_brain.py \
    --node-id visual-0 \
    --module-type visual

11.3 ログの監視

# ログレベルを DEBUG に設定してトレース
export LOG_LEVEL=DEBUG
python examples/run_zenoh_distributed_brain.py ...

---

12. 結論

分散脳シミュレーションにおいて、ノードに LLM が明示的に指定されなかった場合の動作は以下の通り確認された：

✅ 検証完了事項

1. 自動フォールバック機構が完全に実装されている
2. API からの読込失敗時に自動的にデフォルト初期化

3. 複数段階のフォールバック機構で信頼性を確保

4. 各ノードタイプに最適化されたデフォルト値

5. Lang-Main: SpikingEvoTextLM(vocab_size=30522)
6. Visual: SpikingEvoVisionEncoder(image_size=(28,28))

7. Audio: SpikingEvoAudioEncoder(input_features=13)

8. 堅牢なエラーハンドリング

9. API 接続失敗時も動作を継続
10. ログに詳細な情報を記録

11. 復旧可能な設計

12. 本番環境での使用に適した実装

13. タイムアウト設定（5秒）
14. デバイス自動選択（CUDA/MPS/CPU）
15. 段階的なフェーズ初期化

🎯 システム堅牢性

シナリオ別の動作確認: - ✅ DB あり、LLM あり: DB からロード - ✅ DB あり、LLM なし: デフォルト初期化 - ✅ DB なし、LLM あり: デフォルト初期化 - ✅ DB なし、LLM なし: デフォルト初期化

結論: 全シナリオで正常に初期化される 安全な実装