路径变量说明（本文档通用）：

• \x3CTASK_ROOT> = ~/.openclaw/workspace/skills/agent-task-list
• \x3CTASK_SCRIPTS> = \x3CTASK_ROOT>/scripts
• \x3CTASK_DATA> = ~/.openclaw/workspace/agent-tasks

Agent 独立任务列表系统

概述

为每个 Agent 创建独立的任务列表管理，支持：

• 独立任务队列：每个 Agent 拥有自己的任务队列
• 任务优先级调度：支持任务优先级排序和调度
• 状态实时同步：任务状态变化实时同步
• 历史记录：完整任务执行历史记录

核心概念

Agent A
├── current_task: 当前执行任务
├── pending_tasks: 待办任务队列（按优先级排序）
├── completed_tasks: 已完成任务历史
└── failed_tasks: 失败任务记录

Agent B
├── current_task: 当前执行任务
├── pending_tasks: 待办任务队列
├── completed_tasks: 已完成任务历史
└── failed_tasks: 失败任务记录

一、任务数据模型

1.1 任务类型

1.2 任务属性

interface Task {
  // 基础属性
  id: string;              // 任务 ID（唯一）
  name: string;            // 任务名称
  description: string;     // 任务描述
  
  // 状态属性
  status: TaskStatus;      // 任务状态
  priority: number;        // 优先级（1-10，数字越大优先级越高）
  
  // 时间属性
  assigned_at: string;     // 分配时间（ISO 8601）
  started_at?: string;     // 开始时间
  completed_at?: string;   // 完成时间
  failed_at?: string;      // 失败时间
  
  // 关联属性
  agent_id: string;        // 关联 Agent ID
  
  // 扩展属性
  metadata?: Record\x3Cstring, any>;  // 扩展元数据
  error_message?: string;          // 失败时的错误信息
  retry_count?: number;            // 重试次数
}

type TaskStatus = 
  | 'pending'      // 待执行
  | 'running'      // 执行中
  | 'completed'    // 已完成
  | 'failed'       // 失败
  | 'cancelled';   // 已取消

1.3 Agent 任务列表结构

interface AgentTaskList {
  agent_id: string;
  agent_name: string;
  current_task?: Task;
  pending_tasks: Task[];
  completed_tasks: Task[];
  failed_tasks: Task[];
  created_at: string;
  updated_at: string;
}

二、数据存储结构

2.1 目录结构

~/.openclaw/workspace/agent-tasks/
├── agents/
│   ├── agent-coder/
│   │   └── task-list.json
│   ├── agent-planner/
│   │   └── task-list.json
│   └── ...
├── history/
│   ├── 2026-04/
│   │   ├── task-001.json
│   │   └── task-002.json
│   └── ...
├── task-counter.txt      # 任务 ID 计数器
└── index.json           # 全局索引

2.2 任务列表文件（task-list.json）

{
  "agent_id": "agent-coder",
  "agent_name": "代码开发 Agent",
  "current_task": {
    "id": "task-001",
    "name": "实现用户登录功能",
    "description": "实现基于 JWT 的用户登录功能",
    "status": "running",
    "priority": 8,
    "assigned_at": "2026-04-10T10:00:00+08:00",
    "started_at": "2026-04-10T10:05:00+08:00",
    "agent_id": "agent-coder"
  },
  "pending_tasks": [
    {
      "id": "task-002",
      "name": "实现用户注册功能",
      "description": "实现用户注册功能，包含邮箱验证",
      "status": "pending",
      "priority": 7,
      "assigned_at": "2026-04-10T10:10:00+08:00",
      "agent_id": "agent-coder"
    }
  ],
  "completed_tasks": [],
  "failed_tasks": [],
  "created_at": "2026-04-10T09:00:00+08:00",
  "updated_at": "2026-04-10T10:05:00+08:00"
}

2.3 全局索引（index.json）

{
  "agents": [
    {
      "agent_id": "agent-coder",
      "agent_name": "代码开发 Agent",
      "task_count": {
        "current": 1,
        "pending": 5,
        "completed": 10,
        "failed": 2
      },
      "last_updated": "2026-04-10T10:05:00+08:00"
    }
  ],
  "total_tasks": {
    "current": 1,
    "pending": 5,
    "completed": 10,
    "failed": 2
  }
}

三、任务分配算法

3.1 优先级调度

def get_next_task(pending_tasks: List[Task]) -> Optional[Task]:
    """
    从待办队列中获取下一个任务
    
    调度策略：
    1. 按优先级降序排序
    2. 优先级相同时，按分配时间升序排序（先进先出）
    """
    if not pending_tasks:
        return None
    
    # 排序：优先级降序，分配时间升序
    sorted_tasks = sorted(
        pending_tasks,
        key=lambda t: (-t['priority'], t['assigned_at'])
    )
    
    return sorted_tasks[0]

3.2 任务分配流程

1. 创建任务 → pending_tasks
2. Agent 空闲时 → 从 pending_tasks 获取最高优先级任务
3. 任务开始执行 → pending_tasks → current_task
4. 任务完成 → current_task → completed_tasks
5. 任务失败 → current_task → failed_tasks

四、任务状态流转

4.1 状态机

        创建任务
          ↓
      [pending]
          ↓ (Agent 开始执行)
      [running]
          ↓
    ┌─────┴─────┐
    ↓           ↓
[completed]  [failed]
    ↓           ↓
  归档历史    记录错误

4.2 状态转换规则

五、API 设计

5.1 任务管理 API

创建任务

# 创建新任务
python3 \x3CTASK_SCRIPTS>/task_manager.py create \
  --agent \x3Cagent_id> \
  --name "\x3C任务名称>" \
  --description "\x3C任务描述>" \
  --priority \x3C1-10>

返回：

{
  "status": "success",
  "task_id": "task-001",
  "message": "任务已添加到待办队列"
}

获取 Agent 任务列表

# 获取指定 Agent 的任务列表
python3 \x3CTASK_SCRIPTS>/task_manager.py list --agent \x3Cagent_id>

# 获取所有 Agent 的任务概览
python3 \x3CTASK_SCRIPTS>/task_manager.py list-all

返回：

{
  "agent_id": "agent-coder",
  "current_task": {...},
  "pending_tasks": [...],
  "completed_tasks": [...],
  "failed_tasks": [...]
}

开始执行任务

# Agent 开始执行下一个待办任务
python3 \x3CTASK_SCRIPTS>/task_manager.py start --agent \x3Cagent_id>

返回：

{
  "status": "success",
  "task_id": "task-001",
  "task": {...},
  "message": "任务已开始执行"
}

完成任务

# 标记当前任务为完成
python3 \x3CTASK_SCRIPTS>/task_manager.py complete --agent \x3Cagent_id>

返回：

{
  "status": "success",
  "task_id": "task-001",
  "message": "任务已完成"
}

标记任务失败

# 标记当前任务为失败
python3 \x3CTASK_SCRIPTS>/task_manager.py fail \
  --agent \x3Cagent_id> \
  --error "\x3C错误信息>"

返回：

{
  "status": "success",
  "task_id": "task-001",
  "message": "任务已标记为失败"
}

取消任务

# 取消当前任务或待办任务
python3 \x3CTASK_SCRIPTS>/task_manager.py cancel \
  --agent \x3Cagent_id> \
  --task-id \x3Ctask_id>

获取任务详情

# 获取指定任务的详细信息
python3 \x3CTASK_SCRIPTS>/task_manager.py get \
  --task-id \x3Ctask_id>

重试失败任务

# 将失败任务重新加入待办队列
python3 \x3CTASK_SCRIPTS>/task_manager.py retry \
  --agent \x3Cagent_id> \
  --task-id \x3Ctask_id>

5.2 查询 API

查询任务

# 按状态查询
python3 \x3CTASK_SCRIPTS>/task_manager.py query \
  --agent \x3Cagent_id> \
  --status \x3Cpending|running|completed|failed>

# 按优先级查询
python3 \x3CTASK_SCRIPTS>/task_manager.py query \
  --agent \x3Cagent_id> \
  --min-priority \x3C1-10>

# 按时间范围查询
python3 \x3CTASK_SCRIPTS>/task_manager.py query \
  --agent \x3Cagent_id> \
  --from \x3CISO8601> \
  --to \x3CISO8601>

统计信息

# 获取任务统计
python3 \x3CTASK_SCRIPTS>/task_manager.py stats --agent \x3Cagent_id>

返回：

{
  "agent_id": "agent-coder",
  "total": {
    "current": 1,
    "pending": 5,
    "completed": 10,
    "failed": 2
  },
  "success_rate": 0.83,
  "avg_completion_time": "25m"
}

六、使用示例

6.1 创建任务

# 为代码 Agent 创建一个任务
python3 \x3CTASK_SCRIPTS>/task_manager.py create \
  --agent agent-coder \
  --name "实现用户登录功能" \
  --description "实现基于 JWT 的用户登录功能，包含密码加密和 Token 生成" \
  --priority 8

输出：

{
  "status": "success",
  "task_id": "task-001",
  "message": "任务已添加到 agent-coder 的待办队列"
}

6.2 Agent 开始工作

# Agent 检查是否有待办任务并开始执行
python3 \x3CTASK_SCRIPTS>/task_manager.py start --agent agent-coder

输出：

{
  "status": "success",
  "task_id": "task-001",
  "task": {
    "id": "task-001",
    "name": "实现用户登录功能",
    "status": "running",
    "priority": 8
  },
  "message": "任务已开始执行"
}

6.3 完成任务

# 任务执行完成后标记为完成
python3 \x3CTASK_SCRIPTS>/task_manager.py complete --agent agent-coder

输出：

{
  "status": "success",
  "task_id": "task-001",
  "message": "任务已完成，已归档到历史记录"
}

6.4 查看任务列表

# 查看 Agent 的任务列表
python3 \x3CTASK_SCRIPTS>/task_manager.py list --agent agent-coder

输出：

{
  "agent_id": "agent-coder",
  "agent_name": "代码开发 Agent",
  "current_task": null,
  "pending_tasks": [
    {
      "id": "task-002",
      "name": "实现用户注册功能",
      "priority": 7
    }
  ],
  "completed_tasks": [
    {
      "id": "task-001",
      "name": "实现用户登录功能",
      "completed_at": "2026-04-10T11:30:00+08:00"
    }
  ],
  "failed_tasks": []
}

七、脚本说明

7.1 task_manager.py（核心管理脚本）

# 任务创建
python3 \x3CTASK_SCRIPTS>/task_manager.py create \
  --agent \x3Cagent_id> \
  --name \x3Cname> \
  --description \x3Cdesc> \
  --priority \x3C1-10>

# 任务列表
python3 \x3CTASK_SCRIPTS>/task_manager.py list --agent \x3Cagent_id>
python3 \x3CTASK_SCRIPTS>/task_manager.py list-all

# 任务执行
python3 \x3CTASK_SCRIPTS>/task_manager.py start --agent \x3Cagent_id>
python3 \x3CTASK_SCRIPTS>/task_manager.py complete --agent \x3Cagent_id>
python3 \x3CTASK_SCRIPTS>/task_manager.py fail --agent \x3Cagent_id> --error \x3Cerror>
python3 \x3CTASK_SCRIPTS>/task_manager.py cancel --agent \x3Cagent_id> --task-id \x3Ctask_id>

# 任务查询
python3 \x3CTASK_SCRIPTS>/task_manager.py get --task-id \x3Ctask_id>
python3 \x3CTASK_SCRIPTS>/task_manager.py query --agent \x3Cagent_id> --status \x3Cstatus>

# 任务重试
python3 \x3CTASK_SCRIPTS>/task_manager.py retry --agent \x3Cagent_id> --task-id \x3Ctask_id>

# 统计信息
python3 \x3CTASK_SCRIPTS>/task_manager.py stats --agent \x3Cagent_id>

# 历史记录
python3 \x3CTASK_SCRIPTS>/task_manager.py history --agent \x3Cagent_id> --limit \x3Cn>

八、注意事项

1. 任务 ID 唯一性：全局任务 ID 必须唯一，通过计数器保证
2. 并发安全：多 Agent 同时操作时需加锁
3. 数据持久化：所有状态变更需立即写入文件
4. 历史记录：完成任务自动归档到 history 目录
5. 优先级范围：1-10，数字越大优先级越高

九、扩展功能

9.1 任务依赖

未来可扩展任务依赖关系：

{
  "depends_on": ["task-001", "task-002"]
}

9.2 任务分组

支持任务分组/标签：

{
  "tags": ["feature", "urgent"],
  "group": "user-auth-module"
}

9.3 任务通知

任务状态变化时发送通知：

• 任务开始执行
• 任务完成
• 任务失败

意见反馈

欢迎提交 Issue 或 Pull Request！

Changelog

2026-04-10

新增

• 独立任务列表系统设计
• 任务数据模型定义
• 任务分配算法
• 任务状态流转逻辑
• 任务管理 API
• 核心管理脚本