AI Agent在智能生产中的应用：多智能体协同调度与优化案例

SuperAGI2025

293人浏览 · 2026-04-13 20:24:52

SuperAGI2025 · 2026-04-13 20:24:52 发布

AI Agent在智能生产中的应用：多智能体协同调度与优化案例

关键词

AI Agent、多智能体系统、智能生产、协同调度、优化算法、智能制造、工业4.0

摘要

在工业4.0和智能制造的浪潮下，AI Agent技术正逐渐成为智能生产系统的核心驱动力。本文将深入探讨AI Agent在智能生产中的应用，特别聚焦于多智能体协同调度与优化问题。我们将从基础概念出发，通过生动形象的比喻解释复杂技术，结合数学模型和算法实现，最后通过实际案例展示其应用价值。文章不仅为读者提供理论知识，还包含实用的代码示例和最佳实践，帮助读者理解和应用这一前沿技术。

1. 背景介绍

核心概念

在开始深入探讨之前，让我们先明确几个核心概念：

AI Agent（智能体）：指能够感知环境、做出决策并执行动作的自主实体，具有自主性、反应性、主动性和社交能力等特征。
多智能体系统（MAS）：由多个相互作用的智能体组成的系统，这些智能体通过协作或竞争来解决复杂问题。
智能生产：利用人工智能、物联网、大数据等先进技术，实现生产过程的自动化、智能化和优化。
协同调度：在资源受限的情况下，通过协调多个智能体的活动，实现整体目标的最优化。
优化算法：用于在可行解空间中寻找最优或近似最优解的算法，如遗传算法、强化学习等。

问题背景

传统的生产调度系统往往依赖于集中式控制和预先设定的规则，在面对复杂多变的生产环境时显得力不从心。随着产品定制化需求增加、生产周期缩短、资源约束增强，企业亟需更灵活、更智能的生产调度方案。

想象一下传统工厂的生产场景：一条流水线可能由数十台设备组成，每种产品需要经过多个工序，每个工序需要特定的设备和工时。一旦某个设备发生故障、某个订单加急或者原材料供应出现问题，整个生产计划就可能被打乱，调度员需要花费大量时间重新安排生产计划，而且往往难以找到最优解。

这就像是交通高峰期的城市道路，每个车辆都是一个独立的"参与者"，如果没有有效的交通管理系统，整个城市交通就会陷入瘫痪。但如果每个车辆都能智能地感知周围环境、与其他车辆沟通，并自主做出最优决策，那么交通效率将大幅提升。AI Agent在智能生产中的应用，正是要实现这样的愿景。

问题描述

在智能生产环境中，我们面临的主要问题可以概括为以下几个方面：

动态不确定性：生产环境中存在各种不确定因素，如设备故障、订单变更、质量波动等，传统方法难以应对。
多目标优化：生产调度通常需要同时考虑多个目标，如最小化生产周期、降低生产成本、提高设备利用率等，这些目标之间往往存在冲突。
分布式决策：现代生产系统往往由多个分布式单元组成，需要在分散决策的同时保持整体协调。
实时响应：对于生产现场的突发情况，需要能够快速做出响应和调整。
人机协作：在许多场景中，自动化设备与人工操作需要协同工作，如何实现高效的人机协作也是一个重要挑战。

问题解决思路

AI Agent技术为解决上述问题提供了新的思路。通过将生产系统中的各种资源（如设备、物料、人员、订单等）抽象为智能体，利用多智能体系统的协同机制，可以实现更加灵活、智能的生产调度与优化。

这就像是组建一支高效的团队：每个团队成员（智能体）都有自己的专长和职责，他们能够自主完成任务，同时又能与其他成员有效沟通协作。当遇到问题时，他们可以快速调整策略，共同找到最佳解决方案。

在接下来的章节中，我们将深入探讨这一解决方案的核心概念、技术原理、实现方法以及实际应用案例。

边界与外延

在开始深入探讨之前，我们需要明确本文的讨论边界：

边界：本文主要聚焦于离散制造业中的多智能体协同调度问题，特别是车间作业调度（Job Shop Scheduling）和柔性作业车间调度（Flexible Job Shop Scheduling）场景。
外延：虽然本文重点讨论生产调度，但AI Agent技术在智能生产中的应用远不止于此，还包括设备维护、质量控制、供应链管理、仓储物流等多个领域。相关的技术方法也可以迁移应用到其他领域，如交通调度、电网管理、服务资源分配等。

2. 核心概念解析

2.1 AI Agent：智能生产中的"自主员工"

让我们从最基础的概念开始——什么是AI Agent？想象一下你在一家现代化工厂工作，这里有一位特别的"员工"：

它能够通过传感器"看到"周围的环境（感知能力）
它有自己的目标和任务（目的性）
它能够根据环境变化自主做出决策（自主性）
它可以和其他"同事"交流协作（社交能力）
它能够从经验中学习，不断提高工作效率（学习能力）

这就是AI Agent的形象化描述。在学术上，我们可以将AI Agent定义为：能够感知环境、自主做出决策并执行动作，以实现特定目标的计算实体。

智能体的核心属性

一个典型的AI Agent具有以下核心属性：

自主性（Autonomy）：智能体能够在没有人类直接干预的情况下运行，控制自己的行为和内部状态。
反应性（Reactivity）：智能体能够感知环境，并对环境的变化做出及时反应。
主动性（Pro-activeness）：智能体不仅能够对环境做出反应，还能够主动地采取行动实现目标。
社交能力（Social Ability）：智能体能够与其他智能体（或人类）进行交互、沟通和协作。

这四个特性可以用我们熟悉的团队协作场景来理解：一个优秀的团队成员应该能够独立完成工作（自主性）、及时应对突发情况（反应性）、主动承担任务和解决问题（主动性），并且善于与他人合作（社交能力）。

智能体的基本架构

从技术实现的角度看，AI Agent通常包含以下几个核心组件：

感知模块：负责获取环境信息，如传感器数据、其他智能体的状态等。
推理/决策模块：根据感知到的信息、知识库和目标，进行推理和决策。
动作执行模块：将决策转化为具体的动作，作用于环境。
知识库：存储智能体的知识、经验和规则。
通信模块：负责与其他智能体进行信息交换和协调。
目标/效用函数：定义智能体的目标和评价标准。

2.2 多智能体系统：智能生产中的"协作团队"

单个智能体的能力是有限的，就像一个人无法完成所有工作一样。在复杂的生产环境中，我们需要多个智能体协同工作，这就形成了多智能体系统（Multi-Agent System，MAS）。

什么是多智能体系统？

多智能体系统是由多个相互作用的智能体组成的系统，这些智能体通过协作、协调或竞争来解决单个智能体无法解决的复杂问题。

这就像一个足球队：每个球员都是一个智能体，他们有自己的位置和职责（守门员、前锋、中场、后卫），有个人技术和特点，但更重要的是他们能够通过传球、跑位、配合来赢得比赛。

多智能体系统的特点

分布性：智能体在物理上或逻辑上是分布的，没有全局控制中心。
异构性：不同的智能体可能具有不同的能力、知识和目标。
交互性：智能体之间通过通信、协作、协商等方式进行交互。
自组织：系统能够通过智能体的局部交互产生全局的有序行为。
涌现性：系统整体表现出单个智能体不具备的特性和能力。

多智能体系统的类型

根据智能体之间的关系和系统结构，多智能体系统可以分为以下几类：

系统类型	特点	适用场景	示例
协作型多智能体系统	智能体具有共同的目标，通过协作实现整体最优	生产调度、交通管理	车间设备协同作业
竞争型多智能体系统	智能体具有相互冲突的目标，通过竞争获得自身利益	资源分配、拍卖	供应链中的价格谈判
混合系统	同时存在协作和竞争关系	复杂市场环境	企业内部各部门协作与竞争
分层系统	智能体按层次组织，高层智能体协调低层智能体	大型复杂系统	多级生产管理系统

2.3 智能生产系统：从自动化到智能化

传统的生产系统主要依靠自动化设备和集中控制系统，而智能生产系统则引入了AI Agent和多智能体技术，实现了从自动化到智能化的跨越。

传统生产系统 vs 智能生产系统

让我们通过对比来理解两者的区别：

特性	传统生产系统	智能生产系统
控制方式	集中式控制	分布式协同
适应性	固定流程，适应能力弱	灵活调整，适应能力强
决策机制	预先设定的规则	自主学习和决策
资源利用	静态配置	动态优化配置
容错能力	单点故障影响大	分布式容错，鲁棒性强
人机关系	人操作机器	人机协作，共同决策

这就好比传统的指挥控制系统和现代的扁平化团队：传统系统依赖上级的指令，而现代系统则让每个团队成员都能发挥主动性和创造性，同时保持整体协同。

智能生产中的智能体类型

在智能生产系统中，我们可以将各种资源和实体抽象为不同类型的智能体：

资源智能体（Resource Agent）：代表生产资源，如机床、机器人、运输设备等。
- 能力：了解自身状态、处理任务、报告故障
- 目标：最大化自身利用率、最小化能耗
产品智能体（Product Agent）：代表在制品或待加工产品。
- 能力：携带加工工艺信息、请求资源、跟踪生产进度
- 目标：按时完成加工、保证质量
订单智能体（Order Agent）：代表客户订单。
- 能力：分解订单、协调生产、跟踪交货期
- 目标：按时交货、满足客户需求
运输智能体（Transport Agent）：负责物料和产品的运输。
- 能力：规划路径、避障、优化运输路线
- 目标：高效运输、减少等待时间
管理智能体（Management Agent）：负责全局协调和优化。
- 能力：监控系统状态、协调冲突、优化全局目标
- 目标：系统整体效率最优化

这些智能体之间的关系和交互构成了智能生产系统的核心。

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

现在，让我们将这些概念整合起来，构建AI Agent在智能生产中应用的概念结构：

这个概念结构展示了AI Agent在智能生产系统中的四个核心层次：

环境层：包括物理设备、物料系统、人员和信息系统等生产环境要素。
感知层：负责收集传感器数据、智能体状态、任务信息和环境变化等数据。
决策层：由各种类型的智能体组成，它们通过交互和协作做出决策。
执行层：将决策转化为实际的控制动作，如设备控制、物料运输、任务分配等。

2.5 概念之间的关系：核心属性维度对比

为了更清晰地理解不同类型智能体的特点，让我们从多个核心属性维度进行对比：

智能体类型	自主性	反应性	主动性	社交性	主要目标	决策时间尺度	知识范围
资源智能体	高	高	中	中	最大化资源利用率	短（秒-分钟）	局部
产品智能体	中	中	高	高	按时完成加工	中（分钟-小时）	局部
订单智能体	中	低	高	高	满足客户需求	长（小时-天）	全局
运输智能体	高	高	中	中	高效运输	短（秒-分钟）	局部
管理智能体	中	中	高	高	系统整体优化	中-长（分钟-天）	全局

从这个对比表中，我们可以看出不同类型智能体的特点和定位：

资源智能体和运输智能体需要快速响应环境变化，因此具有较高的自主性和反应性，决策时间尺度较短，主要关注局部目标。
产品智能体和订单智能体更加关注目标的实现，因此具有较高的主动性和社交性，决策时间尺度较长，需要与其他智能体进行更多的协作。
管理智能体则负责全局协调和优化，需要平衡各种目标，决策时间尺度覆盖中长范围，知识范围最广。

2.6 概念联系的ER实体关系图

为了更好地理解这些概念之间的关系，让我们使用ER（实体关系）图来表示：

这个ER图展示了智能生产系统中各类型智能体之间的主要关系：

订单智能体将订单分解为多个产品智能体。
产品智能体向资源智能体请求加工服务，向运输智能体请求运输服务。
资源智能体之间可能存在协作关系。
管理智能体监控、协调和优化其他所有智能体。

通过这些关系，不同类型的智能体能够有效地协同工作，共同实现生产目标。

2.7 智能体交互关系图

除了静态的实体关系，智能体之间的动态交互也非常重要。让我们来看一个典型的智能体交互流程：

这个交互流程展示了一个典型的订单从接收到完成的全过程：

订单智能体向管理智能体提交新订单。
订单智能体创建产品智能体。
产品智能体与资源智能体协商加工安排。
产品智能体与运输智能体协商运输安排。
产品智能体跟踪整个生产过程，协调各资源和运输。
生产完成后，产品智能体向订单智能体报告。

在实际生产中，这个交互过程可能会更加复杂，可能会出现资源冲突、设备故障、订单变更等情况，这就需要智能体具有更强的适应能力和协商能力。

3. 技术原理与实现

在理解了核心概念之后，让我们深入探讨AI Agent在智能生产中应用的技术原理与实现方法。我们将从数学模型、算法原理到代码实现，一步步构建多智能体协同调度系统。

3.1 生产调度问题的数学模型

首先，让我们用数学语言来描述生产调度问题。在智能生产中，最经典的调度问题是柔性作业车间调度问题（Flexible Job Shop Scheduling Problem, FJSP），它比传统的作业车间调度问题更加复杂，也更贴近实际生产场景。

问题定义

柔性作业车间调度问题可以定义为：

有 $n$ 个作业（Job） $\{J_1, J_2, \ldots, J_n\}$ 需要加工
每个作业 $J_i$ 包含 $n_i$ 个工序（Operation） $Oi1,Oi2,…,OiniO_{i1}, O_{i2}, \ldots, O_{in_i}$
有 $m$ 台机器（Machine） $\{M_1, M_2, \ldots, M_m\}$ 可以使用
每个工序 $O_{ij}$ 可以在一组可选机器 $Mij⊆MM_{ij} \subseteq M$ 中的任意一台上加工
工序 $O_{ij}$ 在机器 $M_k$ 上的加工时间为 $p_{ijk}$
工序之间存在工艺约束：同一作业的工序必须按顺序加工，即 $O_{ij}$ 必须在 $O_{i(j-1)}$ 完成后才能开始
机器约束：同一台机器在同一时间只能加工一个工序

我们的目标是找到一个调度方案，使得某些性能指标最优化，常见的性能指标包括：

最大完成时间（Makespan）：所有作业完成时间的最大值，即 $Cmax=max⁡{Ci∣i=1,2,…,n}C_{\text{max}} = \max\{C_i \mid i = 1, 2, \ldots, n\}$ ，其中 $C_i$ 是作业 $J_i$ 的完成时间。
总流经时间（Total Flow Time）：所有作业从开始到完成的时间之和，即 $∑i=1n(Ci−ri)\sum_{i=1}^n (C_i - r_i)$ ，其中 $r_i$ 是作业 $J_i$ 的就绪时间。
总拖期时间（Total Tardiness）：所有作业超过交货期的时间之和，即 $∑i=1nmax⁡(0,Ci−di)\sum_{i=1}^n \max(0, C_i - d_i)$ ，其中 $d_i$ 是作业 $J_i$ 的交货期。
机器利用率（Machine Utilization）：机器的有效加工时间与总可用时间的比率。

在本文中，我们主要以最小化最大完成时间（Makespan）为目标。

数学模型

我们可以用以下数学模型来形式化描述FJSP问题：

变量定义：

$x_{ijk}$ ：布尔变量，表示工序 $O_{ij}$ 是否在机器 $M_k$ 上加工， $x_{ijk} = 1$ 表示是， $x_{ijk} = 0$ 表示否。
$s_{ij}$ ：工序 $O_{ij}$ 的开始加工时间。
$c_{ij}$ ：工序 $O_{ij}$ 的完成加工时间，显然 $cij=sij+∑k=1mpijkxijkc_{ij} = s_{ij} + \sum_{k=1}^m p_{ijk} x_{ijk}$ 。
$y_{ijkab}$ ：布尔变量，表示工序 $O_{ij}$ 和 $O_{ab}$ 在同一台机器 $M_k$ 上的加工顺序， $y_{ijkab} = 1$ 表示 $O_{ij}$ 在 $O_{ab}$ 之前加工， $y_{ijkab} = 0$ 表示反之。

目标函数：
$\text{minimize} \quad C_{\text{max}} = \max\{c_{in_i} \mid i = 1, 2, \ldots, n\}$

约束条件：

机器分配约束：每个工序只能在一台机器上加工
$\sum_{k \in M_{ij}} x_{ijk} = 1, \quad \forall i, j$
工艺顺序约束：同一作业的工序必须按顺序加工
$s_{ij} \geq c_{i(j-1)}, \quad \forall i, j \geq 2$
加工时间约束：工序的完成时间等于开始时间加上加工时间
$c_{ij} = s_{ij} + \sum_{k=1}^m p_{ijk} x_{ijk}, \quad \forall i, j$
机器能力约束：同一台机器在同一时间只能加工一个工序
$s_{ab} \geq c_{ij} - Q(1 - y_{ijkab}), \quad \forall k, (i,j) \neq (a,b)$
$s_{ij} \geq c_{ab} - Q y_{ijkab}, \quad \forall k, (i,j) \neq (a,b)$
其中 $Q$ 是一个足够大的常数。
最大完成时间约束：所有作业的完成时间不超过 $CmaxC_{\text{max}}$
$c_{in_i} \leq C_{\text{max}}, \quad \forall i$
非负约束
$s_{ij} \geq 0, \quad \forall i, j$

这个数学模型完整地描述了柔性作业车间调度问题，但求解这个问题是NP-hard的，也就是说，随着问题规模的增大，求解时间呈指数增长。因此，我们需要使用启发式算法或元启发式算法来寻找近似最优解。

3.2 多智能体协同决策机制

在多智能体系统中，如何让多个智能体有效地协同决策是一个核心问题。在生产调度场景中，我们可以采用以下几种协同决策机制：

1. 合同网协议（Contract Net Protocol, CNP）

合同网协议是最经典的多智能体协商机制之一，它模拟了市场经济中的招标-投标-中标过程。

在生产调度场景中，合同网协议的工作流程如下：

招标（Announcement）：当产品智能体需要某个工序的加工资源时，它会向所有可用的资源智能体发送招标信息，包括工序要求、时间窗口、评价标准等。
投标（Bidding）：收到招标信息的资源智能体根据自身状态和能力，评估是否能够满足要求，然后提交标书，包括可用时间、加工成本、质量保证等信息。
评标（Awarding）：产品智能体收到所有标书后，根据预设的评价标准（如最短加工时间、最低成本、最高可靠性等）选择最合适的资源智能体。
授标（Acceptance）：产品智能体向中标的资源智能体发送授标信息，双方签订"合同"，明确各自的权利和义务。
履约（Execution）：资源智能体按照合同要求完成加工任务，并向产品智能体报告进展。

合同网协议的优点是简单、灵活、易于实现，能够很好地适应动态环境。但它也有一些缺点，如通信开销大、可能出现局部最优、协商时间长等。

2. 拍卖机制（Auction Mechanism）

拍卖机制是另一种常用的资源分配方法，它通过竞争的方式实现资源的优化配置。在生产调度中，常见的拍卖形式包括：

英式拍卖（English Auction）：从低价开始，逐渐加价，直到只有一个竞标者。
荷兰式拍卖（Dutch Auction）：从高价开始，逐渐降价，直到有竞标者接受。
密封投标拍卖（Sealed-Bid Auction）：所有竞标者同时提交标书，最高/最低标价者中标。
组合拍卖（Combinatorial Auction）：允许竞标者对资源组合进行投标。

在多智能体生产调度中，我们可以让产品智能体竞标资源，或者让资源智能体竞标任务，根据具体场景选择合适的拍卖形式。

3. 博弈论方法（Game Theory）

博弈论为分析多智能体决策问题提供了数学框架。在生产调度中，智能体之间既存在合作关系，也存在竞争关系，这可以用博弈论来建模。

我们可以将生产调度问题建模为一个非合作博弈，每个智能体都是一个玩家，它们的策略是选择加工机器和时间，收益是完成任务的效率和成本。我们的目标是找到一个纳什均衡（Nash Equilibrium），即没有智能体可以通过单方面改变策略来提高自己的收益。

然而，纳什均衡不一定是全局最优的，为了实现系统整体最优，我们可以引入一些机制，如社会福利函数（Social Welfare Function）、**激励机制（Incentive Mechanism）**等，引导智能体做出对全局有利的决策。

4. 强化学习方法（Reinforcement Learning）

强化学习是一种让智能体通过与环境交互来学习最优策略的方法。在多智能体系统中，我们可以使用**多智能体强化学习（Multi-Agent Reinforcement Learning, MARL）**来训练智能体的协同决策能力。

在MARL中，每个智能体都有自己的状态空间、动作空间和奖励函数，但它们的状态和奖励会受到其他智能体的影响。智能体通过不断尝试不同的动作，观察环境的反馈（奖励），逐步学习到最优的策略。

常见的MARL算法包括：

独立Q学习（Independent Q-Learning）：每个智能体独立地学习Q值函数，忽略其他智能体的存在。
联合动作学习者（Joint Action Learners）：智能体考虑其他智能体的动作，学习联合动作的价值。
团队Q学习（Team Q-Learning）：所有智能体共享同一个奖励函数，学习团队最优策略。
深度确定性策略梯度（DDPG）和多智能体DDPG（MADDPG）：使用深度神经网络来近似策略和价值函数，适用于连续动作空间。

在生产调度场景中，我们可以使用MARL来训练智能体，让它们学会如何在动态环境中高效地协同工作。

3.3 多智能体系统的通信机制

智能体之间的有效通信是实现协同的基础。在多智能体系统中，我们需要设计合适的通信机制，包括通信协议、通信语言和通信拓扑等。

1. 通信协议

通信协议定义了智能体之间通信的规则和流程，常见的通信协议包括：

KQML（Knowledge Query and Manipulation Language）：一种基于言语行为理论（Speech Act Theory）的通信语言，支持查询、断言、请求、承诺等多种通信行为。
FIPA-ACL（Foundation for Intelligent Physical Agents - Agent Communication Language）：由FIPA组织制定的标准智能体通信语言，与KQML类似，但更加规范和标准化。
消息队列协议（如MQTT、AMQP）：适用于物联网和分布式系统的轻量级通信协议，支持发布-订阅模式，适合大规模多智能体系统。

在生产调度场景中，我们可以使用KQML或FIPA-ACL来实现智能体之间的高层语义通信，同时使用MQTT或AMQP来实现底层的消息传输。

2. 通信拓扑

通信拓扑定义了智能体之间的连接方式，常见的通信拓扑包括：

完全图（Complete Graph）：每个智能体都可以直接与其他所有智能体通信。优点是信息传播快，缺点是通信开销大，扩展性差。
星型拓扑（Star Topology）：有一个中心智能体，其他智能体都只与中心智能体通信。优点是易于管理，缺点是中心节点是单点故障。
环形拓扑（Ring Topology）：智能体按环形连接，每个智能体只与左右邻居通信。优点是通信负载均衡，缺点是信息传播延迟长。
网状拓扑（Mesh Topology）：智能体之间根据需要建立连接，是一种更灵活的拓扑结构。
分层拓扑（Hierarchical Topology）：智能体按层次组织，高层智能体协调低层智能体，适合大规模系统。

在生产调度场景中，我们通常会采用混合拓扑结构：在局部范围内使用完全图或网状拓扑以提高响应速度，在全局范围内使用分层拓扑以提高扩展性。

3. 基于事件的通信

在动态生产环境中，基于事件的通信是一种有效的方式。智能体可以订阅感兴趣的事件（如设备故障、订单变更、任务完成等），当事件发生时，相关智能体会收到通知并做出响应。

这种方式可以减少不必要的通信开销，提高系统的响应速度。同时，它也符合智能体的反应性特性，使智能体能够及时应对环境变化。

3.4 算法流程图

现在，让我们将前面介绍的概念和机制整合起来，设计一个多智能体协同调度算法的流程：

这个流程图展示了多智能体协同调度系统的主要工作流程：

系统初始化：创建各种智能体实例，进行注册和初始化。
订单处理：接收新订单，创建订单智能体，分解订单为多个产品智能体。
工序调度：对于每个工序，产品智能体通过招标-投标机制选择合适的资源智能体。
执行与监控：安排运输，开始加工，监控加工状态。
异常处理：处理设备故障、订单变更等异常事件。
循环：持续处理新订单和异常事件。

在实际实现中，这个流程可能会更加复杂，需要考虑更多的细节和边界情况。

3.5 算法源代码：Python实现

现在，让我们使用Python来实现一个简化版的多智能体协同调度系统。为了便于理解和演示，我们会做一些简化，但保留核心的概念和机制。

首先，我们需要导入必要的库：

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

接下来，我们定义一些基本的数据类和枚举类型：

class AgentStatus(Enum):
    IDLE = "IDLE"
    BUSY = "BUSY"
    FAULT = "FAULT"

class MessageType(Enum):
    ANNOUNCE = "ANNOUNCE"  # 招标
    BID = "BID"            # 投标
    AWARD = "AWARD"        # 授标
    ACCEPT = "ACCEPT"      # 接受
    REJECT = "REJECT"      # 拒绝
    INFORM = "INFORM"      # 通知
    REQUEST = "REQUEST"    # 请求

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

@dataclass
class Operation:
    job_id: str
    op_id: str
    op_index: int
    available_machines: List[str]
    processing_times: Dict[str, float]  # machine_id -> time
    predecessor: Optional[str] = None  # 前驱工序ID
    successor: Optional[str] = None    # 后继工序ID

@dataclass
class Job:
    job_id: str
    operations: List[Operation]
    arrival_time: float
    due_date: float
    priority: int = 1

@dataclass
class Schedule:
    job_id: str
    op_id: str
    machine_id: str
    start_time: float
    end_time: float

现在，我们定义一个基础的智能体类：

class BaseAgent(ABC):
    def __init__(self, agent_id: str):
        self.agent_id = agent_id
        self.status = AgentStatus.IDLE
        self.message_queue = []  # 消息队列
        self.knowledge_base = {}  # 知识库
        self.outbox = []  # 发件箱
        self.current_time = 0.0
        
    def receive_message(self, message: Message):
        """接收消息"""
        heapq.heappush(self.message_queue, (message.timestamp, message))
    
    def send_message(self, message: Message):
        """发送消息"""
        self.outbox.append(message)
    
    @abstractmethod
    def perceive(self):
        """感知环境"""
        pass
    
    @abstractmethod
    def reason(self):
        """推理决策"""
        pass
    
    @abstractmethod
    def act(self):
        """执行动作"""
        pass
    
    def step(self, current_time: float):
        """执行一步：感知-推理-行动"""
        self.current_time = current_time
        self.perceive()
        self.reason()
        self.act()
        
        # 清空发件箱
        outbox = self.outbox.copy()
        self.outbox.clear()
        return outbox

接下来，我们定义资源智能体类，代表生产设备：

class ResourceAgent(BaseAgent):
    def __init__(self, agent_id: str, machine_type: str):
        super().__init__(agent_id)
        self.machine_type = machine_type
        self.schedule = []  # 已安排的任务
        self.fault_probability = 0.01  # 故障概率
        self.maintenance_time = 0.0  # 维护时间
        self.utilization = 0.0  # 利用率
        
    def perceive(self):
        """感知环境：检查是否有新消息，检查自身状态"""
        # 处理消息队列
        while self.message_queue:
            msg_time, message = heapq.heappop(self.message_queue)
            if msg_time > self.current_time:
                # 消息还没到时间，放回队列
                heapq.heappush(self.message_queue, (msg_time, message))
                break
            
            if message.msg_type == MessageType.ANNOUNCE:
                # 收到招标信息，评估是否投标
                self._evaluate_announcement(message)
            elif message.msg_type == MessageType.AWARD:
                # 收到授标信息，确认是否接受
                self._evaluate_award(message)
        
        # 随机模拟故障
        if self.status == AgentStatus.IDLE and random.random() < self.fault_probability:
            self.status = AgentStatus.FAULT
            self.knowledge_base['fault_time'] = self.current_time
            self.knowledge_base['repair_time'] = self.current_time + random.uniform(10, 30)  # 维修需要10-30时间单位
    
    def _evaluate_announcement(self, message: Message):
        """评估招标信息，决定是否投标"""
        content = message.content
        op_id = content['op_id']
        available_machines = content['available_machines']
        
        # 检查是否可以处理该工序
        if self.agent_id not in available_machines:
            return
        
        # 检查当前状态
        if self.status == AgentStatus.FAULT:
            return
        
        # 计算可用时间窗口
        processing_time = content['processing_times'][self.agent_id]
        earliest_start = self._get_earliest_available_time()
        earliest_finish = earliest_start + processing_time
        
        # 计算投标价格（这里使用完成时间作为主要指标）
        bid_price = earliest_finish
        
        # 发送投标书
        bid_content = {
            'op_id': op_id,
            'machine_id': self.agent_id,
            'earliest_start': earliest_start,
            'earliest_finish': earliest_finish,
            'processing_time': processing_time,
            'bid_price': bid_price,
            'reliability': 0.95  # 可靠性
        }
        
        bid_message = Message(
            sender_id=self.agent_id,
            receiver_id=message.sender_id,
            msg_type=MessageType.BID,
            content=bid_content
        )
        self.send_message(bid_message)
    
    def _evaluate_award(self, message: Message):
        """评估授标信息"""
        content = message.content
        op_id = content['op_id']
        start_time = content['start_time']
        end_time = content['end_time']
        
        # 再次确认是否可以接受
        if self.status == AgentStatus.FAULT:
            # 拒绝
            reject_message = Message(
                sender_id=self.agent_id,
                receiver_id=message.sender_id,
                msg_type=MessageType.REJECT,
                content={'op_id': op_id, 'reason': 'Machine is faulty'}
            )
            self.send_message(reject_message)
            return
        
        # 接受，添加到调度表
        self.schedule.append(Schedule(
            job_id=content['job_id'],
            op_id=op_id,
            machine_id=self.agent_id,
            start_time=start_time,
            end_time=end_time
        ))
        
        # 更新状态
        if start_time <= self.current_time < end_time:
            self.status = AgentStatus.BUSY
        
        # 发送接受消息
        accept_message = Message(
            sender_id=self.agent_id,
            receiver_id=message.sender_id,
            msg_type=MessageType.ACCEPT,
            content={'op_id': op_id}
        )
        self.send_message(accept_message)
    
    def _get_earliest_available_time(self) -> float:
        """获取最早可用时间"""
        if not self.schedule:
            return self.current_time
        
        # 找出最后一个任务的结束时间
        last_end = max(s.end_time for s in self.schedule)
        return max(self.current_time, last_end)
    
    def reason(self):
        """推理决策：更新状态，决定下一步动作"""
        # 检查故障是否修复
        if self.status == AgentStatus.FAULT:
            if self.current_time >= self.knowledge_base.get('repair_time', float('inf')):
                self.status = AgentStatus.IDLE
        
        # 更新忙碌/空闲状态
        if self.status != AgentStatus.FAULT:
            has_active_task = any(
                s.start_time <= self.current_time < s.end_time 
                for s in self.schedule
            )
            self.status = AgentStatus.BUSY if has_active_task else AgentStatus.IDLE
    
    def act(self):
        """执行动作：这里主要是模拟加工过程"""
        # 可以在这里添加实际的设备控制逻辑
        pass
    
    def get_utilization(self, time_window: Tuple[float, float]) -> float:
        """计算利用率"""
        start, end = time_window
        total_time = end - start
        if total_time <= 0:
            return 0.0
        
        busy_time = 0.0
        for s in self.schedule:
            s_start = max(s.start_time, start)
            s_end = min(s.end_time, end)
            if s_start < s_end:
                busy_time += s_end - s_start
        
        return busy_time / total_time