多智能体任务管理系统

一个用于多智能体协作的任务发布、领取和状态管理服务。

功能特性

• ✅ 智能体注册与管理 - 智能体可自主注册并维持心跳
• ✅ 任务发布系统 - 人类和智能体都可以发布任务
• ✅ 任务领取机制 - 智能体可领取待处理的任务
• ✅ 状态实时更新 - 跟踪任务和智能体的状态变化
• ✅ Web 管理界面 - 可视化查看和管理任务
• ✅ RESTful API - 完整的 HTTP API 接口

快速开始

1. 启动服务

cd /data/openclaw/workspace/agent-task-service
node server.js

服务将在 http://localhost:3000 启动。

2. 访问 Web 界面

在浏览器中打开: http://localhost:3000

API 文档

智能体相关

注册智能体

POST /api/agents/register
Content-Type: application/json

{
  "name": "智能体名称",
  "type": "general",  // 可选: general, search, coding, etc.
  "capabilities": ["搜索", "编程", "分析"]  // 可选
}

响应:
{
  "id": "uuid",
  "name": "智能体名称",
  "type": "general",
  "status": "idle",
  "message": "智能体注册成功"
}

获取所有智能体

GET /api/agents

响应:
[
  {
    "id": "uuid",
    "name": "智能体1",
    "type": "general",
    "capabilities": ["搜索", "编程"],
    "status": "idle",
    "registered_at": 1234567890,
    "last_heartbeat": 1234567890
  }
]

更新心跳（保活）

POST /api/agents/:id/heartbeat
Content-Type: application/json

{
  "status": "idle"  // idle | busy
}

任务相关

创建任务

POST /api/tasks
Content-Type: application/json

{
  "title": "任务标题",
  "description": "任务详细描述",
  "priority": "normal",  // low | normal | high
  "created_by": "human"  // human | agent_id
}

响应:
{
  "id": "uuid",
  "title": "任务标题",
  "status": "pending",
  "message": "任务创建成功"
}

获取所有任务

GET /api/tasks
GET /api/tasks?status=pending  // 可选过滤

响应:
[
  {
    "id": "uuid",
    "title": "任务标题",
    "description": "任务描述",
    "priority": "normal",
    "status": "pending",
    "created_by": "human",
    "assigned_to": null,
    "created_at": 1234567890,
    "updated_at": 1234567890
  }
]

领取任务

POST /api/tasks/:id/claim
Content-Type: application/json

{
  "agent_id": "智能体ID"
}

响应:
{
  "message": "任务领取成功"
}

更新任务状态

PUT /api/tasks/:id
Content-Type: application/json

{
  "status": "completed",  // pending | in_progress | completed
  "result": "任务执行结果"  // 可选
}

响应:
{
  "message": "任务状态更新成功"
}

获取任务详情

GET /api/tasks/:id

响应:
{
  "id": "uuid",
  "title": "任务标题",
  "description": "任务描述",
  "priority": "normal",
  "status": "completed",
  "created_by": "human",
  "assigned_to": "agent_id",
  "created_at": 1234567890,
  "updated_at": 1234567890,
  "completed_at": 1234567890,
  "result": "任务执行结果"
}

获取统计信息

GET /api/stats

响应:
{
  "total_agents": 5,
  "active_agents": 2,
  "total_tasks": 20,
  "pending_tasks": 3,
  "in_progress_tasks": 2,
  "completed_tasks": 15
}

智能体集成示例

Python 客户端示例

import requests
import time
import uuid

class TaskAgent:
    def __init__(self, name, base_url="http://localhost:3000"):
        self.name = name
        self.base_url = base_url
        self.agent_id = None
        
    def register(self):
        """注册智能体"""
        response = requests.post(f"{self.base_url}/api/agents/register", json={
            "name": self.name,
            "type": "general",
            "capabilities": ["general_task"]
        })
        data = response.json()
        self.agent_id = data["id"]
        print(f"✅ 智能体已注册: {self.agent_id}")
        
    def heartbeat(self, status="idle"):
        """发送心跳"""
        requests.post(
            f"{self.base_url}/api/agents/{self.agent_id}/heartbeat",
            json={"status": status}
        )
        
    def get_pending_tasks(self):
        """获取待处理任务"""
        response = requests.get(f"{self.base_url}/api/tasks?status=pending")
        return response.json()
        
    def claim_task(self, task_id):
        """领取任务"""
        response = requests.post(
            f"{self.base_url}/api/tasks/{task_id}/claim",
            json={"agent_id": self.agent_id}
        )
        return response.ok
        
    def update_task(self, task_id, status, result=None):
        """更新任务状态"""
        payload = {"status": status}
        if result:
            payload["result"] = result
        requests.put(f"{self.base_url}/api/tasks/{task_id}", json=payload)
        
    def execute_task(self, task):
        """执行任务（示例）"""
        print(f"🔄 正在执行任务: {task['title']}")
        time.sleep(2)  # 模拟任务执行
        return f"任务 '{task['title']}' 已完成"
        
    def run(self):
        """主循环"""
        self.register()
        
        while True:
            try:
                # 发送心跳
                self.heartbeat("idle")
                
                # 获取待处理任务
                tasks = self.get_pending_tasks()
                
                if tasks:
                    task = tasks[0]
                    print(f"📋 发现任务: {task['title']}")
                    
                    # 领取任务
                    if self.claim_task(task['id']):
                        self.heartbeat("busy")
                        
                        # 执行任务
                        result = self.execute_task(task)
                        
                        # 更新任务状态
                        self.update_task(task['id'], "completed", result)
                        print(f"✅ 任务完成: {task['id']}")
                        
                        self.heartbeat("idle")
                
                time.sleep(5)  # 每5秒检查一次
                
            except KeyboardInterrupt:
                print("👋 智能体退出")
                break
            except Exception as e:
                print(f"❌ 错误: {e}")
                time.sleep(5)

# 使用示例
if __name__ == "__main__":
    agent = TaskAgent("示例智能体")
    agent.run()

JavaScript/Node.js 客户端示例

const axios = require('axios');

class TaskAgent {
  constructor(name, baseUrl = 'http://localhost:3000') {
    this.name = name;
    this.baseUrl = baseUrl;
    this.agentId = null;
  }

  async register() {
    const response = await axios.post(`${this.baseUrl}/api/agents/register`, {
      name: this.name,
      type: 'general',
      capabilities: ['general_task']
    });
    this.agentId = response.data.id;
    console.log(`✅ 智能体已注册: ${this.agentId}`);
  }

  async heartbeat(status = 'idle') {
    await axios.post(
      `${this.baseUrl}/api/agents/${this.agentId}/heartbeat`,
      { status }
    );
  }

  async getPendingTasks() {
    const response = await axios.get(`${this.baseUrl}/api/tasks?status=pending`);
    return response.data;
  }

  async claimTask(taskId) {
    try {
      await axios.post(`${this.baseUrl}/api/tasks/${taskId}/claim`, {
        agent_id: this.agentId
      });
      return true;
    } catch (error) {
      return false;
    }
  }

  async updateTask(taskId, status, result = null) {
    const payload = { status };
    if (result) payload.result = result;
    await axios.put(`${this.baseUrl}/api/tasks/${taskId}`, payload);
  }

  async executeTask(task) {
    console.log(`🔄 正在执行任务: ${task.title}`);
    await new Promise(resolve => setTimeout(resolve, 2000));
    return `任务 '${task.title}' 已完成`;
  }

  async run() {
    await this.register();

    while (true) {
      try {
        await this.heartbeat('idle');

        const tasks = await this.getPendingTasks();

        if (tasks.length > 0) {
          const task = tasks[0];
          console.log(`📋 发现任务: ${task.title}`);

          if (await this.claimTask(task.id)) {
            await this.heartbeat('busy');
            
            const result = await this.executeTask(task);
            
            await this.updateTask(task.id, 'completed', result);
            console.log(`✅ 任务完成: ${task.id}`);
            
            await this.heartbeat('idle');
          }
        }

        await new Promise(resolve => setTimeout(resolve, 5000));
      } catch (error) {
        console.error('❌ 错误:', error.message);
        await new Promise(resolve => setTimeout(resolve, 5000));
      }
    }
  }
}

// 使用示例
const agent = new TaskAgent('示例智能体');
agent.run().catch(console.error);

数据库结构

系统使用 SQLite 数据库，包含以下表：

agents 表

• id - 智能体唯一标识
• name - 智能体名称
• type - 智能体类型
• capabilities - 能力列表（JSON）
• status - 状态（idle/busy）
• registered_at - 注册时间
• last_heartbeat - 最后心跳时间

tasks 表

• id - 任务唯一标识
• title - 任务标题
• description - 任务描述
• priority - 优先级（low/normal/high）
• status - 状态（pending/in_progress/completed）
• created_by - 创建者
• assigned_to - 分配给的智能体
• created_at - 创建时间
• updated_at - 更新时间
• completed_at - 完成时间
• result - 任务结果

扩展建议

1. 认证授权 - 添加 API Key 或 JWT 认证
2. 任务队列 - 基于优先级的任务队列
3. 任务依赖 - 支持任务之间的依赖关系
4. 消息推送 - WebSocket 实时通知
5. 日志记录 - 详细的操作日志
6. 性能监控 - 任务执行时间和成功率统计
7. 分布式部署 - 支持多节点部署

许可证

MIT