智能主管 Agent：驱动分布式系统效率的奖惩机制设计
各位专家、同仁，大家好。
今天，我将与大家深入探讨一个在分布式系统设计中极具影响力，但常被低估其潜力的模式——“主管模式”（The Supervisor Pattern）。我们不仅将回顾它的基本概念，更将聚焦于一个高级且实用的扩展：如何为我们的主管 Agent（Supervisor Agent）设计并实现一套智能的“奖惩机制”，以期更主动、更精细化地驱动下属工作节点（Worker Node）的执行效率和整体系统性能。
在一个日益复杂的分布式环境中，仅仅依靠故障恢复和任务分配已不足以应对高并发、高弹性、高效率的需求。我们需要一个能够评估、激励、甚至纠正下属行为的“智能主管”。这不仅仅是关于容错，更是关于性能优化和资源调度。
1. 主管模式（The Supervisor Pattern）的核心理念
主管模式，顾名思义，其核心思想是建立一种层级管理结构。在分布式系统中，这意味着一个或多个主管 Agent 负责管理、监控并协调一组下属 Worker Node 的行为。传统的主管模式主要关注以下几点：
- 任务分发 (Task Dispatching)：将传入的工作任务分配给合适的下属节点。
- 状态监控 (Status Monitoring)：持续跟踪下属节点的工作状态（空闲、忙碌、故障等）。
- 故障恢复 (Fault Tolerance)：当下属节点发生故障时，主管能够检测到并采取相应措施，如重启节点、重新分配任务、隔离问题节点等。
- 资源协调 (Resource Coordination)：确保任务和资源得到有效匹配。
然而，这些功能更多地停留在“被动管理”的层面。当某个 Worker Node 表现不佳时，主管可能只是简单地将任务重新分配给其他节点，或者在多次失败后将其标记为不可用。这种方式缺乏对下属节点进行“行为塑造”的能力，无法主动提升整体效率。
我们的目标是超越传统，引入一套“奖惩机制”，将主管 Agent 升级为一个能够主动优化系统性能的“智能决策者”。
2. 构建智能主管 Agent 的基石：奖惩机制的要素
要设计一个具备奖惩机制的主管 Agent，我们首先需要明确“奖”和“惩”的内涵，以及它们所依据的评估标准。
2.1 奖惩的定义与目标
在分布式系统中，奖惩并非传统意义上的物质奖励或行政处罚，而是对 Worker Node 行为的动态调整和资源倾斜。
- 奖励 (Reward)：
- 目标：鼓励高效、稳定、低资源消耗的 Worker Node。
- 形式：
- 优先任务分配：优先将高价值、高优先级的任务分配给表现优异的节点。
- 资源倾斜：在资源竞争时，给予表现优异的节点更高的资源配额或处理能力（在模拟中可以体现为任务执行速度的提升）。
- 更少监控：对稳定节点减少监控频率，节约主管自身资源（高级功能）。
- 特殊任务授权：赋予处理特定复杂或关键任务的权限。
- 惩罚 (Punishment)：
- 目标：纠正低效、不稳定、高错误率或资源滥用的 Worker Node。
- 形式：
- 任务降级/减少：减少分配给表现不佳节点的任务数量或降低其任务优先级。
- 资源限制：降低其可用的资源配额或处理能力。
- 临时暂停 (Pause)：将节点置于暂停状态，不再分配任务，等待其状态恢复或进行人工干预。
- 隔离/停用 (Deactivate)：在长期表现不佳或严重故障后，将其从可用节点池中移除。
- 再训练/重置 (Retrain/Reset)：模拟对其进行内部优化或配置重置。
2.2 性能评估指标 (Performance Metrics)
奖惩机制的有效性，直接取决于我们能否准确、客观地评估 Worker Node 的性能。以下是一些关键的评估指标：
| 指标名称 | 描述 | 理想趋势 | 权重建议 (可调) |
|---|---|---|---|
| 平均任务完成时间 | Worker Node 完成任务的平均耗时。越短表示效率越高。 | 降低 | 中高 |
| 任务错误率 | 完成任务中失败任务所占的比例。越低表示稳定性越好。 | 降低 | 高 |
| 任务吞吐量 | 单位时间内完成的任务数量。越高表示处理能力越强。 | 升高 | 中高 |
| 平均资源利用率 | Worker Node 在执行任务时平均消耗的 CPU、内存等资源。需要结合任务类型判断，过高或过低都可能存在问题。 | 适中 | 中 |
| 任务完成质量 | （如果可衡量）任务输出的准确性、完整性等。 | 升高 | 高 |
| 可用性/在线时长 | Worker Node 在线并可接受任务的时间比例。 | 升高 | 中 |
这些指标需要被持续收集、聚合，并最终转换为一个统一的“性能评分”（Performance Score），作为主管 Agent 决策的依据。
3. 架构设计：智能主管 Agent 系统组成
为了实现上述奖惩机制，我们的系统需要包含以下核心组件：
SupervisorAgent(主管 Agent)：系统的核心，负责任务调度、监控、性能评估和奖惩决策。WorkerNode(下属工作节点)：执行具体任务的实体，需要能够报告自身状态和任务结果，并接受主管的指令。Task(任务)：系统中最基本的工作单元，包含任务描述、优先级、预估耗时等信息。PerformanceMetric(性能指标收集器)：每个 Worker Node 对应一个，用于记录和计算其各项性能数据。DecisionEngine(决策引擎)：内嵌于SupervisorAgent中，根据PerformanceMetric计算的得分，结合预设的策略（阈值），决定对 Worker Node 进行奖惩。
这些组件通过清晰的接口进行交互，形成一个闭环的反馈系统。
3.1 系统交互流程
- 任务提交：外部系统或用户向
SupervisorAgent提交Task。 - 任务入队：
SupervisorAgent将Task加入优先级队列。 - 任务调度：
SupervisorAgent从队列中取出最高优先级的任务，根据 Worker Node 的当前状态和性能评分，选择最优的空闲WorkerNode进行分配。 - 任务执行：
WorkerNode接收任务并开始执行，更新自身状态。 - 性能监控：
SupervisorAgent定期或通过回调机制，从WorkerNode获取任务执行结果（成功/失败、耗时、资源消耗等）。 - 指标更新：
SupervisorAgent将获取的数据更新到对应WorkerNode的PerformanceMetric对象中。 - 性能评估：
PerformanceMetric根据最新数据，计算并更新WorkerNode的综合性能评分。 - 奖惩决策：
SupervisorAgent的DecisionEngine根据性能评分，对照预设的奖惩阈值，决定是否对WorkerNode实施奖励或惩罚。 - 策略应用：
SupervisorAgent向受影响的WorkerNode发送指令，调整其resource_allocation_factor、状态或任务优先级等。 - 反馈循环：
WorkerNode的行为受奖惩影响，进而影响其后续任务执行效率和性能指标，形成持续优化。
4. 详细设计与代码实现
接下来，我们将使用 Python 来模拟实现上述组件，展示其核心逻辑。
4.1 任务（Task）定义
任务是工作单元的抽象，包含唯一标识、描述、优先级、预估耗时、状态等信息。
from enum import Enum
import uuid
import time
from typing import Any
# 任务状态枚举
class TaskStatus(Enum):
PENDING = "PENDING"
IN_PROGRESS = "IN_PROGRESS"
COMPLETED = "COMPLETED"
FAILED = "FAILED"
# 任务优先级枚举
class TaskPriority(Enum):
LOW = 1
MEDIUM = 2
HIGH = 3
CRITICAL = 4
# 任务类
class Task:
def __init__(self, task_id: str, description: str, priority: TaskPriority, estimated_time: float):
self.task_id = task_id
self.description = description
self.priority = priority
self.estimated_time = estimated_time # 预估执行时间 (秒)
self.status = TaskStatus.PENDING
self.assigned_worker_id: str = None
self.start_time: float = None
self.end_time: float = None
self.result: Any = None
self.error: str = None
@staticmethod
def new_task(description: str, priority: TaskPriority, estimated_time: float):
"""创建新任务的工厂方法"""
return Task(str(uuid.uuid4()), description, priority, estimated_time)
def __lt__(self, other):
"""用于优先级队列的比较，优先级数字越小越优先（Critical最高）"""
return self.priority.value > other.priority.value4.2 性能指标收集器（PerformanceMetric）
这个类负责收集并计算单个 Worker Node 的各项性能指标。我们使用 deque 来存储最近 N 次任务的数据，以计算平均值和错误率。
from collections import deque
import time
class PerformanceMetric:
def __init__(self, worker_id: str, history_capacity: int = 50):
self.worker_id = worker_id
self.completion_times = deque(maxlen=history_capacity) # 存储任务完成耗时
self.error_flags = deque(maxlen=history_capacity) # 存储 0 为成功，1 为失败
self.resource_usages = deque(maxlen=history_capacity) # 存储资源使用率 (模拟值)
self.total_completed_tasks = 0
self.total_failed_tasks = 0
self.last_score_update_time = time.time()
def record_completion(self, duration: float, resource_usage: float = 0.5):
"""记录任务成功完成的数据"""
self.completion_times.append(duration)
self.error_flags.append(0) # 成功
self.resource_usages.append(resource_usage)
self.total_completed_tasks += 1
def record_failure(self, duration: float = None, resource_usage: float = 0.5):
"""记录任务失败的数据"""
if duration is not None:
self.completion_times.append(duration) # 即使失败，也可能耗时
else:
self.completion_times.append(0) # 如果没有耗时数据，记录0
self.error_flags.append(1) # 失败
self.resource_usages.append(resource_usage)
self.total_failed_tasks += 1
def get_average_completion_time(self) -> float:
"""计算平均任务完成时间"""
return sum(self.completion_times) / len(self.completion_times) if self.completion_times else 0.0
def get_error_rate(self) -> float:
"""计算错误率"""
return sum(self.error_flags) / len(self.error_flags) if self.error_flags else 0.0
def get_average_resource_usage(self) -> float:
"""计算平均资源使用率"""
return sum(self.resource_usages) / len(self.resource_usages) if self.resource_usages else 0.0
def get_throughput(self) -> int:
"""计算吞吐量（这里简单以历史记录中的完成任务数近似）"""
# 更精确的吞吐量需要时间窗口计算，这里简化为最近完成的任务数
return self.total_completed_tasks - self.total_failed_tasks # 净完成任务数
def get_overall_score(self) -> float:
"""
计算Worker Node的综合性能评分。
这是一个加权评分，需要根据实际业务场景调整权重和计算方式。
分数越高表示性能越好。
"""
avg_time = self.get_average_completion_time()
error_rate = self.get_error_rate()
avg_resource_usage = self.get_average_resource_usage()
throughput = self.get_throughput()
# 初始基础分数
score = 100.0
# 惩罚项：
# 错误率：错误率越高，惩罚越大
score -= error_rate * 80.0 # 错误率对分数影响较大
# 平均完成时间：时间越长，惩罚越大。这里假设理想时间为5秒，超过则惩罚
if avg_time > 0:
time_penalty = max(0, (avg_time - 5.0) * 2.0) # 超过5秒每秒扣2分
score -= time_penalty
# 资源使用率：过高或过低都可能不是最优
# 假设理想资源使用率为0.5-0.7，偏离则惩罚
if avg_resource_usage > 0:
resource_penalty = abs(avg_resource_usage - 0.6) * 30.0 # 偏离0.6越多，惩罚越大
score -= resource_penalty
# 奖励项：
# 吞吐量：净完成任务数越多，奖励越大
score += throughput * 2.0 # 每净完成一个任务奖励2分
# 确保分数在合理范围
return max(0.0, min(100.0, score))4.3 下属工作节点（WorkerNode）
Worker Node 模拟实际执行任务的服务器或进程。它有自己的能力（capabilities），并会受到主管 Agent 奖惩机制的影响。
import threading
import time
import random
from typing import Dict, Any
# Worker状态枚举
class WorkerStatus(Enum):
IDLE = "IDLE"
BUSY = "BUSY"
PAUSED = "PAUSED" # 被主管暂停
DEACTIVATED = "DEACTIVATED" # 被主管永久停用
class WorkerNode:
def __init__(self, worker_id: str, capabilities: Dict[str, Any]):
self.worker_id = worker_id
self.capabilities = capabilities # 如 {'type': 'CPU_intensive', 'speed': 1.0, 'error_rate_mod': 1.0}
self.status = WorkerStatus.IDLE
self.current_task: Task = None
self.task_thread: threading.Thread = None # 模拟异步执行任务
self.performance_score = 100.0 # 初始性能评分，主管会更新
self.resource_allocation_factor = 1.0 # 资源分配因子，用于模拟奖惩效果
self.penalty_count = 0 # 被惩罚次数
self.reward_count = 0 # 被奖励次数
def assign_task(self, task: Task):
"""主管分配任务给Worker"""
if self.status != WorkerStatus.IDLE:
raise Exception(f"Worker {self.worker_id} is not idle. Current status: {self.status.name}")
self.current_task = task
self.status = WorkerStatus.BUSY
self.current_task.status = TaskStatus.IN_PROGRESS
self.current_task.assigned_worker_id = self.worker_id
self.current_task.start_time = time.time()
# 启动一个新线程模拟任务执行
self.task_thread = threading.Thread(target=self._execute_task, args=(task,))
self.task_thread.start()
# print(f"Worker {self.worker_id} started task {task.task_id}")
def _execute_task(self, task: Task):
"""模拟任务执行逻辑，考虑Worker能力、资源因子和随机性"""
# 任务实际执行时间 = 预估时间 / (Worker速度 * 资源分配因子) * 随机波动
base_speed = self.capabilities.get('speed', 1.0)
time_to_complete = task.estimated_time / (base_speed * self.resource_allocation_factor)
time_to_complete *= random.uniform(0.8, 1.2) # 引入 20% 的随机波动
# 模拟潜在的失败
base_error_chance = 0.1 # 默认10%失败率
error_rate_mod = self.capabilities.get('error_rate_mod', 1.0)
actual_error_chance = base_error_chance * error_rate_mod
if random.random() < actual_error_chance: # 有一定几率失败
time.sleep(time_to_complete * random.uniform(0.5, 1.5)) # 失败也可能耗时
task.status = TaskStatus.FAILED
task.error = f"Simulated error by {self.worker_id} during {task.task_id} execution."
# print(f"Worker {self.worker_id} FAILED task {task.task_id}")
else:
time.sleep(time_to_complete)
task.status = TaskStatus.COMPLETED
task.result = f"Task {task.task_id} completed successfully by {self.worker_id}."
# print(f"Worker {self.worker_id} completed task {task.task_id}")
task.end_time = time.time()
# 任务完成后，Worker状态回到IDLE，但主管会通过监控线程来回收任务结果
# 这里只是将Worker的current_task置空，并不会立刻改变Worker状态，
# 因为Supervisor需要先处理这个任务结果并更新metrics。
# _execute_task完成后，WorkerNode会等待Supervisor的下一个调度周期将其状态置回IDLE。
def apply_reward(self, factor: float):
"""应用奖励，增加资源分配因子，上限2.0倍"""
old_factor = self.resource_allocation_factor
self.resource_allocation_factor = min(2.0, self.resource_allocation_factor * (1 + factor))
self.reward_count += 1
print(f" [Reward] Worker {self.worker_id} rewarded. Factor: {old_factor:.2f} -> {self.resource_allocation_factor:.2f}")
def apply_punishment(self, factor: float):
"""应用惩罚，减少资源分配因子，下限0.1倍，并可能暂停"""
old_factor = self.resource_allocation_factor
self.resource_allocation_factor = max(0.1, self.resource_allocation_factor * (1 - factor))
self.penalty_count += 1
print(f" [Punish] Worker {self.worker_id} punished. Factor: {old_factor:.2f} -> {self.resource_allocation_factor:.2f}")
if self.penalty_count >= 3 and self.resource_allocation_factor < 0.5:
# 连续多次惩罚或性能过低，考虑暂停
if self.status != WorkerStatus.PAUSED:
self.status = WorkerStatus.PAUSED
print(f" [Action] Worker {self.worker_id} paused due to repeated poor performance.")
def get_status(self) -> WorkerStatus:
"""获取当前Worker状态"""
return self.status
def is_idle(self) -> bool:
"""检查Worker是否空闲并可接受任务"""
return self.status == WorkerStatus.IDLE and self.current_task is None
def reactivate(self):
"""重新激活被暂停的Worker"""
if self.status == WorkerStatus.PAUSED:
self.status = WorkerStatus.IDLE
self.penalty_count = 0 # 重置惩罚计数
print(f" [Action] Worker {self.worker_id} reactivated.")4.4 主管 Agent（SupervisorAgent）
这是系统的核心，协调所有 Worker Node，进行任务调度、性能监控和奖惩决策。
import queue
import threading
import time
from typing import List, Dict, Tuple
import random
class SupervisorAgent:
def __init__(self, name: str, reward_threshold: float = 85.0, punishment_threshold: float = 50.0,
reward_factor: float = 0.1, punishment_factor: float = 0.1):
self.name = name
self.workers: Dict[str, WorkerNode] = {}
# 优先级队列，存储 (优先级值, 任务对象)。优先级值越小，优先级越高。
# 由于TaskPriority枚举值是1-4，CRITICAL是4，所以用负数来让高优先级任务排在前面。
self.task_queue = queue.PriorityQueue()
self.completed_tasks: List[Task] = [] # 记录已完成/失败的任务
self.performance_metrics: Dict[str, PerformanceMetric] = {}
self.monitoring_interval = 3 # 监控循环间隔 (秒)
self.reward_threshold = reward_threshold # 达到此分数以上奖励
self.punishment_threshold = punishment_threshold # 低于此分数惩罚
self.reward_factor = reward_factor # 奖励时增加的资源因子比例
self.punishment_factor = punishment_factor # 惩罚时减少的资源因子比例
self._running = False
self._monitor_thread = None
self._dispatcher_thread = None
print(f"Supervisor '{self.name}' initialized.")
def add_worker(self, worker: WorkerNode):
"""添加Worker Node到主管的管理范围"""
self.workers[worker.worker_id] = worker
self.performance_metrics[worker.worker_id] = PerformanceMetric(worker.worker_id)
print(f"Worker '{worker.worker_id}' added to supervisor.")
def submit_task(self, task: Task):
"""提交任务到主管的任务队列"""
self.task_queue.put((-task.priority.value, task)) # 优先级高的（数值小）任务优先
print(f"Task '{task.task_id[:8]}' submitted with priority {task.priority.name}.")
def start(self):
"""启动主管的监控和调度循环"""
self._running = True
self._monitor_thread = threading.Thread(target=self._monitoring_loop)
self._dispatcher_thread = threading.Thread(target=self._dispatch_loop)
self._monitor_thread.start()
self._dispatcher_thread.start()
print(f"Supervisor '{self.name}' started monitoring and dispatching.")
def stop(self):
"""停止主管的所有活动"""
self._running = False
if self._monitor_thread:
self._monitor_thread.join()
if self._dispatcher_thread:
self._dispatcher_thread.join()
print(f"Supervisor '{self.name}' stopped.")
def _dispatch_loop(self):
"""任务调度循环：从队列取任务，分配给最佳Worker"""
while self._running:
try:
# 尝试获取任务，如果队列为空则等待1秒
priority_value, task = self.task_queue.get(timeout=1)
# 查找空闲且未暂停的Worker
available_workers = [
w for w in self.workers.values()
if w.is_idle() and w.status != WorkerStatus.PAUSED and w.status != WorkerStatus.DEACTIVATED
]
if not available_workers:
# 如果没有可用Worker，将任务放回队列并等待
# print(f"No available workers for task {task.task_id[:8]}. Re-queueing.")
self.task_queue.put((priority_value, task))
time.sleep(0.5)
continue
# 调度策略：选择性能评分最高的Worker
# 更复杂的策略可以考虑Worker能力、任务类型匹配等
best_worker = max(available_workers, key=lambda w: w.performance_score)
if best_worker:
# print(f"Dispatching task '{task.task_id[:8]}' to worker '{best_worker.worker_id}' (Score: {best_worker.performance_score:.2f}).")
best_worker.assign_task(task)
else:
# 理论上不会发生，因为available_workers非空
print(f"No suitable worker found for task {task.task_id[:8]}. Re-queueing.")
self.task_queue.put((priority_value, task))
time.sleep(0.5)
except queue.Empty:
# 队列为空，短暂休眠
time.sleep(0.1)
except Exception as e:
print(f"Error in dispatch loop: {e}")
time.sleep(1)
def _monitoring_loop(self):
"""监控循环：检查Worker任务状态，更新性能指标，执行奖惩决策"""
while self._running:
time.sleep(self.monitoring_interval)
print(f"n--- Monitoring Cycle (Time: {time.time():.2f}, Task Queue: {self.task_queue.qsize()}) ---")
for worker_id, worker in self.workers.items():
# 1. 检查Worker的当前任务是否完成或失败
if worker.current_task and
worker.current_task.status in [TaskStatus.COMPLETED, TaskStatus.FAILED]:
task = worker.current_task
duration = task.end_time - task.start_time if task.end_time and task.start_time else 0.0
metrics = self.performance_metrics[worker_id]
resource_usage = random.uniform(0.3, 0.8) # 模拟资源使用
if task.status == TaskStatus.COMPLETED:
metrics.record_completion(duration, resource_usage)
print(f"Worker {worker_id} completed task {task.task_id[:8]} in {duration:.2f}s.")
else: # TaskStatus.FAILED
metrics.record_failure(duration, resource_usage)
print(f"Worker {worker_id} FAILED task {task.task_id[:8]} in {duration:.2f}s. Error: {task.error}")
self.completed_tasks.append(task) # 记录已处理任务
worker.current_task = None # 清除Worker上的任务引用
# 关键：Worker状态回到IDLE，以便重新调度
if worker.status == WorkerStatus.BUSY:
worker.status = WorkerStatus.IDLE
# 2. 更新性能评分并应用奖惩
metrics = self.performance_metrics[worker_id]
current_score = metrics.get_overall_score()
worker.performance_score = current_score # 更新Worker自身的评分，供调度器使用
print(f"Worker {worker_id} status: {worker.status.name}, Score: {current_score:.2f}, Factor: {worker.resource_allocation_factor:.2f}, "
f"Completed: {metrics.total_completed_tasks}, Failed: {metrics.total_failed_tasks}, "
f"AvgTime: {metrics.get_average_completion_time():.2f}, ErrorRate: {metrics.get_error_rate():.2f}")
if worker.status not in [WorkerStatus.PAUSED, WorkerStatus.DEACTIVATED]:
if current_score >= self.reward_threshold:
worker.apply_reward(self.reward_factor)
elif current_score <= self.punishment_threshold:
worker.apply_punishment(self.punishment_factor)
# 3. 处理被暂停的Worker：根据情况进行再激活
if worker.status == WorkerStatus.PAUSED:
# 如果暂停的Worker性能显著提升，并且任务队列有积压，考虑重新激活
if current_score > (self.punishment_threshold + 20) and self.task_queue.qsize() > 0:
worker.reactivate()
print(f" [Reactivate] Worker {worker_id} reactivated due to improved score and task demand.")
print("--- End Monitoring Cycle ---")
def get_worker_status(self) -> Dict[str, Tuple[WorkerStatus, float, float]]:
"""获取所有Worker的当前状态、性能评分和资源因子"""
return {
worker_id: (worker.status, worker.performance_score, worker.resource_allocation_factor)
for worker_id, worker in self.workers.items()
}
def get_task_queue_size(self) -> int:
"""获取任务队列中待处理任务的数量"""
return self.task_queue.qsize()
5. 模拟与示范
现在，让我们通过一个模拟场景来观察智能主管 Agent 如何运作。
# Main simulation block
if __name__ == "__main__":
print("----------------------------------------------------------------------")
print("Starting Supervisor Pattern Demonstration with Reward/Punishment Mechanism")
print("----------------------------------------------------------------------")
# 1. 初始化主管 Agent
# 设置奖励阈值85分，惩罚阈值50分，奖励因子15%，惩罚因子20%
supervisor = SupervisorAgent("MainSupervisor",
reward_threshold=85.0,
punishment_threshold=50.0,
reward_factor=0.15,
punishment_factor=0.2)
# 2. 添加 Worker Node，模拟不同性能的下属
# Worker-A: 普通 Worker
worker1 = WorkerNode("Worker-A", {'type': 'General', 'speed': 1.0, 'error_rate_mod': 1.0})
# Worker-B: 快速 Worker，初始速度快
worker2 = WorkerNode("Worker-B", {'type': 'Fast', 'speed': 1.5, 'error_rate_mod': 0.8}) # 错误率略低
# Worker-C: 慢速 Worker，初始速度慢
worker3 = WorkerNode("Worker-C", {'type': 'Slow', 'speed': 0.7, 'error_rate_mod': 1.2}) # 错误率略高
# Worker-D: 稳定 Worker，速度一般但错误率很低
worker4 = WorkerNode("Worker-D", {'type': 'Reliable', 'speed': 1.1, 'error_rate_mod': 0.3})
supervisor.add_worker(worker1)
supervisor.add_worker(worker2)
supervisor.add_worker(worker3)
supervisor.add_worker(worker4)
# 3. 启动主管 Agent
supervisor.start()
# 4. 提交一批任务，观察初始阶段行为
print("n--- Submitting initial tasks (Batch 1) ---")
for i in range(1, 11):
# 随机分配优先级和预估时间
priority = random.choice([TaskPriority.LOW, TaskPriority.MEDIUM, TaskPriority.HIGH])
estimated_time = random.uniform(2, 6) # 任务耗时2-6秒
task = Task.new_task(f"Data Processing {i}", priority, estimated_time)
supervisor.submit_task(task)
time.sleep(0.3) # 模拟任务陆续到来
print("n--- Running simulation for 20 seconds (observing initial performance) ---")
time.sleep(20)
# 5. 提交第二批任务，观察奖惩机制的效果
print("n--- Submitting more tasks (Batch 2) to observe long-term effects ---")
for i in range(11, 26):
priority = random.choice([TaskPriority.MEDIUM, TaskPriority.HIGH, TaskPriority.CRITICAL])
estimated_time = random.uniform(3, 8) # 任务耗时3-8秒
task = Task.new_task(f"Analytics Report {i}", priority, estimated_time)
supervisor.submit_task(task)
time.sleep(0.4)
print("n--- Running simulation for another 30 seconds ---")
time.sleep(30)
# 6. 提交一些紧急任务，看高分Worker是否优先处理
print("n--- Submitting Critical Tasks ---")
for i in range(1, 4):
task = Task.new_task(f"CRITICAL Alert {i}", TaskPriority.CRITICAL, random.uniform(1, 3))
supervisor.submit_task(task)
time.sleep(0.1)
print("n--- Running simulation for final 15 seconds ---")
time.sleep(15)
# 7. 停止主管 Agent
supervisor.stop()
print("n----------------------------------------------------------------------")
print("--- Final System Status ---")
print("----------------------------------------------------------------------")
for worker_id, (status, score, factor) in supervisor.get_worker_status().items():
worker_obj = supervisor.workers[worker_id]
metrics = supervisor.performance_metrics[worker_id]
print(f"Worker {worker_id}:")
print(f" Status: {status.name}")
print(f" Final Score: {score:.2f}")
print(f" Resource Factor: {factor:.2f}")
print(f" Tasks Completed: {metrics.total_completed_tasks}")
print(f" Tasks Failed: {metrics.total_failed_tasks}")
print(f" Avg Completion Time (last {metrics.completion_times.maxlen} tasks): {metrics.get_average_completion_time():.2f}s")
print(f" Error Rate (last {metrics.error_flags.maxlen} tasks): {metrics.get_error_rate():.2f}")
print(f" Reward/Penalty Count: {worker_obj.reward_count} / {worker_obj.penalty_count}")
print("-" * 30)
print(f"nTotal tasks processed by system: {len(supervisor.completed_tasks)}")
print(f"Tasks remaining in queue: {supervisor.get_task_queue_size()}")
print("----------------------------------------------------------------------")
print("Simulation Finished.")
print("----------------------------------------------------------------------")在运行上述代码时，您会观察到：
- 初始阶段：任务会根据优先级和Worker的初始性能随机分配。
- 中期演进：
Worker-B(Fast) 和Worker-D(Reliable) 由于其固有的优势，会积累更高的性能分数，并获得奖励，其resource_allocation_factor会增加，导致它们执行任务更快。主管也会更频繁地将任务分配给它们。Worker-C(Slow) 由于速度慢和错误率高，性能分数会逐渐下降，并受到惩罚，其resource_allocation_factor会降低，导致执行更慢。如果分数持续过低，甚至可能被主管暂停。Worker-A(General) 的表现可能居中，根据任务负载和随机性，其分数会波动。
- 后期影响：被奖励的 Worker 可能会在任务调度中获得更高的优先级，即使有其他 Worker 空闲，主管也可能优先选择高分 Worker。被惩罚甚至暂停的 Worker 则会减少任务分配，从而减少对整体系统性能的负面影响。当任务队列积压严重时，主管可能会重新激活表现有所改善的暂停 Worker。
6. 进阶考量与实际应用
上述实现是一个基础框架，在实际生产环境中，还需要考虑更多高级特性：
- 动态阈值调整：奖惩阈值不应固定不变，可以根据系统整体负载、资源利用率或SLA（服务等级协议）要求进行动态调整。例如，在高峰期，可以适当放宽惩罚阈值以确保任务完成率。
- 强化学习决策引擎：将
DecisionEngine替换为基于强化学习（Reinforcement Learning, RL）的 Agent。RL Agent 可以通过与环境（Worker Node和任务流）的交互，自主学习最优的奖惩策略，以最大化长期系统效率或最小化成本。这比硬编码的阈值和因子更加灵活和强大。 - 资源抽象与管理：在真实场景中，
resource_allocation_factor会映射到实际的 CPU 核数、内存配额、网络带宽、GPU 算力等。主管 Agent 需要与底层的资源调度系统（如 Kubernetes、YARN 或云服务商的 API）集成。 - 异构 Worker：分布式系统中可能存在不同类型、不同成本、不同能力的 Worker Node。主管 Agent 需要能够识别这些差异，并根据任务需求和 Worker 特性进行更智能的匹配（例如，将 GPU 密集型任务分配给带有 GPU 的 Worker）。
- 主管自身的健壮性：主管 Agent 自身也需要高可用性，可以采用主备模式、集群模式或去中心化机制来避免单点故障。
- 安全性与防作弊：防止 Worker Node 伪造性能数据或通过特定行为“欺骗”主管以获得奖励。这需要更严格的数据校验和更复杂的行为模式分析。
- 可观测性 (Observability)：提供丰富的日志、指标和可视化界面，方便运维人员理解系统运行状态、Worker 性能变化以及主管的决策逻辑。
7. 驱动效率的智能飞轮
通过引入奖惩机制，我们将“主管模式”从一个被动的故障恢复和任务分配器，提升为一个主动的效率优化引擎。这个智能主管 Agent 能够持续监控下属 Worker 的表现，通过动态调整资源分配和任务优先级，鼓励优秀表现，纠正不良行为。这形成了一个正向的反馈循环：高效的 Worker 获得更多资源和任务，进一步提升整体吞吐量；低效的 Worker 被限制或暂停，减少了对系统的负面影响。这种“智能飞轮”效应，将显著提升分布式系统的弹性、效率和资源利用率。