macf_runner.py
PCFA 运行及 RNG 流评估
原文件 · 227 行 · 9838 字节
runner/pcfa/macf_runner.py
SHA-256: a1b7c1894eeb156ebc62d07b7bc811593fb07aae693413b2770a42cee64913e0
1"""PCFA runner for MACF.23This runner executes the non-learning Project Manager-oriented Coalition4Formation Algorithm and logs episode-level metrics.5"""67import os8import pickle9import json1011import numpy as np12from .base_runner import Runner13from algorithms.algorithm.pcfa_policy import PCFAPolicy141516class MACFEnvRunner(Runner):17 """Runner for PCFA baseline."""1819 def __init__(self, config):20 super(MACFEnvRunner, self).__init__(config)2122 self.episode = 023 self.info_buffer = []24 print(self.log_dir)2526 # Retrieve special task IDs from the underlying environment.27 child_env = self.envs.envs[0]28 core_env = getattr(child_env, 'env', child_env)29 idle_task_id = core_env.idle_task_id30 explore_task_id = core_env.explore_task_id31 resupply_task_id = core_env.resupply_task_id3233 share_observation_space = (34 self.envs.share_observation_space[0]35 if self.use_centralized_V36 else self.envs.observation_space[0]37 )3839 self.policy = PCFAPolicy(40 config,41 self.envs.observation_space[0],42 share_observation_space,43 self.envs.action_space[0],44 device=config.device,45 )46 self.policy.set_task_layout(47 delivery_task_num=getattr(48 core_env, 'delivery_task_num', getattr(core_env, 'target_num', explore_task_id)49 ),50 explore_task_start=getattr(core_env, 'explore_task_start', explore_task_id),51 explore_task_end=getattr(core_env, 'explore_task_end', explore_task_id + 1),52 idle_task_id=idle_task_id,53 resupply_task_id=resupply_task_id,54 use_coalition_delivery=getattr(core_env, 'use_coalition_delivery', False),55 )5657 if self.n_rollout_threads != 1:58 raise ValueError(59 "PCFA evaluation currently requires n_rollout_threads=1 so "60 "terminal episodes are not mixed after vector-env auto-reset."61 )6263 # Dummy RNN states kept only to satisfy the policy interface.64 self.rnn_states = np.zeros(65 (66 self.episode_length + 1,67 self.n_rollout_threads,68 self.num_agents,69 self.recurrent_N,70 self.hidden_size,71 ),72 dtype=np.float32,73 )74 self.rnn_states_critic = np.zeros_like(self.rnn_states)7576 # Last actions per env thread; used to keep inactive agents on course.77 self.last_actions = np.full(78 (self.n_rollout_threads, self.num_agents),79 self.policy.idle_task_id,80 dtype=int,81 )8283 def run(self):84 episodes = int(self.num_env_steps) // self.episode_length // self.n_rollout_threads85 os.makedirs(os.path.join(self.log_dir, 'infos'), exist_ok=True)86 episode_results = []87 next_initial_state = None8889 for self.episode in range(episodes):90 if next_initial_state is None:91 obs, active_masks, available_actions = self.envs.reset(self.episode)92 else:93 # DummyVecEnv has already reset a terminal environment and94 # returned the next episode's initial observation. Reuse it95 # so the deterministic seed stream does not skip scenarios.96 obs, active_masks, available_actions = next_initial_state97 next_initial_state = None98 self.info_buffer = []99 self.last_actions[:] = self.policy.idle_task_id100 episode_reward = 0.0101 episode_task_completed = 0102 episode_sent = 0103 episode_sent_goal = 0104 steps_taken = 0105 last_info = None106107 for step in range(self.episode_length):108 new_actions = self._compute_actions(obs, active_masks, available_actions)109 # Inactive agents keep their previous action to avoid switching to idle.110 active = np.asarray(active_masks, dtype=bool).reshape(111 self.n_rollout_threads, self.num_agents112 )113 actions = np.where(active, new_actions, self.last_actions)114 self.last_actions = actions.copy()115116 obs, rewards, dones, infos, active_masks, _bad_mask, available_actions = self.envs.step(actions)117 self.info_buffer.append(infos)118 episode_reward += float(rewards.mean())119 episode_sent += infos[0].get('sent', 0)120 steps_taken = step + 1121 last_info = infos[0]122 if bool(np.all(dones[0])):123 next_initial_state = (obs, active_masks, available_actions)124 break125126 delivery_tasks = last_info['tasks'][: self.policy.delivery_task_num]127 episode_task_completed = sum(1 for t in delivery_tasks if t['completed'])128 episode_task_discovered = sum(1 for t in delivery_tasks if t['discovered'])129 episode_sent_goal = last_info.get('sent_goal', 1)130 success = float(episode_task_completed == self.policy.delivery_task_num)131 task_completion_ratio = float(episode_task_completed) / max(132 1, self.policy.delivery_task_num133 )134 discovered_ratio = float(episode_task_discovered) / max(135 1, self.policy.delivery_task_num136 )137 sent_percentage = float(episode_sent) / max(episode_sent_goal, 1)138 episode_results.append({139 'episode': int(self.episode),140 'success': success,141 'task_completed': int(episode_task_completed),142 'task_completion_ratio': task_completion_ratio,143 'task_discovered': int(episode_task_discovered),144 'discovered_ratio': discovered_ratio,145 'sent': float(episode_sent),146 'sent_goal': float(episode_sent_goal),147 'sent_percentage': sent_percentage,148 'episode_length': int(steps_taken),149 'episode_reward': float(episode_reward),150 })151152 if self.episode % self.log_interval == 0:153 print(154 f"episode {self.episode}/{episodes} | reward {episode_reward:.3f} | "155 f"completed {episode_task_completed} | sent {episode_sent}/{episode_sent_goal} | "156 f"success {success:.0f} | steps {steps_taken}"157 )158 self.writter.add_scalar('reward/episode', episode_reward, self.episode)159 self.writter.add_scalar('reward/task_completed', episode_task_completed, self.episode)160 self.writter.add_scalar('reward/sent_percentage', sent_percentage, self.episode)161 self.writter.add_scalar('eval/success', success, self.episode)162 self.writter.add_scalar(163 'eval/task_completion_ratio', task_completion_ratio, self.episode164 )165 self.writter.add_scalar('eval/discovered_ratio', discovered_ratio, self.episode)166 self.writter.add_scalar('eval/episode_length', steps_taken, self.episode)167168 if self.log_info_interval > 0 and self.episode % self.log_info_interval == 0:169 with open(os.path.join(self.log_dir, 'infos', f'{self.episode:04d}.pkl'), 'wb') as f:170 pickle.dump(self.info_buffer, f)171 print(f'{self.episode:04d}.pkl generated')172173 aggregate = {174 'episodes': len(episode_results),175 'success_rate': float(np.mean([row['success'] for row in episode_results])),176 'task_completion_ratio': float(np.mean([177 row['task_completion_ratio'] for row in episode_results178 ])),179 'discovered_ratio': float(np.mean([180 row['discovered_ratio'] for row in episode_results181 ])),182 'sent_percentage': float(np.mean([183 row['sent_percentage'] for row in episode_results184 ])),185 'episode_length': float(np.mean([186 row['episode_length'] for row in episode_results187 ])),188 }189 output = {190 'algorithm': 'pcfa-delivcoal'191 if self.policy.use_coalition_delivery else 'pcfa',192 'seed': int(getattr(self.config, 'seed', 0)),193 'aggregate': aggregate,194 'episodes': episode_results,195 }196 output_path = os.path.join(str(self.config.run_dir), 'pcfa_results.json')197 temp_path = output_path + '.tmp'198 with open(temp_path, 'w', encoding='utf-8') as file:199 json.dump(output, file, ensure_ascii=False, indent=2)200 os.replace(temp_path, output_path)201202 def _compute_actions(self, obs, active_masks, available_actions):203 env_num = self.n_rollout_threads204 actions = np.full((env_num, self.num_agents), self.policy.idle_task_id, dtype=int)205206 rnn_states_actor = self.rnn_states[0]207 rnn_states_critic = self.rnn_states_critic[0]208 masks = np.ones((env_num, self.num_agents, 1), dtype=np.float32)209210 for env_i in range(env_num):211 env_obs = {k: v[env_i:env_i + 1] for k, v in obs.items()}212 env_available_actions = available_actions[env_i:env_i + 1]213 env_active_masks = active_masks[env_i:env_i + 1]214215 _, acts, _, _, _ = self.policy.get_actions(216 None,217 env_obs,218 rnn_states_actor[env_i:env_i + 1],219 rnn_states_critic[env_i:env_i + 1],220 masks[env_i:env_i + 1],221 env_available_actions,222 env_active_masks,223 deterministic=True,224 )225 actions[env_i] = acts[0]226227 return actions