import copy import numpy as np import pickle import os from collections import deque import yaml eps = 1e-6 from enum import IntEnum class AgentType(IntEnum): Drone = 0 Vehicle = 1 class TaskType(IntEnum): Explore = 0 Deliver = 1 Idle = 2 Resupply = 3 def dist_point_segment(p, a, b): # 点到线段的最短距离；用于判断路径线段是否穿过感知范围或危险圆。 p = np.array(p, dtype=float) a = np.array(a, dtype=float) b = np.array(b, dtype=float) ab = b - a if np.allclose(ab, 0): return np.linalg.norm(p - a) t = np.dot(p - a, ab) / np.dot(ab, ab) t = np.clip(t, 0.0, 1.0) proj = a + t * ab return np.linalg.norm(p - proj) def fast_path(start, goal, danger_points=None, danger_radius=1.0, clearance=0.5, max_iter=8): '''连续空间近似路径规划器。 先尝试直线到达目标；如果路径线段穿过危险圆，则在障碍物附近插入垂直偏移路点绕行。 返回值包含起点和终点的路点列表。 ''' if danger_points is None: danger_points = [] def segment_intersects_circle(a, b, c, r): return dist_point_segment(c, a, b) <= r s = tuple(map(float, start)) g = tuple(map(float, goal)) if np.allclose(s, g): return [s] waypoints = [s, g] iter_count = 0 # 逐段检查路径；某段碰到危险圆时，就在该段中插入一个绕行点。 while iter_count < max_iter: changed = False new_waypoints = [waypoints[0]] for i in range(len(waypoints)-1): a = waypoints[i] b = waypoints[i+1] # 只处理第一个相交障碍，插入路点后下一轮再继续检查。 hit = None for c in danger_points: if segment_intersects_circle(a, b, c, danger_radius + clearance): hit = c break if hit is None: new_waypoints.append(b) continue # 绕行点取“障碍物到线段最近点”的垂直方向偏移。 a_np = np.array(a); b_np = np.array(b); c_np = np.array(hit) ab = b_np - a_np if np.allclose(ab, 0): # 线段退化时没有稳定垂线方向，直接使用一个固定侧向偏移。 dir_perp = np.array([1.0, 0.0]) else: t = np.dot(c_np - a_np, ab) / np.dot(ab, ab) t = np.clip(t, 0.0, 1.0) closest = a_np + t * ab # 正常线段使用垂直方向生成绕行候选点。 dir_perp = np.array([-ab[1], ab[0]]) if np.allclose(dir_perp, 0): dir_perp = np.array([1.0, 0.0]) dir_perp = dir_perp / np.linalg.norm(dir_perp) # 通过比较两侧候选点离其他危险圆的距离，选择更安全的一侧。 # 左右两侧各给一个候选绕行点，选择碰撞风险较小的一侧。 det_dist = danger_radius + clearance cand1 = tuple((closest + dir_perp * det_dist).tolist()) if 'closest' in locals() else tuple((a_np + dir_perp * det_dist).tolist()) cand2 = tuple((closest - dir_perp * det_dist).tolist()) if 'closest' in locals() else tuple((a_np - dir_perp * det_dist).tolist()) # 分数越小表示越不容易贴近其他危险圆。 def collision_score(pt): ssum = 0.0 for c2 in danger_points: d = np.linalg.norm(np.array(pt) - np.array(c2)) - danger_radius ssum += max(0.0, (clearance - d)) return ssum s1 = collision_score(cand1) s2 = collision_score(cand2) detour = cand1 if s1 <= s2 else cand2 # 在当前线段终点前插入绕行点。 new_waypoints.append(detour) new_waypoints.append(b) changed = True waypoints = new_waypoints iter_count += 1 if not changed: break # 去掉几乎重合的路点，避免移动时产生零长度线段。 out = [tuple(map(float, waypoints[0]))] for p in waypoints[1:]: if np.linalg.norm(np.array(p) - np.array(out[-1])) > eps: out.append(tuple(map(float, p))) return out class AgentLastState: """保存上一步智能体状态，奖励函数用它计算切换、距离变化和资源变化。""" __slots__ = ['coalition', 'new_task_found', 'current_task_completed', 'any_task_completed', 'coalition_completed', 'dist', 'load', 'idle'] def __init__(self): self.coalition = EnvCore.idle_task_id self.new_task_found = False self.current_task_completed = False self.any_task_completed = False self.coalition_completed = False self.dist = 0.0 self.load = 0.0 self.idle = True class Agent: """环境内部的智能体实体。 capbilities 四维分别驱动感知、移动、载荷和电量上限；每步事件字段在 step 末尾清零。 """ __slots__ = ['aid', 'position', 'capbilities', 'perception', 'mobility', 'load', 'battery', 'type', 'living', 'idle', 'coalition', 'path', 'last_state', 'dead', 'new_task_found', 'coalition_changed', 'discovery', 'delivery', 'coalition_delivery', 'resupply_ratio', 'init_load', 'resupplied', 'current_task_completed', 'any_task_completed', 'coalition_completed', 'dist', 'unchanged_time'] def __init__(self, aid: int, pos: np.ndarray, capbilities: np.ndarray, rng=None): self.aid = aid self.position = np.array(pos, dtype=float) self.capbilities = capbilities self.perception = float(capbilities[0] * 50.0) self.mobility = float(capbilities[1] * 10) + 10 self.init_load = int(capbilities[2] * 70) + 30 self.load = self.init_load self.battery = int(capbilities[3] * 500) + 500 self.type = rng.choice([AgentType.Drone, AgentType.Vehicle]) if rng is not None else np.random.choice([AgentType.Drone, AgentType.Vehicle]) self.living = True self.idle = True self.coalition = EnvCore.idle_task_id self.path = [] self.dead = False self.new_task_found = False self.current_task_completed = False self.any_task_completed = False self.coalition_completed = False self.coalition_changed = False self.discovery = 0.0 self.delivery = 0.0 self.coalition_delivery = 0.0 self.dist = 0.0 self.unchanged_time = 0.0 self.resupply_ratio = 0.0 self.resupplied = False self.last_state = AgentLastState() def act(self, cid: int): # 只有联盟编号变化时才清空路径；保持同一联盟会累积 unchanged_time。 if cid != self.coalition: self.coalition = cid self.coalition_changed = True self.path = [] self.unchanged_time = 0.0 else: self.unchanged_time += 1.0 if cid == EnvCore.idle_task_id: self.idle = True def consume(self): # 当前模型按步扣电，电量耗尽后变为死亡，死亡事件只在本步奖励中使用。 if self.living: self.battery -= 1 if self.battery == 0: self.living = False self.dead = True def deliver(self, demand): # 交付量受当前载荷和任务剩余需求共同限制；无货交付会给负事件信号。 delivery = np.min([self.load, demand]) self.load -= delivery self.delivery = delivery if delivery > 0 else -10 self.idle = True if self.delivery < 0: pass return delivery def complete(self): # 当前智能体完成自己负责的当前动作，runner 会据此决定是否再次激活该 agent。 self.current_task_completed = True self.idle = True def resupply(self): # 补给恢复电量和载荷，并记录本次补给比例用于奖励。 self.living = True self.dead = False self.idle = True self.battery = int(self.capbilities[3] * 500) + 500 self.resupply_ratio = (self.init_load - self.load) / self.init_load self.resupplied = True self.load = self.init_load def step(self): # 先把当前状态拷贝到 last_state，再清空“本步事件”字段。 # 这样奖励函数可以同时访问上一步状态和本步事件。 for key in self.last_state.__slots__: attr = getattr(self, key) setattr(self.last_state, key, attr) self.dead = False self.new_task_found = False self.current_task_completed = False self.any_task_completed = False self.coalition_completed = False self.coalition_changed = False self.discovery = 0.0 self.delivery = 0.0 self.coalition_delivery = 0.0 self.resupply_ratio = 0.0 self.resupplied = False self.idle = False class Task: """环境内部任务实体。 Deliver 任务初始不可见；Explore/Idle/Resupply 是辅助任务槽，始终可见。 """ __slots__ = [ 'id', 'cap_dim', 'position', 'demand', 'initial_demand', 'urgency', 'req_vec', 'req_ratio', 'type', 'discovered', 'completed', 'discovered_step', 'first_assigned_step', 'response_time' ] def __init__(self, cid: int, pos: np.ndarray, demand: int, urgency: float, cap_dim: int, rng=None): self.id = cid self.cap_dim = cap_dim self.position = np.array(pos, dtype=float) self.demand = demand self.initial_demand = demand self.urgency = urgency self.req_vec = ((rng.rand(cap_dim) if rng is not None else np.random.rand(cap_dim)) * (demand / 100.0)).astype(float) self.req_ratio = np.zeros(cap_dim) self.type = TaskType.Deliver self.discovered = False self.completed = False self.discovered_step = None self.first_assigned_step = None self.response_time = None def calc_req_ratio(self, cap_now): # req_ratio 是当前联盟能力相对需求的比例差，只用于 info/诊断，不直接进入局部观测。 if self.discovered and not self.completed: req_vec = self.req_vec for dim in range(self.cap_dim): if np.isnan(req_vec[dim]) or req_vec[dim] <= 0.0: self.req_ratio[dim] = 0.0 else: self.req_ratio[dim] = (cap_now[dim] - req_vec[dim]) / req_vec[dim] else: self.req_ratio = np.zeros(self.cap_dim) class EnvCore: """MACF 环境核心。 负责连续地图、智能体/任务状态推进、局部 Dict 观测、全局 critic state、奖励和终止条件。 Gym wrapper 只负责把这里的返回值包装成向量环境接口。 """ explore_task_id = -1 idle_task_id = -1 resupply_task_id = -1 def __init__(self, config): # scenario_name 决定是否启用 coalition_delivery 专用任务特征和奖励。 scenario_name = getattr(config, "scenario_name", None) if scenario_name in (None, "", "MyEnv"): scenario_name = getattr(config, "env_name", "macf") self.scenario_name = scenario_name self.use_coalition_delivery = self.scenario_name == "coalition_delivery" self.coalition_delivery_reward_mode = getattr( config, "coalition_delivery_reward_mode", "baseline" ) if self.coalition_delivery_reward_mode not in ("baseline", "shaped"): raise ValueError( "coalition_delivery_reward_mode must be 'baseline' or 'shaped', " f"got {self.coalition_delivery_reward_mode!r}" ) deliver_target_num = config.target_num # 探索动作可以是单一 nearest，也可以拆成多个 sector/survey 动作槽。 self.explore_action_mode = getattr(config, "explore_action_mode", "nearest") if self.explore_action_mode not in ("nearest", "sector", "survey"): raise ValueError(f"Unsupported explore_action_mode: {self.explore_action_mode}") self.num_explore_actions = ( int(getattr(config, "num_explore_actions", 1)) if self.explore_action_mode in ("sector", "survey") else 1 ) if self.num_explore_actions <= 0: raise ValueError("num_explore_actions must be positive.") self._configured_explore_step_size = getattr(config, "explore_step_size", None) self.explore_random_jitter = float(getattr(config, "explore_random_jitter", 0.0)) self.explore_task_start = deliver_target_num self.explore_task_end = deliver_target_num + self.num_explore_actions # 任务槽顺序：真实配送任务 [0, target_num)，探索槽，Idle 槽，Resupply 槽。 self.__class__.explore_task_id = self.explore_task_start self.__class__.idle_task_id = self.explore_task_end self.__class__.resupply_task_id = self.explore_task_end + 1 # 地图是连续二维空间，危险区域用圆心 + 半径表示。 self.map_size = float(config.map_size) # 连续地图边长 / m self.explore_step_size = ( float(self._configured_explore_step_size) if self._configured_explore_step_size is not None else self.map_size * 0.25 ) # 危险圆几何参数；clearance 是绕行路径额外安全距离。 self.danger_radius = float(getattr(config, 'danger_radius', 1.0)) self.clearance = float(getattr(config, 'clearance', 0.5)) # 仍保留 UAV/UGV 数量字段；总智能体数由二者相加得到。 self.uav_num = config.num_uav # 无人机数量 self.ugv_num = config.num_ugv # 无人车数量 self.uav_perception_rg = tuple(config.uav_perception_rg[0:2]) self.ugv_perception_rg = tuple(config.ugv_perception_rg[0:2]) self.delivery_task_num = config.target_num self.task_num = config.target_num + self.num_explore_actions + 2 self.gat_dim = config.gat_dim # GAT 输出维度 self.max_steps = config.max_steps # 最大步数 self.cap_dim = config.cap_dim # 能力维度 self.reconfiguration_mode = str( getattr(config, "reconfiguration_mode", "legacy") ) valid_reconfiguration_modes = { "legacy", "always", "periodic", "event", "urgency", "learned" } if self.reconfiguration_mode not in valid_reconfiguration_modes: raise ValueError( "reconfiguration_mode must be one of " f"{sorted(valid_reconfiguration_modes)}, got " f"{self.reconfiguration_mode!r}" ) self.reconfiguration_interval = int( getattr(config, "reconfiguration_interval", 10) ) if self.reconfiguration_interval <= 0: raise ValueError("reconfiguration_interval must be positive.") self.reconfiguration_urgency_threshold = float( getattr(config, "reconfiguration_urgency_threshold", 0.7) ) self.reconfiguration_urgency_aggregation = str( getattr(config, "reconfiguration_urgency_aggregation", "mean") ) if self.reconfiguration_urgency_aggregation not in ("mean", "sum"): raise ValueError( "reconfiguration_urgency_aggregation must be 'mean' or 'sum'." ) self.reconfiguration_switch_cost = float( getattr(config, "reconfiguration_switch_cost", 0.0) ) if self.reconfiguration_switch_cost < 0.0: raise ValueError("reconfiguration_switch_cost must be non-negative.") self.reward_keys = ['switch_coalition', 'dead', 'discover', 'deliver', 'resupply', 'coalition_deliver', 'coalition_complete', 'dist', 'end'] # 总智能体数，后续 obs/buffer/policy 都以该值为 agent 维度。 self.agent_num = self.uav_num + self.ugv_num self.step_count = 0 # 当前步数 # ego 观测：位置(2)、能力(cap_dim)、当前联盟编号、当前联盟类型、 # 载荷比例、已发现配送任务比例、电量比例。 self.ego_feat_dim = self.cap_dim + 7 if self.use_coalition_delivery: self.coalition_task_feature_mode = str( getattr(config, "coalition_task_feature_mode", "compact") ) if self.coalition_task_feature_mode not in ("compact", "full"): raise ValueError( "coalition_task_feature_mode must be 'compact' or 'full', got " f"{self.coalition_task_feature_mode!r}" ) if self.coalition_task_feature_mode == "full": # 位置、完整能力需求、配送量、期限、探索几何(3)、任务类型。 self.per_task_feat = 2 + self.cap_dim + 1 + 1 + 3 + 1 else: # 历史 6-D schema：位置、配送量、期限、req_vec[0]、任务类型。 self.per_task_feat = 2 + 1 + 1 + 1 + 1 else: self.coalition_task_feature_mode = "full" # 默认任务观测：位置、能力需求、剩余需求、任务类型。 self.per_task_feat = 2 + self.cap_dim + 1 + 1 # other 观测：位置、能力、载荷、电量、联盟编号、智能体类型、存活标志。 self.other_agent_feat_dim = 2 + self.cap_dim + 1 + 1 + 1 + 1 + 1 self.global_agent_feat_dim = self.other_agent_feat_dim # critic 的全局任务状态额外暴露 discovered/completed 标志。 self.global_task_feat_dim = self.per_task_feat + 2 self.global_feat_dim = 1 # flatten 后的 actor 观测维度；Dict 观测实际仍按 ego/tasks/others 三个 key 返回。 self.obs_dim = ( self.ego_feat_dim + self.task_num * self.per_task_feat + self.agent_num * self.other_agent_feat_dim ) # 动作维度等于任务槽数。Idle/Resupply/Explore 都已经是任务槽，不再额外增加动作。 self.action_dim = self.task_num # 事件标志只描述当前 step 内发生的关键事件，奖励和 active_mask 会读取它们。 self.event_new_task_found = 0 self.event_task_completed = [0 for i in range(self.agent_num)] self.event_any_task_completed = 0 self.data_dir = 'data' if config.rewards_config_file: with open(config.rewards_config_file, 'r', encoding='utf-8') as f: rewards_config = yaml.safe_load(f) or {} for key, value in rewards_config.items(): setattr(self, key, value) else: self.pun_change_coalition_event = 0.001 self.pun_change_coalition = 1 self.w_short = 2.0 # 能力不足权重（重罚） self.w_surplus = 0.3 # 能力过多权重（轻罚） self.b_short = 8.0 # 能力不足权重（重罚） self.b_surplus = 20.0 # 能力过多权重（轻罚） self.rwd_ratio_discover_task = 5 self.rwd_ratio_discover_dp = 1.0 self.pun_ag_dead = 10 self.rwd_ratio_send = 1 self.rwd_coalition_ratio_send = 0.2 self.rwd_task_completed = 200 self.rwd_coalition_task_completed = 50 self.rwd_dist_ratio = 3.0 self.pun_time_ratio = 0.1 # legacy GAT 的全局图状态占位；当前 HATT/MAPPO 主要使用 Dict state。 self.global_state = np.zeros((self.agent_num, self.task_num, self.gat_dim)) # reset 相关缓存：jitter 模式复用基础场景，只扰动配送任务位置。 self._seed = None self.rng = None self._base_initialized = False self._base_danger_points = None self._base_tasks = None self._base_agents = None self.task_position_jitter = float(getattr(config, 'task_position_jitter', 0.05)) self.env_reset_mode = getattr(config, 'env_reset_mode', 'jitter') def seed(self, seed): # 设置种子后清空基础场景缓存，保证下一次 reset 从该种子重新生成。 self._seed = int(seed) self.rng = np.random.RandomState(self._seed) self._base_initialized = False self._base_danger_points = None self._base_tasks = None self._base_agents = None return [self._seed] def _rand(self, *args): if self.rng is not None: return self.rng.rand(*args) return np.random.rand(*args) def _randint(self, *args, **kwargs): if self.rng is not None: return self.rng.randint(*args, **kwargs) return np.random.randint(*args, **kwargs) def _choice(self, a): if self.rng is not None: return self.rng.choice(a) return np.random.choice(a) def _explore_sector_goal(self, ag): # sector 模式下，探索任务不是固定点，而是从当前位置朝某个扇区方向推进。 sector = int(ag.coalition - self.explore_task_start) sector = max(0, min(self.num_explore_actions - 1, sector)) angle, sector_width = self._explore_sector_angle(sector) if self.explore_random_jitter > 0.0: jitter = (float(self._rand()) * 2.0 - 1.0) * self.explore_random_jitter * sector_width angle += jitter direction = np.array([np.cos(angle), np.sin(angle)], dtype=float) goal = ag.position + direction * self.explore_step_size return np.clip(goal, 0.0, self.map_size) def _explore_sector_angle(self, sector): sector_width = 2.0 * np.pi / float(self.num_explore_actions) return sector * sector_width, sector_width def _explore_sector_anchor(self, sector): angle, _ = self._explore_sector_angle(sector) direction = np.array([np.cos(angle), np.sin(angle)], dtype=float) center = np.array([0.5 * self.map_size, 0.5 * self.map_size], dtype=float) return np.clip(center + direction * (0.45 * self.map_size), 0.0, self.map_size) def _survey_anchor(self, survey_idx): # survey 模式将地图划分成近似网格，每个探索动作对应一个固定巡查锚点。 cols = int(np.ceil(np.sqrt(self.num_explore_actions))) rows = int(np.ceil(self.num_explore_actions / cols)) col = survey_idx % cols row = survey_idx // cols x = (col + 1.0) / (cols + 1.0) * self.map_size y = (row + 1.0) / (rows + 1.0) * self.map_size return np.array([x, y], dtype=float) def _is_explore_task_id(self, task_id): # 判断某个任务槽是否属于探索动作槽。 return self.explore_task_start <= int(task_id) < self.explore_task_end def _save_base_env(self): # jitter reset 的基础场景快照，包括危险点、任务和智能体初始状态。 self._base_danger_points = copy.deepcopy(self.danger_points) self._base_tasks = copy.deepcopy(self.tasks) self._base_agents = copy.deepcopy(self.agents) def _restore_base_env(self): # 从基础场景快照恢复，避免每个 episode 完全重采样。 self.danger_points = copy.deepcopy(self._base_danger_points) self.tasks = copy.deepcopy(self._base_tasks) self.agents = copy.deepcopy(self._base_agents) def _jitter_task_positions(self): # 只扰动配送任务位置；探索/空闲/补给槽保持固定语义。 if self.task_position_jitter <= 0.0: return scale = self.task_position_jitter * self.map_size for task in self.tasks: if task.type != TaskType.Deliver: continue base_pos = task.position.copy() for _ in range(20): offset = self._rand(2) * 2.0 * scale - scale new_pos = np.clip(base_pos + offset, 0.0, self.map_size) if not any( np.linalg.norm(new_pos - np.array(dp['position'])) <= (self.danger_radius + self.clearance) for dp in self.danger_points ): task.position = new_pos break else: task.position = np.clip(base_pos, 0.0, self.map_size) def _init_map(self): # 连续地图：只记录危险点圆心 self.danger_points = [] # 危险点数量目前固定，历史代码里保留过按面积缩放的方案。 num_danger = 10#max(1, int(self.map_size * self.map_size * 0.002)) for _ in range(num_danger): p = (self._rand(2) * self.map_size).astype(float) self.danger_points.append({ 'position': p, 'discovered': False }) def _init_agents(self): # 随机生成智能体能力和位置；位置不能落在危险圆安全范围内。 self.agents: list[Agent] = [] for i in range(self.agent_num): capbilities = self._rand(self.cap_dim) # 采样连续位置，避开危险点中心 pos = self._rand(2) * self.map_size # 保证不在危险半径内 tries = 0 while any(np.linalg.norm(pos - np.array(dp['position'])) <= (self.danger_radius + self.clearance) for dp in self.danger_points) and tries < 50: pos = self._rand(2) * self.map_size tries += 1 agent = Agent(i, pos, capbilities, rng=self.rng) self.agents.append(agent) def _init_tasks(self): '''随机放置任务，并追加探索、空闲、补给三个类别的辅助任务槽。''' self.tasks: list[Task] = [] # 先生成所有任务槽，后面会覆盖探索/空闲/补给槽。 for i in range(self.task_num): demand_val = int(self._randint(100, 500)) # 50–100 pos = self._rand(2) * self.map_size tries = 0 while any(np.linalg.norm(pos - np.array(dp['position'])) <= (self.danger_radius + self.clearance) for dp in self.danger_points) and tries < 200: pos = self._rand(2) * self.map_size tries += 1 task = Task(i, np.array(pos, dtype=float), demand_val, self._rand(), self.cap_dim, rng=self.rng) if self.use_coalition_delivery and i < self.delivery_task_num: deadline_frac = 0.45 + 0.5 * float(self._rand()) priority = 0.5 + 0.5 * float(self._rand()) task.urgency = deadline_frac task.req_vec = np.array([priority, 0.0, 0.0, 0.0], dtype=float) self.tasks.append(task) # 覆盖探索槽。nearest 模式下探索槽位置为 NaN，step 时动态寻找最近未发现任务。 for sector, explore_id in enumerate(range(self.explore_task_start, self.explore_task_end)): if self.explore_action_mode == "survey": explore_pos = self._survey_anchor(sector) elif self.explore_action_mode == "sector": explore_pos = self._explore_sector_anchor(sector) else: explore_pos = np.array((np.nan, np.nan)) exploring_task = Task(explore_id, explore_pos, 0, 0.5, self.cap_dim) exploring_task.req_vec = np.array([0.25, 0, 0, 0], dtype=float) exploring_task.type = TaskType.Explore exploring_task.discovered = True self.tasks[explore_id] = exploring_task # Idle 槽始终可选，但没有空间目标。 idle_task = Task(self.__class__.idle_task_id, np.array((np.nan, np.nan)), 0, 0, self.cap_dim) idle_task.req_vec = np.array([0, 0, 0, 0]) idle_task.type = TaskType.Idle idle_task.discovered = True self.tasks[self.__class__.idle_task_id] = idle_task # Resupply 槽固定在原点，智能体到达后恢复载荷和电量。 resupply_task = Task(self.__class__.resupply_task_id, np.array((0, 0)), 0, 0, self.cap_dim) resupply_task.req_vec = np.array([0, 0, 0, 0]) resupply_task.type = TaskType.Resupply resupply_task.discovered = True self.tasks[self.__class__.resupply_task_id] = resupply_task def set_reset_mode(self, mode): """动态切换 reset 模式，课程学习会调用。""" assert mode in ("jitter", "full_random") self.env_reset_mode = mode def reset(self, episode=-1): self.step_count = 0 if self.env_reset_mode == "full_random": # 每个 episode 完全重采样地图、任务和智能体。 self._init_map() self._init_tasks() self._init_agents() elif not self._base_initialized: # jitter 模式第一次 reset 先生成基础场景并缓存。 self._init_map() self._init_tasks() self._init_agents() self._save_base_env() self._base_initialized = True else: # jitter 模式后续 episode 恢复基础场景，只扰动配送任务位置。 self._restore_base_env() self._jitter_task_positions() # sent_goal 包含所有任务槽的初始需求；辅助任务需求为 0，不影响总配送目标。 self.sent_goal = np.sum([task.initial_demand for task in self.tasks]) self.event_new_task_found = 0 self.event_task_completed = [0 for i in range(self.agent_num)] self.event_any_task_completed = 0 self.global_state = np.zeros((self.agent_num, self.task_num, self.gat_dim)) self.total_coalition_changes = 0 self.total_reconfiguration_decisions = 0 self._current_decision_mask = np.ones( (self.agent_num,), dtype=np.float32 ) self.total_alive_agent_steps = 0 self.response_times = [] if episode != -1: self.episode = episode observations = self._get_observations() active_mask = np.ones((self.agent_num,), dtype=np.float32) active_mask = active_mask[..., None] available_actions = self._get_available_actions() return observations, active_mask, available_actions#, info def _get_available_actions(self): # 当前策略只能选择“已发现且未完成”的任务槽。 # Explore/Idle/Resupply 在初始化时设置为 discovered=True，因此一般保持可选。 task_mask = np.fromiter( (1.0 if task.discovered and not task.completed else 0.0 for task in self.tasks), dtype=np.float32, count=self.task_num, ) return np.repeat(task_mask[None, :], self.agent_num, axis=0) def _reconfiguration_urgency_pressure(self): """Return the configured aggregate pressure of visible unfinished tasks. A task's pressure grows from ``1-deadline`` at reset to one at its deadline. ``mean`` is scale invariant and is the recommended setting; ``sum`` reproduces the scale-sensitive trigger proposed in the draft. """ time_ratio = self.step_count / max(1.0, float(self.max_steps)) pressures = [] for task in self.tasks: if ( task.type != TaskType.Deliver or not task.discovered or task.completed ): continue deadline_left = max(0.0, float(task.urgency) - time_ratio) pressures.append(1.0 - min(1.0, deadline_left)) if not pressures: return 0.0 if self.reconfiguration_urgency_aggregation == "sum": return float(np.sum(pressures)) return float(np.mean(pressures)) def _get_reconfiguration_active_mask(self, key_event, all_still): """Select agents that receive a fresh policy decision next step.""" living = np.array([ag.living for ag in self.agents], dtype=bool) completed = np.array( [ag.living and ag.current_task_completed for ag in self.agents], dtype=bool, ) if all_still: return living.astype(np.float32) mode = self.reconfiguration_mode if mode == "legacy": return np.array( [ 1.0 if ag.living and ( ag.current_task_completed or (ag.idle and key_event) ) else 0.0 for ag in self.agents ], dtype=np.float32, ) if mode == "always": return living.astype(np.float32) interval_due = self.step_count % self.reconfiguration_interval == 0 if mode == "periodic": reconsider_all = interval_due elif mode == "event": reconsider_all = key_event elif mode == "urgency": reconsider_all = ( self._reconfiguration_urgency_pressure() >= self.reconfiguration_urgency_threshold ) else: # learned: HATT gate filters periodic/event candidate windows. reconsider_all = interval_due or key_event return np.where(reconsider_all, living, completed).astype(np.float32) def discover_along_segment(self, agent_idx, a, b): # 沿移动线段检查感知范围，标记新发现的配送任务和危险点。 # 使用点到线段距离，而不是只看终点距离，避免高速移动穿过目标但漏检。 ag = self.agents[agent_idx] per = float(ag.perception) discovered_tasks = [] discovered_dp = 0 a = np.asarray(a, dtype=float) b = np.asarray(b, dtype=float) ab = b - a denom = float(np.dot(ab, ab)) def point_segment_distances(points): if len(points) == 0: return np.empty((0,), dtype=float) if denom <= eps: return np.linalg.norm(points - a[None, :], axis=1) t = ((points - a[None, :]) @ ab) / denom t = np.clip(t, 0.0, 1.0) proj = a[None, :] + t[:, None] * ab[None, :] return np.linalg.norm(points - proj, axis=1) task_ids = [ idx for idx, task in enumerate(self.tasks) if (not task.discovered) and task.type == TaskType.Deliver and not np.any(np.isnan(task.position)) ] if task_ids: task_positions = np.stack([self.tasks[idx].position for idx in task_ids]) discovered_mask = point_segment_distances(task_positions) <= per for idx in np.asarray(task_ids, dtype=np.int64)[discovered_mask]: task = self.tasks[int(idx)] task.discovered = True task.discovered_step = self.step_count discovered_tasks.append(int(idx)) danger_ids = [ idx for idx, dp in enumerate(self.danger_points) if not dp["discovered"] ] if danger_ids: danger_positions = np.stack( [self.danger_points[idx]["position"] for idx in danger_ids] ) discovered_mask = point_segment_distances(danger_positions) <= per for idx in np.asarray(danger_ids, dtype=np.int64)[discovered_mask]: self.danger_points[int(idx)]["discovered"] = True discovered_dp += 1 return discovered_tasks, discovered_dp def move_along_path(self, path_pts, start_pos, max_dist): # path_pts 是从当前位置到目标的路点列表；max_dist 是本步最多可移动距离。 # 返回新位置和剩余路径。剩余路径首点始终是新位置。 pos = np.array(start_pos, dtype=float) rem = [] # 保证路径从当前位置开始，避免上一帧路径首点和当前位置存在微小偏差。 pts = [tuple(pos)] + [tuple(p) for p in path_pts[1:]] if len(path_pts) > 0 else [tuple(pos)] dist_left = float(max_dist) i = 0 while dist_left > 1e-8 and i < len(pts)-1: a = np.array(pts[i], dtype=float) b = np.array(pts[i+1], dtype=float) seg_len = np.linalg.norm(b - a) if seg_len <= 1e-8: i += 1 continue need = min(seg_len - np.linalg.norm(pos - a), dist_left) # 如果位置不在线段开头，计算剩下在线段上还有多长 to_end = np.linalg.norm(b - pos) move = min(to_end, dist_left) if move >= to_end - 1e-8: # 到达当前线段终点，继续走下一段。 pos = b dist_left -= to_end i += 1 else: # 本步距离不足以到达线段终点，停在线段中间。 direction = (b - pos) / np.linalg.norm(b - pos) pos = pos + direction * move dist_left = 0.0 break # 剩余路径：从当前位置开始，然后接上未走完的后续路点。 remaining = [tuple(pos)] for j in range(i+1, len(pts)): remaining.append(tuple(pts[j])) return pos, remaining def step(self, actions): ''' 参数 ------- actions : List[int] 长度 = self.agent_num 每个元素为智能体选中的任务 ID 动作语义：0..(task_num-1) 对应任务 ID，task_num 表示 Idle（空闲，不选择任何任务）。 为兼容旧版，也接受 -1 表示 Idle。 返回 ------- observations : dict {'ego':..., 'tasks':..., 'others':...} rewards : np.ndarray [N_agents] dones : np.ndarray [N_agents] (全部任务完成或到 max_steps) info : dict 调试、评估和日志指标 ''' from datetime import datetime d = d0 = datetime.now().timestamp() self.step_count += 1 step_reconfiguration_decisions = int( np.count_nonzero(self._current_decision_mask > 0.5) ) self.total_reconfiguration_decisions += step_reconfiguration_decisions self.total_alive_agent_steps += int(sum(ag.living for ag in self.agents)) new_discovered = set() self.t = np.array([0, 0, 0, 0, 0], float) # 1. 联盟决策：动作直接写入每个智能体的 coalition。 # available_actions 在 wrapper/policy 侧限制可选任务；这里仍容忍越界动作并忽略。 for i, raw_a in enumerate(actions): # 兼容 raw_a 是 tensor 或普通 int 的情况 try: a = int(raw_a.item()) except Exception: a = int(raw_a) # 当前动作范围是 [0, task_num)，其中 Idle 已经是一个普通任务槽。 if a < 0 or a >= self.task_num: # 非法动作，忽略 continue self.agents[i].act(a) # 若选择非法任务 ID，忽略，保持原状态 step_coalition_changes = int( sum(ag.coalition_changed for ag in self.agents) ) self.total_coalition_changes += step_coalition_changes # 2. 首次响应时间：任务被发现后，第一次有活着的智能体加入该任务联盟时记录。 for task in self.tasks: if task.type != TaskType.Deliver: continue if not task.discovered or task.completed: continue if task.discovered_step is None: continue if task.first_assigned_step is not None: continue assigned = any( ag.living and ag.coalition == task.id for ag in self.agents ) if assigned: task.first_assigned_step = self.step_count task.response_time = task.first_assigned_step - task.discovered_step self.response_times.append(task.response_time) # 3. 清空本步事件标志；后续发现、完成、死亡等逻辑会重新写入。 self.event_new_task_found = 0 self.event_task_completed = [0 for i in range(self.agent_num)] self.event_any_task_completed = 0 task_caps = np.zeros((self.task_num, self.cap_dim)) coalition_sizes = np.zeros(self.task_num, dtype=int) coalitions = [[] for i in range(self.task_num)] # 4. 依据当前 coalition 聚合每个任务槽上的联盟成员能力和成员列表。 # 这份聚合既用于任务完成逻辑，也用于奖励/诊断中的能力匹配度。 for i, ag in enumerate(self.agents): c = ag.coalition if c is None or c == -1 or c == self.task_num: continue task_caps[c] += ag.capbilities coalition_sizes[c] += 1 coalitions[c].append(ag.aid) # 5. 联盟-任务能力匹配度诊断。 # 当前惩罚分摊代码仍保留但未实际写入 rewards，用于后续恢复或分析。 for j, task in enumerate(self.tasks): # 不考虑 Idle 任务和补给任务 if j == self.__class__.idle_task_id or j == self.__class__.resupply_task_id: continue # 只对 active 的、未完成的任务算匹配度 if coalition_sizes[j] == 0: continue if not task.completed: continue req = task.req_vec if req is None: continue cap = task_caps[j] min_val = 0.2 # 避免除零 denom = np.where((cap > -min_val) & (cap < min_val), min_val, cap) diff = cap - req shortage = np.clip(-diff, 0.0, None) # cap < req 的部分 surplus = np.clip( diff, 0.0, None) # cap > req 的部分 # 按平方惩罚，让严重不匹配的惩罚更大 penalty_j = ( self.w_short * (float(np.sum(shortage ** 2)) - self.b_short) + self.w_surplus * (float(np.sum(surplus ** 2)) - self.b_surplus) ) if penalty_j <= 0: continue # 把惩罚均摊给联盟内所有 agent share = penalty_j / float(coalition_sizes[j]) share = np.clip(share, -100, 2) # for i in coalitions[j]: # rewards[i] -= share d = datetime.now().timestamp() self.t[0] = d - d0 d0 = d # rewards_by_step[1] = rewards - rewards_by_step[0] # 6. 为没有剩余路径的非空闲智能体生成目标点和路径。 # 已有路径的智能体继续执行旧路径，避免每步重规划造成抖动。 danger_positions = [dp['position'] for dp in self.danger_points] for i, ag in enumerate(self.agents): # 死亡智能体不规划路径。 if not ag.living: continue # 未走完旧路径时不重新分配路径。 if len(ag.path) != 0: continue task = self.tasks[ag.coalition] # Idle 槽没有空间目标。 if task.type == TaskType.Idle: continue # 根据任务类型确定目标点：探索任务可动态选择，配送/补给使用任务位置。 if task.type == TaskType.Explore: if self.explore_action_mode == "survey": goal = np.array(task.position, dtype=float) elif self.explore_action_mode == "sector": goal = self._explore_sector_goal(ag) else: # nearest 探索：优先朝最近的未发现配送任务移动。 cand_tasks = [task for task in self.tasks if (not task.discovered) and (task.type == TaskType.Deliver)] if cand_tasks: # 按欧氏距离排序，选择最近未发现配送任务。 cand_tasks.sort(key=lambda task: np.linalg.norm(task.position - ag.position)) goal = tuple(cand_tasks[0].position) else: # 若没有未发现配送任务，则随机探索一个地图点。 goal = tuple((self._rand(2) * self.map_size)) goal = np.array(goal, dtype=float) else: # elif coalition.type == TaskType.Deliver or coalition.type == TaskType.Resupply: goal = task.position # 根据危险点生成绕行路径。 ag.path = fast_path(ag.position, goal, danger_positions, danger_radius=self.danger_radius, clearance=self.clearance) # 理论上 fast_path 很少返回空；若失败，本步该智能体不移动。 if len(ag.path) == 0: print(f'step {self.step_count}, agent {i}: cannot find path for exploration agent') # 7. 沿路径移动，并在移动线段上触发探索、交付和补给事件。 sent = 0 completed = 0 for i, ag in enumerate(self.agents): # 死亡智能体不移动。 if not ag.living: continue # 没有路径说明没有空间目标或规划失败。 if len(ag.path) == 0: continue task = self.tasks[ag.coalition] # Idle 智能体保持当前位置。 if task.type == TaskType.Idle: continue # 只要执行移动就消耗电量。 ag.consume() # mobility 决定本步沿路径最多前进多远。 start = ag.position new_pos, remaining = self.move_along_path(ag.path, start, ag.mobility) # 移动过程中可能发现新任务或危险点。 discovered_tasks, discovered_dp = self.discover_along_segment(i, start, new_pos) # 新发现任务会激活探索奖励，并触发 active_mask 让部分智能体重新决策。 if len(discovered_tasks) > 0: ag.discovery = len(discovered_tasks) + 0.1 * discovered_dp self.event_new_task_found = 1 for task_id in discovered_tasks: new_discovered.add(task_id) ag.position = np.array(new_pos, dtype=float) # 如果路径已经走到终点，则清空路径。 ag.path = remaining if len(remaining) > 1 else [] # 配送/补给任务记录到目标的距离，距离奖励依赖 last_state.dist - dist。 if task.type in [TaskType.Deliver, TaskType.Resupply]: dist = np.linalg.norm(ag.position - task.position) ag.dist = float(dist) if task.type == TaskType.Explore: if self.explore_action_mode == "survey" and len(ag.path) == 0: task.completed = True ag.current_task_completed = True elif len(discovered_tasks) > 0:#ag.unchanged_time >= 10: ag.current_task_completed = True # 配送智能体到达任务点后消耗载荷并减少任务 demand。 elif task.type == TaskType.Deliver: if not np.any(np.isnan(task.position)) and ag.dist < 1e-3: ag.complete() send_amount = ag.deliver(task.demand) sent += send_amount task.demand -= send_amount # 交付量作为联盟贡献同步给同联盟成员，用于联盟运输奖励。 for i in coalitions[task.id]: self.agents[i].coalition_delivery += ag.delivery if task.demand <= 0: task.completed = True completed += 1 # 任务完成时，同联盟成员都获得完成事件。 for i in coalitions[task.id]: self.event_task_completed[i] = 1 self.agents[i].coalition_completed = True self.agents[i].complete() self.event_any_task_completed = 1 # 补给智能体到达补给点后恢复载荷和电量。 elif task.type == TaskType.Resupply: if not np.any(np.isnan(task.position)) and ag.dist < 1e-3: ag.complete() ag.resupply() d = datetime.now().timestamp() self.t[3] = d - d0 d0 = d if len(new_discovered) > 0: # 探索发现新任务后，清空探索智能体路径，让它们可以尽快重新决策。 for ag in self.agents: if self.tasks[ag.coalition].type == TaskType.Explore: ag.path = [] # 历史时间惩罚位置保留在这里；当前实际奖励在 _get_rewards 中统一计算。 # for i, ag in enumerate(self.agents): # if ag['living']: # # 时间惩罚 # rewards[i] -= self.pun_time_ratio * (self.step_count / self.max_steps) # 时间惩罚 # 8. 生成下一时刻局部观测。奖励在后面算，但观测已经反映移动/任务状态更新。 observations = self._get_observations() # 9. 终止条件与 active_mask。 # done 表示环境回合终止或单个智能体死亡；bad_mask 区分自然终止和时间截断。 # key_event 会触发空闲且存活的智能体重新决策。 key_event = (self.event_new_task_found == 1 or self.event_any_task_completed == 1) all_task_completed = all(task.completed for task in self.tasks if task.type == TaskType.Deliver) all_still = all([not ag.living or ag.idle for ag in self.agents]) all_agents_dead = all(not ag.living for ag in self.agents) natural_done = all_task_completed or all_agents_dead time_limit = self.step_count >= self.max_steps episode_done = natural_done or time_limit dones = np.array([episode_done or not ag.living for ag in self.agents]) bad_mask = np.array([0.0 if time_limit and not natural_done else 1.0 for ag in self.agents]) bad_mask = bad_mask[..., None] if episode_done: # episode 结束时所有 agent 都激活，方便 runner flush 最后一帧。 active_mask = np.ones((len(self.agents),), dtype=np.float32) else: active_mask = self._get_reconfiguration_active_mask( key_event=key_event, all_still=all_still, ) active_mask = active_mask[..., None] self._current_decision_mask = active_mask[:, 0].copy() available_actions = self._get_available_actions() d = datetime.now().timestamp() self.t[4] = d - d0 d0 = d # 10. 奖励和任务能力比例诊断。 if self.event_any_task_completed == 1: for ag in self.agents: ag.any_task_completed = True for i, task in enumerate(self.tasks): # req_ratio 反映当前联盟能力相对需求的富余/缺口，写入任务对象供诊断使用。 task.calc_req_ratio(task_caps[i,:]) rewards_ret, rewards_info = self._get_rewards(sent, completed, new_discovered, all_task_completed, all_still) if self.reconfiguration_switch_cost > 0.0: switched = np.array( [ ag.coalition_changed and not ag.last_state.idle and not self.tasks[int(ag.last_state.coalition)].completed for ag in self.agents ], dtype=np.float32, ) extra_switch_penalty = self.reconfiguration_switch_cost * switched rewards_ret[:, 0] -= extra_switch_penalty rewards_info[self.reward_keys.index("switch_coalition")] -= ( extra_switch_penalty ) if key_event: pass # 11. info 主要供日志、评估和离线分析；不进入策略网络。 info = { 'agents': [{ 'position': ag.position, 'coalition': ag.coalition, 'living': ag.living, 'capabilities': ag.capbilities, 'load': ag.load, 'unchanged': ag.unchanged_time, } for ag in self.agents], 'tasks': [{ 'position': task.position, 'req_vec': task.req_vec, 'demand': task.demand, 'urgency': task.urgency, 'initial_demand': task.initial_demand, 'type': int(task.type), 'discovered': task.discovered, 'completed': task.completed } for task in self.tasks], 'danger_points': self.danger_points, 't': self.t, 'rewards': rewards_info, 'reward_keys': self.reward_keys, 'sent': sent, 'sent_goal': self.sent_goal, 'agent_caps': np.stack([ag.capbilities for ag in self.agents]), 'task_reqs': np.stack([task.req_vec for task in self.tasks]), 'reconfiguration': { 'mode': self.reconfiguration_mode, 'decisions': step_reconfiguration_decisions, 'coalition_changes': step_coalition_changes, 'total_decisions': self.total_reconfiguration_decisions, 'total_coalition_changes': self.total_coalition_changes, 'total_alive_agent_steps': self.total_alive_agent_steps, 'urgency_pressure': self._reconfiguration_urgency_pressure(), 'mean_response_time': float(np.mean(self.response_times)) if self.response_times else 0.0, }, 'end': self.step_count + 1 if episode_done else 0 } for ag in self.agents: # 最后更新 last_state，并清空本步事件，准备下一次 step。 ag.step() return observations, rewards_ret, dones, info, active_mask, bad_mask, available_actions def _task_feature_matrix(self, include_flags=False, local_visible=False): """构造任务特征矩阵。 local_visible=True 时，未发现或已完成任务在局部观测中置零； include_flags=True 时，为 critic 额外添加 discovered/completed 标志。 """ extra_dim = 2 if include_flags else 0 features = np.zeros((self.task_num, self.per_task_feat + extra_dim), dtype=np.float32) for idx, task in enumerate(self.tasks): if local_visible and (not task.discovered or task.completed): continue pos = task.position if np.isnan(pos[0]) or np.isnan(pos[1]): # Idle 或 nearest Explore 这类无固定空间目标的任务用 0 位置占位。 pos = np.array([0.0, 0.0]) if self.use_coalition_delivery: features[idx, 0:2] = pos / self.map_size if self.coalition_task_feature_mode == "full": features[idx, 2:2 + self.cap_dim] = task.req_vec / 5.0 features[idx, 2 + self.cap_dim] = task.demand / 500.0 deadline_idx = 3 + self.cap_dim type_idx = 7 + self.cap_dim else: features[idx, 2] = task.demand / 500.0 features[idx, 4] = ( float(task.req_vec[0]) if task.req_vec is not None else 0.0 ) deadline_idx = 3 type_idx = 5 if task.type == TaskType.Deliver: features[idx, deadline_idx] = max( 0.0, float(task.urgency) - self.step_count / max(1.0, self.max_steps), ) else: features[idx, deadline_idx] = 1.0 features[idx, type_idx] = float(task.type) / float(max(TaskType)) else: # 默认 MACF 任务特征：位置、能力需求、剩余需求、任务类型。 # req_vec / 5.0 与 HATT residual 中的 hatt_capability_scale 对应。 features[idx, 0:2] = pos / self.map_size features[idx, 2:2 + self.cap_dim] = task.req_vec / 5.0 features[idx, 2 + self.cap_dim] = task.demand / 500.0 features[idx, 3 + self.cap_dim] = float(task.type) / float(max(TaskType)) if include_flags: # 这两个 flag 只给 critic 全局状态，不给 actor 局部任务观测。 features[idx, self.per_task_feat] = 1.0 if task.discovered else 0.0 features[idx, self.per_task_feat + 1] = 1.0 if task.completed else 0.0 return features def _agent_feature_matrix(self): """构造所有智能体的状态矩阵，用于 other 观测和 critic 全局状态。""" features = np.zeros((self.agent_num, self.other_agent_feat_dim), dtype=np.float32) max_coalition = float(max(1, self.task_num - 1)) max_agent_type = float(max(AgentType)) for idx, ag in enumerate(self.agents): # 注意：这里保留 agent 自己；_get_observations 会在 others 中把对角线置零。 features[idx, 0:2] = ag.position / self.map_size features[idx, 2:2 + self.cap_dim] = ag.capbilities features[idx, 2 + self.cap_dim] = float(ag.load) / float(ag.init_load) features[idx, 3 + self.cap_dim] = ag.battery / 1000.0 features[idx, 4 + self.cap_dim] = float(ag.coalition) / max_coalition features[idx, 5 + self.cap_dim] = float(ag.type) / max_agent_type features[idx, 6 + self.cap_dim] = 1.0 if ag.living else 0.0 return features def _get_observations(self): """返回 actor 使用的局部 Dict 观测。 ego: 每个 agent 自己的状态，shape [A, ego_dim] tasks: 每个 agent 看到的任务槽，shape [A, T, task_dim] others: 每个 agent 看到的其他智能体，shape [A, A, other_dim]，自身位置置零 """ task_features = self._task_feature_matrix(local_visible=True) # 当前任务可见性对所有 agent 相同，因此复制到 observer 维。 task_obs = np.repeat(task_features[None, :, :], self.agent_num, axis=0) agent_features = self._agent_feature_matrix() other_obs = np.repeat(agent_features[None, :, :], self.agent_num, axis=0) self_indices = np.arange(self.agent_num) # others 的第 idx 个 observer 不应看到“另一个自己”，对角线置零。 other_obs[self_indices, self_indices, :] = 0.0 discovered_deliver = sum( 1 for task in self.tasks if task.discovered and task.type == TaskType.Deliver ) discovered_ratio = float(discovered_deliver) / float(self.delivery_task_num) max_coalition = float(max(1, self.task_num - 1)) max_task_type = float(max(TaskType)) ego_obs = np.zeros((self.agent_num, self.ego_feat_dim), dtype=np.float32) for idx, ag in enumerate(self.agents): coalition = self.tasks[ag.coalition] # ego 中保留当前联盟编号和联盟类型，让 actor 知道当前正在执行的任务语义。 ego_obs[idx, 0:2] = ag.position / self.map_size ego_obs[idx, 2:2 + self.cap_dim] = ag.capbilities ego_obs[idx, 2 + self.cap_dim] = float(ag.coalition) / max_coalition ego_obs[idx, 3 + self.cap_dim] = float(coalition.type) / max_task_type ego_obs[idx, 4 + self.cap_dim] = float(ag.load) / float(ag.init_load) ego_obs[idx, 5 + self.cap_dim] = discovered_ratio ego_obs[idx, 6 + self.cap_dim] = ag.battery / 1000.0 return { "ego": ego_obs, "tasks": task_obs.astype(np.float32, copy=False), "others": other_obs.astype(np.float32, copy=False), } def get_global_state(self): """返回 critic 使用的特权全局状态。 actor 只能看到局部可见 tasks；critic 可以看到所有任务及 discovered/completed 标志。 """ return { "agents": self._agent_feature_matrix(), "tasks": self._task_feature_matrix(include_flags=True), "global": np.array([self.step_count / self.max_steps], dtype=np.float32), } def _get_rewards(self, sent, completed, new_discovered, all_task_completed, all_still): """选择当前场景的奖励函数，并返回每个 agent 的总奖励和分项奖励。""" if self.use_coalition_delivery: if self.coalition_delivery_reward_mode == "shaped": return self._get_coalition_delivery_rewards_shaped( sent, completed, new_discovered, all_task_completed, all_still ) return self._get_coalition_delivery_rewards( sent, completed, new_discovered, all_task_completed, all_still ) # 默认奖励：保留早期探索/配送/补给/完成/距离塑形逻辑。 any_explore = any(self._is_explore_task_id(ag.coalition) and ag.living for ag in self.agents) # explore_decay = max(0.1, 1 - self.step_count / (0.5 * self.max_steps)) # 已发现任务比例用于衰减探索奖励，前期鼓励发现，后期避免一直探索。 discovered_ratio = float([task.discovered and task.type == TaskType.Deliver for task in self.tasks].count(True)) / self.delivery_task_num ep_check_1 = 200 ep_check_2 = 600 # if self.episode < ep_check_1: # ep_decay = 0.0 # elif self.episode < ep_check_2: # ep_decay = float(self.episode - ep_check_1) / (ep_check_2 - ep_check_1) # else: # ep_decay = 1.0 ep_decay = 1.0 step_decay = self.step_count / self.max_steps # rewards_1/rewards_2 是历史上两套奖励方案的混合接口；当前 ep_decay=1，只使用 rewards_2。 rewards = [{key: 0.0 for key in self.reward_keys} for _ in range(self.agent_num)] rewards_1 = [{key: 0.0 for key in self.reward_keys} for _ in range(self.agent_num)] rewards_2 = [{key: 0.0 for key in self.reward_keys} for _ in range(self.agent_num)] for i, ag in enumerate(self.agents): if not ag.dead and not ag.living: ag.coalition = self.idle_task_id continue coalition_type = self.tasks[ag.coalition].type last_coalition_type = self.tasks[ag.last_state.coalition].type # 切换联盟，且上回合不空闲，惩罚 if ag.coalition_changed and not ag.last_state.idle: # 切换到探索（第一回合不罚） if self.step_count > 0 and coalition_type == TaskType.Explore: # 从运输、补给切换到探索，罚 if last_coalition_type in [TaskType.Deliver, TaskType.Resupply]: rewards_2[i]['switch_coalition'] -= 1.0 # 从运输或补给切换 if last_coalition_type in [TaskType.Deliver, TaskType.Resupply]: rewards_2[i]['switch_coalition'] -= 1.5 if ag.coalition_changed and ag.last_state.idle: # 从空闲状态出发时，用载荷比例决定补给/执行任务是否合理。 load_ratio = ag.load / int(ag.init_load) rewards_2[i]['switch_coalition'] += (-1.0 if coalition_type == TaskType.Resupply else 1.0) \ * np.sinh((load_ratio - 0.5) * 2) / np.sinh(1.0) * 2.0 # if coalition_type == TaskType.Resupply: # if load_ratio < 0.4: # rewards_2[i]['switch_coalition'] += ((0.4 - load_ratio) ** 2) * 5 # else: # rewards_2[i]['switch_coalition'] += load_ratio - 0.5 # 死亡惩罚 if ag.dead: rewards_2[i]['dead'] -= 5.0 # 探索奖励 # 发现事件奖励按当前发现比例衰减；没有任何探索者时额外惩罚，避免全体过早放弃探索。 rewards_2[i]['discover'] = ag.discovery * (0.1/(discovered_ratio + 0.1) if discovered_ratio<0.25 else 0.1/0.35) rewards_2[i]['discover'] -= 0.2 if not any_explore and discovered_ratio <= 0.9 else 0.0 # 运输奖励 # deliver 是本步真实交付量，直接转成稠密奖励。 rewards_2[i]['deliver'] = ag.delivery * 0.1 # 补给奖励 # 补给越充分奖励越高；无效补给会通过 -1 基线体现代价。 rewards_2[i]['resupply'] += ag.resupply_ratio * 2.0 - 1.0 if ag.resupplied else 0.0 # 联盟运输奖励 # rewards_2[i]['coalition_deliver'] = ag.coalition_delivery * 0.07 # 联盟完成任务奖励 # rewards_2[i]['coalition_complete'] += ag.coalition_completed * 4.0 base = 20.0 time_coef = 20.0 # 单任务完成奖励预留变量；当前没有写入分项，只保留计算位置。 completion_reward = (base + time_coef * (1.0 - self.step_count / self.max_steps)) if ag.coalition_completed else 0 if all_task_completed: # 全部配送任务完成时给全体终局奖励，越早完成奖励越高。 rewards_2[i]['coalition_complete'] += (1 - self.step_count / self.max_steps) * 20 + 20 # 距离奖励 if not ag.coalition_changed \ and coalition_type != TaskType.Idle \ and coalition_type != TaskType.Explore: unchanged_decay = (ag.unchanged_time if ag.unchanged_time < 5.0 else 5.0) / 5.0 rewards_2[i]['dist'] = 0.002 * (ag.last_state.dist - ag.dist) # 时间惩罚 # 每步固定惩罚起到时间成本作用，所有 agent 都会承担。 rewards_2[i]['dist'] -= 0.2 if all_still: # 全体静止通常表示策略陷入空转，额外惩罚。 rewards_2[i]['dist'] -= 1.0 # 回合结束平均奖励 # end 分项按当前总交付量和完成数给所有 agent 共享进度信号。 rewards_2[i]['end'] += sent / self.sent_goal * 10 rewards_2[i]['end'] += completed / self.delivery_task_num * 50 rewards_2[i]['end'] /= 2.0 for i in range(self.agent_num): for key in self.reward_keys: # 当前 ep_decay=1，因此 rewards_1 不影响实际奖励。 rewards[i][key] = (1 - ep_decay) * rewards_1[i][key] + ep_decay * rewards_2[i][key] rewards_ret = np.zeros(self.agent_num, dtype=float) rewards_info = np.zeros((len(self.reward_keys), self.agent_num), dtype=float) for i, rewards_ag in enumerate(rewards): for key, rw in rewards_ag.items(): # rewards_ret 给 PPO 使用；rewards_info 保留每个分项，供日志和评估分析。 rewards_ret[i] += rw rewards_info[self.reward_keys.index(key)][i] = rw rewards_ret = rewards_ret[..., None] # rewards_info = np.concatenate([rewards_info, np.array([ag.coalition for ag in self.agents]).reshape(1, -1)]) return rewards_ret, rewards_info def _get_coalition_delivery_rewards(self, sent, completed, new_discovered, all_task_completed, all_still): """coalition_delivery 场景的基础奖励版本。 强调配送进度、按时完成、缺货补给和未完成任务的终局惩罚。 """ rewards = [{key: 0.0 for key in self.reward_keys} for _ in range(self.agent_num)] discovered_ratio = float( [task.discovered and task.type == TaskType.Deliver for task in self.tasks].count(True) ) / self.delivery_task_num any_explore = any(self._is_explore_task_id(ag.coalition) and ag.living for ag in self.agents) remaining_demand = sum( max(0.0, float(task.demand)) for task in self.tasks if task.type == TaskType.Deliver and not task.completed ) unfinished_tasks = sum( 1 for task in self.tasks if task.type == TaskType.Deliver and not task.completed ) has_remaining_delivery = remaining_demand > 1e-6 for i, ag in enumerate(self.agents): if not ag.dead and not ag.living: ag.coalition = self.idle_task_id continue task = self.tasks[ag.coalition] coalition_type = task.type last_coalition_type = self.tasks[ag.last_state.coalition].type if ag.coalition_changed and not ag.last_state.idle: rewards[i]["switch_coalition"] -= 0.05 if last_coalition_type == TaskType.Deliver and coalition_type != TaskType.Deliver: rewards[i]["switch_coalition"] -= 0.05 if ag.dead: rewards[i]["dead"] -= 5.0 rewards[i]["discover"] += ag.discovery * 1.0 if not any_explore and discovered_ratio < 0.9: rewards[i]["discover"] -= 0.05 if ag.delivery > 0: rewards[i]["deliver"] += 0.05 * ag.delivery elif ag.delivery < 0: rewards[i]["deliver"] -= 0.5 if ag.resupplied: rewards[i]["resupply"] += 0.8 * ag.resupply_ratio if ag.coalition_completed: completed_task = self.tasks[ag.coalition] priority = float(completed_task.req_vec[0]) if completed_task.req_vec is not None else 1.0 deadline_step = float(completed_task.urgency) * self.max_steps on_time_bonus = max(0.0, (deadline_step - self.step_count) / max(1.0, self.max_steps)) rewards[i]["coalition_complete"] += 10.0 * priority + 10.0 * on_time_bonus if has_remaining_delivery: rewards[i]["dist"] -= 0.005 if coalition_type == TaskType.Idle: rewards[i]["dist"] -= 0.03 if coalition_type == TaskType.Explore and discovered_ratio >= 0.99: rewards[i]["discover"] -= 0.05 if ag.load <= 0 and coalition_type != TaskType.Resupply: rewards[i]["dist"] -= 0.04 if coalition_type != TaskType.Idle: rewards[i]["dist"] -= 0.015 if coalition_type in (TaskType.Deliver, TaskType.Resupply): rewards[i]["dist"] += 0.001 * (ag.last_state.dist - ag.dist) if all_still: rewards[i]["dist"] -= 0.2 if self.step_count >= self.max_steps or all_task_completed: shortage_penalty = ( 12.0 * remaining_demand / max(1.0, float(self.sent_goal)) + 4.0 * unfinished_tasks / max(1.0, float(self.delivery_task_num)) ) completion_bonus = 15.0 if all_task_completed else 0.0 for item in rewards: item["end"] += completion_bonus - shortage_penalty rewards_ret = np.zeros(self.agent_num, dtype=float) rewards_info = np.zeros((len(self.reward_keys), self.agent_num), dtype=float) for i, rewards_ag in enumerate(rewards): for key, rw in rewards_ag.items(): rewards_ret[i] += rw rewards_info[self.reward_keys.index(key)][i] = rw return rewards_ret[..., None], rewards_info def _get_coalition_delivery_rewards_shaped(self, sent, completed, new_discovered, all_task_completed, all_still): """coalition_delivery 场景的 shaped 奖励版本。 在基础版本上加强全局配送进度、剩余 backlog 和 deadline pressure 的连续塑形。 """ rewards = [{key: 0.0 for key in self.reward_keys} for _ in range(self.agent_num)] delivery_tasks = [task for task in self.tasks if task.type == TaskType.Deliver] discovered_ratio = float( [task.discovered for task in delivery_tasks].count(True) ) / max(1, self.delivery_task_num) any_explore = any(self._is_explore_task_id(ag.coalition) and ag.living for ag in self.agents) remaining_demand = sum( max(0.0, float(task.demand)) for task in delivery_tasks if not task.completed ) unfinished_tasks = sum(1 for task in delivery_tasks if not task.completed) has_remaining_delivery = remaining_demand > 1e-6 time_ratio = self.step_count / max(1.0, float(self.max_steps)) sent_goal = max(1.0, float(self.sent_goal)) progress_reward = 25.0 * float(sent) / sent_goal backlog_cost = 0.025 * (remaining_demand / sent_goal) * (0.5 + time_ratio) urgency_pressure = 0.0 for task in delivery_tasks: if task.completed: continue remaining_ratio = max(0.0, float(task.demand)) / max(1.0, float(task.initial_demand)) priority = float(task.req_vec[0]) if task.req_vec is not None else 1.0 deadline_overrun = max(0.0, time_ratio - float(task.urgency)) urgency_pressure += priority * remaining_ratio * (0.25 * time_ratio + deadline_overrun) urgency_cost = 0.02 * urgency_pressure / max(1, self.delivery_task_num) for i, ag in enumerate(self.agents): if not ag.dead and not ag.living: ag.coalition = self.idle_task_id continue task = self.tasks[ag.coalition] coalition_type = task.type last_coalition_type = self.tasks[ag.last_state.coalition].type if ag.coalition_changed and not ag.last_state.idle: rewards[i]["switch_coalition"] -= 0.05 if last_coalition_type == TaskType.Deliver and coalition_type != TaskType.Deliver: rewards[i]["switch_coalition"] -= 0.05 if ag.dead: rewards[i]["dead"] -= 5.0 rewards[i]["discover"] += ag.discovery * 1.0 if not any_explore and discovered_ratio < 0.9: rewards[i]["discover"] -= 0.05 if has_remaining_delivery and coalition_type == TaskType.Explore and discovered_ratio >= 0.99: rewards[i]["discover"] -= 0.08 if ag.delivery > 0: rewards[i]["deliver"] += 0.04 * ag.delivery elif ag.delivery < 0: rewards[i]["deliver"] -= 1.0 if ag.resupplied: resource_gain = float(ag.resupply_ratio) rewards[i]["resupply"] += 1.0 * resource_gain if resource_gain <= 0.05: rewards[i]["resupply"] -= 0.5 * (1.0 - resource_gain / 0.05) if ag.coalition_completed: completed_task = self.tasks[ag.coalition] priority = float(completed_task.req_vec[0]) if completed_task.req_vec is not None else 1.0 deadline_step = float(completed_task.urgency) * self.max_steps on_time_bonus = max(0.0, (deadline_step - self.step_count) / max(1.0, self.max_steps)) rewards[i]["coalition_complete"] += 8.0 * priority + 8.0 * on_time_bonus if progress_reward != 0.0: rewards[i]["coalition_deliver"] += progress_reward if has_remaining_delivery: rewards[i]["dist"] -= backlog_cost + urgency_cost if coalition_type == TaskType.Idle: rewards[i]["dist"] -= 0.04 if ag.load <= 0 and coalition_type != TaskType.Resupply: rewards[i]["dist"] -= 0.04 if coalition_type != TaskType.Idle: rewards[i]["dist"] -= 0.02 if coalition_type in (TaskType.Deliver, TaskType.Resupply): rewards[i]["dist"] += 0.001 * (ag.last_state.dist - ag.dist) if all_still: rewards[i]["dist"] -= 0.2 if self.step_count >= self.max_steps or all_task_completed: shortage_penalty = ( 14.0 * remaining_demand / sent_goal + 4.0 * unfinished_tasks / max(1.0, float(self.delivery_task_num)) ) completion_bonus = 15.0 if all_task_completed else 0.0 for item in rewards: item["end"] += completion_bonus - shortage_penalty rewards_ret = np.zeros(self.agent_num, dtype=float) rewards_info = np.zeros((len(self.reward_keys), self.agent_num), dtype=float) for i, rewards_ag in enumerate(rewards): for key, rw in rewards_ag.items(): rewards_ret[i] += rw rewards_info[self.reward_keys.index(key)][i] = rw return rewards_ret[..., None], rewards_info