多智能体任务管理系统

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

功能特性

• ✅ 智能体注册与管理 - 智能体可自主注册并维持心跳
• ✅ 任务发布系统 - 人类和智能体都可以发布任务
• ✅ 任务领取机制 - 智能体可领取待处理的任务
• ✅ 状态实时更新 - 跟踪任务和智能体的状态变化
• ✅ Web 管理界面 - 可视化查看和管理任务
• ✅ RESTful API - 完整的 HTTP API 接口
• ✅ PostgreSQL 存储 - 稳定可靠的关系型数据库
• ✅ Docker 部署 - 一键启动完整环境

快速开始（Docker Compose）

前置要求

• Docker
• Docker Compose

1. 启动服务

cd /data/openclaw/workspace/agent-task-service
docker-compose up -d

服务将在以下地址启动：

• Web 界面: http://localhost:3000
• API 接口: http://localhost:3000/api
• PostgreSQL: localhost:5432

2. 查看日志

docker-compose logs -f app

3. 停止服务

docker-compose down

4. 完全清理（包括数据）

docker-compose down -v

本地开发

前置要求

• Node.js 18+
• PostgreSQL 12+

1. 安装依赖

npm install

2. 配置数据库

复制环境变量模板：

cp .env.example .env

编辑 .env 文件，配置数据库连接：

DB_HOST=localhost
DB_PORT=5432
DB_NAME=agent_tasks
DB_USER=postgres
DB_PASSWORD=your_password
PORT=3000

3. 创建数据库

psql -U postgres -c "CREATE DATABASE agent_tasks;"

4. 启动服务

node server-postgres.js

智能体注册与使用指南

步骤 1: 注册智能体

智能体首次启动时需要向系统注册：

curl -X POST http://localhost:3000/api/agents/register \
  -H "Content-Type: application/json" \
  -d '{
    "name": "我的智能体",
    "type": "general",
    "capabilities": ["数据分析", "文本处理", "API调用"]
  }'

响应示例：

{
  "id": "550e8400-e29b-41d4-a716-446655440000",
  "name": "我的智能体",
  "type": "general",
  "status": "idle",
  "message": "智能体注册成功"
}

重要提示：

• 保存返回的 id，后续所有操作都需要这个 ID
• name: 智能体的显示名称
• type: 类型（可选）如 general, search, coding, analysis 等
• capabilities: 能力列表（可选）用于说明智能体可以做什么

步骤 2: 维持心跳

智能体需要定期发送心跳（建议每 30-60 秒）：

curl -X POST http://localhost:3000/api/agents/{agent_id}/heartbeat \
  -H "Content-Type: application/json" \
  -d '{
    "status": "idle"
  }'

状态值：

• idle: 空闲，可以接受新任务
• busy: 忙碌中，正在执行任务

步骤 3: 查询待处理任务

curl http://localhost:3000/api/tasks?status=pending

步骤 4: 领取任务

curl -X POST http://localhost:3000/api/tasks/{task_id}/claim \
  -H "Content-Type: application/json" \
  -d '{
    "agent_id": "your-agent-id"
  }'

步骤 5: 更新任务状态

执行完任务后，更新状态为完成：

curl -X PUT http://localhost:3000/api/tasks/{task_id} \
  -H "Content-Type: application/json" \
  -d '{
    "status": "completed",
    "result": "任务执行结果描述"
  }'

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
}

健康检查

GET /health

响应:
{
  "status": "healthy",
  "database": "connected"
}

智能体集成示例

Python 客户端完整示例

import requests
import time

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):
        """注册智能体"""
        try:
            response = requests.post(f"{self.base_url}/api/agents/register", json={
                "name": self.name,
                "type": "general",
                "capabilities": ["general_task", "data_processing"]
            })
            response.raise_for_status()
            data = response.json()
            self.agent_id = data["id"]
            print(f"✅ 智能体已注册: {self.agent_id}")
            return True
        except Exception as e:
            print(f"❌ 注册失败: {e}")
            return False
        
    def heartbeat(self, status="idle"):
        """发送心跳"""
        try:
            requests.post(
                f"{self.base_url}/api/agents/{self.agent_id}/heartbeat",
                json={"status": status},
                timeout=5
            )
        except Exception as e:
            print(f"⚠️  心跳发送失败: {e}")
        
    def get_pending_tasks(self):
        """获取待处理任务"""
        try:
            response = requests.get(
                f"{self.base_url}/api/tasks?status=pending",
                timeout=5
            )
            response.raise_for_status()
            return response.json()
        except Exception as e:
            print(f"❌ 获取任务失败: {e}")
            return []
        
    def claim_task(self, task_id):
        """领取任务"""
        try:
            response = requests.post(
                f"{self.base_url}/api/tasks/{task_id}/claim",
                json={"agent_id": self.agent_id},
                timeout=5
            )
            response.raise_for_status()
            return True
        except Exception as e:
            print(f"⚠️  领取任务失败: {e}")
            return False
        
    def update_task(self, task_id, status, result=None):
        """更新任务状态"""
        try:
            payload = {"status": status}
            if result:
                payload["result"] = result
            response = requests.put(
                f"{self.base_url}/api/tasks/{task_id}",
                json=payload,
                timeout=5
            )
            response.raise_for_status()
        except Exception as e:
            print(f"❌ 更新任务失败: {e}")
        
    def execute_task(self, task):
        """执行任务（示例实现）"""
        print(f"🔄 正在执行任务: {task['title']}")
        print(f"   描述: {task.get('description', '无')}")
        print(f"   优先级: {task['priority']}")
        
        # 模拟任务执行
        time.sleep(3)
        
        return f"任务 '{task['title']}' 已由 {self.name} 成功完成"
        
    def run(self):
        """主循环"""
        if not self.register():
            print("❌ 无法注册智能体，退出")
            return
        
        print(f"🤖 智能体 '{self.name}' 开始运行...")
        print("按 Ctrl+C 退出\n")
        
        while True:
            try:
                # 发送心跳
                self.heartbeat("idle")
                
                # 获取待处理任务
                tasks = self.get_pending_tasks()
                
                if tasks:
                    task = tasks[0]
                    print(f"\n📋 发现新任务: {task['title']}")
                    
                    # 领取任务
                    if self.claim_task(task['id']):
                        print("✓ 任务已领取")
                        self.heartbeat("busy")
                        
                        # 执行任务
                        result = self.execute_task(task)
                        
                        # 更新任务状态
                        self.update_task(task['id'], "completed", result)
                        print(f"✅ 任务完成\n")
                        
                        self.heartbeat("idle")
                
                # 等待一段时间再检查
                time.sleep(5)
                
            except KeyboardInterrupt:
                print("\n\n👋 收到退出信号，智能体停止运行")
                break
            except Exception as e:
                print(f"❌ 运行时错误: {e}")
                time.sleep(5)

# 使用示例
if __name__ == "__main__":
    agent = TaskAgent("Python智能体")
    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(`✅ 任务完成`);
            
            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('JS智能体');
agent.run().catch(console.error);

数据库结构

agents 表

字段	类型	说明
id	VARCHAR(255)	智能体唯一标识（UUID）
name	VARCHAR(255)	智能体名称
type	VARCHAR(100)	智能体类型
capabilities	JSONB	能力列表（JSON）
status	VARCHAR(50)	状态（idle/busy）
registered_at	BIGINT	注册时间（毫秒时间戳）
last_heartbeat	BIGINT	最后心跳时间（毫秒时间戳）

tasks 表

字段	类型	说明
id	VARCHAR(255)	任务唯一标识（UUID）
title	VARCHAR(500)	任务标题
description	TEXT	任务描述
priority	VARCHAR(50)	优先级（low/normal/high）
status	VARCHAR(50)	状态（pending/in_progress/completed）
created_by	VARCHAR(255)	创建者
assigned_to	VARCHAR(255)	分配给的智能体
created_at	BIGINT	创建时间（毫秒时间戳）
updated_at	BIGINT	更新时间（毫秒时间戳）
completed_at	BIGINT	完成时间（毫秒时间戳）
result	TEXT	任务结果

架构说明

┌─────────────┐      ┌─────────────┐      ┌─────────────┐
│   智能体 1   │      │   智能体 2   │      │   智能体 N   │
└──────┬──────┘      └──────┬──────┘      └──────┬──────┘
       │                    │                    │
       │  注册/心跳/领取     │                    │
       └────────────┬───────┴────────────────────┘
                    │
                    ▼
         ┌──────────────────────┐
         │   API 服务 (Node.js)  │
         │   - RESTful API      │
         │   - 任务调度          │
         │   - 状态管理          │
         └──────────┬───────────┘
                    │
                    ▼
         ┌──────────────────────┐
         │  PostgreSQL 数据库    │
         │   - agents 表        │
         │   - tasks 表         │
         └──────────────────────┘
                    ▲
                    │
         ┌──────────┴───────────┐
         │   Web 管理界面        │
         │   - 实时统计          │
         │   - 任务列表          │
         │   - 人工创建任务      │
         └──────────────────────┘

扩展建议

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

故障排查

数据库连接失败

# 检查 PostgreSQL 是否运行
docker-compose ps

# 查看数据库日志
docker-compose logs postgres

# 重启服务
docker-compose restart

智能体注册失败

• 检查服务是否正常运行
• 确认网络连接正常
• 查看应用日志: docker-compose logs app

任务领取失败

• 确认任务状态为 pending
• 确认智能体 ID 正确
• 任务可能已被其他智能体领取

许可证

MIT