orchestrator

orchestrator

目的

このスキルは、分散システムにおける複数インスタンスへの委任を処理し、インスタンス A-H へのタスクのルーティング、ファンアウトと応答の集約、セッションの作成／送信、ヘルスチェックを管理します。分散通信クラスター全体で複雑なワークフローを調整するために設計されています。

使用する場面

負荷分散リクエストを A-H に振り分ける、分散ノードからデータを集約する、長時間実行されるセッションを管理する、フォールトトレラントなシステムでインスタンスの健全性を確保するといった、複数のインスタンスを扱うシナリオでこのスキルを使用します。スケーラブルなウェブサービスや並列計算のように、単一インスタンスでの処理では不十分な場合に適用してください。

主な機能

• ルーティング：対象を指定した委任を使用して、特定のインスタンス（例：A-H）にタスクを直接送信します。
• ファンアウト／集約：複数のインスタンスにリクエストをブロードキャストし、応答を収集／集約します。並列クエリなどの操作に対応します。
• セッション管理：sessions_spawn で新しいセッションを作成し、sessions_send でデータを送信して、状態を保持するやり取りを可能にします。
• ヘルスチェック：/api/health/check などのエンドポイントを使用してインスタンスの状態を定期的に確認し、ステータスコード（健全な場合は 200 など）を含む JSON を返します。
• エラー耐性：設定可能なしきい値に基づき、失敗した委任を最大 3 回まで自動的に再試行します。

使用パターン

• 単純なルーティング：対象を指定してインスタンス A にタスクをルーティングします。特定の宛先に向けたワークフローで使用します。
• ファンアウトと集約：インスタンス A-C にリクエストを送信し、結果を集約します。分散検索に最適です。
• セッションベースのワークフロー：セッションを作成してデータを送信し、健全性を監視します。複数段階のプロセスで使用します。
• ヘルス監視ループ：スクリプトに組み込み、60 秒ごとにインスタンスを確認して、必要に応じてルートを変更します。 認証が必要な操作では、必ず環境変数 $ORCHESTRATOR_API_KEY を設定してください。

一般的なコマンド／API

• CLI コマンド：
  • インスタンスへのルーティング：openclaw orchestrator route --instances A B --payload '{"data": "task"}'
  • ファンアウトと集約：openclaw orchestrator fanout --targets A-H --aggregate true --timeout 10s
  • セッションの作成と送信：openclaw orchestrator sessions_spawn --id session1; openclaw orchestrator sessions_send --id session1 --data '{"key": "value"}'
  • ヘルスチェック：openclaw orchestrator health --instances A-H --output json
• API エンドポイント：
  • POST /api/orchestrator/route：本文：{"instances": ["A", "B"], "payload": {"data": "task"}}、ヘッダー：Authorization: Bearer $ORCHESTRATOR_API_KEY
  • POST /api/orchestrator/fanout：本文：{"targets": ["A", "C"], "aggregate": true}、集約された JSON 配列を返します。
  • POST /api/orchestrator/sessions/spawn：本文：{"sessionId": "session1"}、レスポンス：セッショントークン。
  • GET /api/orchestrator/health：クエリ：?instances=A-H、戻り値：{"A": "healthy", "B": "down"}。
• コードスニペット：

  import requests
  headers = {'Authorization': f'Bearer {os.environ["ORCHESTRATOR_API_KEY"]}'}
  response = requests.post('http://api.openclaw/orchestrator/route', json={'instances': ['A'], 'payload': {'data': 'task'}}, headers=headers)
  

  export ORCHESTRATOR_API_KEY=your_key_here
  openclaw orchestrator fanout --targets A-H --aggregate true > output.json
  

• 設定形式：ペイロードには JSON を使用します。例：{"instances": ["A"], "timeout": 5}。config.json などのファイルに保存し、--config config.json で渡します。

統合に関する注意

コマンドを実行する前に、環境変数 $ORCHESTRATOR_API_KEY を設定して統合してください。分散通信クラスターでは、クラスターの API を介してスキルが登録されていることを確認してください（例：本文 {"skillId": "orchestrator"} を指定した POST /api/cluster/register）。Python などの言語では SDK ラッパーを使用してください。pip install openclaw-sdk でインストールできます。埋め込む際は、ヒント文字列をメタデータに含めてください：orchestrate delegate multi-instance session spawn send fanout aggregate instances。ルーティングとセッションの永続性を確認するため、サンドボックス環境で統合をテストしてください。

エラー処理

• 一般的なエラー：
  • 認証の失敗（例：401 Unauthorized）：$ORCHESTRATOR_API_KEY が設定され、有効であることを確認してから、一度再試行してください。
  • インスタンスに接続できない（例：503 Service Unavailable）：CLI では --retry 3 を使用するか、コード内で指数バックオフを処理してください。
  • 集約のタイムアウト：--timeout 10s を設定し、スクリプトでは try/except でエラーを捕捉してください。
• 具体的な手順：
  • ルーティングエラーの場合：レスポンスの JSON を解析してエラーコード（例：「instance_down」）を確認し、正常なインスタンスにフォールバックしてください。
  • コード内では：

    try:
        response = requests.post(url, headers=headers)
        response.raise_for_status()
    except requests.exceptions.HTTPError as e:
        print(f"Error: {e} - Retrying...")
        # Implement retry logic here
    

  • デバッグできるよう、エラーは必ずタイムスタンプとインスタンスの詳細とともに記録してください。

グラフの関係

• 関連項目：distributed-comms クラスター（親）、sessions スキル（セッション管理の依存関係）、multi-instance スキル（委任を担うピア）。
• 依存関係：インスタンスの監視には health-checks モジュールが必要です。
• 接続：ルーティングには委任タグを統合し、distributed-comms のファンアウト操作と集約します。