AI Agent Harness多语言交互管控：构建跨语言智能协作系统

1. 标题 (Title)

• AI Agent Harness多语言交互管控：构建跨语言智能协作系统
• 从理论到实践：AI Agent多语言交互管控架构详解与实现
• 打破语言壁垒：AI Agent Harness如何实现高效的多语言交互与管控
• 智能协作新范式：深入理解AI Agent的多语言交互管控机制
• 构建通用AI Agent系统：多语言交互管控的设计思路与技术实践

2. 引言 (Introduction)

痛点引入 (Hook)

想象一下，你正在构建一个由多个AI Agent组成的智能系统，每个Agent都有其独特的专长——有的擅长数据分析，有的精通自然语言处理，有的则专注于决策优化。然而，当这些Agent需要协同工作时，却面临着一个巨大的挑战：它们使用着不同的"语言"进行沟通。有的Agent用Python编写，有的用JavaScript，还有的可能是用Rust或Go开发的。它们的数据格式、通信协议和交互方式都各不相同，就像一群来自不同国家的人试图在没有翻译的情况下进行复杂的商业谈判。

文章内容概述 (What)

本文将带你深入探索AI Agent Harness的多语言交互管控系统。我们将从概念基础开始，逐步构建一个能够支持多种编程语言Agent协同工作的管控框架。你将学习如何设计灵活的通信协议、实现高效的消息转换机制、构建可靠的任务调度系统，以及确保整个多Agent系统的安全性和可扩展性。

读者收益 (Why)

读完本文，你将能够：

• 理解AI Agent Harness多语言交互管控的核心概念和架构
• 设计并实现一个基本的多语言Agent通信系统
• 掌握不同编程语言Agent之间的数据格式转换技术
• 了解如何处理多Agent协作中的冲突和错误
• 获得构建可扩展、高可用多Agent系统的实践经验

3. 准备工作 (Prerequisites)

在开始我们的探索之旅之前，让我们确保你已经做好了充分的准备：

技术栈/知识

• 熟悉至少两种编程语言（我们将使用Python、JavaScript和Go作为示例）
• 理解基本的分布式系统概念（如消息队列、服务发现、负载均衡等）
• 了解RESTful API和WebSocket等通信协议
• 熟悉JSON、Protocol Buffers或其他数据序列化格式
• 对AI Agent的基本概念有一定了解

环境/工具

• 已安装Python 3.8+、Node.js 14+和Go 1.16+
• 一个代码编辑器（推荐VS Code）
• Postman或类似的API测试工具
• Docker（可选，用于容器化部署）
• Git（用于版本控制）

4. 核心内容：手把手实战 (Step-by-Step Tutorial)

步骤一：理解AI Agent Harness多语言交互管控的核心概念

在我们开始编码之前，让我们先建立一个坚实的理论基础。理解这些核心概念将帮助我们更好地设计和实现我们的系统。

什么是AI Agent Harness？

AI Agent Harness是一个用于管理和协调多个AI Agent的框架或平台。它提供了一套工具和机制，使得不同的Agent能够发现彼此、通信、协作，并共同完成复杂的任务。“Harness"这个词在这里的意思是"驾驭"或"利用”，形象地表达了这个框架的作用——将多个独立的Agent像马匹一样驾驭起来，让它们协同工作。

多语言交互管控的挑战

当我们的Agent使用不同的编程语言开发时，我们面临着几个主要挑战：

1. 通信障碍：不同语言有不同的网络库和通信机制
2. 数据格式不兼容：每种语言处理数据的方式不同，需要统一的数据序列化格式
3. 类型系统差异：静态类型语言和动态类型语言之间的类型转换问题
4. 执行环境差异：不同语言的运行时环境和资源管理方式不同
5. 错误处理机制不同：每种语言都有自己的错误处理范式

核心概念结构

让我们通过一个实体关系图来理解AI Agent Harness多语言交互管控系统的核心组件及其关系：

从这个ER图中，我们可以看到系统的核心组件包括：

• Hub：中央协调器，负责管理Agent和路由消息
• Agent：执行具体任务的智能体
• Message：Agent之间通信的载体
• Task：需要完成的工作单元
• Language Runtime：不同编程语言的运行环境
• Protocol：定义通信规则的协议
• Registry：用于Agent发现和注册的组件

步骤二：设计系统架构

现在我们已经理解了核心概念，让我们来设计一个灵活且可扩展的多语言交互管控系统架构。

分层架构设计

我们将采用分层架构来组织我们的系统，这样可以确保各部分之间的关注点分离，提高系统的可维护性和可扩展性。

让我们详细了解每一层的作用：

1. 应用层：提供用户接口和API，允许用户提交任务、监控系统状态和与Agent交互。
2. 协调层：负责任务调度、Agent管理和工作流协调，是系统的"大脑"。
3. 通信层：处理消息传递和事件通知，确保不同组件之间能够可靠通信。
4. 适配层：为不同编程语言的Agent提供统一的接口，处理语言特定的细节。
5. Agent层：包含各种专门的Agent，执行具体的任务。

核心架构决策

在设计过程中，我们需要做出几个关键的架构决策：

1. 通信协议选择：我们将使用HTTP/REST用于简单的请求-响应交互，WebSocket用于实时双向通信，以及消息队列（如RabbitMQ或NATS）用于异步消息传递。
2. 数据序列化格式：我们将主要使用JSON，因为它广泛支持且易于调试，但对于性能敏感的场景，我们也会提供Protocol Buffers支持。
3. Agent注册与发现：我们将实现一个简单的服务注册表，Agent可以在启动时注册自己，并查询其他Agent的信息。
4. 容错机制：我们将设计重试、断路器和超时机制，以确保系统在部分组件失败时仍能继续工作。
5. 安全性：我们将实现身份验证、授权和消息加密，以保护系统免受未授权访问和攻击。

步骤三：实现基础通信系统

现在让我们开始实现系统的核心部分——通信系统。我们将从创建一个简单的Hub开始，然后实现不同语言的Agent适配器。

3.1 实现Hub（使用Python）

Hub是我们系统的中央协调器，负责管理Agent注册、消息路由和任务分配。我们将使用Python和FastAPI来实现Hub，因为FastAPI提供了高性能的异步处理能力和自动API文档生成。

首先，让我们创建项目结构并安装必要的依赖：

mkdir ai-agent-harness
cd ai-agent-harness
mkdir hub python-agent js-agent go-agent
cd hub
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate
pip install fastapi uvicorn pydantic python-multipart redis

现在，让我们创建Hub的主要代码：

# hub/main.py
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException, BackgroundTasks
from pydantic import BaseModel, Field
from typing import List, Dict, Any, Optional
import json
import uuid
from datetime import datetime
import asyncio

app = FastAPI(title="AI Agent Harness Hub", version="1.0.0")

# 存储Agent信息
agents = {}
# 存储WebSocket连接
active_connections = {}
# 存储任务
tasks = {}
# 存储消息
messages = []

class AgentRegistration(BaseModel):
    id: str = Field(..., description="Agent唯一标识符")
    name: str = Field(..., description="Agent名称")
    language: str = Field(..., description="Agent开发语言")
    agent_type: str = Field(..., description="Agent类型")
    capabilities: List[str] = Field(..., description="Agent能力列表")
    metadata: Optional[Dict[str, Any]] = Field(default=None, description="额外元数据")

class Message(BaseModel):
    id: str = Field(default_factory=lambda: str(uuid.uuid4()), description="消息唯一标识符")
    sender_id: str = Field(..., description="发送者ID")
    receiver_id: str = Field(..., description="接收者ID")
    message_type: str = Field(..., description="消息类型")
    payload: Dict[str, Any] = Field(..., description="消息内容")
    timestamp: datetime = Field(default_factory=datetime.utcnow, description="时间戳")

class Task(BaseModel):
    id: str = Field(default_factory=lambda: str(uuid.uuid4()), description="任务唯一标识符")
    description: str = Field(..., description="任务描述")
    requirements: List[str] = Field(..., description="任务要求的能力")
    assigned_agent_id: Optional[str] = Field(default=None, description="分配的Agent ID")
    status: str = Field(default="pending", description="任务状态")
    result: Optional[Dict[str, Any]] = Field(default=None, description="任务结果")
    created_at: datetime = Field(default_factory=datetime.utcnow, description="创建时间")
    updated_at: datetime = Field(default_factory=datetime.utcnow, description="更新时间")

@app.on_event("startup")
async def startup_event():
    print("AI Agent Harness Hub starting up...")

@app.on_event("shutdown")
async def shutdown_event():
    print("AI Agent Harness Hub shutting down...")
    # 关闭所有WebSocket连接
    for connection in active_connections.values():
        await connection.close()

@app.post("/agents/register", response_model=Dict[str, Any])
async def register_agent(agent: AgentRegistration):
    """注册一个新的Agent"""
    if agent.id in agents:
        raise HTTPException(status_code=400, detail="Agent with this ID already registered")
    
    agents[agent.id] = {
        "id": agent.id,
        "name": agent.name,
        "language": agent.language,
        "agent_type": agent.agent_type,
        "capabilities": agent.capabilities,
        "metadata": agent.metadata or {},
        "status": "online",
        "registered_at": datetime.utcnow()
    }
    
    return {
        "status": "success",
        "message": "Agent registered successfully",
        "agent": agents[agent.id]
    }

@app.get("/agents", response_model=List[Dict[str, Any]])
async def list_agents():
    """列出所有注册的Agent"""
    return list(agents.values())

@app.get("/agents/{agent_id}", response_model=Dict[str, Any])
async def get_agent(agent_id: str):
    """获取特定Agent的信息"""
    if agent_id not in agents:
        raise HTTPException(status_code=404, detail="Agent not found")
    return agents[agent_id]

@app.delete("/agents/{agent_id}")
async def unregister_agent(agent_id: str):
    """注销Agent"""
    if agent_id not in agents:
        raise HTTPException(status_code=404, detail="Agent not found")
    
    # 关闭Agent的WebSocket连接（如果存在）
    if agent_id in active_connections:
        await active_connections[agent_id].close()
        del active_connections[agent_id]
    
    # 更新Agent状态
    agents[agent_id]["status"] = "offline"
    
    return {"status": "success", "message": "Agent unregistered successfully"}

@app.post("/messages/send", response_model=Dict[str, Any])
async def send_message(message: Message):
    """发送消息给特定Agent"""
    if message.sender_id not in agents:
        raise HTTPException(status_code=404, detail="Sender agent not found")
    
    if message.receiver_id not in agents:
        raise HTTPException(status_code=404, detail="Receiver agent not found")
    
    # 存储消息
    messages.append(message.dict())
    
    # 如果接收者有活动的WebSocket连接，通过WebSocket发送消息
    if message.receiver_id in active_connections:
        try:
            await active_connections[message.receiver_id].send_text(message.json())
        except Exception as e:
            print(f"Error sending message via WebSocket: {e}")
    
    return {
        "status": "success",
        "message": "Message sent successfully",
        "message_id": message.id
    }

@app.get("/messages", response_model=List[Dict[str, Any]])
async def list_messages(limit: int = 100):
    """列出最近的消息"""
    return messages[-limit:]

@app.post("/tasks/create", response_model=Dict[str, Any])
async def create_task(task: Task, background_tasks: BackgroundTasks):
    """创建新任务"""
    # 存储任务
    tasks[task.id] = task.dict()
    
    # 在后台分配任务
    background_tasks.add_task(assign_task, task.id)
    
    return {
        "status": "success",
        "message": "Task created successfully",
        "task_id": task.id
    }

@app.get("/tasks", response_model=List[Dict[str, Any]])
async def list_tasks(status: Optional[str] = None):
    """列出任务，可按状态过滤"""
    if status:
        return [task for task in tasks.values() if task["status"] == status]
    return list(tasks.values())

@app.get("/tasks/{task_id}", response_model=Dict[str, Any])
async def get_task(task_id: str):
    """获取特定任务的信息"""
    if task_id not in tasks:
        raise HTTPException(status_code=404, detail="Task not found")
    return tasks[task_id]

@app.post("/tasks/{task_id}/complete")
async def complete_task(task_id: str, result: Dict[str, Any]):
    """标记任务为完成"""
    if task_id not in tasks:
        raise HTTPException(status_code=404, detail="Task not found")
    
    task = tasks[task_id]
    task["status"] = "completed"
    task["result"] = result
    task["updated_at"] = datetime.utcnow()
    
    return {
        "status": "success",
        "message": "Task completed successfully",
        "task": task
    }

@app.websocket("/ws/{agent_id}")
async def websocket_endpoint(websocket: WebSocket, agent_id: str):
    """WebSocket端点，用于Agent与Hub的实时通信"""
    if agent_id not in agents:
        await websocket.close(code=1008, reason="Agent not registered")
        return
    
    await websocket.accept()
    active_connections[agent_id] = websocket
    
    # 更新Agent状态为在线
    agents[agent_id]["status"] = "online"
    
    try:
        while True:
            data = await websocket.receive_text()
            try:
                message_data = json.loads(data)
                message_type = message_data.get("type")
                
                if message_type == "ping":
                    await websocket.send_json({"type": "pong", "timestamp": datetime.utcnow().isoformat()})
                elif message_type == "message":
                    # 处理Agent发送的消息
                    message = Message(**message_data["payload"])
                    await send_message(message)
                elif message_type == "task_update":
                    # 处理任务更新
                    task_id = message_data["task_id"]
                    status = message_data["status"]
                    if task_id in tasks:
                        tasks[task_id]["status"] = status
                        tasks[task_id]["updated_at"] = datetime.utcnow()
                else:
                    print(f"Unknown message type: {message_type}")
            except json.JSONDecodeError:
                print(f"Invalid JSON received: {data}")
            except Exception as e:
                print(f"Error processing WebSocket message: {e}")
    except WebSocketDisconnect:
        print(f"Agent {agent_id} disconnected")
    finally:
        if agent_id in active_connections:
            del active_connections[agent_id]
        if agent_id in agents:
            agents[agent_id]["status"] = "offline"

async def assign_task(task_id: str):
    """为任务分配合适的Agent"""
    if task_id not in tasks:
        return
    
    task = tasks[task_id]
    requirements = task["requirements"]
    
    # 查找符合要求的在线Agent
    suitable_agents = []
    for agent_id, agent in agents.items():
        if agent["status"] == "online":
            # 检查Agent是否具备所有要求的能力
            has_all_capabilities = all(req in agent["capabilities"] for req in requirements)
            if has_all_capabilities:
                suitable_agents.append(agent_id)
    
    if not suitable_agents:
        print(f"No suitable agents found for task {task_id}")
        task["status"] = "waiting_for_agent"
        return
    
    # 简单的负载均衡：选择任务最少的Agent
    # 实际应用中可以使用更复杂的策略
    agent_task_counts = {agent_id: 0 for agent_id in suitable_agents}
    for t in tasks.values():
        if t["assigned_agent_id"] in agent_task_counts and t["status"] in ["pending", "in_progress"]:
            agent_task_counts[t["assigned_agent_id"]] += 1
    
    selected_agent_id = min(agent_task_counts, key=agent_task_counts.get)
    
    # 分配任务
    task["assigned_agent_id"] = selected_agent_id
    task["status"] = "assigned"
    task["updated_at"] = datetime.utcnow()
    
    # 如果Agent有活动的WebSocket连接，发送任务通知
    if selected_agent_id in active_connections:
        try:
            await active_connections[selected_agent_id].send_json({
                "type": "task_assigned",
                "task": task
            })
        except Exception as e:
            print(f"Error sending task assignment: {e}")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

这个Hub实现了以下核心功能：

1. Agent注册与管理
2. 消息发送与接收
3. 任务创建与分配
4. WebSocket实时通信
5. 基本的任务调度逻辑

3.2 实现Python Agent

现在让我们创建一个Python Agent，它能够与Hub通信并执行任务。

# python-agent/main.py
import asyncio
import websockets
import json
import requests
import uuid
from datetime import datetime
import time

class PythonAgent:
    def __init__(self, agent_id, name, hub_url="http://localhost:8000", capabilities=None):
        self.agent_id = agent_id
        self.name = name
        self.hub_url = hub_url
        self.capabilities = capabilities or ["data_processing", "calculation"]
        self.websocket = None
        self.is_running = False
        
    async def register(self):
        """向Hub注册Agent"""
        registration_data = {
            "id": self.agent_id,
            "name": self.name,
            "language": "python",
            "agent_type": "data_processor",
            "capabilities": self.capabilities,
            "metadata": {
                "version": "1.0.0",
                "environment": "development"
            }
        }
        
        try:
            response = requests.post(f"{self.hub_url}/agents/register", json=registration_data)
            response.raise_for_status()
            print(f"Agent {self.agent_id} registered successfully")
            return True
        except requests.exceptions.RequestException as e:
            print(f"Failed to register agent: {e}")
            return False
    
    async def connect_websocket(self):
        """连接到Hub的WebSocket端点"""
        ws_url = f"ws://localhost:8000/ws/{self.agent_id}".replace("http://", "ws://").replace("https://", "wss://")
        
        try:
            self.websocket = await websockets.connect(ws_url)
            print(f"WebSocket connected for agent {self.agent_id}")
            return True
        except Exception as e:
            print(f"Failed to connect WebSocket: {e}")
            return False
    
    async def send_message(self, receiver_id, message_type, payload):
        """发送消息给另一个Agent"""
        message = {
            "id": str(uuid.uuid4()),
            "sender_id": self.agent_id,
            "receiver_id": receiver_id,
            "message_type": message_type,
            "payload": payload,
            "timestamp": datetime.utcnow().isoformat()
        }
        
        # 通过WebSocket发送消息
        if self.websocket and self.websocket.open:
            await self.websocket.send(json.dumps({
                "type": "message",
                "payload": message
            }))
            print(f"Message sent to {receiver_id}")
        else:
            # 如果WebSocket不可用，通过HTTP API发送
            try:
                response = requests.post(f"{self.hub_url}/messages/send", json=message)
                response.raise_for_status()
                print(f"Message sent via HTTP to {receiver_id}")
            except requests.exceptions.RequestException as e:
                print(f"Failed to send message: {e}")
    
    async def update_task_status(self, task_id, status):
        """更新任务状态"""
        if self.websocket and self.websocket.open:
            await self.websocket.send(json.dumps({
                "type": "task_update",
                "task_id": task_id,
                "status": status
            }))
            print(f"Task {task_id} status updated to {status}")
    
    async def complete_task(self, task_id, result):
        """完成任务并提交结果"""
        try:
            response = requests.post(f"{self.hub_url}/tasks/{task_id}/complete", json=result)
            response.raise_for_status()
            print(f"Task {task_id} completed successfully")
        except requests.exceptions.RequestException as e:
            print(f"Failed to complete task: {e}")
    
    async def process_task(self, task):
        """处理分配的任务"""
        task_id = task["id"]
        print(f"Processing task {task_id}: {task['description']}")
        
        # 更新任务状态为进行中
        await self.update_task_status(task_id, "in_progress")
        
        # 模拟任务处理
        # 实际应用中这里会是具体的业务逻辑
        await asyncio.sleep(2)  # 模拟耗时操作
        
        # 根据任务要求生成结果
        result = {
            "processed_by": self.agent_id,
            "processed_at": datetime.utcnow().isoformat(),
            "data": {
                "status": "success",
                "message": "Task processed successfully",
                "original_requirements": task["requirements"]
            }
        }
        
        # 完成任务
        await self.complete_task(task_id, result)
    
    async def handle_message(self, message):
        """处理收到的消息"""
        print(f"Received message from {message['sender_id']}: {message['payload']}")
        # 这里可以添加消息处理逻辑
    
    async def listen(self):
        """监听WebSocket消息"""
        if not self.websocket:
            print("WebSocket not connected")
            return
        
        self.is_running = True
        
        try:
            while self.is_running:
                try:
                    # 设置超时，以便可以检查is_running标志
                    message = await asyncio.wait_for(self.websocket.recv(), timeout=1.0)
                    data = json.loads(message)
                    message_type = data.get("type")
                    
                    if message_type == "pong":
                        # 忽略pong消息
                        pass
                    elif message_type == "task_assigned":
                        # 处理任务分配
                        task = data["task"]
                        asyncio.create_task(self.process_task(task))
                    elif message_type == "message":
                        # 处理收到的消息
                        await self.handle_message(data["payload"])
                    else:
                        print(f"Unknown message type: {message_type}")
                except asyncio.TimeoutError:
                    # 超时，继续循环
                    continue
                except websockets.exceptions.ConnectionClosed:
                    print("WebSocket connection closed")
                    break
                except Exception as e:
                    print(f"Error processing message: {e}")
        finally:
            self.is_running = False
    
    async def start(self):
        """启动Agent"""
        # 注册Agent
        if not await self.register():
            return
        
        # 连接WebSocket
        if not await self.connect_websocket():
            return
        
        # 开始监听消息
        await self.listen()
    
    async def stop(self):
        """停止Agent"""
        self.is_running = False
        if self.websocket:
            await self.websocket.close()
        print(f"Agent {self.agent_id} stopped")

async def main():
    # 创建并启动Python Agent
    agent = PythonAgent(
        agent_id="python-agent-001",
        name="Python Data Processor",
        capabilities=["data_processing", "calculation", "statistics"]
    )
    
    try:
        await agent.start()
    except KeyboardInterrupt:
        print("\nStopping agent...")
        await agent.stop()

if __name__ == "__main__":
    asyncio.run(main())

这个Python Agent实现了以下功能：

1. 向Hub注册自己
2. 通过WebSocket与Hub保持实时连接
3. 接收并处理分配的任务
4. 与其他Agent通信
5. 定期发送心跳以保持连接活跃

3.3 实现JavaScript Agent

现在让我们创建一个JavaScript Agent，展示如何使用不同的编程语言实现Agent。

// js-agent/package.json
{
  "name": "js-agent",
  "version": "1.0.0",
  "description": "JavaScript AI Agent for AI Agent Harness",
  "main": "main.js",
  "type": "module",
  "scripts": {
    "start": "node main.js"
  },
  "dependencies": {
    "axios": "^1.4.0",
    "ws": "^8.13.0",
    "uuid": "^9.0.0"
  }
}

// js-agent/main.js
import axios from 'axios';
import WebSocket from 'ws';
import { v4 as uuidv4 } from 'uuid';

class JavaScriptAgent {
  constructor(agentId, name, hubUrl = 'http://localhost:8000', capabilities = []) {
    this.agentId = agentId;
    this.name = name;
    this.hubUrl = hubUrl;
    this.capabilities = capabilities || ['nlp', 'text_processing', 'sentiment_analysis'];
    this.websocket = null;
    this.isRunning = false;
    this.pingInterval = null;
  }

  async register() {
    // 向Hub注册Agent
    const registrationData = {
      id: this.agentId,
      name: this.name,
      language: 'javascript',
      agent_type: 'nlp_processor',
      capabilities: this.capabilities,
      metadata: {
        version: '1.0.0',
        environment: 'development'
      }
    };

    try {
      const response = await axios.post(`${this.hubUrl}/agents/register`, registrationData);
      console.log(`Agent ${this.agentId} registered successfully`);
      return true;
    } catch (error) {
      console.error(`Failed to register agent: ${error.message}`);
      return false;
    }
  }

  connectWebSocket() {
    // 连接到Hub的WebSocket端点
    const wsUrl = `${this.hubUrl.replace('http://', 'ws://').replace('https://', 'wss://')}/ws/${this.agentId}`;
    
    return new Promise((resolve, reject) => {
      this.websocket = new WebSocket(wsUrl);
      
      this.websocket.on('open', () => {
        console.log(`WebSocket connected for agent ${this.agentId}`);
        this.startPing();
        resolve(true);
      });
      
      this.websocket.on('message', (data) => {
        this.handleMessage(data.toString());
      });
      
      this.websocket.on('error', (error) => {
        console.error(`WebSocket error: ${error.message}`);
        reject(error);
      });
      
      this.websocket.on('close', (code, reason) => {
        console.log(`WebSocket closed: ${code} - ${reason}`);
        this.stopPing();
        this.websocket = null;
      });
    });
  }

  startPing() {
    // 定期发送ping消息保持连接活跃
    this.pingInterval = setInterval(() => {
      if (this.websocket && this.websocket.readyState === WebSocket.OPEN) {
        this.websocket.send(JSON.stringify({
          type: 'ping'
        }));
      }
    }, 30000); // 每30秒发送一次ping
  }

  stopPing() {
    if (this.pingInterval) {
      clearInterval(this.pingInterval);
      this.pingInterval = null;
    }
  }

  sendMessage(receiverId, messageType, payload) {
    // 发送消息给另一个Agent
    const message = {
      id: uuidv4(),
      sender_id: this.agentId,
      receiver_id: receiverId,
      message_type: messageType,
      payload: payload,
      timestamp: new Date().toISOString()
    };
    
    if (this.websocket && this.websocket.readyState === WebSocket.OPEN) {
      this.websocket.send(JSON.stringify({
        type: 'message',
        payload: message
      }));
      console.log(`Message sent to ${receiverId}`);
    } else {
      // 如果WebSocket不可用，通过HTTP API发送
      axios.post(`${this.hubUrl}/messages/send`, message)
        .then(() => console.log(`Message sent via HTTP to ${receiverId}`))
        .catch(error => console.error(`Failed to send message: ${error.message}`));
    }
  }

  updateTaskStatus(taskId, status) {
    // 更新任务状态
    if (this.websocket && this.websocket.readyState === WebSocket.OPEN) {
      this.websocket.send(JSON.stringify({
        type: 'task_update',
        task_id: taskId,
        status: status
      }));
      console.log(`Task ${taskId} status updated to ${status}`);
    }
  }

  async completeTask(taskId, result) {
    // 完成任务并提交结果
    try {
      await axios.post(`${this.hubUrl}/tasks/${taskId}/complete`, result);
      console.log(`Task ${taskId} completed successfully`);
    } catch (error) {
      console.error(`Failed to complete task: ${error.message}`);
    }
  }

  async processTask(task) {
    // 处理分配的任务
    const taskId = task.id;
    console.log(`Processing task ${taskId}: ${task.description}`);
    
    // 更新任务状态为进行中
    this.updateTaskStatus(taskId, 'in_progress');
    
    // 模拟任务处理
    // 实际应用中这里会是具体的业务逻辑
    await new Promise(resolve => setTimeout(resolve, 2000)); // 模拟耗时操作
    
    // 根据任务要求生成结果
    const result = {
      processed_by: this.agentId,
      processed_at: new Date().toISOString(),
      data: {
        status: 'success',
        message: 'Task processed successfully',
        original_requirements: task.requirements,
        nlp_results: {
          sentiment: 'positive',
          entities: ['AI', 'Agent', 'NLP']
        }
      }
    };
    
    // 完成任务
    await this.completeTask(taskId, result);
  }

  handleReceivedMessage(message) {
    // 处理收到的消息
    console.log(`Received message from ${message.sender_id}:`, message.payload);
    // 这里可以添加消息处理逻辑
  }

  handleMessage(data) {
    // 处理WebSocket消息
    try {
      const message = JSON.parse(data);
      const messageType = message.type;
      
      switch (messageType) {
        case 'pong':
          // 忽略pong消息
          break;
        case 'task_assigned':
          // 处理任务分配
          this.processTask(message.task);
          break;
        case 'message':
          // 处理收到的消息
          this.handleReceivedMessage(message.payload);
          break;
        default:
          console.log(`Unknown message type: ${messageType}`);
      }
    } catch (error) {
      console.error(`Error processing message: ${error.message}`);
    }
  }

  async start() {
    // 启动Agent
    console.log(`Starting JavaScript Agent ${this.agentId}...`);
    
    // 注册Agent
    const registered = await this.register();
    if (!registered) {
      console.error('Failed to register agent. Exiting.');
      return;
    }
    
    // 连接WebSocket
    try {
      await this.connectWebSocket();
      this.isRunning = true;
      console.log(`Agent ${this.agentId} started successfully`);
    } catch (error) {
      console.error('Failed to connect WebSocket. Exiting.');
      return;
    }
  }

  stop() {
    // 停止Agent
    console.log(`Stopping agent ${this.agentId}...`);
    this.isRunning = false;
    this.stopPing();
    
    if (this.websocket) {
      this.websocket.close();
    }
    
    console.log(`Agent ${this.agentId} stopped`);
  }
}

// 创建并启动JavaScript Agent
const agent = new JavaScriptAgent(
  'js-agent-001',
  'JavaScript NLP Processor',
  'http://localhost:8000',
  ['nlp', 'text_processing', 'sentiment_analysis']
);

// 启动Agent
agent.start();

// 优雅关闭
process.on('SIGINT', () => {
  console.log('\nReceived SIGINT. Shutting down gracefully...');
  agent.stop();
  process.exit(0);
});

这个JavaScript Agent与Python Agent功能类似，但使用了Node.js和JavaScript生态系统的库。它展示了如何使用不同的编程语言实现相同的Agent接口。

3.4 实现简单的测试客户端

现在让我们创建一个简单的测试客户端，用于创建任务并观察系统的运行情况。

# hub/test_client.py
import requests
import time
import json

HUB_URL = "http://localhost:8000"

def list_agents():
    """列出所有注册的Agent"""
    response = requests.get(f"{HUB_URL}/agents")
    if response.status_code == 200:
        agents = response.json()
        print(f"Registered Agents ({len(agents)}):")
        for agent in agents:
            print(f"  - {agent['name']} ({agent['id']}) - {agent['language']}")
            print(f"    Status: {agent['status']}")
            print(f"    Capabilities: {', '.join(agent['capabilities'])}")
        return agents
    else:
        print(f"Failed to list agents: {response.status_code}")
        return []

def create_task(description, requirements):
    """创建新任务"""
    task_data = {
        "description": description,
        "requirements": requirements
    }
    
    response = requests.post(f"{HUB_URL}/tasks/create", json=task_data)
    if response.status_code == 200:
        result = response.json()
        task_id = result["task_id"]
        print(f"Task created successfully: {task_id}")
        return task_id
    else:
        print(f"Failed to create task: {response.status_code} - {response.text}")
        return None

def list_tasks(status=None):
    """列出任务"""
    params = {}
    if status:
        params["status"] = status
    
    response = requests.get(f"{HUB_URL}/tasks", params=params)
    if response.status_code == 200:
        tasks = response.json()
        print(f"Tasks ({len(tasks)}):")
        for task in tasks:
            print(f"  - {task['id']}: {task['description']}")
            print(f"    Status: {task['status']}")
            if task['assigned_agent_id']:
                print(f"    Assigned to: {task['assigned_agent_id']}")
        return tasks
    else:
        print(f"Failed to list tasks: {response.status_code}")
        return []

def get_task(task_id):
    """获取任务详情"""
    response = requests.get(f"{HUB_URL}/tasks/{task_id}")
    if response.status_code == 200:
        task = response.json()
        print(f"Task Details:")
        print(f"  ID: {task['id']}")
        print(f"  Description: {task['description']}")
        print(f"  Status: {task['status']}")
        print(f"  Requirements: {', '.join(task['requirements'])}")
        if task['assigned_agent_id']:
            print(f"  Assigned to: {task['assigned_agent_id']}")
        if task['result']:
            print(f"  Result: {json.dumps(task['result'], indent=4)}")
        return task
    else:
        print(f"Failed to get task: {response.status_code}")
        return None

def main():
    print("=" * 60)
    print("AI Agent Harness Test Client")
    print("=" * 60)
    
    # 等待用户输入
    while True:
        print("\nOptions:")
        print("1. List agents")
        print("2. Create task")
        print("3. List tasks")
        print("4. Get task details")
        print("5. Exit")
        
        choice = input("Enter your choice (1-5): ")
        
        if choice == "1":
            list_agents()
        elif choice == "2":
            description = input("Enter task description: ")
            requirements_input = input("Enter requirements (comma-separated): ")
            requirements = [req.strip() for req in requirements_input.split(",")]
            create_task(description, requirements)
        elif choice == "3":
            status_filter = input("Filter by status (leave empty for all): ").strip()
            list_tasks(status_filter if status_filter else None)
        elif choice == "4":
            task_id = input("Enter task ID: ")
            get_task(task_id)
        elif choice == "5":
            print("Exiting...")
            break
        else:
            print("Invalid choice. Please try again.")

if __name__ == "__main__":
    main()

这个测试客户端提供了一个简单的命令行界面，允许我们：

1. 列出所有注册的Agent
2. 创建新任务
3. 查看任务列表
4. 获取特定任务的详细信息

步骤四：测试系统

现在我们已经实现了Hub和两种不同语言的Agent，让我们测试一下系统是否正常工作。

4.1 启动Hub

首先，让我们启动Hub：

cd hub
source venv/bin/activate  # Windows: venv\Scripts\activate
python main.py

Hub将在 http://localhost:8000 上启动，你也可以通过 http://localhost:8000/docs 访问自动生成的API文档。

4.2 启动Python Agent

在另一个终端窗口中，启动Python Agent：

cd python-agent
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate
pip install websockets requests
python main.py

4.3 启动JavaScript Agent

在第三个终端窗口中，启动JavaScript Agent：

cd js-agent
npm install
npm start

4.4 运行测试客户端

在第四个终端窗口中，运行测试客户端：

cd hub
source venv/bin/activate  # Windows: venv\Scripts\activate
python test_client.py

现在你可以使用测试客户端来创建任务并观察系统的运行情况。尝试创建一个需要"data_processing"能力的任务，看看Python Agent是否会接手处理；再创建一个需要"nlp"能力的任务，看看JavaScript Agent是否会处理它。

步骤五：自定义与美化

现在我们已经有了一个基本的工作系统，让我们通过添加一些自定义功能和改进用户体验来美化它。

5.1 添加Web界面

让我们为Hub添加一个简单的Web界面，使用户可以更方便地监控和管理系统。

首先，在hub目录中创建一个templates文件夹和一个static文件夹：

cd hub
mkdir templates static

现在，让我们更新main.py以支持模板渲染：

# 在