为低代码 Agent 设计 Harness 可视化编排接口

1. 核心概念

1.1 低代码 Agent

在当今快速发展的技术生态系统中，“低代码 Agent” 代表了一种革命性的软件开发范式。低代码 Agent 是一种智能软件实体，它可以通过可视化界面和最小化的手工编码来构建、配置和部署。这种方法使得不仅专业开发者，甚至业务用户也能创建复杂的自动化工作流和智能应用。

低代码 Agent 的核心优势在于其抽象层级，它隐藏了底层技术复杂性，同时提供了足够的灵活性来实现各种业务场景。这些 Agent 可以集成 AI 能力、API 调用、数据处理和用户交互，形成一个完整的智能解决方案。

1.2 Harness 框架

“Harness” 在这个语境中指的是一个专门为低代码 Agent 设计的管理和编排框架。它提供了一套完整的工具和服务，用于 Agent 的创建、测试、部署、监控和优化。Harness 框架的设计理念是提供一个统一的环境，让不同类型的 Agent 可以协同工作，共享资源，并遵循一致的标准。

Harness 不仅仅是一个开发工具，更是一个完整的生命周期管理平台。它处理 Agent 之间的通信、状态管理、错误处理、性能监控等关键方面，使得开发者可以专注于业务逻辑而不是基础设施问题。

1.3 可视化编排接口

可视化编排接口是 Harness 框架的用户界面层，它允许用户通过拖拽、连接和配置图形元素来设计 Agent 的工作流程。这种编排方式使得复杂的逻辑关系变得直观可见，大大降低了理解和设计的难度。

一个好的可视化编排接口应该具备以下特点：

• 直观的拖放操作
• 清晰的流程可视化
• 丰富的组件库
• 实时预览和调试能力
• 版本控制和协作功能
• 与代码编辑器的无缝集成

2. 问题背景

2.1 传统开发方式的局限性

在传统的软件开发模式中，构建智能应用和自动化流程通常需要大量的专业编码工作。这种方式存在几个明显的问题：

1. 开发周期长：从需求分析到最终部署，传统开发可能需要数周甚至数月的时间。
2. 技术门槛高：需要掌握多种编程语言、框架和工具，限制了参与者的范围。
3. 维护成本高：代码库随着时间推移变得复杂，修改和扩展变得困难。
4. 协作困难：业务用户和技术团队之间存在沟通障碍，需求理解容易出现偏差。

2.2 Agent 技术的兴起

随着人工智能和自动化技术的发展，Agent 作为一种新的软件实体逐渐受到关注。Agent 具有自主性、反应性、主动性和社交能力等特点，非常适合构建复杂的智能系统。

然而，当前的 Agent 开发仍然面临许多挑战：

• 缺乏统一的开发标准和框架
• Agent 之间的协作和通信机制复杂
• 调试和监控困难
• 部署和扩展成本高

2.3 低代码平台的发展

低代码开发平台在过去几年中取得了显著的发展，它们通过可视化设计和模型驱动开发的方式，大大提高了软件开发的效率。然而，大多数现有的低代码平台主要关注传统的企业应用开发，对于智能 Agent 的支持还非常有限。

3. 问题描述

3.1 当前低代码 Agent 开发的痛点

尽管低代码和 Agent 技术都在快速发展，但将两者有效结合仍然面临许多挑战：

1. 缺乏专门的编排工具：现有的低代码平台通常不支持 Agent 特有的概念，如感知、推理、行动和协作。
2. 可视化表达能力有限：Agent 的行为往往是异步、并发和事件驱动的，传统的流程图难以表达这些复杂的交互模式。
3. 集成复杂性：将 AI 能力、外部 API、数据源和用户界面集成到一个 Agent 中需要大量的配置和编码工作。
4. 测试和调试困难：Agent 的行为往往是非确定性的，传统的测试方法难以覆盖所有可能的场景。
5. 性能和可扩展性问题：随着 Agent 数量和复杂度的增加，系统的性能和可扩展性成为重大挑战。

3.2 需求分析

为了解决上述问题，我们需要设计一个专门为低代码 Agent 打造的 Harness 可视化编排接口，它应该满足以下需求：

1. 支持 Agent 概念模型：提供直观的方式来表达 Agent 的感知、推理、决策和行动等核心概念。
2. 强大的可视化表达能力：能够表达并发、异步、事件驱动等复杂的交互模式。
3. 简化集成过程：提供预置的连接器和模板，简化与各种服务和系统的集成。
4. 增强的调试和监控能力：提供实时可视化的调试工具和全面的监控仪表板。
5. 优化的性能和可扩展性：设计高效的运行时引擎，支持大规模 Agent 部署和执行。

4. 问题解决

4.1 设计理念

我们的解决方案基于以下核心设计理念：

1. 以 Agent 为中心：所有设计决策都围绕 Agent 的概念模型和行为特征展开。
2. 可视化优先：尽可能通过可视化方式表达复杂概念，减少编码需求。
3. 渐进式披露：根据用户的专业水平，逐步披露系统的复杂性。
4. 可扩展性：设计模块化的架构，支持功能扩展和定制。
5. 开发者体验：注重工具的易用性和效率，提供流畅的开发体验。

4.2 架构概览

我们的 Harness 可视化编排接口采用分层架构，主要包括以下几个层次：

1. 用户界面层：提供可视化编排界面、仪表盘和管理工具。
2. 设计时层：处理流程设计、验证和转换。
3. 运行时层：负责 Agent 的执行、协调和监控。
4. 集成层：提供与外部系统和服务的连接能力。
5. 基础设施层：提供底层的计算、存储和网络资源。

4.3 核心功能模块

为了解决前述问题，我们设计了以下核心功能模块：

1. 可视化编排器：提供直观的拖放界面，支持 Agent 流程的设计和配置。
2. 组件库：包含预构建的 Agent 组件，如感知器、推理引擎、动作执行器等。
3. 集成连接器：简化与外部系统、API 和服务的集成。
4. 调试和测试工具：提供实时调试、模拟和测试环境。
5. 监控和分析仪表板：实时监控 Agent 的性能和行为，提供分析和优化建议。
6. 协作和版本控制：支持团队协作和流程版本管理。

5. 边界与外延

5.1 系统边界

我们的 Harness 可视化编排接口主要关注以下范围：

• Agent 的设计、配置和编排
• Agent 的执行和监控
• Agent 之间的协作和通信
• 与外部系统的集成

不在本系统直接范围内的包括：

• 底层 AI 模型的训练（但支持预训练模型的使用）
• 基础设施的管理（但可以与现有云平台集成）
• 完整的应用程序开发（专注于 Agent 部分）

5.2 技术栈选择

为了实现我们的设计目标，我们选择了以下技术栈：

• 前端：React、TypeScript、GraphQL、React Flow（用于可视化编排）
• 后端：Go、gRPC、Kubernetes（用于编排和扩展）
• 数据存储：PostgreSQL（结构化数据）、Redis（缓存和状态管理）、MongoDB（非结构化数据）
• 消息队列：NATS（用于 Agent 间通信）
• 监控和可观测性：Prometheus、Grafana、OpenTelemetry

5.3 与现有系统的集成

我们的设计考虑了与现有低代码平台、Agent 框架和云服务的集成：

• 低代码平台：提供 API 和插件机制，与 Mendix、OutSystems、Power Apps 等集成
• Agent 框架：支持 LangChain、AutoGPT、BabyAGI 等流行 Agent 框架
• 云服务：与 AWS、Azure、GCP 等云平台深度集成
• DevOps 工具：与 Jenkins、GitLab CI/CD、Terraform 等工具集成

6. 概念结构与核心要素组成

6.1 Agent 概念模型

在我们的系统中，Agent 由以下核心要素组成：

1. 感知器（Perceptors）：负责从环境中获取信息
2. 知识库（Knowledge Base）：存储 Agent 的知识和状态
3. 推理引擎（Reasoning Engine）：处理信息并做出决策
4. 动作执行器（Actuators）：执行具体的行动
5. 通信模块（Communication Module）：处理与其他 Agent 的交互

6.2 编排概念模型

可视化编排接口基于以下概念构建：

1. 节点（Nodes）：代表 Agent 流程中的单个步骤或操作
2. 连接（Edges）：表示节点之间的数据流和控制流
3. 容器（Containers）：用于组织和分组相关节点
4. 变量（Variables）：在流程中传递和处理的数据
5. 触发器（Triggers）：启动或影响流程执行的事件
6. 条件（Conditions）：控制流程分支的逻辑判断

6.3 组件分类

我们的组件库将包含以下几类组件：

1. 基础组件：提供基本的流程控制功能

   • 开始/结束节点
   • 条件分支
   • 循环
   • 等待
   • 并行执行

2. AI 组件：集成各种 AI 能力

   • 语言模型交互
   • 图像识别
   • 语音处理
   • 预测分析
   • 推荐系统

3. 数据处理组件：处理各种数据操作

   • 数据查询
   • 数据转换
   • 数据验证
   • 数据聚合
   • 数据可视化

4. 集成组件：连接外部系统和服务

   • API 调用
   • 数据库操作
   • 云服务集成
   • 消息队列
   • Webhook

5. 交互组件：处理用户交互

   • 表单
   • 通知
   • 审批流程
   • 聊天界面
   • 仪表盘

7. 概念之间的关系

7.1 概念核心属性维度对比

为了更好地理解不同概念之间的关系，我们从多个维度对核心概念进行对比：

概念	主要职责	执行方式	状态管理	错误处理	可重用性
感知器	信息获取	主动/被动	短期	容错	高
推理引擎	决策制定	主动	长期	重试/降级	中
动作执行器	行动执行	主动	短期	事务/补偿	高
节点	步骤执行	同步/异步	流程内	异常处理	中
连接	数据传递	单向/双向	无	重试	低
容器	组织管理	顺序/并行	容器内	范围隔离	中

7.2 ER 实体关系图

以下是系统核心概念的实体关系图：

7.3 交互关系图

以下是系统核心组件的交互关系图：

8. 数学模型

8.1 Agent 行为模型

我们可以用数学模型来描述 Agent 的行为。一个 Agent 可以看作是一个从感知到行动的映射函数：

$$A:P\to Act$$

其中，$P$ 表示感知空间，$Act$ 表示行动空间。

更详细地，我们可以将 Agent 的决策过程建模为一个部分可观察马尔可夫决策过程（POMDP）：

$$⟨S,A,T,R,\Omega ,O,\gamma ⟩$$

其中：

• $S$ 是状态集合
• $A$ 是行动集合
• $T:S\times A\times S\to [0,1]$ 是转移概率函数
• $R:S\times A\to \mathbb{R}$ 是奖励函数
• $\Omega$ 是观察集合
• $O:S\times A\times \Omega \to [0,1]$ 是观察概率函数
• $\gamma \in [0,1)$ 是折扣因子

8.2 流程编排模型

对于流程编排，我们可以使用有向图来表示：

$$G=(V,E,\lambda )$$

其中：

• $V$ 是节点集合，代表流程中的步骤
• $E\subseteq V\times V$ 是边集合，代表步骤之间的依赖关系
• $\lambda :V\to \Lambda$ 是节点标记函数，为每个节点分配类型和属性

我们还可以定义流程的执行语义：

$$\sigma \stackrel{v}{\to }{\sigma }^{\prime }$$

表示在状态 $\sigma$ 下执行节点 $v$ 后转换到状态 ${\sigma }^{\prime }$。

8.3 性能模型

为了优化系统性能，我们可以建立以下性能模型：

$$T_{total}=\sum _{i=1}^n{T}_{exec}(v_i)+\sum _{j=1}^m{T}_{comm}(e_j)+T_{overhead}$$

其中：

• $T_{total}$ 是总执行时间
• $T_{exec}(v_i)$ 是节点 $v_i$ 的执行时间
• $T_{comm}(e_j)$ 是边 $e_j$ 的通信时间
• $T_{overhead}$ 是系统开销

对于可扩展性，我们可以使用阿姆达尔定律：

$$S(n)=\frac{1}{(1-P)+\frac{P}{n}}$$

其中：

• $S(n)$ 是使用 $n$ 个处理单元时的加速比
• $P$ 是可并行化部分的比例

9. 算法流程图

9.1 流程执行引擎算法

以下是流程执行引擎的核心算法流程图：

9.2 Agent 协调算法

以下是多 Agent 协调的核心算法流程图：

10. 算法源代码

10.1 流程执行引擎核心实现

以下是流程执行引擎的核心 Python 实现：

import asyncio
import uuid
from typing import Dict, List, Any, Optional, Set
from dataclasses import dataclass, field
from enum import Enum
import logging

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class NodeStatus(Enum):
    """节点状态枚举"""
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
    SKIPPED = "skipped"

@dataclass
class Node:
    """流程节点"""
    id: str
    type: str
    properties: Dict[str, Any] = field(default_factory=dict)
    next_nodes: List[str] = field(default_factory=list)
    prev_nodes: List[str] = field(default_factory=list)

@dataclass
class Workflow:
    """流程定义"""
    id: str
    nodes: Dict[str, Node] = field(default_factory=dict)
    start_node_id: Optional[str] = None
    variables: Dict[str, Any] = field(default_factory=dict)

@dataclass
class ExecutionContext:
    """执行上下文"""
    workflow_id: str
    execution_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    node_statuses: Dict[str, NodeStatus] = field(default_factory=dict)
    node_outputs: Dict[str, Any] = field(default_factory=dict)
    variables: Dict[str, Any] = field(default_factory=dict)
    errors: List[Dict[str, Any]] = field(default_factory=list)

class NodeExecutor:
    """节点执行器基类"""
    
    async def execute(self, node: Node, context: ExecutionContext) -> Any:
        """执行节点逻辑"""
        raise NotImplementedError("Subclasses must implement execute method")
    
    async def rollback(self, node: Node, context: ExecutionContext) -> None:
        """回滚节点操作"""
        pass  # 默认不执行任何操作

class WorkflowEngine:
    """流程执行引擎"""
    
    def __init__(self):
        self._node_executors: Dict[str, NodeExecutor] = {}
        self._execution_contexts: Dict[str, ExecutionContext] = {}
    
    def register_executor(self, node_type: str, executor: NodeExecutor) -> None:
        """注册节点执行器"""
        self._node_executors[node_type] = executor
        logger.info(f"Registered executor for node type: {node_type}")
    
    async def execute_workflow(self, workflow: Workflow) -> ExecutionContext:
        """执行流程"""
        logger.info(f"Starting execution of workflow: {workflow.id}")
        
        # 创建执行上下文
        context = ExecutionContext(
            workflow_id=workflow.id,
            variables=workflow.variables.copy()
        )
        self._execution_contexts[context.execution_id] = context
        
        try:
            # 初始化所有节点状态为待处理
            for node_id in workflow.nodes:
                context.node_statuses[node_id] = NodeStatus.PENDING
            
            # 从起始节点开始执行
            if workflow.start_node_id:
                await self._execute_node_and_next(workflow, workflow.start_node_id, context)
            else:
                raise ValueError("Workflow has no start node")
            
            logger.info(f"Workflow execution completed: {workflow.id}")
        except Exception as e:
            logger.error(f"Workflow execution failed: {workflow.id}, error: {str(e)}")
            context.errors.append({
                "type": "workflow_error",
                "message": str(e)
            })
        
        return context
    
    async def _execute_node_and_next(
        self, 
        workflow: Workflow, 
        node_id: str, 
        context: ExecutionContext
    ) -> None:
        """执行节点并处理后续节点"""
        # 检查节点是否已经处理过
        if context.node_statuses[node_id] != NodeStatus.PENDING:
            logger.debug(f"Skipping already processed node: {node_id}")
            return
        
        # 检查前置条件是否满足
        node = workflow.nodes[node_id]
        if not await self._check_preconditions(node, context):
            logger.debug(f"Preconditions not met for node: {node_id}")
            return
        
        # 执行节点
        context.node_statuses[node_id] = NodeStatus.RUNNING
        logger.info(f"Executing node: {node_id}")
        
        try:
            executor = self._node_executors.get(node.type)
            if not executor:
                raise ValueError(f"No executor registered for node type: {node.type}")
            
            # 执行节点逻辑
            output = await executor.execute(node, context)
            context.node_outputs[node_id] = output
            context.node_statuses[node_id] = NodeStatus.COMPLETED
            logger.info(f"Node completed: {node_id}")
            
            # 处理后续节点
            for next_node_id in node.next_nodes:
                await self._execute_node_and_next(workflow, next_node_id, context)
                
        except Exception as e:
            logger.error(f"Node execution failed: {node_id}, error: {str(e)}")
            context.node_statuses[node_id] = NodeStatus.FAILED
            context.errors.append({
                "node_id": node_id,
                "type": "node_error",
                "message": str(e)
            })
    
    async def _check_preconditions(self, node: Node, context: ExecutionContext) -> bool:
        """检查节点的前置条件是否满足"""
        # 检查所有前置节点是否已完成
        for prev_node_id in node.prev_nodes:
            if context.node_statuses.get(prev_node_id) != NodeStatus.COMPLETED:
                return False
        return True
    
    def get_execution_context(self, execution_id: str) -> Optional[ExecutionContext]:
        """获取执行上下文"""
        return self._execution_contexts.get(execution_id)

# 示例节点执行器
class LogNodeExecutor(NodeExecutor):
    """日志节点执行器"""
    
    async def execute(self, node: Node, context: ExecutionContext) -> Any:
        message = node.properties.get("message", "No message")
        logger.info(f"Log node output: {message}")
        return {"logged": True, "message": message}

class DelayNodeExecutor(NodeExecutor):
    """延迟节点执行器"""
    
    async def execute(self, node: Node, context: ExecutionContext) -> Any:
        delay_seconds = node.properties.get("delay_seconds", 1)
        logger.info(f"Delaying for {delay_seconds} seconds")
        await asyncio.sleep(delay_seconds)
        return {"delayed": True, "seconds": delay_seconds}

class ConditionNodeExecutor(NodeExecutor):
    """条件节点执行器"""
    
    async def execute(self, node: Node, context: ExecutionContext) -> Any:
        condition = node.properties.get("condition", "true")
        
        # 简单的条件评估（实际应用中应使用更安全的方法）
        try:
            # 在实际应用中，应该使用更安全的表达式评估方法
            # 这里只是为了演示
            result = eval(condition, {}, context.variables)
            logger.info(f"Condition evaluated: {condition} = {result}")
            
            # 根据条件结果选择不同的后续节点
            # 这里假设节点属性中有 true_next 和 false_next
            if result:
                true_next = node.properties.get("true_next")
                if true_next and true_next in node.next_nodes:
                    # 过滤节点的 next_nodes，只保留 true_next
                    node.next_nodes = [true_next]
            else:
                false_next = node.properties.get("false_next")
                if false_next and false_next in node.next_nodes:
                    # 过滤节点的 next_nodes，只保留 false_next
                    node.next_nodes = [false_next]
            
            return {"condition": condition, "result": result}
        except Exception as e:
            logger.error(f"Error evaluating condition: {str(e)}")
            raise

# 示例使用
async def main():
    # 创建流程执行引擎
    engine = WorkflowEngine()
    
    # 注册节点执行器
    engine.register_executor("log", LogNodeExecutor())
    engine.register_executor("delay", DelayNodeExecutor())
    engine.register_executor("condition", ConditionNodeExecutor())
    
    # 创建一个简单的流程
    workflow = Workflow(
        id="example-workflow",
        variables={"count": 0}
    )
    
    # 创建节点
    start_node = Node(
        id="start",
        type="log",
        properties={"message": "Workflow started"},
        next_nodes=["delay1"]
    )
    
    delay_node = Node(
        id="delay1",
        type="delay",
        properties={"delay_seconds": 2},
        prev_nodes=["start"],
        next_nodes=["condition1"]
    )
    
    condition_node = Node(
        id="condition1",
        type="condition",
        properties={
            "condition": "count < 5",
            "true_next": "increment",
            "false_next": "end"
        },
        prev_nodes=["delay1"],
        next_nodes=["increment", "end"]
    )
    
    # 为了演示，这里我们用 log 节点代替 increment 节点
    increment_node = Node(
        id="increment",
        type="log",
        properties={"message": "Incrementing count"},
        prev_nodes=["condition1"],
        next_nodes=["delay1"]
    )
    
    end_node = Node(
        id="end",
        type="log",
        properties={"message": "Workflow ended"},
        prev_nodes=["condition1"]
    )
    
    # 添加节点到流程
    workflow.nodes = {
        "start": start_node,
        "delay1": delay_node,
        "condition1": condition_node,
        "increment": increment_node,
        "end": end_node
    }
    
    workflow.start_node_id = "start"
    
    # 执行流程
    context = await engine.execute_workflow(workflow)
    
    # 打印结果
    print(f"Execution ID: {context.execution_id}")
    print(f"Workflow ID: {context.workflow_id}")
    print("Node statuses:")
    for node_id, status in context.node_statuses.items():
        print(f"  {node_id}: {status.value}")
    
    if context.errors:
        print("Errors:")
        for error in context.errors:
            print(f"  {error}")

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

10.2 Agent 协调器核心实现

以下是 Agent 协调器的核心 Python 实现：

import asyncio
import uuid
import json
from typing import Dict, List, Any, Optional, Set, Callable
from dataclasses import dataclass, field
from enum import Enum
import logging

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class MessageType(Enum):
    """消息类型枚举"""
    REGISTER = "register"
    REGISTER_RESPONSE = "register_response"
    EVENT = "event"
    REQUEST = "request"
    RESPONSE = "response"
    BROADCAST = "broadcast"

class AgentStatus(Enum):
    """Agent 状态枚举"""
    INACTIVE = "inactive"
    ACTIVE = "active"
    BUSY = "busy"
    ERROR = "error"

@dataclass
class Message:
    """消息数据结构"""
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    type: MessageType
    sender_id: str
    receiver_id: Optional[str] = None
    content: Dict[str, Any] = field(default_factory=dict)
    timestamp: float = field(default_factory=lambda: asyncio.get_event_loop().time())

@dataclass
class AgentInfo:
    """Agent 信息"""
    id: str
    name: str
    capabilities: List[str] = field(default_factory=list)
    status: AgentStatus = AgentStatus.INACTIVE
    metadata: Dict[str, Any] = field(default_factory=dict)

class Agent:
    """Agent 基类"""
    
    def __init__(self, agent_id: str, name: str):
        self.id = agent_id
        self.name = name
        self.capabilities: List[str] = []
        self.knowledge_base: Dict[str, Any] = {}
        self.coordinator: Optional[AgentCoordinator] = None
        self.message_handlers: Dict[MessageType, List[Callable]] = {}
        self.event_subscriptions: Set[str] = set()
        self.status = AgentStatus.INACTIVE
    
    async def start(self) -> None:
        """启动 Agent"""
        logger.info(f"Starting agent: {self.name} ({self.id})")
        self.status = AgentStatus.ACTIVE
        await self.on_start()
    
    async def stop(self) -> None:
        """停止 Agent"""
        logger.info(f"Stopping agent: {self.name} ({self.id})")
        self.status = AgentStatus.INACTIVE
        await self.on_stop()
    
    async def on_start(self) -> None:
        """Agent 启动时的回调"""
        pass
    
    async def on_stop(self) -> None:
        """Agent 停止时的回调"""
        pass
    
    async def on_message(self, message: Message) -> None:
        """处理收到的消息"""
        logger.debug(f"Agent {self.id} received message: {message.type} from {message.sender_id}")
        
        # 调用注册的消息处理器
        if message.type in self.message_handlers:
            for handler in self.message_handlers[message.type]:
                try:
                    await handler(message)
                except Exception as e:
                    logger.error(f"Error in message handler: {str(e)}")
    
    def register_message_handler(self, message_type: MessageType, handler: Callable) -> None:
        """注册消息处理器"""
        if message_type not in self.message_handlers:
            self.message_handlers[message_type] = []
        self.message_handlers[message_type].append(handler)
    
    async def send_message(self, message: Message) -> None:
        """发送消息"""
        if self.coordinator:
            await self.coordinator.route_message(message)
        else:
            logger.error("Agent not connected to a coordinator")
    
    async def register_capability(self, capability: str) -> None:
        """注册能力"""
        if capability not in self.capabilities:
            self.capabilities.append(capability)
            logger.info(f"Agent {self.id} registered capability: {capability}")
    
    async def subscribe_to_event(self, event_type: str) -> None:
        """订阅事件"""
        self.event_subscriptions.add(event_type)
        if self.coordinator:
            await self.coordinator.subscribe_agent_to_event(self.id, event_type)
    
    async def unsubscribe_from_event(self, event_type: str) -> None:
        """取消订阅事件"""
        if event_type in self.event_subscriptions:
            self.event_subscriptions.remove(event_type)
            if self.coordinator:
                await self.coordinator.unsubscribe_agent_from_event(self.id, event_type)
    
    async def emit_event(self, event_type: str, event_data: Dict[str, Any]) -> None:
        """发出事件"""
        if self.coordinator:
            await self.coordinator.publish_event(self.id, event_type, event_data)

class AgentCoordinator:
    """Agent 协调器"""
    
    def __init__(self):
        self.agents: Dict[str, AgentInfo] = {}
        self.agent_connections: Dict[str, Agent] = {}
        self.event_subscribers: Dict[str, Set[str]] = {}
        self.message_queue: asyncio.Queue = asyncio.Queue()
        self.request_response_map: Dict[str, asyncio.Future] = {}
        self._running = False
    
    async def start(self) -> None:
        """启动协调器"""
        logger.info("Starting agent coordinator")
        self._running = True
        asyncio.create_task(self._process_messages())
    
    async def stop(self) -> None:
        """停止协调器"""
        logger.info("Stopping agent coordinator")
        self._running = False
    
    async def register_agent(self, agent: Agent) -> str:
        """注册 Agent"""
        agent_info = AgentInfo(
            id=agent.id,
            name=agent.name,
            capabilities=agent.capabilities.copy()
        )
        
        self.agents[agent.id] = agent_info
        self.agent_connections[agent.id] = agent
        agent.coordinator = self
        
        logger.info(f"Registered agent: {agent.name} ({agent.id})")
        
        # 发送注册响应
        response = Message(
            type=MessageType.REGISTER_RESPONSE,
            sender_id="coordinator",
            receiver_id=agent.id,
            content={"status": "registered", "agent_id": agent.id}
        )
        await self.route_message(response)
        
        return agent.id
    
    async def unregister_agent(self, agent_id: str) -> None:
        """注销 Agent"""
        if agent_id in self.agents:
            agent = self.agent_connections.get(agent_id)
            if agent:
                agent.coordinator = None
            
            del self.agents[agent_id]
            del self.agent_connections[agent_id]
            
            # 从事件订阅者中移除
            for event_type, subscribers in self.event_subscribers.items():
                if agent_id in subscribers:
                    subscribers.remove(agent_id)
            
            logger.info(f"Unregistered agent: {agent_id}")
    
    async def route_message(self, message: Message) -> None:
        """路由消息"""
        await self.message_queue.put(message)
    
    async def _process_messages(self) -> None:
        """处理消息队列中的消息"""
        while self._running:
            try:
                # 使用超时来定期检查运行状态
                message = await asyncio.wait_for(self.message_queue.get(), timeout=0.1)
                await self._handle_message(message)
                self.message_queue.task_done()
            except asyncio.TimeoutError:
                continue
            except Exception as e:
                logger.error(f"Error processing message: {str(e)}")
    
    async def _handle_message(self, message: Message) -> None:
        """处理单个消息"""
        logger.debug(f"Coordinator handling message: {message.type} from {message.sender_id}")
        
        if message.receiver_id:
            # 直接消息
            if message.receiver_id in self.agent_connections:
                await self.agent_connections[message.receiver_id].on_message(message)
            elif message.receiver_id == "coordinator":
                # 发送给协调器的消息
                await self._handle_coordinator_message(message)
            else:
                logger.warning(f"Receiver agent not found: {message.receiver_id}")
        else:
            # 广播消息
            for agent_id, agent in self.agent_connections.items():
                if agent_id != message.sender_id:
                    await agent.on_message(message)
    
    async def _handle_coordinator_message(self, message: Message) -> None:
        """处理发送给协调器的消息"""
        if message.type == MessageType.REGISTER:
            # 处理注册消息（通常通过 register_agent 方法处理）
            pass
        elif message.type == MessageType.REQUEST:
            # 处理请求消息
            await self._handle_request(message)
    
    async def _handle_request(self, message: Message) -> None:
        """处理请求消息"""
        request_type = message.content.get("request_type")
        
        if request_type == "find_agents":
            # 查找具有特定能力的 Agent
            capabilities = message.content.get("capabilities", [])
            matching_agents = [
                agent_info for agent_info in self.agents.values()
                if any(cap in agent_info.capabilities for cap in capabilities)
            ]
            
            # 发送响应
            response = Message(
                type=MessageType.RESPONSE,
                sender_id="coordinator",
                receiver_id=message.sender_id,
                content={
                    "request_id": message.id,
                    "agents": [
                        {"id": agent.id, "name": agent.name, "capabilities": agent.capabilities}
                        for agent in matching_agents
                    ]
                }
            )
            await self.route_message(response)
    
    async def subscribe_agent_to_event(self, agent_id: str, event_type: str) -> None:
        """订阅 Agent 到事件"""
        if event_type not in self.event_subscribers:
            self.event_subscribers[event_type] = set()
        self.event_subscribers[event_type].add(agent_id)
        logger.info(f"Agent {agent_id} subscribed to event: {event_type}")
    
    async def unsubscribe_agent_from_event(self, agent_id: str, event_type: str) -> None:
        """取消 Agent 的事件订阅"""
        if event_type in self.event_subscribers and agent_id in self.event_subscribers[event_type]:
            self.event_subscribers[event_type].remove(agent_id)
            logger.info(f"Agent {agent_id} unsubscribed from event: {event_type}")
    
    async def publish_event(self, sender_id: str, event_type: str, event_data: Dict[str, Any]) -> None:
        """发布事件"""
        logger.info(f"Publishing event: {event_type} from {sender_id}")
        
        # 创建事件消息
        event_message = Message(
            type=MessageType.EVENT,
            sender_id=sender_id,
            content={
                "event_type": event_type,
                "data": event_data
            }
        )
        
        # 发送给订阅了该事件的所有 Agent
        if event_type in self.event_subscribers:
            for agent_id in self.event_subscribers[event_type]:
                if agent_id != sender_id:
                    event_message.receiver_id = agent_id
                    await self.route_message(event_message)
        
        # 也发送给所有订阅了所有事件的 Agent（如果有这种机制的话）
    
    async def request_agent_collaboration(
        self, 
        requesting_agent_id: str, 
        target_agent_id: str, 
        task_description: Dict[str, Any]
    ) -> Optional[Dict[str, Any]]:
        """请求 Agent 协作"""
        if target_agent_id not in self.agents:
            logger.warning(f"Target agent not found: {target_agent_id}")
            return None
        
        # 创建请求消息
        request_id = str(uuid.uuid4())
        request = Message(
            id=request_id,
            type=MessageType.REQUEST,
            sender_id=requesting_agent_id,
            receiver_id=target_agent_id,
            content=task_description
        )
        
        # 创建一个 Future 来等待响应
        response_future = asyncio.Future()
        self.request_response_map[request_id] = response_future
        
        # 发送请求
        await self.route_message(request)
        
        try:
            # 等待响应（超时时间为 30 秒）
            response = await asyncio.wait_for(response_future, timeout=30.0)
            return response.content
        except asyncio.TimeoutError:
            logger.warning(f"Collaboration request timed out: {request_id}")
            return None
        finally:
            # 清理 Future
            if request_id in self.request_response_map:
                del self.request_response_map[request_id]

# 示例 Agent 实现
class DataCollectorAgent(Agent):
    """数据收集 Agent"""
    
    def __init__(self, agent_id: str, name: str):
        super().__init__(agent_id, name)
        self.capabilities = ["data_collection", "data_storage"]
    
    async def on_start(self) -> None:
        await self.register_capability("data_collection")
        await self.register_capability("data_storage")
        await self.subscribe_to_event("data_request")
        self.register_message_handler(MessageType.REQUEST, self._handle_data_request)
    
    async def _handle_data_request(self, message: Message) -> None:
        """处理数据请求"""
        request_type = message.content.get("request_type")
        if request_type == "collect_data":
            # 模拟数据收集
            data_source = message.content.get("data_source", "default")
            logger.info(f"Collecting data from: {data_source}")
            
            # 模拟耗时操作
            await asyncio.sleep(1)
            
            # 发送响应
            response = Message(
                type=MessageType.RESPONSE,
                sender_id=self.id,
                receiver_id=message.sender_id,
                content={
                    "request_id": message.id,
                    "data": {
                        "source": data_source,
                        "items": [1, 2, 3, 4, 5],
                        "collected_at": asyncio.get_event_loop().time()
                    }
                }
            )
            await self.send_message(response)

class DataProcessorAgent(Agent):
    """数据处理 Agent"""
    
    def __init__(self, agent_id: str, name: str):
        super().__init__(agent_id, name)
        self.capabilities = ["data_processing", "analysis"]
    
    async def on_start(self) -> None:
        await self.register_capability("data_processing")
        await self.register_capability("analysis")
        await self.subscribe_to_event("process_data")
        self.register_message_handler(MessageType.EVENT, self._handle_event)
        self.register_message_handler(MessageType.REQUEST, self._handle_process_request)
    
    async def _handle_event(self, message: Message) -> None:
        """处理事件"""
        event_type = message.content.get("event_type")
        if event_type == "process_data":
            data = message.content.get("data", {})
            logger.info(f"Received process_data event with data: {data}")
    
    async def _handle_process_request(self, message: Message) -> None:
        """处理处理请求"""