任务路由（Task Router）

面向 OpenClaw 多智能体系统的分布式任务队列。提供中心化协调、异步任务交接与基于能力的路由调度。

快速上手

# 安装技能
clawhub install task-router

# 注册智能体
task agent register watson --capabilities "research analysis" --max-concurrent 3

# 创建任务
task create --type research --title "竞品分析" --priority high

# 路由器通过心跳自动运行
# 查看任务状态
task list --status pending
task show task-abc123

功能说明

核心功能：

• 入队（Enqueue）：从任意会话（主会话或子智能体）创建任务
• 路由（Route）：根据智能体能力匹配并分发任务
• 追踪（Track）：监控任务全生命周期（pending → active → complete/failed）
• 异步协调（Async Coordination）：移交工作后异步检查结果
• 死信处理（Dead Letter）：自动处理超时与失败任务
• 再均衡（Rebalance）：迁移卡住的任务，启用备选策略重试

典型用例：

• 心跳触发创建研究类任务 → 自动路由至 research 智能体
• 主智能体派生任务 → 异步执行，稍后检查结果
• 多步骤工作流：任务 A 输出 → 作为任务 B 输入
• 智能体故障 → 任务自动重新分配至备用智能体
• 在具备相同能力的多个智能体间实现负载均衡

配置

目录结构

~/.openclaw/task-router/
├── config.yaml           # 路由器设置、超时参数
├── agents.yaml           # 智能体注册表 + 能力声明
├── queue/                # 任务状态存储
│   ├── pending/          # 等待分配
│   ├── active/           # 已分配给智能体
│   ├── completed/        # 成功完成
│   └── failed/           # 失败且重试次数已耗尽
└── logs/
    └── router.log        # 路由决策日志

config.yaml

router:
  check_interval: 30           # 路由器两次运行间隔（秒）
  default_ttl: 3600            # 任务默认超时时间（秒）
  max_retries: 2
  
  strategies:
    default: least-loaded      # 可选值：round-robin | least-loaded | priority
    by_type:
      research: least-loaded
      image_gen: round-robin
      urgent: priority

  health:
    agent_timeout: 300           # 智能体无响应超时（秒），超时后标记为不健康
    task_timeout:
      warning: 1800              # 30 分钟时发出告警
      critical: 3600             # 1 小时后标记为失败

  notifications:
    on_complete: true
    on_fail: true
    channels: [main_session]   # 可扩展支持 Discord 等渠道

agents.yaml（自动维护）

agents:
  watson:
    id: watson
    emoji: 🔬
    capabilities: [research, analysis, web_search]
    max_concurrent: 3
    current_tasks: [task-abc123, task-def456]
    stats:
      completed: 47
      failed: 2
      avg_duration: 180
    health:
      last_ping: 2026-02-13T09:15:00Z
      status: healthy
    
  picasso:
    id: picasso
    emoji: 🎨
    capabilities: [image_gen, image_edit]
    max_concurrent: 2
    current_tasks: []

任务 Schema

id: task-abc123
type: research               # 与智能体 capability 匹配
title: Research Gameye competitors
description: Deep competitive analysis

payload:
  query: Gameye vs competitors
  sources: [web, apollo]
  output_format: markdown
  
created_by: main             # 创建该任务的会话标签
assigned_to: watson          # 路由前为 null
assigned_by: router          # router | manual | agent

created_at: 2026-02-13T09:00:00Z
assigned_at: 2026-02-13T09:05:00Z
started_at: 2026-02-13T09:06:00Z
completed_at: null
expires_at: 2026-02-13T10:00:00Z  # created_at + ttl

priority: high               # low | normal | high | urgent
ttl: 3600                    # 秒
retries: 0
max_retries: 2

dependencies: []             # 阻塞直至所列任务全部完成
blocked_by: []               # 由路由器填充

status: assigned             # pending | assigned | running | complete | failed
result: null                 # 结果文件路径
error: null                  # 失败时的错误信息

metadata:
  source: heartbeat
  tags: [gameye, competitors]

CLI 命令

任务管理

# 创建任务
task create --type research 
  --title "Research Gameye competitors" 
  --data '{"query": "Gameye pricing"}' 
  --priority high 
  --ttl 3600

# 创建带依赖关系的任务
task create --type analysis 
  --title "Analyze research results" 
  --depends-on task-abc123

# 列出任务
task list                    # 所有未完成任务
task list --status pending
task list --assigned-to watson
task list --type research --limit 10
task list --created-after 2026-02-13

# 查看任务详情
task show task-abc123        # 完整详情 + 结果预览

# 任务操作
task cancel task-abc123      # 取消 pending 或 active 状态任务
task retry task-abc123       # 将 failed 任务移回 pending 状态
task reprioritize task-abc123 --priority urgent

# 结果操作
task result task-abc123      # 查看结果文件内容
task export --status completed --since 2026-02-13 > ~/reports/tasks.ndjson

智能体管理

# 注册智能体（路由前必需）
task agent register watson 
  --capabilities "research analysis web_search" 
  --max-concurrent 3 
  --emoji 🔬

# 更新智能体
task agent update watson --add-capability "competitive-analysis"
task agent update watson --max-concurrent 5

# 检查智能体健康状态
task agent status watson     # 当前任务、健康状态、统计信息
task agent ping watson       # 健康检查 ping

# 列出智能体
task agent list              # 所有已注册智能体
task agent list --capable-of research

# 注销智能体
task agent unregister watson --reassign-tasks

路由器控制

# 状态查询
task router status           # 队列深度、智能体健康状态

# 控制流程
task router pause            # 暂停新任务分配
task router resume           # 恢复路由
task router rebalance        # 重新分配卡住的任务

# 维护操作  
task router cleanup          # 归档旧的已完成任务
task router drain            # 完成所有 active 任务，不再接受新 pending 任务

程序化 API

import * as Task from "~/.openclaw/task-router/sdk";

// === 创建任务 ===

// 简单创建
const task = await Task.create({
  type: "research",
  title: "Competitor analysis",
  payload: { query: "Gameye vs competitors" }
});

// 带选项创建
const task = await Task.create({
  type: "image_gen",
  title: "Generate hero image",
  payload: { prompt: "Futuristic game server...", size: "1024x1024" },
  priority: "high",
  ttl: 1800,
  max_retries: 1,
  dependencies: [previousTaskId],  // 先等待这些任务完成
  created_by: "main"
});

// === 查询任务 ===

// 获取状态
const status = await Task.status(task.id);
// { id, status, assigned_to, created_at, expires_at }

// 等待完成（阻塞式）
const result = await Task.wait(task.id, { timeout: 300, pollInterval: 5 });

// 按条件查询列表
const pending = await Task.query({
  status: "pending",
  type: "research",
  priority: "high",
  limit: 10
});

const myTasks = await Task.query({
  created_by: "main",
  status: ["assigned", "running"]
});

// === 智能体集成 ===

// 智能体主动领取任务（禁用自动分配时使用）
await Task.claim({
  agentId: "watson",
  capableOf: "research",
  limit: 1
});

// 标记任务完成
await Task.complete(task.id, {
  result_path: "~/agents/watson/memory/results/competitors.md",
  summary: "Found 5 competitors: GameLift, Multiplay, Hathora, Edgegap, Agones",
  metadata: { competitors_count: 5 }
});

// 标记任务失败
await Task.fail(task.id, {
  reason: "API quota exceeded",
  retryable: true  // 将自动重试
});

// === 多步骤工作流 ===

// 串行任务链
const analysisTask = await Task.chain(researchTask.id, {
  type: "analysis",
  title: "Analyze research findings",
  payload: { input_task: researchTask.id }
});

// 并行任务
const tasks = await Task.parallel([
  { type: "research", title: "Research A", payload: {} },
  { type: "research", title: "Research B", payload: {} },
  { type: "research", title: "Research C", payload: {} }
]);
await Task.waitAll(tasks.map(t => t.id));

// === 智能体会话集成 ===

// 通过任务路由器派生（推荐用于异步工作）
const spawnResult = await Task.spawn({
  taskId: task.id,
  agentId: "watson",      // 可选：若省略则自动路由
  useSession: true      // 使用 sessions_spawn 而非 sessions_send
});

// 路由器将调用：
// sessions_spawn({ agentId, task, label: task.id })

HEARTBEAT 集成

创建 ~/.openclaw/workspace/HEARTBEAT.md：

# Task Router Heartbeat

## 路由器周期（每 30 秒运行一次）
```typescript
import * as Task from "~/.openclaw/task-router/sdk";

// 1. 自动路由 pending 任务
const routed = await Task.router.cycle();
if (routed.length > 0) {
  Task.log(`Routed ${routed.length} tasks:`, routed.map(t => `${t.id} → ${t.assigned_to}`));
}

// 2. 检查超时任务
const timeouts = await Task.router.checkTimeouts();
for (const task of timeouts) {
  if (task.retries < task.max_retries) {
    Task.log(`Retrying ${task.id} after timeout`);
    await Task.retry(task.id);
  } else {
    Task.log(`Dead lettering ${task.id}`);
    await Task.router.moveToDeadLetter(task);
  }
}

// 3. 检查智能体健康状态
const unhealthy = await Task.agents.checkHealth();
for (const agent of unhealthy) {
  // 重新分配其任务
  await Task.router.reassignFrom(agent.id);
}

// 4. 完成任务通知
const recent = await Task.query({
  status: "completed",
  completed_after: Date.now() - 60000  // 过去一分钟内
});
for (const task of recent) {
  if (task.created_by === "main") {
    sessions_send({
      message: `✅ Task complete: ${task.title}\nResult: ${task.result}`
    });
  }
}

路由策略

# config.yaml
strategies:
  default: least-loaded
  
  # 按任务类型指定策略
  by_type:
    research: least-loaded      # 避免单个研究员过载
    image_gen: round-robin      # 均衡 GPU 利用率
    urgent: priority             # 总是选择最优智能体
    
  # 自定义规则
  rules:
    - if: priority == urgent
      then: fastest
    - if: tags includes "sticky"
      then: sticky

任务生命周期详解

PENDING ──assign──→ ASSIGNED ──ack──→ RUNNING ──complete──→ COMPLETE
   │           │          │              │                      │
   │           │          │              └──fail────┐            │
   │           │          │                        ↓            │
   │           │          │                     FAILED ──retry──┘
 timeout      │       timeout        (if retries < max)        │
   │           │          │                (else dead letter)    │
   │           │          │                                     │
   └───────────┴──────────┘─────────────────────────────────────┘

状态定义：

• pending：已创建，等待路由器分配
• assigned：已路由至智能体，等待其确认接收
• running：智能体已确认，正在执行中
• complete：成功完成，结果可用
• failed：最终失败（重试次数已耗尽）
• dead_letter：永久失败，需人工介入审查

死信队列（Dead Letter Queue）

当任务重试次数耗尽时：

~/.openclaw/task-router/dead-letter/
├── task-failed-001.yaml       # 含最终错误状态的任务定义
├── task-failed-002.yaml
└── index.yaml                 # 供管理员审查的汇总信息

# 查看失败任务
task dead-letter list
task dead-letter show task-failed-001

# 可执行操作
task dead-letter retry task-failed-001      # 强制重试
task dead-letter reassign task-failed-001 --to watson
task dead-letter archive task-failed-001   # 确认并归档

最佳实践

任务设计：

• 保持 payload 可 JSON 序列化（避免循环引用）
• 在 payload 中包含输出格式提示
• 使用 dependencies 实现真正严格的任务顺序
• 设置合理的 TTL（避免无限期阻塞）

智能体设计：

• 初始注册能力时应尽量精确窄泛
• 设置保守的 max_concurrent 值
• HEARTBEAT 应定期检查已分配任务
• 务必完成「确认 → 完成 / 失败」完整闭环

协同模式：

• 使用 Task.spawn() 实现 fire-and-forget 场景
• 使用 Task.wait() 在用户需要即时结果时阻塞等待
• 将复杂逻辑拆分为链式依赖任务，而非单一巨型任务
• 由路由器统一处理重试逻辑，智能体无需自行实现

多智能体示例

// 用户提问：“调研竞品并生成一份演示文稿”

// 主智能体（即你）负责编排：

// 1. 创建研究任务
const research = await Task.create({
  type: "research",
  title: "Research Gameye competitors",
  payload: { query: "Gameye vs competitors" }
});

// 2. 创建分析任务（依赖 research）
const analysis = await Task.create({
  type: "analysis",
  title: "Analyze competitive landscape",
  dependencies: [research.id],
  payload: { input_task: research.id }
});

// 3. 创建图像任务（独立、异步）
const images = await Task.parallel([
  { type: "image_gen", title: "Competitor comparison chart", payload: {} },
  { type: "image_gen", title: "Market positioning diagram", payload: {} }
]);

// 4. 创建演示文稿任务（依赖上述全部任务）
const deck = await Task.create({
  type: "presentation",
  title: "Gameye competitive analysis deck",
  dependencies: [analysis.id, ...images.map(i => i.id)],
  payload: { 
    research: analysis.id,
    images: images.map(i => i.result)
  }
});

// 5. 等待最终结果
const result = await Task.wait(deck.id, { timeout: 600 });

// 最终效果：协调 4 个智能体、执行 5 个任务、自动处理依赖关系

故障排查

# 无任务被路由？
task router status           # 是否处于 paused 状态？
task agent list             # 是否存在健康智能体？
task list --status pending  # 是否有任务在等待？

# 任务卡住？
task show task-abc123        # 检查 assigned_to、started_at 字段
task agent status watson     # 智能体是否健康？

# 智能体未领取任务？
# - 检查该智能体的 HEARTBEAT 是否包含任务轮询逻辑
# - 验证 sessions_send 向该智能体发送消息是否正常

# 失败过多？
task dead-letter list       # 审查失败模式
task router logs            # 查看路由决策日志

# 清空全部（慎用）
task router drain           # 等待所有 active 任务完成
task list --status pending | xargs task cancel
task dead-letter clear

运行要求

• OpenClaw 环境，需支持 sessions_send / sessions_spawn / sessions_list
• 各智能体需在 HEARTBEAT.md 中正确实现任务轮询逻辑
• 可选：若 HEARTBEAT 不可靠，可通过 cron 定时运行路由器

未来扩展方向

• 指标（Metrics）：兼容 Prometheus 的统计指标
• Web UI：本地 Web 控制台（localhost:3333）
• 插件（Plugins）：Slack/Discord 通知支持
• 优先级队列（Priority Queues）