orchestrator

orchestrator

用途

此技能用于处理分布式系统中的多实例委派，将任务路由到实例 A-H，管理扇出和响应聚合，启动/发送会话，并执行健康检查。它专为协调分布式通信集群中的复杂工作流而设计。

适用场景

当场景涉及多个实例时，可使用此技能，例如在 A-H 之间进行请求负载均衡、聚合分布式节点的数据、管理长时间运行的会话，或确保容错系统中的实例健康状况。在单实例处理无法满足需求时应用此技能，例如可扩展的 Web 服务或并行计算。

主要功能

• 路由：使用定向委派将任务发送到指定实例（例如 A-H）。
• 扇出/聚合：向多个实例广播请求并收集/聚合响应，支持并行查询等操作。
• 会话管理：使用 sessions_spawn 启动新会话，并通过 sessions_send 发送数据，从而支持有状态交互。
• 健康检查：定期通过 /api/health/check 等端点验证实例状态，并返回包含状态码的 JSON（例如，健康状态为 200）。
• 错误恢复能力：根据可配置的阈值，自动重试失败的委派，最多 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 中与扇出操作聚合。