企业级Multi-Agent实施指南：从POC到生产环境的迁移策略

在当今快速发展的AI领域，Multi-Agent系统（多智能体系统）正在成为企业数字化转型的关键驱动力。本文将全面解析企业级Multi-Agent系统的实施路径，从概念验证（POC）到生产环境的完整迁移策略。我们将深入探讨核心概念、技术原理、架构设计、实际案例以及未来发展趋势，帮助企业决策者和技术团队理解并成功实施这一前沿技术。通过生动的比喻、详细的代码示例和实用的最佳实践，本文将为读者提供一份全

AI学长带你学AI

42人浏览 · 2026-04-12 21:57:00

AI学长带你学AI · 2026-04-12 21:57:00 发布

企业级Multi-Agent实施指南：从POC到生产环境的迁移策略

关键词

Multi-Agent系统
企业级应用
POC到生产迁移
分布式系统
智能协作
架构设计
生产部署

摘要

1. 背景介绍

1.1 主题背景和重要性

在人工智能技术快速发展的今天，单一AI模型虽然在特定任务上表现出色，但在处理复杂、动态、多维度的企业级问题时往往力不从心。这就像一个超级天才虽然在某个领域无人能敌，但要经营一家跨国企业，他需要一支专业团队，每个成员各司其职，协同工作。Multi-Agent系统正是这样一支"AI团队"，由多个智能体组成，每个智能体具有特定的能力和目标，通过协作解决复杂问题。

企业对Multi-Agent系统的兴趣正在迅速增长。根据Gartner的预测，到2025年，超过60%的大型企业将在其运营中使用Multi-Agent系统，以提高效率、创新能力和竞争力。这种增长源于企业面临的挑战日益复杂，传统的单一模型或自动化方案已无法满足需求。

1.2 目标读者

本文主要面向以下读者群体：

企业技术决策者：了解Multi-Agent系统的商业价值和实施路径
AI/ML工程师：学习Multi-Agent系统的技术实现和最佳实践
架构师：设计可扩展、可靠的Multi-Agent系统架构
产品经理：理解如何将Multi-Agent技术转化为产品功能
研究人员：探索Multi-Agent系统的前沿应用和未来方向

无论您是刚接触Multi-Agent系统的新手，还是有一定经验的从业者，本文都将提供有价值的见解和实用的指导。

1.3 核心问题或挑战

虽然Multi-Agent系统前景广阔，但从POC到生产环境的迁移过程中面临着诸多挑战：

系统复杂性：多个智能体之间的交互使系统变得复杂，难以预测和调试
可扩展性：随着智能体数量的增加，如何保持系统性能和响应时间
可靠性：确保系统在部分智能体失效时仍能继续工作
安全性：防止恶意智能体或外部攻击破坏系统
资源管理：有效分配计算资源，避免资源争夺
监控与维护：建立有效的监控机制，及时发现和解决问题
投资回报：证明Multi-Agent系统的商业价值，确保持续投入

在本文中，我们将逐一探讨这些挑战，并提供切实可行的解决方案。

2. 核心概念解析

2.1 什么是Multi-Agent系统？

让我们用一个生活化的比喻来理解Multi-Agent系统。想象一家现代化的餐厅，它不是由一个全能的超级员工运营，而是由一支专业团队组成：

前台接待：欢迎顾客，安排座位
服务员：记录订单，传递给厨房，送上食物
厨师：准备菜品
采购员：确保食材供应
经理：协调各个环节，处理特殊情况

每个角色都有自己的专长和职责，但他们需要相互沟通、协作，才能为顾客提供优质的用餐体验。这就是一个Multi-Agent系统的生动写照！

在技术术语中，Multi-Agent系统（MAS）是由多个相互作用的智能体组成的计算机系统。每个智能体都是一个自主的实体，能够感知环境、做出决策并采取行动，以实现其目标。同时，智能体之间通过通信、协作和协调来解决单个智能体无法解决的复杂问题。

2.2 核心概念详解

2.2.1 智能体（Agent）

智能体是Multi-Agent系统的基本构建块。一个智能体应该具备以下特性：

自主性：能够在没有人类直接干预的情况下运行，控制自己的行为和内部状态
反应性：能够感知环境（可能是物理世界、数字世界或其他智能体）并及时响应环境变化
主动性：不仅能够对环境做出反应，还能够通过主动采取行动实现目标
社交能力：能够与其他智能体（可能还有人类）进行交互和通信

我们可以将智能体想象成餐厅中的每个员工，他们都有自己的专业技能、工作目标和决策权，但同时也需要与其他人合作。

2.2.2 环境（Environment）

环境是智能体存在和运作的"空间"。它可以是物理环境（如机器人工作的工厂车间），也可以是数字环境（如互联网、数据库或模拟系统）。环境具有以下关键特征：

可观察性：智能体能在多大程度上感知环境的状态
动态性：环境是否会随时间变化，以及变化的速度
确定性：行动的后果是否可预测
离散性/连续性：环境状态和可能的行动是离散的还是连续的

在我们的餐厅比喻中，整个餐厅（包括顾客、食材、设备等）就是环境。

2.2.3 交互与通信（Interaction and Communication）

智能体之间的交互和通信是Multi-Agent系统的核心。没有有效的交互，系统就只是一群独立工作的个体，而不是一个协调的团队。交互可以采取多种形式：

直接通信：智能体之间直接交换信息，如通过消息传递
间接通信：智能体通过修改环境来传递信息，就像蚂蚁通过信息素沟通
协作：多个智能体共同努力实现共同目标
协调：管理智能体之间的依赖关系，避免冲突
协商：智能体通过谈判解决利益冲突，达成共识

在餐厅中，服务员向厨师传达订单就是直接通信；服务员通过在订单上做特殊标记来表明紧急程度是间接通信；整个团队共同努力提供优质服务是协作；经理安排工作时间是协调；处理顾客的特殊要求时，服务员和厨师之间的讨论是协商。

2.2.4 组织与架构（Organization and Architecture）

Multi-Agent系统的组织方式决定了智能体如何相互关联以及决策权如何分配。常见的组织架构包括：

分层架构：智能体按层次组织，上层智能体监督和协调下层智能体
联邦架构：智能体组成联盟，每个联盟有自己的协调机制
市场架构：智能体通过市场机制（如竞价）分配任务和资源
混合架构：结合多种组织方式的优点

在餐厅中，经理领导整个团队是分层架构的体现；而如果餐厅有多个分店，每个分店相对独立运营但共享某些资源，则类似于联邦架构。

2.3 概念间的关系和相互作用

为了更清晰地理解Multi-Agent系统中各概念之间的关系，让我们通过以下表格和图表来进行说明。

2.3.1 核心概念属性对比表

概念	核心属性	主要目标	关键行为	自主性程度
智能体	自主性、反应性、主动性、社交性	完成特定任务	感知、推理、决策、行动	高
环境	可观察性、动态性、确定性、离散/连续性	提供智能体运作的空间	状态变化、提供反馈	无
交互与通信	信息交换、协作、协调、协商	实现共同目标、解决冲突	消息传递、环境修改	混合
组织与架构	结构、决策权分配、协调机制	优化系统性能、实现可扩展性	组织规则、协调策略	中到高

2.3.2 Multi-Agent系统实体关系图

2.3.3 Multi-Agent系统交互关系图

2.4 Multi-Agent系统的类型

根据不同的分类标准，Multi-Agent系统可以分为多种类型：

按智能体能力分类：
- 同构系统：所有智能体具有相同的能力和目标
- 异构系统：智能体具有不同的能力和/或目标
按交互方式分类：
- 合作系统：智能体有共同目标，相互协作
- 竞争系统：智能体目标冲突，相互竞争
- 混合系统：既有合作也有竞争
按架构分类：
- 集中式系统：有一个中央控制器协调所有智能体
- 分布式系统：没有中央控制器，智能体自主决策和协调
- 混合系统：结合集中式和分布式特点

回到餐厅比喻，一家所有服务员都经过相同培训、承担相同职责的餐厅是同构系统；而有不同专长服务员（如专门负责酒水、专门负责儿童顾客）的餐厅是异构系统。一家经营良好的餐厅主要是合作系统，但也可能存在销售提成等竞争因素。

2.5 概念结构与核心要素组成

一个完整的Multi-Agent系统通常包含以下核心要素：

智能体群：多个具有特定能力和角色的智能体
环境模型：对智能体所处环境的表示和理解
通信机制：智能体之间交换信息的方式和协议
协调机制：管理智能体间交互、避免冲突的方法
知识表示：智能体存储和使用知识的方式
学习机制：智能体从经验中改进性能的能力
监控与评估系统：跟踪系统性能、评估目标达成情况

在设计Multi-Agent系统时，需要仔细考虑这些要素的设计和实现，确保它们能够协同工作，实现系统的整体目标。

3. 技术原理与实现

3.1 Multi-Agent系统的工作原理

理解Multi-Agent系统的工作原理，我们可以从单个智能体的决策过程入手，然后扩展到多个智能体的交互。

3.1.1 单个智能体的决策循环

典型的智能体决策循环包括以下步骤：

感知：智能体通过传感器获取环境信息
推理：智能体处理感知到的信息，更新内部状态，并根据目标和知识进行推理
决策：智能体选择下一步行动
行动：智能体通过执行器执行选定的行动，影响环境

这个循环不断重复，形成智能体的基本工作流程。我们可以用以下数学模型来描述这个过程：

设智能体在时间 $t$ 的内部状态为 $S_t$ ，感知到的环境信息为 $E_t$ ，采取的行动为 $A_t$ ，目标函数为 $G$ 。则智能体的决策过程可以表示为：

感知阶段： $Et=Perceive(Environment)E_t = \text{Perceive}(\text{Environment})$
状态更新： $St=UpdateState(St−1,Et)S_t = \text{UpdateState}(S_{t-1}, E_t)$
推理阶段： $Bt=Reason(St,K)B_t = \text{Reason}(S_t, K)$ ，其中 $K$ 是智能体的知识库
决策阶段： $At=Decide(Bt,G)A_t = \text{Decide}(B_t, G)$
行动阶段： $Environment=Act(At,Environment)\text{Environment} = \text{Act}(A_t, \text{Environment})$

3.1.2 多智能体的交互模型

当多个智能体共存时，它们的决策循环会相互影响。一个智能体的行动可能会改变环境，进而影响其他智能体的感知和决策。这种交互可以用以下模型表示：

假设有 $n$ 个智能体，第 $i$ 个智能体在时间 $t$ 的状态为 $S_t^i$ ，行动为 $A_t^i$ 。环境状态为 $E_t$ 。则系统的演化可以表示为：

$Et+1=τ(Et,At1,At2,…,Atn)E_{t+1} = \tau(E_t, A_t^1, A_t^2, \ldots, A_t^n)$

$St+1i=σi(Sti,ζi(Et+1))S_{t+1}^i = \sigma_i(S_t^i, \zeta_i(E_{t+1}))$

$At+1i=πi(St+1i)A_{t+1}^i = \pi_i(S_{t+1}^i)$

其中：

$τ\tau$ 是环境转移函数，描述环境如何响应智能体的行动
$σi\sigma_i$ 是第 $i$ 个智能体的状态更新函数
$ζi\zeta_i$ 是第 $i$ 个智能体的感知函数
$πi\pi_i$ 是第 $i$ 个智能体的策略函数，决定其在给定状态下采取的行动

这个模型展示了多智能体系统的核心复杂性：每个智能体的决策不仅影响环境，还会通过环境影响其他智能体的决策，形成一个复杂的反馈循环。

3.2 核心算法与技术

3.2.1 智能体决策算法

智能体的决策算法是其"大脑"，决定了它如何选择行动。以下是几种常用的决策算法：

基于规则的系统：使用"如果-那么"规则来决定行动，简单直接，但灵活性有限
决策树：通过树状结构表示决策过程，可解释性强
强化学习：智能体通过与环境交互学习最优策略，适合动态、不确定环境
规划算法：如经典规划、时序规划等，智能体预先规划行动序列
概率推理：如贝叶斯网络，用于处理不确定性

3.2.2 多智能体协调算法

当多个智能体需要协同工作时，协调算法变得至关重要：

合同网协议：智能体通过招标-投标机制分配任务
分布式约束优化：将问题建模为约束优化问题，智能体协作寻找最优解
拍卖机制：模拟市场拍卖，用于资源分配和任务分配
投票机制：智能体通过投票达成群体决策
博弈论方法：使用纳什均衡等概念分析和设计智能体交互

3.2.3 通信协议

有效的通信是多智能体协作的基础：

FIPA-ACL：智能体通信语言的标准，定义了消息类型和语义
KQML：知识查询与操作语言，早期的智能体通信语言
MQTT：轻量级消息队列协议，适合资源受限的智能体
REST/HTTP：简单通用的通信方式，适合Web环境

3.3 算法流程图

让我们用流程图来可视化多智能体系统的工作流程，从任务接收到任务完成的整个过程。

3.4 算法源代码实现

现在，让我们通过Python代码实现一个简单的Multi-Agent系统。我们将创建一个任务处理场景，其中包含不同类型的智能体协作完成任务。

import time
import random
from abc import ABC, abstractmethod
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
from enum import Enum

# 定义消息类型
class MessageType(Enum):
    TASK_REQUEST = "task_request"
    TASK_ACCEPT = "task_accept"
    TASK_REJECT = "task_reject"
    TASK_COMPLETE = "task_complete"
    STATUS_QUERY = "status_query"
    STATUS_RESPONSE = "status_response"
    HELP_REQUEST = "help_request"
    HELP_OFFER = "help_offer"

# 消息类
@dataclass
class Message:
    sender_id: str
    receiver_id: str
    msg_type: MessageType
    content: Any
    timestamp: float = None
    
    def __post_init__(self):
        if self.timestamp is None:
            self.timestamp = time.time()

# 智能体状态类
class AgentStatus(Enum):
    IDLE = "idle"
    BUSY = "busy"
    ERROR = "error"
    OFFLINE = "offline"

# 任务类
@dataclass
class Task:
    task_id: str
    description: str
    requirements: Dict[str, float]  # 资源需求
    dependencies: List[str]  # 依赖的任务ID
    estimated_time: float
    priority: int = 1  # 1-5，5最高
    status: str = "pending"  # pending, in_progress, completed, failed
    assigned_agent: Optional[str] = None
    start_time: Optional[float] = None
    end_time: Optional[float] = None

# 智能体基类
class Agent(ABC):
    def __init__(self, agent_id: str, capabilities: List[str], resources: Dict[str, float]):
        self.agent_id = agent_id
        self.capabilities = capabilities  # 能力列表
        self.resources = resources  # 可用资源
        self.status = AgentStatus.IDLE
        self.inbox: List[Message] = []
        self.outbox: List[Message] = []
        self.current_task: Optional[Task] = None
        self.knowledge_base: Dict[str, Any] = {}  # 知识库
        self.performance_history: List[Dict[str, Any]] = []  # 性能历史
    
    def receive_message(self, message: Message):
        """接收消息"""
        self.inbox.append(message)
    
    def send_message(self, message: Message):
        """发送消息（实际上是将消息放入发件箱，由环境负责传递）"""
        self.outbox.append(message)
    
    @abstractmethod
    def perceive(self, environment):
        """感知环境"""
        pass
    
    @abstractmethod
    def reason(self):
        """推理和决策"""
        pass
    
    @abstractmethod
    def act(self, environment):
        """执行行动"""
        pass
    
    def can_handle_task(self, task: Task) -> bool:
        """检查是否能处理给定任务"""
        # 检查能力匹配
        # 简化示例，实际中可能需要更复杂的匹配逻辑
        for req in task.requirements:
            if req not in self.resources or self.resources[req] < task.requirements[req]:
                return False
        return True
    
    def start_task(self, task: Task):
        """开始执行任务"""
        self.status = AgentStatus.BUSY
        self.current_task = task
        task.status = "in_progress"
        task.assigned_agent = self.agent_id
        task.start_time = time.time()
        print(f"Agent {self.agent_id} started task {task.task_id}")
    
    def complete_task(self, task: Task, success: bool = True):
        """完成任务"""
        task.status = "completed" if success else "failed"
        task.end_time = time.time()
        
        # 记录性能数据
        actual_time = task.end_time - task.start_time
        self.performance_history.append({
            "task_id": task.task_id,
            "time_taken": actual_time,
            "success": success,
            "timestamp": time.time()
        })
        
        print(f"Agent {self.agent_id} completed task {task.task_id} "
              f"with {'success' if success else 'failure'}")
        
        # 释放资源，更新状态
        self.current_task = None
        self.status = AgentStatus.IDLE
    
    def update(self, environment):
        """智能体的主更新循环"""
        # 感知环境
        self.perceive(environment)
        
        # 处理消息
        self.process_messages()
        
        # 推理和决策
        self.reason()
        
        # 执行行动
        self.act(environment)
    
    def process_messages(self):
        """处理收到的消息"""
        for message in self.inbox:
            self.handle_message(message)
        self.inbox.clear()  # 清空收件箱
    
    @abstractmethod
    def handle_message(self, message: Message):
        """处理单条消息"""
        pass

# 专门的任务执行智能体
class WorkerAgent(Agent):
    def __init__(self, agent_id: str, capabilities: List[str], resources: Dict[str, float], 
                 specialization: str = None):
        super().__init__(agent_id, capabilities, resources)
        self.specialization = specialization  # 专长领域
        self.efficiency_factor = 1.0  # 效率因子
        self.task_progress = 0.0  # 当前任务进度 (0-1)
    
    def perceive(self, environment):
        """感知环境状态"""
        # 在实际系统中，这里会收集环境信息
        self.knowledge_base["environment_status"] = environment.get_status()
        self.knowledge_base["time"] = time.time()
    
    def reason(self):
        """推理和决策过程"""
        # 如果当前有任务，更新进度
        if self.current_task and self.status == AgentStatus.BUSY:
            elapsed_time = time.time() - self.current_task.start_time
            estimated_total = self.current_task.estimated_time * self.efficiency_factor
            
            if elapsed_time >= estimated_total:
                self.task_progress = 1.0
            else:
                self.task_progress = min(elapsed_time / estimated_total, 0.99)  # 避免浮点误差
            
            # 检查是否需要帮助
            if self.task_progress < 0.5 and elapsed_time > estimated_total * 0.7:
                self.request_help()
    
    def act(self, environment):
        """执行行动"""
        # 如果任务已完成，标记为完成
        if self.current_task and self.task_progress >= 1.0:
            # 模拟一些随机失败
            success = random.random() > 0.1  # 90%成功率
            self.complete_task(self.current_task, success)
            
            # 通知协调者
            if success:
                self.send_message(Message(
                    sender_id=self.agent_id,
                    receiver_id="coordinator",
                    msg_type=MessageType.TASK_COMPLETE,
                    content={"task_id": self.current_task.task_id, "success": True}
                ))
    
    def handle_message(self, message: Message):
        """处理收到的消息"""
        if message.msg_type == MessageType.TASK_REQUEST:
            task = message.content
            if self.status == AgentStatus.IDLE and self.can_handle_task(task):
                # 接受任务
                self.send_message(Message(
                    sender_id=self.agent_id,
                    receiver_id=message.sender_id,
                    msg_type=MessageType.TASK_ACCEPT,
                    content={"task_id": task.task_id}
                ))
            else:
                # 拒绝任务
                self.send_message(Message(
                    sender_id=self.agent_id,
                    receiver_id=message.sender_id,
                    msg_type=MessageType.TASK_REJECT,
                    content={"task_id": task.task_id, "reason": "busy or incompatible"}
                ))
        
        elif message.msg_type == MessageType.TASK_ACCEPT:
            # 协调者会处理这个
            pass
        
        elif message.msg_type == MessageType.HELP_OFFER:
            # 接受或拒绝帮助
            if self.current_task and self.task_progress < 0.7:
                print(f"Agent {self.agent_id} accepting help from {message.sender_id}")
                self.efficiency_factor *= 0.8  # 提高效率
    
    def request_help(self):
        """请求其他智能体的帮助"""
        if self.current_task:
            print(f"Agent {self.agent_id} requesting help for task {self.current_task.task_id}")
            self.send_message(Message(
                sender_id=self.agent_id,
                receiver_id="coordinator",
                msg_type=MessageType.HELP_REQUEST,
                content={"task_id": self.current_task.task_id}
            ))

# 协调者智能体
class CoordinatorAgent(Agent):
    def __init__(self, agent_id: str):
        super().__init__(agent_id, ["coordination"], {})
        self.tasks: Dict[str, Task] = {}  # 所有任务
        self.unassigned_tasks: List[Task] = []  # 未分配的任务
        self.agent_registry: Dict[str, Dict[str, Any]] = {}  # 注册的智能体
        self.assignment_attempts: Dict[str, int] = {}  # 任务分配尝试次数
    
    def register_agent(self, agent_id: str, capabilities: List[str], resources: Dict[str, float]):
        """注册一个智能体"""
        self.agent_registry[agent_id] = {
            "capabilities": capabilities,
            "resources": resources,
            "status": AgentStatus.IDLE,
            "current_task": None
        }
    
    def add_task(self, task: Task):
        """添加新任务"""
        self.tasks[task.task_id] = task
        self.unassigned_tasks.append(task)
        self.assignment_attempts[task.task_id] = 0
        print(f"Coordinator received new task: {task.task_id}")
    
    def perceive(self, environment):
        """感知环境"""
        self.knowledge_base["environment_status"] = environment.get_status()
        
        # 更新已知智能体的状态
        for agent_id in self.agent_registry:
            # 在实际系统中，这里会查询或接收智能体状态更新
            pass
    
    def reason(self):
        """推理和决策"""
        # 找出可以分配的任务（依赖已满足的）
        assignable_tasks = []
        for task in self.unassigned_tasks:
            # 检查依赖是否已满足
            dependencies_met = True
            for dep_id in task.dependencies:
                if dep_id not in self.tasks or self.tasks[dep_id].status != "completed":
                    dependencies_met = False
                    break
            
            if dependencies_met:
                assignable_tasks.append(task)
        
        # 按优先级排序
        assignable_tasks.sort(key=lambda t: t.priority, reverse=True)
        
        # 找出空闲的智能体
        idle_agents = [agent_id for agent_id, info in self.agent_registry.items() 
                      if info["status"] == AgentStatus.IDLE]
        
        # 尝试匹配任务和智能体
        self.knowledge_base["pending_assignments"] = []
        for task in assignable_tasks:
            if not idle_agents:
                break
            
            # 找到最合适的智能体
            best_agent = None
            best_score = -1
            
            for agent_id in idle_agents:
                agent_info = self.agent_registry[agent_id]
                
                # 简单的评分机制，实际应用中可能更复杂
                score = 0
                for req, amount in task.requirements.items():
                    if req in agent_info["resources"]:
                        # 资源富余越多，分数越高
                        score += agent_info["resources"][req] / amount
                
                # 考虑经验/历史表现
                # 简化示例，实际中会基于performance_history计算
                
                if score > best_score:
                    best_score = score
                    best_agent = agent_id
            
            if best_agent:
                self.knowledge_base["pending_assignments"].append((task, best_agent))
                idle_agents.remove(best_agent)
    
    def act(self, environment):
        """执行行动"""
        # 分配任务
        if "pending_assignments" in self.knowledge_base:
            for task, agent_id in self.knowledge_base["pending_assignments"]:
                self.assign_task(task, agent_id)
            
            del self.knowledge_base["pending_assignments"]
    
    def assign_task(self, task: Task, agent_id: str):
        """分配任务给智能体"""
        print(f"Coordinator assigning task {task.task_id} to agent {agent_id}")
        
        # 更新内部状态
        self.agent_registry[agent_id]["status"] = AgentStatus.BUSY
        self.agent_registry[agent_id]["current_task"] = task.task_id
        self.unassigned_tasks.remove(task)
        self.assignment_attempts[task.task_id] += 1
        
        # 发送任务分配消息
        self.send_message(Message(
            sender_id=self.agent_id,
            receiver_id=agent_id,
            msg_type=MessageType.TASK_REQUEST,
            content=task
        ))
    
    def handle_message(self, message: Message):
        """处理收到的消息"""
        if message.msg_type == MessageType.TASK_ACCEPT:
            task_id = message.content["task_id"]
            agent_id = message.sender_id
            
            if task_id in self.tasks:
                task = self.tasks[task_id]
                # 查找环境中的智能体并让它开始任务
                # 在实际系统中，这需要访问环境或有其他机制
                print(f"Agent {agent_id} accepted task {task_id}")
        
        elif message.msg_type == MessageType.TASK_REJECT:
            task_id = message.content["task_id"]
            agent_id = message.sender_id
            reason = message.content["reason"]
            
            print(f"Agent {agent_id} rejected task {task_id}: {reason}")
            
            # 重置分配状态
            if task_id in self.tasks:
                task = self.tasks[task_id]
                if agent_id in self.agent_registry:
                    self.agent_registry[agent_id]["status"] = AgentStatus.IDLE
                    self.agent_registry[agent_id]["current_task"] = None
                
                # 将任务放回未分配列表，如果没有超过尝试次数
                if self.assignment_attempts.get(task_id, 0) < 3:
                    self.unassigned_tasks.append(task)
                else:
                    task.status = "failed"
                    print(f"Task {task_id} failed after multiple assignment attempts")
        
        elif message.msg_type == MessageType.TASK_COMPLETE:
            task_id = message.content["task_id"]
            agent_id = message.sender_id
            success = message.content["success"]
            
            # 更新智能体状态
            if agent_id in self.agent_registry:
                self.agent_registry[agent_id]["status"] = AgentStatus.IDLE
                self.agent_registry[agent_id]["current_task"] = None
            
            if success:
                print(f"Coordinator notified of task {task_id} completion")
            else:
                print(f"Coordinator notified of task {task_id} failure")
                # 可以选择重新分配或标记为失败
        
        elif message.msg_type == MessageType.HELP_REQUEST:
            task_id = message.content["task_id"]
            requester_id = message.sender_id
            
            print(f"Coordinator received help request for task {task_id} from {requester_id}")
            
            # 寻找可以提供帮助的空闲智能体
            for agent_id, info in self.agent_registry.items():
                if agent_id != requester_id and info["status"] == AgentStatus.IDLE:
                    # 发送帮助请求
                    self.send_message(Message(
                        sender_id=self.agent_id,
                        receiver_id=agent_id,
                        msg_type=MessageType.HELP_OFFER,
                        content={"task_id": task_id, "requester_id": requester_id}
                    ))
                    break

# 环境类
class Environment:
    def __init__(self):
        self.agents: Dict[str, Agent] = {}
        self.state: Dict[str, Any] = {
            "time": time.time(),
            "resources": {},
            "events": []
        }
    
    def add_agent(self, agent: Agent):
        """向环境中添加智能体"""
        self.agents[agent.agent_id] = agent
        
        # 如果是协调者，自动注册其他智能体
        if isinstance(agent, CoordinatorAgent):
            for other_id, other_agent in self.agents.items():
                if other_id != agent.agent_id:
                    agent.register_agent(other_id, other_agent.capabilities, other_agent.resources)
    
    def get_status(self):
        """获取环境状态"""
        return {
            "time": time.time(),
            "agent_count": len(self.agents),
            "active_agents": sum(1 for a in self.agents.values() if a.status == AgentStatus.BUSY)
        }
    
    def distribute_messages(self):
        """分发智能体之间的消息"""
        # 收集所有待发消息
        all_messages = []
        for agent in self.agents.values():
            all_messages.extend(agent.outbox)
            agent.outbox.clear()  # 清空发件箱
        
        # 分发消息
        for message in all_messages:
            if message.receiver_id in self.agents:
                self.agents[message.receiver_id].receive_message(message)
            elif message.receiver_id == "all":
                for agent in self.agents.values():
                    if agent.agent_id != message.sender_id:
                        agent.receive_message(message)
    
    def step(self):
        """环境的一个时间步"""
        self.state["time"] = time.time()
        
        # 更新所有智能体
        for agent in self.agents.values():
            if agent.status != AgentStatus.OFFLINE:
                agent.update(self)
        
        # 分发消息
        self.distribute_messages()
    
    def run(self, steps: int = 10, delay: float = 0.5):
        """运行环境多个时间步"""
        print(f"Environment starting run for {steps} steps")
        for i in range(steps):
            print(f"\n--- Step {i+1} ---")
            self.step()
            time.sleep(delay)
        print("\n--- Run completed ---")

# 示例使用
def main():
    # 创建环境
    env = Environment()
    
    # 创建协调者
    coordinator = CoordinatorAgent("coordinator")
    env.add_agent(coordinator)
    
    # 创建几个工作智能体
    resources1 = {"cpu": 4.0, "memory": 8.0, "io": 2.0}
    agent1 = WorkerAgent("worker1", ["data_processing", "analysis"], resources1)
    env.add_agent(agent1)
    
    resources2 = {"cpu": 2.0, "memory": 4.0, "io": 5.0}
    agent2 = WorkerAgent("worker2", ["data_transfer", "storage"], resources2)
    env.add_agent(agent2)
    
    resources3 = {"cpu": 8.0, "memory": 16.0, "io": 3.0}
    agent3 = WorkerAgent("worker3", ["machine_learning", "heavy_computation"], resources3)
    env.add_agent(agent3)
    
    # 创建任务
    task1 = Task(
        task_id="task_001",
        description="Collect and preprocess data",
        requirements={"cpu": 2.0, "memory": 4.0, "io": 3.0},
        dependencies=[],
        estimated_time=3.0,
        priority=3
    )
    
    task2 = Task(
        task_id="task_002",
        description="Train machine learning model",
        requirements={"cpu": 6.0, "memory": 12.0, "io": 1.0},
        dependencies=["task_001"],
        estimated_time=5.0,
        priority=5
    )
    
    task3 = Task(
        task_id="task_003",
        description="Store results in database",
        requirements={"cpu": 1.0, "memory": 2.0, "io": 4.0},
        dependencies=["task_002"],
        estimated_time=2.0,
        priority=4
    )
    
    # 将任务添加到协调者
    coordinator.add_task(task1)
    coordinator.add_task(task2)
    coordinator.add_task(task3)
    
    # 运行模拟
    env.run(steps=20, delay=1.0)
    
    # 打印最终状态
    print("\n--- Final State ---")
    print(f"Task 1 status: {task1.status}, assigned to: {task1.assigned_agent}")
    print(f"Task 2 status: {task2.status}, assigned to: {task2.assigned_agent}")
    print(f"Task 3 status: {task3.status}, assigned to: {task3.assigned_agent}")

if __name__ == "__main__":
    main()