智能体化ABM：从Mesa实战到统计模型检验的完整指南

在复杂系统建模与仿真领域，基于代理的模型（Agent-based Models, ABM）正经历一场深刻的范式变革。传统ABM通常将代理行为预设为固定规则，而新兴的"智能体化"（Agentic）ABM则致力于赋予代理更高层次的自主决策与学习能力。这种转变不仅提升了模型对现实世界复杂性的刻画精度，也对模型可行性、性能评估及验证方法提出了全新挑战。本文将深入探讨智能体化ABM的实现路径，结合Mesa框架实战演示，并系统介绍统计模型检验（Statistical Model Checking）在这一前沿领域的应用方案。

1. 智能体化ABM的核心概念与演进背景

1.1 从传统ABM到智能体化ABM的范式转变

传统ABM中的代理行为多基于简单规则（如条件判断、状态机），虽然能够模拟群体涌现现象，但在处理动态环境、适应性学习等场景时显得力不从心。智能体化ABM的核心突破在于引入强化学习、深度学习等AI技术，使代理具备环境感知、策略优化和长期规划能力。这种转变使得模型能够更真实地模拟人类决策、市场博弈等复杂行为，为社会科学、经济学、生态学等领域提供更强大的研究工具。

1.2 智能体化ABM的关键特征与实现层级

智能体化ABM通常具备以下核心特征：

• 自主目标导向：代理能够根据内部状态和外部环境自主设定并追求目标
• 学习与适应：通过经验积累调整行为策略，具备在线或离线学习能力
• 社会推理：能够对其他代理的行为意图进行预测和响应
• 长期规划：考虑行动的长远后果，进行多步决策优化

实现层级可从简单到复杂分为：

1. 反应层：基于感知-行动循环的即时响应
2. 认知层：包含信念-期望-意图（BDI）架构的推理能力
3. 学习层：集成机器学习算法的自适应行为
4. 元认知层：具备自我监控和策略调整的高级智能

1.3 智能体化ABM的典型应用场景

智能体化ABM已在多个领域展现巨大潜力：

• 金融市场模拟：智能交易代理学习市场模式并调整投资策略
• 流行病传播研究：个体代理根据风险感知自主调整防护行为
• 城市交通优化：自动驾驶代理学习最优路径选择策略
• 供应链管理：智能物流代理动态调整库存和配送方案

2. 环境准备与Mesa框架基础

2.1 环境配置与依赖管理

智能体化ABM开发推荐使用Python生态，其中Mesa框架提供了完善的ABM开发基础。以下是标准环境配置：

# 创建虚拟环境 python -m venv agentic_abm_env source agentic_abm_env/bin/activate # Linux/Mac # agentic_abm_env\Scripts\activate # Windows # 安装核心依赖 pip install mesa pip install torch==2.0.1 # 深度学习支持 pip install gymnasium==0.28.1 # 强化学习环境 pip install scikit-learn==1.3.0 # 机器学习工具 pip install pandas==2.0.3 # 数据分析

2.2 Mesa框架架构解析

Mesa采用模块化设计，核心组件包括：

• Model类：定义模型整体逻辑和时间推进机制
• Agent类：封装个体代理的行为和状态
• Space模块：管理代理的空间关系和交互网络
• Visualization模块：提供多种可视化方案
• DataCollection模块：支持运行时数据采集

# 基础Mesa模型结构示例 import mesa import numpy as np class SmartAgent(mesa.Agent): def __init__(self, unique_id, model, learning_rate=0.01): super().__init__(unique_id, model) self.q_table = {} # Q-learning表 self.learning_rate = learning_rate self.state = None self.reward = 0 def step(self): """代理决策步骤""" current_state = self.get_state() action = self.choose_action(current_state) self.perform_action(action) next_state = self.get_state() self.update_q_value(current_state, action, self.reward, next_state) def get_state(self): """获取当前环境状态""" # 实现状态感知逻辑 return self.model.get_agent_environment(self) def choose_action(self, state): """基于当前状态选择行动""" if state not in self.q_table: self.q_table[state] = {action: 0 for action in self.possible_actions} # ε-greedy策略 if np.random.random() < self.model.epsilon: return np.random.choice(self.possible_actions) else: return max(self.q_table[state], key=self.q_table[state].get) class AgenticModel(mesa.Model): def __init__(self, N, width, height, epsilon=0.1): super().__init__() self.num_agents = N self.grid = mesa.space.MultiGrid(width, height, True) self.schedule = mesa.time.RandomActivation(self) self.epsilon = epsilon # 创建智能代理 for i in range(self.num_agents): agent = SmartAgent(i, self) self.schedule.add(agent) x = self.random.randrange(self.grid.width) y = self.random.randrange(self.grid.height) self.grid.place_agent(agent, (x, y)) def step(self): """模型时间步推进""" self.schedule.step()

3. 智能体化ABM的可行性分析

3.1 技术可行性评估框架

实现智能体化ABM需要综合评估多个维度的可行性：

计算复杂度分析：

• 状态空间大小：代理感知的环境状态维度
• 行动空间复杂度：可选行动的组合爆炸问题
• 学习算法收敛性：训练时间和稳定性要求
• 内存与存储需求：Q表、神经网络参数等存储开销

算法可行性矩阵：

3.2 资源需求与优化策略

智能体化ABM对计算资源的需求显著高于传统ABM，需要针对性优化：

# 资源优化示例：经验回放与目标网络 import collections import torch.nn as nn class ReplayBuffer: def __init__(self, capacity): self.buffer = collections.deque(maxlen=capacity) def push(self, state, action, reward, next_state, done): self.buffer.append((state, action, reward, next_state, done)) def sample(self, batch_size): indices = np.random.choice(len(self.buffer), batch_size, replace=False) states, actions, rewards, next_states, dones = zip(*[self.buffer[idx] for idx in indices]) return np.array(states), np.array(actions), np.array(rewards), np.array(next_states), np.array(dones) class DQN(nn.Module): def __init__(self, input_dim, output_dim, hidden_dim=128): super(DQN, self).__init__() self.network = nn.Sequential( nn.Linear(input_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, output_dim) ) def forward(self, x): return self.network(x) # 目标网络更新策略 def update_target_network(target_net, policy_net, tau=0.005): """软更新目标网络参数""" for target_param, policy_param in zip(target_net.parameters(), policy_net.parameters()): target_param.data.copy_(tau * policy_param.data + (1.0 - tau) * target_param.data)

3.3 模型可解释性与验证挑战

智能体化ABM的黑箱特性带来可解释性挑战，需要建立专门的验证框架：

• 行为轨迹分析：记录关键决策点的状态-行动序列
• 重要性采样：识别对结果影响最大的代理行为模式
• 敏感性分析：评估模型对超参数和初始条件的依赖程度
• 对比实验设计：与规则基ABM的结果进行系统性比较

4. 性能评估指标体系

4.1 计算性能度量标准

智能体化ABM的性能评估需要多维度指标：

# 性能监控装饰器 import time import functools from memory_profiler import memory_usage def performance_monitor(func): @functools.wraps(func) def wrapper(*args, **kwargs): # 时间性能 start_time = time.time() # 内存性能 mem_usage_before = memory_usage(-1, interval=0.1, timeout=1)[0] result = func(*args, **kwargs) execution_time = time.time() - start_time mem_usage_after = memory_usage(-1, interval=0.1, timeout=1)[0] memory_increase = mem_usage_after - mem_usage_before print(f"函数 {func.__name__} 执行时间: {execution_time:.4f}秒") print(f"内存增长: {memory_increase:.4f} MB") return result return wrapper class PerformanceMetrics: def __init__(self): self.metrics = { 'step_time': [], 'memory_usage': [], 'convergence_rate': [], 'reward_progression': [] } def record_step_metrics(self, step_time, memory_usage, reward=None): self.metrics['step_time'].append(step_time) self.metrics['memory_usage'].append(memory_usage) if reward is not None: self.metrics['reward_progression'].append(reward) def calculate_convergence(self, window_size=100): """计算奖励收敛性""" if len(self.metrics['reward_progression']) < window_size: return None recent_rewards = self.metrics['reward_progression'][-window_size:] return np.std(recent_rewards) / (abs(np.mean(recent_rewards)) + 1e-8)

4.2 模型质量评估维度

除了计算性能，模型质量评估同样重要：

有效性指标：

• 预测准确性：与真实数据或基准模型的对比
• 泛化能力：在未见数据上的表现
• 稳健性：对噪声和扰动的抵抗能力

实用性指标：

• 训练效率：达到满意性能所需的计算资源
• 可扩展性：代理数量增加时的性能衰减程度
• 易用性：模型配置和调参的复杂度

4.3 多智能体系统特有性能挑战

智能体化ABM中的多代理交互带来独特性能问题：

# 通信优化与并行计算 from multiprocessing import Pool import threading class ParallelAgentScheduler: def __init__(self, num_processes=4): self.num_processes = num_processes self.agent_batches = [] def batch_agents(self, agents, batch_size): """将代理分批处理""" self.agent_batches = [agents[i:i + batch_size] for i in range(0, len(agents), batch_size)] def parallel_step(self, batch): """并行执行代理步进""" results = [] for agent in batch: result = agent.step() results.append(result) return results def run_parallel(self): """并行运行所有代理""" with Pool(self.num_processes) as pool: results = pool.map(self.parallel_step, self.agent_batches) return [item for sublist in results for item in sublist] # 通信效率优化 class EfficientCommunication: def __init__(self, communication_range): self.range = communication_range self.message_queues = {} def broadcast_message(self, sender, message, recipients): """有限范围广播通信""" for recipient in recipients: if self.distance(sender, recipient) <= self.range: if recipient not in self.message_queues: self.message_queues[recipient] = [] self.message_queues[recipient].append(message) def distance(self, agent1, agent2): """计算代理间距离""" pos1 = agent1.pos pos2 = agent2.pos return np.sqrt((pos1[0]-pos2[0])**2 + (pos1[1]-pos2[1])**2)

5. 统计模型检验理论与实践

5.1 统计模型检验基础概念

统计模型检验（Statistical Model Checking, SMC）通过统计推理验证系统属性，特别适合处理复杂系统的验证问题：

核心优势：

• 避免状态空间爆炸问题
• 提供概率性保证而非绝对确定性
• 适用于黑箱或部分可观测系统
• 能够处理连续时间和随机行为

基本流程：

1. 形式化规约：用时序逻辑描述待验证属性
2. 样本生成：运行模型获得行为轨迹
3. 假设检验：基于样本统计推断属性成立概率
4. 结果解释：给出置信区间和错误边界

5.2 SMC在智能体化ABM中的实现

# 统计模型检验框架实现 import scipy.stats as stats from scipy import special class StatisticalModelChecker: def __init__(self, model, confidence=0.95, error_margin=0.05): self.model = model self.confidence = confidence self.error_margin = error_margin self.traces = [] def generate_trace(self, max_steps=1000): """生成单条行为轨迹""" trace = [] self.model.reset() for step in range(max_steps): self.model.step() state = self.model.get_global_state() trace.append(state) if self.model.termination_condition(): break return trace def check_property(self, property_func, num_samples=1000): """检验特定属性""" positive_samples = 0 for i in range(num_samples): trace = self.generate_trace() if property_func(trace): positive_samples += 1 # 计算置信区间 p_hat = positive_samples / num_samples z_value = stats.norm.ppf(1 - (1 - self.confidence) / 2) margin = z_value * np.sqrt(p_hat * (1 - p_hat) / num_samples) lower_bound = max(0, p_hat - margin) upper_bound = min(1, p_hat + margin) return { 'estimated_probability': p_hat, 'confidence_interval': (lower_bound, upper_bound), 'sample_size': num_samples, 'confidence_level': self.confidence } def sequential_probability_ratio_test(self, property_func, p0, p1, alpha=0.05, beta=0.05): """序贯概率比检验（SPRT）""" A = (1 - beta) / alpha B = beta / (1 - alpha) logA = np.log(A) logB = np.log(B) log_likelihood_ratio = 0 sample_count = 0 while True: trace = self.generate_trace() sample_count += 1 property_holds = property_func(trace) if property_holds: log_likelihood_ratio += np.log(p1 / p0) else: log_likelihood_ratio += np.log((1 - p1) / (1 - p0)) if log_likelihood_ratio >= logA: return {'decision': 'accept H1', 'samples': sample_count} elif log_likelihood_ratio <= logB: return {'decision': 'accept H0', 'samples': sample_count} # 常用时序逻辑属性模板 class TemporalProperties: @staticmethod def eventually(trace, condition, within_steps=None): """最终性：◇condition""" for i, state in enumerate(trace): if condition(state): if within_steps is None or i <= within_steps: return True return False @staticmethod def always(trace, condition): """始终性：□condition""" return all(condition(state) for state in trace) @staticmethod def until(trace, condition1, condition2): """直到：condition1 U condition2""" condition1_held = False for state in trace: if condition2(state): return True if not condition1(state): return False condition1_held = True return condition1_held

5.3 高级SMC技术与优化策略

重要性采样优化：

class ImportanceSamplingSMC: def __init__(self, model, importance_sampler): self.model = model self.sampler = importance_sampler def likelihood_ratio_estimation(self, property_func, num_samples): """似然比估计""" weighted_sum = 0 total_weight = 0 for _ in range(num_samples): # 从建议分布采样 trace, weight = self.sampler.generate_weighted_trace() if property_func(trace): weighted_sum += weight total_weight += weight return weighted_sum / total_weight if total_weight > 0 else 0 class AdaptiveImportanceSampler: def __init__(self, initial_params): self.params = initial_params self.performance_history = [] def update_parameters(self, traces, property_func): """基于历史性能自适应调整采样参数""" successful_traces = [t for t in traces if property_func(t)] if successful_traces: # 分析成功轨迹的特征，调整采样策略 self.analyze_success_patterns(successful_traces) def analyze_success_patterns(self, traces): """分析导致属性成立的关键模式""" # 实现模式识别逻辑 pass

6. 完整实战案例：智能城市交通模拟

6.1 项目需求与架构设计

构建一个智能城市交通模拟系统，其中车辆代理学习最优路径选择策略，交通灯代理动态调整信号时序：

系统组件：

• 道路网络：有向图表示城市道路
• 车辆代理：强化学习路径选择
• 交通灯代理：优化信号控制策略
• 环境模拟：交通流、拥堵传播
• 评估模块：通行时间、拥堵指数等指标

6.2 核心实现代码

# 智能交通模型完整实现 import networkx as nx from mesa.visualization.ModularVisualization import ModularServer from mesa.visualization.modules import CanvasGrid, ChartModule class SmartVehicle(mesa.Agent): def __init__(self, unique_id, model, origin, destination): super().__init__(unique_id, model) self.origin = origin self.destination = destination self.current_node = origin self.path = [] self.travel_time = 0 self.learning_model = RouteLearningModel() def step(self): if self.current_node == self.destination: return # 到达目的地 if not self.path: self.plan_route() next_node = self.path.pop(0) road_segment = self.model.road_network[self.current_node][next_node] # 检查道路容量和拥堵情况 if road_segment['traffic'] < road_segment['capacity']: self.move_to(next_node) self.travel_time += road_segment['base_time'] * self.congestion_factor(road_segment) else: # 拥堵等待或重新规划 self.handle_congestion() def plan_route(self): """基于学习模型规划路径""" self.path = self.learning_model.get_optimal_route( self.current_node, self.destination, self.model.get_traffic_conditions() ) def congestion_factor(self, road_segment): """计算拥堵影响因子""" utilization = road_segment['traffic'] / road_segment['capacity'] return 1 + 0.5 * utilization ** 2 class TrafficLight(mesa.Agent): def __init__(self, unique_id, model, intersection, phases): super().__init__(unique_id, model) self.intersection = intersection self.phases = phases self.current_phase = 0 self.phase_timer = 0 self.learning_controller = SignalController() def step(self): # 基于实时交通流调整信号时序 traffic_conditions = self.assess_traffic_conditions() optimal_phase = self.learning_controller.select_phase(traffic_conditions) if optimal_phase != self.current_phase: self.transition_phase(optimal_phase) self.phase_timer += 1 if self.phase_timer >= self.phases[self.current_phase]['duration']: self.advance_phase() def assess_traffic_conditions(self): """评估各方向交通需求""" conditions = {} for direction in ['north', 'south', 'east', 'west']: queue_length = self.get_queue_length(direction) conditions[direction] = queue_length return conditions class SmartCityModel(mesa.Model): def __init__(self, road_network, num_vehicles, traffic_lights): super().__init__() self.road_network = road_network self.grid = mesa.space.NetworkGrid(road_network) self.schedule = mesa.time.SimultaneousActivation(self) # 创建交通灯 for tl_id, tl_config in traffic_lights.items(): traffic_light = TrafficLight(tl_id, self, tl_config['intersection'], tl_config['phases']) self.schedule.add(traffic_light) self.grid.place_agent(traffic_light, tl_config['intersection']) # 创建车辆 for i in range(num_vehicles): origin, destination = self.generate_od_pair() vehicle = SmartVehicle(i, self, origin, destination) self.schedule.add(vehicle) self.grid.place_agent(vehicle, origin) def generate_od_pair(self): """生成起点-终点对""" nodes = list(self.road_network.nodes()) return self.random.choice(nodes), self.random.choice(nodes) def get_traffic_conditions(self): """获取全局交通状况""" conditions = {} for edge in self.road_network.edges(): conditions[edge] = { 'traffic': len(self.grid.get_cell_list_contents([edge])), 'capacity': self.road_network[edge[0]][edge[1]]['capacity'] } return conditions # 强化学习路径规划模块 class RouteLearningModel: def __init__(self, learning_rate=0.1, discount_factor=0.9): self.q_values = {} self.learning_rate = learning_rate self.discount_factor = discount_factor def get_optimal_route(self, start, end, traffic_conditions): """基于Q-learning获取最优路径""" # 实现路径学习算法 path = self.a_star_search(start, end, traffic_conditions) return path def update_q_values(self, experience): """根据经验更新Q值""" state, action, reward, next_state = experience current_q = self.get_q_value(state, action) max_next_q = max([self.get_q_value(next_state, a) for a in self.get_actions(next_state)]) new_q = current_q + self.learning_rate * (reward + self.discount_factor * max_next_q - current_q) self.set_q_value(state, action, new_q)

6.3 模型验证与性能分析

# 交通模型验证框架 class TrafficModelValidator: def __init__(self, model): self.model = model self.metrics = { 'average_travel_time': [], 'congestion_level': [], 'throughput': [] } def validate_model_properties(self): """验证关键模型属性""" properties = { 'deadlock_free': self.check_deadlock_freedom(), 'fairness': self.check_fairness(), 'liveness': self.check_liveness(), 'safety': self.check_safety() } return properties def check_deadlock_freedom(self): """检验死锁自由性""" def property_func(trace): # 检查是否存在永久阻塞的车辆 for step_data in trace: if any(vehicle['blocked_time'] > 100 for vehicle in step_data['vehicles']): return False return True checker = StatisticalModelChecker(self.model) return checker.check_property(property_func) def check_fairness(self): """检验公平性：所有车辆最终都能到达目的地""" def property_func(trace): final_state = trace[-1] return all(vehicle['arrived'] for vehicle in final_state['vehicles']) checker = StatisticalModelChecker(self.model) return checker.check_property(property_func, num_samples=500) def performance_benchmark(self, num_runs=10): """性能基准测试""" results = [] for run in range(num_runs): self.model.reset() performance_data = self.run_simulation() results.append(performance_data) return self.analyze_benchmark_results(results) # 可视化与监控 def create_visualization(): """创建模型可视化界面""" grid = CanvasGrid(lambda agent: agent.portrayal(), 50, 50, 500, 500) chart = ChartModule([{ "Label": "AverageTravelTime", "Color": "Black" }], data_collector_name="datacollector") server = ModularServer(SmartCityModel, [grid, chart], "Smart City Traffic Model", {"N": 100, "width": 50, "height": 50}) return server

7. 常见问题与解决方案

7.1 训练不稳定与收敛问题

智能体化ABM训练过程中常见的不稳定现象及应对策略：

问题现象：奖励震荡剧烈

• 可能原因：学习率过高、探索策略过于激进
• 解决方案：自适应学习率调整、保守探索策略

class AdaptiveLearningRate: def __init__(self, initial_lr=0.01, decay_factor=0.99, min_lr=1e-5): self.lr = initial_lr self.decay_factor = decay_factor self.min_lr = min_lr self.stable_count = 0 def update_based_on_performance(self, reward_std): """基于奖励稳定性调整学习率""" if reward_std < 0.1: # 奖励稳定 self.stable_count += 1 if self.stable_count > 10: self.lr = max(self.min_lr, self.lr * self.decay_factor) else: self.stable_count = 0

问题现象：代理行为模式单一

• 可能原因：过早收敛到局部最优、探索不足
• 解决方案：周期性探索增强、课程学习策略

class CurriculumExploration: def __init__(self, base_epsilon=0.1, curriculum_stages=5): self.base_epsilon = base_epsilon self.stages = curriculum_stages self.current_stage = 0 def get_exploration_rate(self, training_progress): """基于训练进度调整探索率""" stage_progress = training_progress * self.stages self.current_stage = min(int(stage_progress), self.stages - 1) # 早期阶段探索率较高，后期逐渐降低 stage_epsilon = self.base_epsilon * (1.0 - self.current_stage / self.stages) return max(0.01, stage_epsilon)

7.2 计算性能瓶颈优化

大规模智能体化ABM的性能优化技巧：

内存优化策略：

class MemoryEfficientAgents: def __init__(self): self.shared_knowledge = {} # 共享知识库减少重复存储 def use_compressed_representations(self, state): """使用压缩状态表示""" # 哈希状态为紧凑表示 return hash(tuple(sorted(state.items()))) def implement_experience_replay(self, buffer_size=10000): """经验回放与优先级采样""" self.replay_buffer = PrioritizedReplayBuffer(buffer_size) def clear_transient_data(self): """定期清理临时数据""" import gc gc.collect() class PrioritizedReplayBuffer: def __init__(self, capacity, alpha=0.6): self.capacity = capacity self.alpha = alpha self.buffer = [] self.priorities = np.zeros(capacity) self.position = 0 def add(self, experience, priority): """添加经验并设置优先级""" if len(self.buffer) < self.capacity: self.buffer.append(experience) else: self.buffer[self.position] = experience self.priorities[self.position] = priority self.position = (self.position + 1) % self.capacity def sample(self, batch_size, beta=0.4): """基于优先级采样""" priorities = self.priorities[:len(self.buffer)] probabilities = priorities ** self.alpha probabilities /= probabilities.sum() indices = np.random.choice(len(self.buffer), batch_size, p=probabilities) experiences = [self.buffer[idx] for idx in indices] # 重要性采样权重 total = len(self.buffer) weights = (total * probabilities[indices]) ** (-beta) weights /= weights.max() return experiences, indices, weights

7.3 多代理协调与通信挑战

智能体化ABM中代理间协调的常见问题：

通信开销控制：

class EfficientMultiAgentCommunication: def __init__(self, communication_cost_weight=0.1): self.cost_weight = communication_cost_weight self.message_log = [] def should_communicate(self, information_gain, communication_cost): """权衡通信收益与成本""" return information_gain > self.cost_weight * communication_cost def implement_broadcast_filtering(self, message, potential_receivers): """智能广播过滤""" relevant_receivers = [ agent for agent in potential_receivers if self.is_message_relevant(agent, message) ] return relevant_receivers def is_message_relevant(self, agent, message): """判断消息对代理的相关性""" # 基于代理角色、位置、目标等判断相关性 return (agent.role in message['target_roles'] and self.distance(agent, message['source']) < message['range'])

8. 最佳实践与工程建议

8.1 模型设计原则

构建高质量智能体化ABM的系统化方法：

模块化设计：

• 代理行为与学习算法分离
• 环境模拟与交互逻辑解耦
• 评估指标独立可配置
• 可视化与核心逻辑分层

可复现性保障：

class ReproducibleExperiment: def __init__(self, seed=42): self.seed = seed self.set_random_seeds() def set_random_seeds(self): """设置所有随机数生成器种子""" import random random.seed(self.seed) np.random.seed(self.seed) torch.manual_seed(self.seed) def save_experiment_config(self, config): """保存实验配置""" import json with open(f'experiment_config_{self.seed}.json', 'w') as f: json.dump(config, f, indent=2) def log_training_artifacts(self, model, metrics, artifacts_dir): """保存训练过程产物""" os.makedirs(artifacts_dir, exist_ok=True) # 保存模型参数 torch.save(model.state_dict(), f'{artifacts_dir}/model_weights.pth') # 保存训练曲线 self.plot_metrics(metrics, f'{artifacts_dir}/training_metrics.png')

8.2 性能调优策略

系统化的性能优化方法论：

计算资源分配优化：

class ResourceAwareScheduler: def __init__(self, available_memory_gb=8, available_cores=4): self.memory_limit = available_memory_gb * 1024 ** 3 # 转换为字节 self.core_count = available_cores self.agent_batch_size = self.calculate_optimal_batch_size() def calculate_optimal_batch_size(self): """基于可用资源计算最优批处理大小""" memory_per_agent = 50 * 1024 * 1024 # 估计每个代理内存占用50MB max_agents_by_memory = self.memory_limit // memory_per_agent max_agents_by_cores = self.core_count * 10 # 每个核心处理10个代理 return min(max_agents_by_memory, max_agents_by_cores) def dynamic_batch_adjustment(self, current_memory_usage): """动态调整批处理大小""" memory_ratio = current_memory_usage / self.memory_limit if memory_ratio > 0.8: self.agent_batch_size = max(1, int(self.agent_batch_size * 0.8)) elif memory_ratio < 0.5: self.agent_batch_size = min(100, int(self.agent_batch_size * 1.2))

分布式计算支持：

class DistributedABMFramework: def __init__(self, num_workers=4): self.num_workers = num_workers self.worker_pool = None def initialize_workers(self): """初始化工作进程池""" from multiprocessing import Process, Queue self.worker_queues = [Queue() for _ in range(self.num_workers)] self.result_queues = [Queue() for _ in range(self.num_workers)] self.workers = [] for i in range(self.num_workers): p = Process(target=self.worker_loop, args=(i, self.worker_queues[i], self.result_queues[i])) p.start() self.workers.append(p) def distribute_agents(self, agents): """将代理分布到不同工作进程""" agent_batches = np.array_split(agents, self.num_workers) for i, batch in enumerate(agent_batches): self.worker_queues[i].put(