任务分配算法概述

任务分配的重要性

在多智能体系统中，任务分配是一个核心问题。有效的任务分配能够：

• 最大化系统整体性能
• 平衡各Agent的工作负载
• 最小化任务完成时间
• 优化资源利用率

任务分配的挑战

• 动态环境下的实时分配
• Agent能力和状态的异构性
• 通信开销的控制
• 任务优先级和约束的处理

负载均衡算法

算法原理

负载均衡算法的目标是将任务均匀分配给各个Agent，避免某些Agent过载而其他Agent空闲。

负载均衡算法实现

import heapq
import random
from typing import List, Dict, Any, Tuple
from datetime import datetime
import time
class LoadBalancer:
"""负载均衡器"""
def __init__(self, strategy: str = 'round_robin'):
"""
初始化负载均衡器
Args:
strategy: 负载均衡策略 ('round_robin', 'least_connections', 'weighted_round_robin', 'random')
"""
self.strategy = strategy
self.agents = []
self.agent_stats = {}
self.task_history = []
self.round_robin_index = 0
def register_agent(self, agent_id: str, capacity: int = 10, weight: int = 1):
"""注册Agent"""
self.agents.append({
'id': agent_id,
'capacity': capacity,
'weight': weight,
'current_load': 0,
'total_processed': 0,
'last_activity': time.time()
})
self.agent_stats[agent_id] = {
'requests_handled': 0,
'avg_response_time': 0,
'success_rate': 1.0
}
print(f"注册Agent: {agent_id}, 容量: {capacity}, 权重: {weight}")
def get_available_agents(self) -> List[Dict[str, Any]]:
"""获取可用Agent列表"""
available = []
for agent in self.agents:
if agent['current_load'] < agent['capacity']:
available.append(agent)
return available
def assign_task(self, task: Dict[str, Any]) -> str:
"""分配任务"""
available_agents = self.get_available_agents()
if not available_agents:
raise Exception("没有可用的Agent来处理任务")
if self.strategy == 'round_robin':
assigned_agent = self._round_robin_selection(available_agents)
elif self.strategy == 'least_connections':
assigned_agent = self._least_connections_selection(available_agents)
elif self.strategy == 'weighted_round_robin':
assigned_agent = self._weighted_round_robin_selection(available_agents)
elif self.strategy == 'random':
assigned_agent = self._random_selection(available_agents)
else:
assigned_agent = available_agents[0]  # 默认
# 更新Agent负载
for agent in self.agents:
if agent['id'] == assigned_agent['id']:
agent['current_load'] += 1
agent['total_processed'] += 1
agent['last_activity'] = time.time()
break
# 更新统计信息
self.agent_stats[assigned_agent['id']]['requests_handled'] += 1
assignment_record = {
'task_id': task['id'],
'assigned_to': assigned_agent['id'],
'strategy': self.strategy,
'timestamp': time.time(),
'agent_load_before': assigned_agent['current_load'] - 1,
'agent_load_after': assigned_agent['current_load']
}
self.task_history.append(assignment_record)
print(f"任务 {task['id']} 分配给 Agent {assigned_agent['id']} (负载: {assigned_agent['current_load']}/{assigned_agent['capacity']})")
return assigned_agent['id']
def _round_robin_selection(self, available_agents: List[Dict[str, Any]]) -> Dict[str, Any]:
"""轮询选择"""
if not available_agents:
return None
# 简单轮询
selected = available_agents[self.round_robin_index % len(available_agents)]
self.round_robin_index += 1
return selected
def _least_connections_selection(self, available_agents: List[Dict[str, Any]]) -> Dict[str, Any]:
"""最少连接选择"""
if not available_agents:
return None
# 选择当前负载最小的Agent
return min(available_agents, key=lambda x: x['current_load'])
def _weighted_round_robin_selection(self, available_agents: List[Dict[str, Any]]) -> Dict[str, Any]:
"""加权轮询选择"""
if not available_agents:
return None
# 计算总权重
total_weight = sum(agent['weight'] for agent in available_agents)
# 计算累积权重并选择
current_weight = 0
threshold = random.randint(0, total_weight - 1)
for agent in available_agents:
current_weight += agent['weight']
if current_weight > threshold:
return agent
# 如果没有找到，返回第一个
return available_agents[0]
def _random_selection(self, available_agents: List[Dict[str, Any]]) -> Dict[str, Any]:
"""随机选择"""
if not available_agents:
return None
return random.choice(available_agents)
def complete_task(self, task_id: str, agent_id: str, success: bool = True):
"""标记任务完成"""
# 更新Agent负载
for agent in self.agents:
if agent['id'] == agent_id:
agent['current_load'] = max(0, agent['current_load'] - 1)
break
# 更新统计信息
if agent_id in self.agent_stats:
stats = self.agent_stats[agent_id]
total_requests = stats['requests_handled']
successful_requests = stats['requests_handled'] if success else stats['requests_handled'] - 1
stats['success_rate'] = successful_requests / max(1, total_requests)
def get_load_distribution(self) -> Dict[str, Any]:
"""获取负载分布信息"""
total_capacity = sum(agent['capacity'] for agent in self.agents)
total_load = sum(agent['current_load'] for agent in self.agents)
utilization = total_load / max(1, total_capacity)
agent_loads = [(agent['id'], agent['current_load'], agent['capacity']) for agent in self.agents]
return {
'total_utilization': utilization,
'agent_loads': agent_loads,
'total_agents': len(self.agents),
'busy_agents': len([a for a in self.agents if a['current_load'] > 0])
}
# 演示负载均衡算法
def demo_load_balancer():
print("=== 负载均衡算法演示 ===\n")
# 测试不同策略
strategies = ['round_robin', 'least_connections', 'weighted_round_robin', 'random']
for strategy in strategies:
print(f"--- 策略: {strategy} ---")
lb = LoadBalancer(strategy=strategy)
# 注册不同容量和权重的Agent
lb.register_agent('Agent_High_Capacity', capacity=15, weight=3)
lb.register_agent('Agent_Medium_Capacity', capacity=10, weight=2)
lb.register_agent('Agent_Low_Capacity', capacity=5, weight=1)
# 生成任务并分配
for i in range(10):
task = {'id': f'task_{strategy}_{i}', 'data': f'data_{i}'}
try:
assigned_agent = lb.assign_task(task)
except Exception as e:
print(f"任务分配失败: {e}")
break
# 显示负载分布
distribution = lb.get_load_distribution()
print(f"总体利用率: {distribution['total_utilization']:.2%}")
for agent_id, load, capacity in distribution['agent_loads']:
print(f"  {agent_id}: {load}/{capacity} ({load/capacity:.2%})")
print()

基于能力的任务分配

算法原理

基于能力的分配考虑Agent的专业技能和任务需求的匹配程度。

基于能力的分配实现

import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
class CapabilityBasedTaskAllocator:
"""基于能力的任务分配器"""
def __init__(self):
self.agents = {}  # agent_id -> agent_info
self.tasks = {}   # task_id -> task_info
self.compatibility_matrix = None
self.assignment_history = []
def register_agent(self, agent_id: str, capabilities: Dict[str, float],
performance_history: List[Dict[str, Any]] = None):
"""
注册Agent
Args:
agent_id: Agent ID
capabilities: 能力字典，格式为 {capability: proficiency_score}
performance_history: 性能历史记录
"""
self.agents[agent_id] = {
'capabilities': capabilities,
'performance_history': performance_history or [],
'current_tasks': [],
'total_completed': 0,
'avg_quality': 0.0
}
print(f"注册Agent: {agent_id}, 能力: {list(capabilities.keys())}")
def register_task(self, task_id: str, requirements: Dict[str, float], priority: int = 1):
"""
注册任务
Args:
task_id: 任务ID
requirements: 任务要求，格式为 {capability: required_level}
priority: 任务优先级
"""
self.tasks[task_id] = {
'requirements': requirements,
'priority': priority,
'assigned_to': None,
'status': 'pending'
}
print(f"注册任务: {task_id}, 要求: {list(requirements.keys())}, 优先级: {priority}")
def calculate_compatibility(self, agent_capabilities: Dict[str, float],
task_requirements: Dict[str, float]) -> float:
"""
计算Agent能力与任务要求的兼容性分数
Returns:
兼容性分数 (0-1)
"""
if not task_requirements:
return 1.0  # 如果没有特定要求，默认完全兼容
# 计算加权兼容性分数
total_weight = 0
matched_weight = 0
for capability, required_level in task_requirements.items():
agent_level = agent_capabilities.get(capability, 0)
weight = required_level  # 使用要求级别作为权重
# 计算匹配度 (如果Agent能力>=要求，则完全匹配，否则按比例)
match_score = min(1.0, agent_level / max(required_level, 0.001))
matched_weight += weight * match_score
total_weight += weight
if total_weight == 0:
return 1.0  # 如果没有要求，认为完全兼容
return matched_weight / total_weight
def calculate_performance_score(self, agent_id: str) -> float:
"""计算Agent的历史性能分数"""
agent = self.agents[agent_id]
if not agent['performance_history']:
return 0.8  # 默认性能分数
# 计算加权平均性能（最近的性能更重要）
total_score = 0
total_weight = 0
decay_factor = 0.9  # 衰减因子
for i, perf_record in enumerate(agent['performance_history']):
weight = decay_factor ** i
score = perf_record.get('quality', 0.8)  # 默认质量分数
total_score += score * weight
total_weight += weight
return total_score / total_weight if total_weight > 0 else 0.8
def assign_single_task(self, task_id: str, selection_strategy: str = 'compatibility_only') -> str:
"""
分配单个任务
Args:
task_id: 任务ID
selection_strategy: 选择策略 ('compatibility_only', 'performance_weighted', 'hybrid')
Returns:
分配的Agent ID
"""
if task_id not in self.tasks:
raise ValueError(f"任务不存在: {task_id}")
task = self.tasks[task_id]
if task['status'] != 'pending':
raise ValueError(f"任务状态不是待分配: {task['status']}")
# 找到所有兼容的Agent
compatible_agents = []
for agent_id, agent in self.agents.items():
compatibility = self.calculate_compatibility(
agent['capabilities'],
task['requirements']
)
if compatibility > 0:  # 至少有部分兼容性
performance_score = self.calculate_performance_score(agent_id)
# 根据策略计算最终得分
if selection_strategy == 'compatibility_only':
final_score = compatibility
elif selection_strategy == 'performance_weighted':
final_score = compatibility * 0.7 + performance_score * 0.3
else:  # hybrid
current_load = len(agent['current_tasks'])
max_load = 10  # 假设最大负载为10
load_factor = 1 - (current_load / max_load)  # 负载越低，分数越高
final_score = (compatibility * 0.5 + performance_score * 0.3 + load_factor * 0.2)
compatible_agents.append((agent_id, final_score))
if not compatible_agents:
print(f"没有Agent能够处理任务: {task_id}")
return None
# 选择得分最高的Agent
compatible_agents.sort(key=lambda x: x[1], reverse=True)
selected_agent_id = compatible_agents[0][0]
# 执行分配
self.agents[selected_agent_id]['current_tasks'].append(task_id)
task['assigned_to'] = selected_agent_id
task['status'] = 'assigned'
assignment_record = {
'task_id': task_id,
'assigned_to': selected_agent_id,
'compatibility_score': self.calculate_compatibility(
self.agents[selected_agent_id]['capabilities'],
task['requirements']
),
'performance_score': self.calculate_performance_score(selected_agent_id),
'strategy': selection_strategy,
'timestamp': time.time()
}
self.assignment_history.append(assignment_record)
print(f"任务 {task_id} 分配给 {selected_agent_id} (兼容性: {compatible_agents[0][1]:.2f})")
return selected_agent_id
def batch_assign_tasks(self, selection_strategy: str = 'hybrid') -> Dict[str, str]:
"""批量分配所有待分配任务"""
assignments = {}
# 按优先级排序任务
pending_tasks = [
(task_id, task['priority'])
for task_id, task in self.tasks.items()
if task['status'] == 'pending'
]
pending_tasks.sort(key=lambda x: x[1], reverse=True)  # 高优先级优先
for task_id, priority in pending_tasks:
assigned_agent = self.assign_single_task(task_id, selection_strategy)
if assigned_agent:
assignments[task_id] = assigned_agent
return assignments
def complete_task(self, task_id: str, quality: float = 0.9, success: bool = True):
"""完成任务"""
if task_id not in self.tasks:
raise ValueError(f"任务不存在: {task_id}")
task = self.tasks[task_id]
if not task['assigned_to']:
raise ValueError(f"任务未分配: {task_id}")
agent_id = task['assigned_to']
agent = self.agents[agent_id]
# 移除任务
if task_id in agent['current_tasks']:
agent['current_tasks'].remove(task_id)
# 更新性能历史
agent['performance_history'].append({
'task_id': task_id,
'quality': quality,
'success': success,
'timestamp': time.time()
})
# 更新任务状态
task['status'] = 'completed' if success else 'failed'
print(f"任务 {task_id} 由 {agent_id} 完成 (质量: {quality:.2f})")
# 演示基于能力的分配
def demo_capability_based_allocation():
print("=== 基于能力的任务分配演示 ===\n")
allocator = CapabilityBasedTaskAllocator()
# 注册Agent
allocator.register_agent(
'DataScientist_001',
{
'machine_learning': 0.9,
'data_analysis': 0.8,
'statistical_modeling': 0.9,
'python': 0.8
}
)
allocator.register_agent(
'WebDeveloper_001',
{
'web_development': 0.9,
'javascript': 0.8,
'react': 0.7,
'api_integration': 0.8
}
)
allocator.register_agent(
'DevOpsEngineer_001',
{
'cloud_deployment': 0.9,
'docker': 0.8,
'kubernetes': 0.7,
'ci_cd': 0.9
}
)
# 注册任务
allocator.register_task(
'task_ml_model',
{
'machine_learning': 0.8,
'data_analysis': 0.7,
'python': 0.6
},
priority=2
)
allocator.register_task(
'task_web_api',
{
'web_development': 0.7,
'javascript': 0.8,
'api_integration': 0.6
},
priority=1
)
allocator.register_task(
'task_deploy_system',
{
'cloud_deployment': 0.8,
'docker': 0.7,
'ci_cd': 0.8
},
priority=3
)
# 测试不同分配策略
strategies = ['compatibility_only', 'performance_weighted', 'hybrid']
for strategy in strategies:
print(f"\n--- 策略: {strategy} ---")
# 重新初始化任务状态
for task_id in allocator.tasks:
allocator.tasks[task_id]['status'] = 'pending'
allocator.tasks[task_id]['assigned_to'] = None
# 清空Agent当前任务
for agent_id in allocator.agents:
allocator.agents[agent_id]['current_tasks'] = []
# 批量分配
assignments = allocator.batch_assign_tasks(strategy)
print("分配结果:")
for task_id, agent_id in assignments.items():
task = allocator.tasks[task_id]
agent = allocator.agents[agent_id]
compatibility = allocator.calculate_compatibility(
agent['capabilities'],
task['requirements']
)
print(f"  {task_id} -> {agent_id} (兼容性: {compatibility:.2f})")

基于效用的任务分配

算法原理

基于效用的分配考虑分配的总体效益，通常使用拍卖机制或优化算法来最大化系统效用。

基于效用的分配实现

from scipy.optimize import linear_sum_assignment
class UtilityBasedTaskAllocator:
"""基于效用的任务分配器"""
def __init__(self):
self.agents = {}
self.tasks = {}
self.utility_matrix = None
self.assignments = {}
def register_agent(self, agent_id: str, capabilities: Dict[str, float],
cost_structure: Dict[str, float] = None):
"""注册Agent"""
self.agents[agent_id] = {
'capabilities': capabilities,
'cost_structure': cost_structure or {'base_cost': 10, 'per_unit_cost': 1},
'current_utility': 0,
'max_tasks': 5  # 每个Agent最多处理5个任务
}
def register_task(self, task_id: str, requirements: Dict[str, float],
value: float = 1.0, complexity: float = 1.0):
"""注册任务"""
self.tasks[task_id] = {
'requirements': requirements,
'value': value,  # 任务价值
'complexity': complexity,  # 任务复杂度
'status': 'unassigned'
}
def calculate_utility(self, agent_id: str, task_id: str) -> float:
"""
计算Agent执行任务的效用
Returns:
效用值（可以为负值）
"""
agent = self.agents[agent_id]
task = self.tasks[task_id]
# 计算兼容性得分
compatibility_score = 0
total_weight = 0
for req_cap, req_level in task['requirements'].items():
agent_level = agent['capabilities'].get(req_cap, 0)
weight = req_level
compatibility_score += min(1.0, agent_level / max(req_level, 0.001)) * weight
total_weight += weight
if total_weight > 0:
compatibility_score /= total_weight
else:
compatibility_score = 1.0
# 计算成本
base_cost = agent['cost_structure']['base_cost']
per_unit_cost = agent['cost_structure']['per_unit_cost']
cost = base_cost + task['complexity'] * per_unit_cost
# 计算净效用
net_utility = task['value'] * compatibility_score - cost
return net_utility
def build_utility_matrix(self) -> np.ndarray:
"""构建效用矩阵"""
agent_ids = list(self.agents.keys())
task_ids = list(self.tasks.keys())
matrix = np.zeros((len(agent_ids), len(task_ids)))
for i, agent_id in enumerate(agent_ids):
for j, task_id in enumerate(task_ids):
matrix[i][j] = self.calculate_utility(agent_id, task_id)
self.utility_matrix = matrix
return matrix
def optimize_assignment(self) -> Dict[str, str]:
"""使用匈牙利算法优化分配"""
if not self.agents or not self.tasks:
return {}
agent_ids = list(self.agents.keys())
task_ids = list(self.tasks.keys())
# 构建效用矩阵
utility_matrix = self.build_utility_matrix()
# 使用匈牙利算法求解最大权重匹配
# 注意：linear_sum_assignment是最小化，所以我们需要取负值
row_indices, col_indices = linear_sum_assignment(-utility_matrix)
assignments = {}
for agent_idx, task_idx in zip(row_indices, col_indices):
agent_id = agent_ids[agent_idx]
task_id = task_ids[task_idx]
# 检查效用是否为正（负效用不应该分配）
if utility_matrix[agent_idx][task_idx] > 0:
assignments[task_id] = agent_id
self.agents[agent_id]['current_utility'] += utility_matrix[agent_idx][task_idx]
self.tasks[task_id]['status'] = 'assigned'
self.tasks[task_id]['assigned_to'] = agent_id
self.assignments = assignments
return assignments
def get_total_utility(self) -> float:
"""获取总效用"""
total = 0
for task_id, agent_id in self.assignments.items():
total += self.calculate_utility(agent_id, task_id)
return total
def print_allocation_summary(self):
"""打印分配摘要"""
print("\n=== 效用优化分配结果 ===")
print(f"总效用: {self.get_total_utility():.2f}")
print("\n分配详情:")
for task_id, agent_id in self.assignments.items():
utility = self.calculate_utility(agent_id, task_id)
print(f"  {task_id} -> {agent_id} (效用: {utility:.2f})")
print("\nAgent负载:")
agent_loads = {agent_id: 0 for agent_id in self.agents}
for task_id, agent_id in self.assignments.items():
agent_loads[agent_id] += 1
for agent_id, load in agent_loads.items():
print(f"  {agent_id}: {load} 个任务")
# 演示基于效用的分配
def demo_utility_based_allocation():
print("\n=== 基于效用的任务分配演示 ===")
allocator = UtilityBasedTaskAllocator()
# 注册Agent
allocator.register_agent(
'ExpertAgent_001',
{'high_complexity': 0.9, 'critical_tasks': 0.8},
{'base_cost': 20, 'per_unit_cost': 2}
)
allocator.register_agent(
'StandardAgent_001',
{'medium_complexity': 0.7, 'standard_tasks': 0.8},
{'base_cost': 10, 'per_unit_cost': 1}
)
allocator.register_agent(
'BasicAgent_001',
{'simple_tasks': 0.6, 'routine_work': 0.7},
{'base_cost': 5, 'per_unit_cost': 0.5}
)
# 注册任务
allocator.register_task('task_critical', {'high_complexity': 0.8, 'critical_tasks': 0.9}, value=100, complexity=2.0)
allocator.register_task('task_standard', {'medium_complexity': 0.7, 'standard_tasks': 0.6}, value=50, complexity=1.0)
allocator.register_task('task_simple', {'simple_tasks': 0.8, 'routine_work': 0.7}, value=20, complexity=0.5)
allocator.register_task('task_complex', {'high_complexity': 0.9}, value=80, complexity=1.5)
# 执行优化分配
assignments = allocator.optimize_assignment()
# 打印结果
allocator.print_allocation_summary()

动态任务重分配

算法原理

在动态环境中，需要根据系统状态变化重新分配任务。

动态重分配实现

class DynamicTaskReallocator:
"""动态任务重分配器"""
def __init__(self, original_allocator):
self.original_allocator = original_allocator
self.reassignment_threshold = 0.2  # 重新分配阈值
self.reassignment_history = []
def should_reassign(self, agent_id: str, task_id: str) -> Tuple[bool, float]:
"""判断是否需要重新分配任务"""
current_utility = self.original_allocator.calculate_utility(agent_id, task_id)
# 检查是否有其他Agent能提供显著更好的效用
best_alternative_utility = float('-inf')
best_alternative_agent = None
for alt_agent_id in self.original_allocator.agents:
if alt_agent_id != agent_id:
alt_utility = self.original_allocator.calculate_utility(alt_agent_id, task_id)
if alt_utility > best_alternative_utility:
best_alternative_utility = alt_utility
best_alternative_agent = alt_agent_id
# 如果替代方案比当前方案好超过阈值，则建议重新分配
if best_alternative_agent and (best_alternative_utility - current_utility) > self.reassignment_threshold:
improvement = best_alternative_utility - current_utility
return True, improvement
return False, 0.0
def check_reassignments(self) -> List[Dict[str, Any]]:
"""检查所有分配，确定需要重新分配的任务"""
reassignment_candidates = []
for task_id, agent_id in self.original_allocator.assignments.items():
should_reassign, improvement = self.should_reassign(agent_id, task_id)
if should_reassign:
candidate = {
'task_id': task_id,
'current_agent': agent_id,
'recommended_agent': self._find_best_agent_for_task(task_id),
'improvement': improvement,
'timestamp': time.time()
}
reassignment_candidates.append(candidate)
return reassignment_candidates
def _find_best_agent_for_task(self, task_id: str) -> str:
"""为任务找到最佳Agent"""
best_utility = float('-inf')
best_agent = None
for agent_id in self.original_allocator.agents:
utility = self.original_allocator.calculate_utility(agent_id, task_id)
if utility > best_utility:
best_utility = utility
best_agent = agent_id
return best_agent
def perform_reassignment(self, candidates: List[Dict[str, Any]]) -> Dict[str, str]:
"""执行重新分配"""
reassignments = {}
for candidate in candidates:
task_id = candidate['task_id']
new_agent = candidate['recommended_agent']
# 从原Agent移除任务
old_agent = self.original_allocator.assignments[task_id]
old_agent_obj = self.original_allocator.agents[old_agent]
if task_id in old_agent_obj['current_tasks']:
old_agent_obj['current_tasks'].remove(task_id)
# 分配给新Agent
self.original_allocator.agents[new_agent]['current_tasks'].append(task_id)
self.original_allocator.tasks[task_id]['assigned_to'] = new_agent
self.original_allocator.assignments[task_id] = new_agent
reassignment_record = {
'task_id': task_id,
'from': old_agent,
'to': new_agent,
'improvement': candidate['improvement'],
'timestamp': time.time()
}
self.reassignment_history.append(reassignment_record)
reassignments[task_id] = new_agent
print(f"重新分配: {task_id} 从 {old_agent} 到 {new_agent} (改善: {candidate['improvement']:.2f})")
return reassignments
def demo_dynamic_reallocation():
print("\n=== 动态任务重分配演示 ===")
# 首先运行效用优化分配
allocator = UtilityBasedTaskAllocator()
# 注册Agent
allocator.register_agent(
'Agent_Initial_001',
{'type_a': 0.5, 'type_b': 0.3},
{'base_cost': 10, 'per_unit_cost': 1}
)
allocator.register_agent(
'Agent_Initial_002',
{'type_a': 0.2, 'type_b': 0.8},
{'base_cost': 12, 'per_unit_cost': 1.2}
)
allocator.register_agent(
'Agent_Improved_001',  # 这个Agent后来能力提升了
{'type_a': 0.9, 'type_b': 0.1},  # 现在更适合type_a任务
{'base_cost': 8, 'per_unit_cost': 0.8}
)
# 注册任务
allocator.register_task('task_type_a', {'type_a': 0.8}, value=50, complexity=1.0)
allocator.register_task('task_type_b', {'type_b': 0.7}, value=40, complexity=1.0)
# 初始分配
print("初始分配:")
initial_assignments = allocator.optimize_assignment()
allocator.print_allocation_summary()
# 模拟Agent能力提升
print("\n模拟Agent能力提升...")
allocator.register_agent(
'Agent_Improved_001',
{'type_a': 0.9, 'type_b': 0.1},
{'base_cost': 8, 'per_unit_cost': 0.8}
)
# 检查是否需要重新分配
reallocator = DynamicTaskReallocator(allocator)
candidates = reallocator.check_reassignments()
if candidates:
print(f"\n发现 {len(candidates)} 个重新分配候选:")
for candidate in candidates:
print(f"  {candidate['task_id']}: {candidate['current_agent']} -> {candidate['recommended_agent']} (改善: {candidate['improvement']:.2f})")
# 执行重新分配
reassignments = reallocator.perform_reassignment(candidates)
print(f"\n重新分配后:")
allocator.print_allocation_summary()
else:
print("\n无需重新分配")
if __name__ == "__main__":
demo_load_balancer()
demo_capability_based_allocation()
demo_utility_based_allocation()
demo_dynamic_reallocation()

算法比较与选择

各算法特点对比

算法类型	优点	缺点	适用场景
负载均衡	简单高效，防止过载	忽略任务和Agent特性	同质化Agent环境
基于能力	精确匹配，高质量输出	计算复杂，需要详细能力信息	异构Agent环境
基于效用	全局最优，考虑成本效益	计算复杂度高	价值敏感的应用

选择策略

• 环境稳定性：稳定的环境适合复杂算法，动态环境适合简单算法
• 任务特性：复杂任务适合基于能力分配，简单任务适合负载均衡
• Agent异构性：异构Agent适合基于能力或效用分配

实践练习

• 实现一个混合分配算法，结合负载均衡和能力匹配的优点。
• 扩展基于效用的分配器，添加时间窗口约束。
• 创建一个自适应分配系统，根据历史性能自动调整分配策略。

总结

本节课详细介绍了多智能体系统中的任务分配算法，包括负载均衡、基于能力和基于效用的分配方法。有效的任务分配是多智能体系统性能的关键因素，需要根据具体应用场景选择合适的算法。

学AI大模型的正确顺序，千万不要搞错了

🤔2026年AI风口已来！各行各业的AI渗透肉眼可见，超多公司要么转型做AI相关产品，要么高薪挖AI技术人才，机遇直接摆在眼前！

有往AI方向发展，或者本身有后端编程基础的朋友，直接冲AI大模型应用开发转岗超合适！

就算暂时不打算转岗，了解大模型、RAG、Prompt、Agent这些热门概念，能上手做简单项目，也绝对是求职加分王🔋

📝给大家整理了超全最新的AI大模型应用开发学习清单和资料，手把手帮你快速入门！👇👇

学习路线:

✅大模型基础认知—大模型核心原理、发展历程、主流模型（GPT、文心一言等）特点解析
✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑
✅开发基础能力—Python进阶、API接口调用、大模型开发框架（LangChain等）实操
✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用
✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代
✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经

以上6大模块，看似清晰好上手，实则每个部分都有扎实的核心内容需要吃透！

我把大模型的学习全流程已经整理📚好了！抓住AI时代风口，轻松解锁职业新可能，希望大家都能把握机遇，实现薪资/职业跃迁～

这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】