构建AI智能体：Agent2Agent多智能体系统：基础通信与任务协作实现

class MessageType(Enum):
    TASK_REQUEST = "task_request"      # 任务请求
    TASK_RESPONSE = "task_response"    # 任务响应  
    GREETING = "greeting"              # 问候消息

@dataclass
class A2AMessage:
    message_id: str           # 消息唯一ID
    sender_id: str           # 发送者ID
    receiver_id: str         # 接收者ID
    message_type: MessageType # 消息类型
    content: Dict[str, Any]   # 消息内容
    timestamp: float         # 时间戳

class SimpleTransport:
    def __init__(self):
        self.agents: Dict[str, callable] = {}  # 智能体注册表

def register_agent(self, agent_id: str, message_handler: callable):
    self.agents[agent_id] = message_handler

async def send_message(self, message: A2AMessage):
    # 检查接收者是否存在
    # 模拟网络延迟
    # 调用接收者的消息处理函数

class SimpleAgent:
    def __init__(self, agent_id: str, name: str, transport: SimpleTransport):
        self.agent_id = agent_id
        self.name = name
        self.transport = transport
        self.conversation_history: List[A2AMessage] = []

async def send_message(self, receiver_id: str, message_type: MessageType, content: Dict[str, Any]):
    # 创建标准化消息
    # 记录到对话历史
    # 通过传输层发送

async def receive_message(self, message: A2AMessage):
    # 记录接收到的消息
    # 提供基础处理（子类可扩展）

class TaskManagerAgent(SimpleAgent):
    def __init__(self, agent_id: str, name: str, transport: SimpleTransport):
        super().__init__(agent_id, name, transport)
        self.pending_tasks: Dict[str, Dict] = {}   # 待处理任务
        self.completed_tasks: Dict[str, Any] = {}  # 已完成任务

async def create_task(self, task_description: str, worker_id: str):
    # 生成唯一任务ID
    # 保存任务信息到待处理列表
    # 发送任务请求给指定工作者

async def receive_message(self, message: A2AMessage):
    await super().receive_message(message)  # 基础处理
    if message.message_type == MessageType.TASK_RESPONSE:
        await self.handle_task_response(message)
    elif message.message_type == MessageType.GREETING:
        await self.handle_greeting(message)

class WorkerAgent(SimpleAgent):
    def __init__(self, agent_id: str, name: str, transport: SimpleTransport, skills: List[str]):
        super().__init__(agent_id, name, transport)
        self.skills = skills
        self.tasks_completed = 0

async def handle_task_request(self, message: A2AMessage):
    # 解析任务信息
    # 模拟任务执行（异步等待）
    # 生成任务结果
    # 发送任务完成响应

def process_task(self, description: str) -> str:
    if "计算" in description or "数学" in description:
        return f"计算完成: 42 (使用 {self.skills} 技能)"
    # 其他技能分支...

async def demo_agent_communication():
    # 创建传输层和智能体实例
    # 演示问候消息交换
    # 演示任务分配和执行流程
    # 展示统计信息

async def advanced_demo():
    # 创建多个专业智能体
    # 并行分配不同类型任务
    # 展示跨领域协作

import asyncio
import time
import uuid
from dataclasses import dataclass
from typing import Dict, Any, List, Optional
from enum import Enum

# ==================== 基础定义 ====================

class MessageType(Enum):
    """消息类型枚举"""
    TASK_REQUEST = "task_request"      # 任务请求
    TASK_RESPONSE = "task_response"    # 任务响应
    GREETING = "greeting"              # 问候消息

@dataclass
class A2AMessage:
    """
    A2A消息类
    定义智能体间通信的标准消息格式
    """
    message_id: str           # 消息唯一ID
    sender_id: str           # 发送者ID
    receiver_id: str         # 接收者ID
    message_type: MessageType # 消息类型
    content: Dict[str, Any]   # 消息内容
    timestamp: float         # 时间戳
    
    def to_dict(self) -> Dict[str, Any]:
        """将消息转换为字典格式，便于传输"""
        return {
            "message_id": self.message_id,
            "sender_id": self.sender_id,
            "receiver_id": self.receiver_id,
            "message_type": self.message_type.value,
            "content": self.content,
            "timestamp": self.timestamp
        }

class SimpleTransport:
    """
    简单传输层
    负责智能体间的消息传递（模拟网络通信）
    """
    
    def __init__(self):
        # 存储注册的智能体和它们的消息处理函数
        self.agents: Dict[str, callable] = {}
    
    def register_agent(self, agent_id: str, message_handler: callable):
        """
        注册智能体到传输层
        
        Args:
            agent_id: 智能体ID
            message_handler: 消息处理函数
        """
        self.agents[agent_id] = message_handler
        print(f" 传输层: 智能体 {agent_id} 已注册")
    
    async def send_message(self, message: A2AMessage):
        """
        发送消息到目标智能体
        
        Args:
            message: 要发送的消息
        """
        print(f" 传输层: {message.sender_id} -> {message.receiver_id} "
              f"[{message.message_type.value}]")
        
        # 检查接收者是否存在
        if message.receiver_id not in self.agents:
            print(f" 错误: 智能体 {message.receiver_id} 未注册")
            return
        
        # 获取接收者的消息处理函数
        receiver_handler = self.agents[message.receiver_id]
        
        # 模拟网络延迟
        await asyncio.sleep(0.1)
        
        # 将消息传递给接收者
        await receiver_handler(message)

# ==================== 智能体定义 ====================

class SimpleAgent:
    """
    简单智能体基类
    所有智能体的共同属性和行为
    """
    
    def __init__(self, agent_id: str, name: str, transport: SimpleTransport):
        """
        初始化智能体
        
        Args:
            agent_id: 智能体唯一标识
            name: 智能体名称
            transport: 传输层实例
        """
        self.agent_id = agent_id
        self.name = name
        self.transport = transport
        self.conversation_history: List[A2AMessage] = []
        
        # 向传输层注册自己
        self.transport.register_agent(self.agent_id, self.receive_message)
        
        print(f" 智能体初始化: {self.name} (ID: {self.agent_id})")
    
    async def send_message(self, receiver_id: str, message_type: MessageType, content: Dict[str, Any]):
        """
        发送消息给其他智能体
        
        Args:
            receiver_id: 接收者ID
            message_type: 消息类型
            content: 消息内容
        """
        # 创建消息对象
        message = A2AMessage(
            message_id=str(uuid.uuid4()),  # 生成唯一ID
            sender_id=self.agent_id,
            receiver_id=receiver_id,
            message_type=message_type,
            content=content,
            timestamp=time.time()
        )
        
        # 保存到对话历史
        self.conversation_history.append(message)
        
        # 通过传输层发送消息
        await self.transport.send_message(message)
    
    async def receive_message(self, message: A2AMessage):
        """
        接收消息的通用处理（子类可以重写）
        
        Args:
            message: 接收到的消息
        """
        print(f" {self.name} 收到来自 {message.sender_id} 的消息: {message.message_type.value}")
        
        # 保存到对话历史
        self.conversation_history.append(message)
    
    def get_conversation_summary(self):
        """获取对话摘要"""
        return f"{self.name} 进行了 {len(self.conversation_history)} 次消息交换"

class TaskManagerAgent(SimpleAgent):
    """
    任务管理器智能体
    负责创建和分配任务
    """
    
    def __init__(self, agent_id: str, name: str, transport: SimpleTransport):
        super().__init__(agent_id, name, transport)
        self.pending_tasks: Dict[str, Dict] = {}  # 待处理任务
        self.completed_tasks: Dict[str, Any] = {} # 已完成任务
    
    async def create_task(self, task_description: str, worker_id: str):
        """
        创建新任务并分配给工作者
        
        Args:
            task_description: 任务描述
            worker_id: 工作者智能体ID
        """
        task_id = f"task_{int(time.time())}"
        
        print(f" {self.name} 创建任务: {task_description}")
        
        # 保存任务信息
        self.pending_tasks[task_id] = {
            "description": task_description,
            "worker": worker_id,
            "status": "assigned",
            "created_time": time.time()
        }
        
        # 发送任务请求
        await self.send_message(
            receiver_id=worker_id,
            message_type=MessageType.TASK_REQUEST,
            content={
                "task_id": task_id,
                "description": task_description,
                "priority": "normal"
            }
        )
    
    async def receive_message(self, message: A2AMessage):
        """
        处理接收到的消息（重写父类方法）
        """
        await super().receive_message(message)  # 调用父类基础处理
        
        if message.message_type == MessageType.TASK_RESPONSE:
            # 处理任务响应
            await self.handle_task_response(message)
        elif message.message_type == MessageType.GREETING:
            # 处理问候消息
            await self.handle_greeting(message)
    
    async def handle_task_response(self, message: A2AMessage):
        """
        处理任务完成响应
        
        Args:
            message: 任务响应消息
        """
        task_id = message.content.get("task_id")
        result = message.content.get("result")
        status = message.content.get("status")
        
        if task_id in self.pending_tasks:
            if status == "completed":
                # 任务完成
                self.completed_tasks[task_id] = {
                    **self.pending_tasks[task_id],
                    "result": result,
                    "completion_time": time.time()
                }
                del self.pending_tasks[task_id]
                
                print(f" {self.name} 收到任务完成通知: {task_id}")
                print(f"   结果: {result}")
            else:
                print(f" 任务 {task_id} 失败: {result}")
    
    async def handle_greeting(self, message: A2AMessage):
        """
        处理问候消息
        
        Args:
            message: 问候消息
        """
        greeting_text = message.content.get("text", "")
        print(f" {self.name} 收到问候: {greeting_text}")

class WorkerAgent(SimpleAgent):
    """
    工作者智能体
    负责执行具体任务
    """
    
    def __init__(self, agent_id: str, name: str, transport: SimpleTransport, skills: List[str]):
        """
        初始化工作者智能体
        
        Args:
            skills: 技能列表
        """
        super().__init__(agent_id, name, transport)
        self.skills = skills
        self.tasks_completed = 0
    
    async def receive_message(self, message: A2AMessage):
        """
        处理接收到的消息
        """
        await super().receive_message(message)
        
        if message.message_type == MessageType.TASK_REQUEST:
            # 处理任务请求
            await self.handle_task_request(message)
    
    async def handle_task_request(self, message: A2AMessage):
        """
        处理任务请求并执行任务
        
        Args:
            message: 任务请求消息
        """
        task_id = message.content.get("task_id")
        description = message.content.get("description")
        
        print(f" {self.name} 开始处理任务: {description}")
        
        # 模拟任务执行时间
        await asyncio.sleep(1)
        
        # 生成任务结果（根据任务描述和技能）
        result = self.process_task(description)
        
        # 发送任务完成响应
        await self.send_message(
            receiver_id=message.sender_id,
            message_type=MessageType.TASK_RESPONSE,
            content={
                "task_id": task_id,
                "status": "completed",
                "result": result,
                "worker_skills_used": self.skills
            }
        )
        
        self.tasks_completed += 1
        print(f" {self.name} 完成任务: {description}")
    
    def process_task(self, description: str) -> str:
        """
        处理具体任务（根据描述生成结果）
        
        Args:
            description: 任务描述
            
        Returns:
            任务结果
        """
        if "计算" in description or "数学" in description:
            return f"计算完成: 42 (使用 {self.skills} 技能)"
        elif "分析" in description:
            return f"分析报告: 数据趋势良好"
        elif "编写" in description or "代码" in description:
            return f"代码已编写: def solution(): return 'Hello World'"
        else:
            return f"任务 '{description}' 已处理完成"

# ==================== 演示程序 ====================

async def demo_agent_communication():
    """
    演示智能体间通信的主函数
    """
    print("=" * 50)
    print(" Agent2Agent 通信演示开始")
    print("=" * 50)
    
    # 创建传输层实例
    transport = SimpleTransport()
    
    # 创建智能体实例
    task_manager = TaskManagerAgent(
        agent_id="manager_001", 
        name="任务管理器", 
        transport=transport
    )
    
    worker_agent = WorkerAgent(
        agent_id="worker_001",
        name="AI工作者", 
        transport=transport,
        skills=["Python编程", "数据分析", "机器学习"]
    )
    
    # 等待一下让注册完成
    await asyncio.sleep(0.1)
    
    print("\n" + "="*30 + " 场景1: 简单问候 " + "="*30)
    
    # 场景1: 简单问候
    await task_manager.send_message(
        receiver_id=worker_agent.agent_id,
        message_type=MessageType.GREETING,
        content={"text": "你好，我是任务管理器！"}
    )
    
    await asyncio.sleep(0.5)  # 等待消息处理
    
    print("\n" + "="*30 + " 场景2: 任务分配 " + "="*30)
    
    # 场景2: 任务分配和执行
    tasks = [
        "计算复杂的数学公式",
        "分析销售数据趋势", 
        "编写Python数据处理代码",
        "生成月度报告摘要"
    ]
    
    for i, task_desc in enumerate(tasks, 1):
        print(f"\n--- 任务 {i} ---")
        await task_manager.create_task(task_desc, worker_agent.agent_id)
        await asyncio.sleep(1.5)  # 等待任务完成
    
    print("\n" + "="*30 + " 结果统计 " + "="*30)
    
    # 显示结果统计
    print(f" 任务管理器统计:")
    print(f"   总消息数: {len(task_manager.conversation_history)}")
    print(f"   完成任务: {len(task_manager.completed_tasks)}")
    print(f"   待处理任务: {len(task_manager.pending_tasks)}")
    
    print(f"\n 工作者统计:")
    print(f"   总消息数: {len(worker_agent.conversation_history)}")
    print(f"   完成任务: {worker_agent.tasks_completed}")
    
    print("\n" + "="*50)
    print(" Agent2Agent 通信演示完成!")
    print("=" * 50)

async def advanced_demo():
    """
    高级演示：多个智能体协作
    """
    print("\n" + "*"*20 + " 高级演示 " + "*"*20)
    
    transport = SimpleTransport()
    
    # 创建多个智能体
    manager = TaskManagerAgent("manager_1", "高级管理器", transport)
    
    # 创建具有不同技能的多个工作者
    developer = WorkerAgent("worker_1", "开发专家", transport, 
                           ["Python", "Java", "系统设计"])
    analyst = WorkerAgent("worker_2", "数据分析师", transport,
                         ["统计分析", "数据可视化", "机器学习"])
    writer = WorkerAgent("worker_3", "技术作家", transport,
                        ["文档编写", "技术说明", "API文档"])
    
    await asyncio.sleep(0.1)
    
    # 并行分配任务给不同的工作者
    tasks = [
        ("开发新的数据处理模块", developer.agent_id),
        ("分析用户行为数据", analyst.agent_id),
        ("编写项目文档", writer.agent_id),
        ("优化算法性能", developer.agent_id),
    ]
    
    # 同时启动所有任务
    task_coroutines = [
        manager.create_task(desc, worker_id) 
        for desc, worker_id in tasks
    ]
    
    await asyncio.gather(*task_coroutines)
    
    # 等待所有任务完成
    await asyncio.sleep(3)
    
    print(f"\n 高级演示完成!")
    print(f"   管理器处理了 {len(manager.conversation_history)} 条消息")
    print(f"   完成了 {len(manager.completed_tasks)} 个跨领域任务")

if __name__ == "__main__":
    # 运行基础演示
    asyncio.run(demo_agent_communication())
    
    # 运行高级演示
    asyncio.run(advanced_demo())