在复杂系统建模与仿真领域基于代理的模型Agent-based Models, ABM正经历一场深刻的范式变革。传统ABM通常将代理行为预设为固定规则而新兴的智能体化AgenticABM则致力于赋予代理更高层次的自主决策与学习能力。这种转变不仅提升了模型对现实世界复杂性的刻画精度也对模型可行性、性能评估及验证方法提出了全新挑战。本文将深入探讨智能体化ABM的实现路径结合Mesa框架实战演示并系统介绍统计模型检验Statistical Model Checking在这一前沿领域的应用方案。1. 智能体化ABM的核心概念与演进背景1.1 从传统ABM到智能体化ABM的范式转变传统ABM中的代理行为多基于简单规则如条件判断、状态机虽然能够模拟群体涌现现象但在处理动态环境、适应性学习等场景时显得力不从心。智能体化ABM的核心突破在于引入强化学习、深度学习等AI技术使代理具备环境感知、策略优化和长期规划能力。这种转变使得模型能够更真实地模拟人类决策、市场博弈等复杂行为为社会科学、经济学、生态学等领域提供更强大的研究工具。1.2 智能体化ABM的关键特征与实现层级智能体化ABM通常具备以下核心特征自主目标导向代理能够根据内部状态和外部环境自主设定并追求目标学习与适应通过经验积累调整行为策略具备在线或离线学习能力社会推理能够对其他代理的行为意图进行预测和响应长期规划考虑行动的长远后果进行多步决策优化实现层级可从简单到复杂分为反应层基于感知-行动循环的即时响应认知层包含信念-期望-意图BDI架构的推理能力学习层集成机器学习算法的自适应行为元认知层具备自我监控和策略调整的高级智能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 torch2.0.1 # 深度学习支持 pip install gymnasium0.28.1 # 强化学习环境 pip install scikit-learn1.3.0 # 机器学习工具 pip install pandas2.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_rate0.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], keyself.q_table[state].get) class AgenticModel(mesa.Model): def __init__(self, N, width, height, epsilon0.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表、神经网络参数等存储开销算法可行性矩阵算法类型状态空间适应性训练效率收敛保证适用场景Q-learning中小离散空间中等有理论保证规则明确环境Deep Q-Network高维连续空间较低局部最优复杂感知任务Policy Gradient连续行动空间中等渐进收敛精细控制任务Multi-agent RL动态交互环境低挑战较大社会模拟场景3.2 资源需求与优化策略智能体化ABM对计算资源的需求显著高于传统ABM需要针对性优化# 资源优化示例经验回放与目标网络 import collections import torch.nn as nn class ReplayBuffer: def __init__(self, capacity): self.buffer collections.deque(maxlencapacity) 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, replaceFalse) 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_dim128): 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, tau0.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, interval0.1, timeout1)[0] result func(*args, **kwargs) execution_time time.time() - start_time mem_usage_after memory_usage(-1, interval0.1, timeout1)[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, rewardNone): 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_size100): 计算奖励收敛性 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_processes4): 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通过统计推理验证系统属性特别适合处理复杂系统的验证问题核心优势避免状态空间爆炸问题提供概率性保证而非绝对确定性适用于黑箱或部分可观测系统能够处理连续时间和随机行为基本流程形式化规约用时序逻辑描述待验证属性样本生成运行模型获得行为轨迹假设检验基于样本统计推断属性成立概率结果解释给出置信区间和错误边界5.2 SMC在智能体化ABM中的实现# 统计模型检验框架实现 import scipy.stats as stats from scipy import special class StatisticalModelChecker: def __init__(self, model, confidence0.95, error_margin0.05): self.model model self.confidence confidence self.error_margin error_margin self.traces [] def generate_trace(self, max_steps1000): 生成单条行为轨迹 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_samples1000): 检验特定属性 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, alpha0.05, beta0.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_stepsNone): 最终性◇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_held5.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): 分析导致属性成立的关键模式 # 实现模式识别逻辑 pass6. 完整实战案例智能城市交通模拟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_rate0.1, discount_factor0.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_samples500) def performance_benchmark(self, num_runs10): 性能基准测试 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_namedatacollector) server ModularServer(SmartCityModel, [grid, chart], Smart City Traffic Model, {N: 100, width: 50, height: 50}) return server7. 常见问题与解决方案7.1 训练不稳定与收敛问题智能体化ABM训练过程中常见的不稳定现象及应对策略问题现象奖励震荡剧烈可能原因学习率过高、探索策略过于激进解决方案自适应学习率调整、保守探索策略class AdaptiveLearningRate: def __init__(self, initial_lr0.01, decay_factor0.99, min_lr1e-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_epsilon0.1, curriculum_stages5): 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_size10000): 经验回放与优先级采样 self.replay_buffer PrioritizedReplayBuffer(buffer_size) def clear_transient_data(self): 定期清理临时数据 import gc gc.collect() class PrioritizedReplayBuffer: def __init__(self, capacity, alpha0.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, beta0.4): 基于优先级采样 priorities self.priorities[:len(self.buffer)] probabilities priorities ** self.alpha probabilities / probabilities.sum() indices np.random.choice(len(self.buffer), batch_size, pprobabilities) experiences [self.buffer[idx] for idx in indices] # 重要性采样权重 total len(self.buffer) weights (total * probabilities[indices]) ** (-beta) weights / weights.max() return experiences, indices, weights7.3 多代理协调与通信挑战智能体化ABM中代理间协调的常见问题通信开销控制class EfficientMultiAgentCommunication: def __init__(self, communication_cost_weight0.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, seed42): 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(fexperiment_config_{self.seed}.json, w) as f: json.dump(config, f, indent2) def log_training_artifacts(self, model, metrics, artifacts_dir): 保存训练过程产物 os.makedirs(artifacts_dir, exist_okTrue) # 保存模型参数 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_gb8, available_cores4): 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_workers4): 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(targetself.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(