evaluate_priority_reconfiguration.py
配对优先级事件、首次动作和闭环执行
原文件 · 596 行 · 21644 字节
scripts/analysis/evaluate_priority_reconfiguration.py
SHA-256: be27bad0754cadad43949f0422de8cc2a9368777f31626b70879f849c489b871
1"""Evaluate frozen policies on a controlled task-priority reveal event.23Each model first creates its own pre-event trajectory on a scenario with nine4visible delivery tasks and one reserved hidden task. At ``event_step`` the5same serialized environment state is forked into three conditions:67* ``high``: reveal the reserved task with priority 1.0;8* ``low``: reveal it with priority 0.5;9* ``high_keep_current``: reveal the high-priority task while retaining the10 previous joint action at the event-triggered decision.1112The high/low pair measures selective policy response. ``high_keep_current``13is an event-response counterfactual baseline, not an oracle or a separate14policy. The environment next invokes the policy at its usual redecision15event, so this branch measures the value of reacting at the reveal itself.16All continuations are deterministic so the three branches share the same17pre-event state and differ only through the registered intervention.18"""1920from __future__ import annotations2122import argparse23import copy24import csv25import gc26import hashlib27import json28import os29import subprocess30import sys31import threading32from concurrent.futures import ThreadPoolExecutor, as_completed33from datetime import datetime34from pathlib import Path3536import numpy as np37import torch38import yaml3940REPO_ROOT = Path(__file__).resolve().parents[2]41sys.path.insert(0, str(REPO_ROOT))4243from scripts.analysis.diagnose_idle_opportunity import ( # noqa: E40244 _distribution,45 _make_runner,46 _parse_scenarios,47 _write_csv,48)49from train.eval import _close_runner # noqa: E402505152SCHEMA_VERSION = 1535455def _atomic_json(path, value):56 path = Path(path)57 temporary = path.with_suffix(path.suffix + ".tmp")58 temporary.write_text(59 json.dumps(value, ensure_ascii=False, indent=2) + "\n",60 encoding="utf-8",61 )62 os.replace(temporary, path)636465def _physical_state_hash(core):66 payload = {67 "step": int(core.step_count),68 "agents": [69 {70 "id": int(agent.aid),71 "position": np.asarray(agent.position, dtype=np.float64).tolist(),72 "living": bool(agent.living),73 "load": float(agent.load),74 "battery": float(agent.battery),75 "coalition": int(agent.coalition),76 }77 for agent in core.agents78 ],79 "tasks": [80 {81 "id": int(task.id),82 "position": np.asarray(task.position, dtype=np.float64).tolist(),83 "demand": float(task.demand),84 "priority": float(task.req_vec[0]),85 "urgency": float(task.urgency),86 "discovered": bool(task.discovered),87 "completed": bool(task.completed),88 }89 for task in core.tasks90 ],91 }92 rendered = json.dumps(payload, sort_keys=True, separators=(",", ":"))93 return hashlib.sha256(rendered.encode("utf-8")).hexdigest()949596def _policy_step(actor, obs, rnn_states, masks, available_actions, active, last):97 distribution, next_rnn = _distribution(98 actor, obs, rnn_states, masks, available_actions99 )100 proposed = distribution.probs.argmax(dim=-1, keepdim=True).cpu().numpy()101 next_rnn = next_rnn.detach().cpu().numpy()102 actions = np.asarray(active) * proposed + (1.0 - np.asarray(active)) * last103 return distribution, actions, next_rnn104105106def _set_reserved_visibility(core, task_id, discovered):107 task = core.tasks[int(task_id)]108 task.discovered = bool(discovered)109 task.completed = False110 if discovered:111 task.discovered_step = int(core.step_count)112 core.event_new_task_found = 1113 else:114 task.discovered_step = None115 task.first_assigned_step = None116 task.response_time = None117118119def _event_observation(env, task_id, priority):120 core = getattr(env, "env", env)121 task = core.tasks[int(task_id)]122 task.req_vec[0] = float(priority)123 _set_reserved_visibility(core, task_id, True)124 obs = core._get_observations()125 available = core._get_available_actions()126 active = np.asarray(127 [1.0 if agent.living else 0.0 for agent in core.agents],128 dtype=np.float32,129 )[:, None]130 core._current_decision_mask = active[:, 0].copy()131 return obs, active, available132133134def _source_departure_diagnostics(core, actions):135 rows = []136 for agent in core.agents:137 old = int(agent.coalition)138 new = int(actions[int(agent.aid), 0])139 if (140 new == old141 or old < 0142 or old >= len(core.tasks)143 or int(core.tasks[old].type) != 1144 ):145 continue146 task = core.tasks[old]147 remaining_other_load = sum(148 max(0.0, float(other.load))149 for other in core.agents150 if other.living151 and int(other.aid) != int(agent.aid)152 and int(other.coalition) == old153 )154 rows.append(155 {156 "agent": int(agent.aid),157 "source_task": old,158 "source_demand": float(task.demand),159 "source_other_committed_load": remaining_other_load,160 "source_fragile": int(161 remaining_other_load + 1e-6 < float(task.demand)162 ),163 }164 )165 return rows166167168def _metrics(core, task_id, start_step, reward_sum):169 deliver = [task for task in core.tasks if int(task.type) == 1]170 completed = sum(bool(task.completed) for task in deliver)171 weighted_total = sum(max(0.0, float(task.req_vec[0])) for task in deliver)172 weighted_done = sum(173 max(0.0, float(task.req_vec[0]))174 for task in deliver175 if task.completed176 )177 target = core.tasks[int(task_id)]178 return {179 "success": int(completed == len(deliver)),180 "tasks_completed": int(completed),181 "task_completion_ratio": float(completed / max(1, len(deliver))),182 "priority_weighted_completion": float(183 weighted_done / max(1e-8, weighted_total)184 ),185 "remaining_demand": float(186 sum(max(0.0, float(task.demand)) for task in deliver)187 ),188 "event_task_completed": int(target.completed),189 "event_task_response_delay": (190 int(target.first_assigned_step - start_step)191 if target.first_assigned_step is not None192 else -1193 ),194 "episode_length": int(core.step_count),195 "team_reward_after_event": float(reward_sum),196 }197198199@torch.no_grad()200def _continue(201 config, actor, env, task_id, obs, active, available, rnn, masks, last, first202):203 core = getattr(env, "env", env)204 actions = np.asarray(first).copy()205 next_rnn = np.asarray(rnn).copy()206 reward_sum = 0.0207 start_step = int(core.step_count)208 while int(core.step_count) < int(config.max_steps):209 obs, rewards, dones, _, active, _, available = env.step(actions)210 reward_sum += float(np.asarray(rewards, dtype=np.float64).sum())211 done = np.asarray(dones, dtype=bool)212 next_rnn[done] = 0.0213 masks = np.ones((config.num_agents, 1), dtype=np.float32)214 masks[done] = 0.0215 rnn = next_rnn216 last = actions217 if bool(np.all(done)):218 break219 _, actions, next_rnn = _policy_step(220 actor, obs, rnn, masks, available, active, last221 )222 return _metrics(core, task_id, start_step, reward_sum)223224225@torch.no_grad()226def _run_model(model, manifest_dir, output_dir, scenarios, cuda, event_step):227 name = str(model["name"])228 config, runner, envs, checkpoint = _make_runner(229 model, manifest_dir, output_dir, cuda230 )231 actor = runner.policy.actor232 env = envs.envs[0]233 rows = []234 try:235 for scenario_id in scenarios:236 seed = int(config.eval_seed) + int(scenario_id)237 np.random.seed(seed)238 torch.manual_seed(seed)239 if config.device.type == "cuda":240 torch.cuda.manual_seed_all(seed)241 env.seed(seed)242 obs, active, available = env.reset(scenario_id)243 core = getattr(env, "env", env)244 deliver_ids = [int(t.id) for t in core.tasks if int(t.type) == 1]245 task_id = int(deliver_ids[0])246 for task in core.tasks:247 if int(task.type) == 1:248 task.discovered = True249 task.discovered_step = 0250 _set_reserved_visibility(core, task_id, False)251 obs = core._get_observations()252 available = core._get_available_actions()253 active = np.ones((config.num_agents, 1), dtype=np.float32)254 rnn = np.zeros(255 (config.num_agents, config.recurrent_N, config.hidden_size),256 dtype=np.float32,257 )258 masks = np.ones((config.num_agents, 1), dtype=np.float32)259 last = np.full(260 (config.num_agents, 1), int(core.idle_task_id), dtype=np.int64261 )262 valid = True263 while int(core.step_count) < int(event_step):264 _, actions, next_rnn = _policy_step(265 actor, obs, rnn, masks, available, active, last266 )267 obs, _, dones, _, active, _, available = env.step(actions)268 rnn = next_rnn269 last = actions270 if core.tasks[task_id].discovered:271 _set_reserved_visibility(core, task_id, False)272 core.event_new_task_found = 0273 obs = core._get_observations()274 available = core._get_available_actions()275 all_still = all(276 (not agent.living) or bool(agent.idle)277 for agent in core.agents278 )279 active = core._get_reconfiguration_active_mask(280 bool(core.event_any_task_completed), all_still281 )[:, None]282 core._current_decision_mask = active[:, 0].copy()283 if bool(np.all(dones)):284 valid = False285 break286 if not valid:287 rows.append(288 {289 "model": name,290 "model_role": model.get("role", "unknown"),291 "ablation": model.get("ablation", "none"),292 "scenario_id": int(scenario_id),293 "scenario_seed": seed,294 "condition": "skipped_early_terminal",295 }296 )297 continue298299 base = copy.deepcopy(env)300 pre_hash = _physical_state_hash(core)301 for condition, priority, defer in (302 ("high", 1.0, False),303 ("low", 0.5, False),304 ("high_keep_current_at_event", 1.0, True),305 ):306 branch = copy.deepcopy(base)307 branch_core = getattr(branch, "env", branch)308 branch_obs, branch_active, branch_available = _event_observation(309 branch, task_id, priority310 )311 distribution, policy_actions, next_rnn = _policy_step(312 actor,313 branch_obs,314 rnn,315 masks,316 branch_available,317 branch_active,318 last,319 )320 probs = distribution.probs.detach().cpu().numpy()321 executed = np.asarray(last if defer else policy_actions).copy()322 source_rows = _source_departure_diagnostics(323 branch_core, policy_actions324 )325 before = [int(agent.coalition) for agent in branch_core.agents]326 immediate_switches = sum(327 int(policy_actions[i, 0]) != before[i]328 for i in range(config.num_agents)329 if branch_core.agents[i].living330 )331 immediate_target = sum(332 int(policy_actions[i, 0]) == task_id333 for i in range(config.num_agents)334 if branch_core.agents[i].living335 )336 outcome = _continue(337 config,338 actor,339 branch,340 task_id,341 branch_obs,342 branch_active,343 branch_available,344 rnn,345 masks,346 last,347 executed,348 )349 rows.append(350 {351 "schema_version": SCHEMA_VERSION,352 "model": name,353 "model_role": model.get("role", "unknown"),354 "training_condition": model.get(355 "training_condition", "unknown"356 ),357 "ablation": model.get("ablation", "none"),358 "train_seed": int(model.get("train_seed", -1)),359 "checkpoint_role": model.get(360 "checkpoint_kind", "validation-best"361 ),362 "checkpoint": str(checkpoint),363 "scenario_id": int(scenario_id),364 "scenario_seed": seed,365 "event_step": int(event_step),366 "condition": condition,367 "priority": priority,368 "pre_event_state_hash": pre_hash,369 "mean_event_task_probability": float(370 probs[:, task_id].mean()371 ),372 "max_event_task_probability": float(373 probs[:, task_id].max()374 ),375 "immediate_target_assignments": immediate_target,376 "immediate_policy_switches": immediate_switches,377 "departures_from_fragile_sources": int(378 sum(row["source_fragile"] for row in source_rows)379 ),380 "departures_from_delivery_sources": len(source_rows),381 **outcome,382 }383 )384 print(f"{name} scenario={scenario_id} complete", flush=True)385 finally:386 _close_runner(runner, envs)387 del runner, envs388 gc.collect()389 if config.device.type == "cuda":390 torch.cuda.empty_cache()391 _write_csv(output_dir / f"{name}.csv", rows)392 return rows393394395def _summarize(rows):396 grouped = {}397 for row in rows:398 if not str(row.get("condition", "")).startswith(("high", "low")):399 continue400 key = (row["model"], row["condition"])401 grouped.setdefault(key, []).append(row)402 result = []403 metrics = (404 "mean_event_task_probability",405 "immediate_target_assignments",406 "immediate_policy_switches",407 "departures_from_fragile_sources",408 "success",409 "task_completion_ratio",410 "priority_weighted_completion",411 "event_task_completed",412 "event_task_response_delay",413 "remaining_demand",414 "episode_length",415 )416 for (model, condition), items in sorted(grouped.items()):417 row = {"model": model, "condition": condition, "n": len(items)}418 for metric in metrics:419 values = [float(item[metric]) for item in items if metric in item]420 row[metric] = float(np.mean(values)) if values else None421 result.append(row)422 return result423424425def _paired_effects(rows):426 indexed = {427 (row.get("model"), row.get("scenario_id"), row.get("condition")): row428 for row in rows429 }430 metrics = (431 "mean_event_task_probability",432 "immediate_target_assignments",433 "immediate_policy_switches",434 "departures_from_fragile_sources",435 "success",436 "task_completion_ratio",437 "priority_weighted_completion",438 "event_task_completed",439 "remaining_demand",440 "episode_length",441 "team_reward_after_event",442 )443 result = []444 pairs = (445 ("high_minus_low", "high", "low"),446 (447 "react_minus_keep_current",448 "high",449 "high_keep_current_at_event",450 ),451 )452 for model, scenario_id, condition in sorted(indexed):453 if condition != "high":454 continue455 for contrast, treatment, control in pairs:456 left = indexed.get((model, scenario_id, treatment))457 right = indexed.get((model, scenario_id, control))458 if left is None or right is None:459 continue460 item = {461 "model": model,462 "scenario_id": scenario_id,463 "contrast": contrast,464 "pre_event_state_hash": left.get("pre_event_state_hash"),465 }466 for metric in metrics:467 item[metric] = float(left[metric]) - float(right[metric])468 result.append(item)469 return result470471472def _worker(args):473 manifest_path = Path(args.manifest).resolve()474 manifest = yaml.safe_load(manifest_path.read_text(encoding="utf-8")) or {}475 model = next(item for item in manifest["models"] if item["name"] == args.model)476 output = Path(args.output).resolve()477 output.mkdir(parents=True, exist_ok=True)478 _run_model(479 model,480 manifest_path.parent,481 output,482 _parse_scenarios(args.scenarios),483 args.cuda,484 args.event_step,485 )486487488def main():489 parser = argparse.ArgumentParser()490 parser.add_argument("--manifest", required=True)491 parser.add_argument("--output", required=True)492 parser.add_argument("--scenarios", default="0:32")493 parser.add_argument("--event-step", type=int, default=150)494 parser.add_argument("--cuda", type=int, default=1)495 parser.add_argument("--max-workers", type=int, default=4)496 parser.add_argument("--model")497 args = parser.parse_args()498 if args.model:499 _worker(args)500 return501502 manifest_path = Path(args.manifest).resolve()503 manifest = yaml.safe_load(manifest_path.read_text(encoding="utf-8")) or {}504 output = Path(args.output).resolve()505 output.mkdir(parents=True, exist_ok=True)506 status_path = output / "status.json"507 status = {508 "name": manifest.get("name", "priority-reconfiguration"),509 "manifest": str(manifest_path),510 "created_at": datetime.now().isoformat(timespec="seconds"),511 "state": "running",512 "jobs": {513 item["name"]: {514 "state": "pending",515 "role": item.get("role", "unknown"),516 "training_condition": item.get("training_condition", "unknown"),517 "ablation": item.get("ablation", "none"),518 }519 for item in manifest["models"]520 },521 }522 _atomic_json(status_path, status)523 status_lock = threading.Lock()524525 command_base = [526 sys.executable,527 str(Path(__file__).resolve()),528 "--manifest",529 str(manifest_path),530 "--output",531 str(output),532 "--scenarios",533 args.scenarios,534 "--event-step",535 str(args.event_step),536 "--cuda",537 str(args.cuda),538 ]539540 def launch(name):541 with status_lock:542 status["jobs"][name]["state"] = "running"543 status["jobs"][name]["started_at"] = datetime.now().isoformat(544 timespec="seconds"545 )546 _atomic_json(status_path, status)547 log = output / f"{name}.log"548 with log.open("w", encoding="utf-8") as stream:549 return subprocess.run(550 command_base + ["--model", name],551 cwd=REPO_ROOT,552 stdout=stream,553 stderr=subprocess.STDOUT,554 env={**os.environ, "PYTHONUNBUFFERED": "1"},555 check=False,556 ).returncode557558 with ThreadPoolExecutor(max_workers=max(1, args.max_workers)) as pool:559 futures = {}560 for item in manifest["models"]:561 name = item["name"]562 futures[pool.submit(launch, name)] = name563 for future in as_completed(futures):564 name = futures[future]565 code = int(future.result())566 with status_lock:567 status["jobs"][name]["state"] = (568 "completed" if code == 0 else "failed"569 )570 status["jobs"][name]["return_code"] = code571 status["jobs"][name]["finished_at"] = datetime.now().isoformat(572 timespec="seconds"573 )574 _atomic_json(status_path, status)575576 all_rows = []577 for item in manifest["models"]:578 path = output / f"{item['name']}.csv"579 if path.exists():580 with path.open(encoding="utf-8", newline="") as stream:581 all_rows.extend(csv.DictReader(stream))582 _write_csv(output / "all_results.csv", all_rows)583 summary = _summarize(all_rows)584 _write_csv(output / "summary.csv", summary)585 _write_csv(output / "paired_effects.csv", _paired_effects(all_rows))586 status["state"] = (587 "completed"588 if all(job["state"] == "completed" for job in status["jobs"].values())589 else "partial_failure"590 )591 status["finished_at"] = datetime.now().isoformat(timespec="seconds")592 _atomic_json(status_path, status)593594595if __name__ == "__main__":596 main()