简介这是一套面向强化学习研究者与多智能体系统开发者的Python开源平台基于OpenAI Gym框架实现多智能体追逃博弈MAPE-G仿真环境适用于机器人协作、无人机对抗、网络安全策略建模等场景适合具备Python编程基础及强化学习入门知识的中高级学习者。资源共41个文件含27个核心Python源码覆盖3D/2D飞行仿真环境envs、dogFightEnv追逃逻辑、test模块验证脚本、9个编译缓存文件pyc、1个README.md说明文档、1个LICENSE授权文件、1个requirements.txt依赖清单及1张环境示意图png整体压缩包仅79KB轻量易部署。已有245人学习下载代码高度模块化支持灵活替换DQN、策略梯度等算法可快速定制智能体角色、调整环境参数并开展训练评估目录结构清晰分层含FlightGearEnv、JsbSimEnv等真实感仿真接口便于理解多智能体动态交互机制与MARL工程实践路径。1. 这不是玩具环境一个能跑通 MARL 训练闭环的多智能体追逃平台专为算法验证和策略对比而生你有没有试过在gym里搭一个多智能体追逃场景结果卡在状态同步上三天或者刚写完 reward 函数发现两个智能体互相“穿模”、reward 爆负、训练曲线像心电图这不是玄学——是缺一个从环境建模、动作解耦、观测隔离到 reward 分层设计都经过实测打磨的底层平台。这个 Python 实现的多智能体追逃博弈强化学习平台不是 demo也不是教学示例而是一个可直接接入 PPO、MAPPO、QMIX 等主流 MARL 算法框架的生产级 gym 兼容环境。它用纯 NumPy PyGame可选渲染实现物理建模支持动态障碍物、异构智能体追者/逃者能力不对称、局部观测掩码、通信带宽限制模拟等真实约束。适合正在做 MARL 算法复现、策略鲁棒性测试、或需要在可控环境下做消融实验的从业者。如果你手头已有 RL 训练脚本替换env gym.make(...)这一行就能跑起来——前提是你得先搞懂它的状态空间怎么切、动作怎么对齐、以及为什么默认不启用全局观测。2. 环境结构与核心机制为什么它不是简单复制单智能体 gym 的逻辑这个平台不是把多个CartPole-v1拼在一起。它重构了 gym 的交互契约单次step()返回的是 N 个智能体的联合动作响应而非单个标量。这意味着你必须理解它的三层抽象环境层world model、智能体层agent interface、接口层gym wrapper。下面拆解关键设计选择及其工程依据。2.1 状态空间连续坐标 相对观测 掩码机制拒绝“上帝视角”环境默认提供两种观测模式global全图坐标仅用于 debug和local默认每个智能体仅感知半径 R 内的其他实体。但真正关键的是relative encoding occlusion mask所有坐标均以当前智能体位置为原点归一化如逃者看到追者位置是(dx, dy)而非(x1, y1)若视线被障碍物阻挡对应目标的观测向量置为[0, 0, 0]x, y, type并额外输出一个occlusion_mask布尔数组每个智能体观测向量长度固定为max_entities × 3 22 是自身速度分量避免变长输入破坏 batch 训练稳定性。提示max_entities在初始化时通过env PursuitEvasionEnv(n_pursuers2, n_evaders1, max_entities5)显式指定它决定了观测向量维度。若运行时报Observation shape mismatch八成是这里没对齐你的网络输入层。2.2 动作空间离散控制 连续微调双模式适配不同算法需求平台支持两种动作模式通过action_modediscrete默认或continuous切换Discrete 模式每个智能体输出 5 类动作 —NOOP,UP,DOWN,LEFT,RIGHT对应单位步长移动。这是为 QMIX、VDN 等值分解算法准备的Continuous 模式输出[vx, vy] ∈ [-1.0, 1.0]²由环境内部做速度裁剪与碰撞检测。适合 PPO、SAC 等策略梯度方法。二者底层物理模型一致采用带阻尼的二阶运动方程$$ \dot{v} a - \gamma v $$其中a是动作施加的加速度γ0.1是默认阻尼系数。这比纯一阶移动更符合真实机器人动力学也使策略学习更具泛化性。# 初始化时显式声明动作模式 env PursuitEvasionEnv( n_pursuers3, n_evaders2, action_modecontinuous, # ← 关键开关 max_speed0.05, # 连续模式下最大速度归一化单位/step dt0.1 # 物理仿真时间步长 )参数说明max_speed防止智能体“瞬移”值过大会导致碰撞检测失效dt越小越精确但计算开销线性上升0.1是精度与效率的实测平衡点切换模式后env.action_space自动变为Box(-1,1,(2,))或Discrete(5)无需手动修改网络头。2.3 Reward 设计分层稀疏 即时惩罚解决 MARL 中的 credit assignment 难题追逃任务天然稀疏——只有抓到或逃脱才给大 reward。但纯稀疏 reward 会让策略梯度消失。本平台采用三级 reward 结构层级触发条件数值范围作用即时层每 step 移动距离 0.01-0.01防止智能体“冻结”协作层追者间距离 cohesion_radius0.02 × (1 - dist/radius)鼓励围堵阵型终局层追者触碰逃者成功捕获5.0主 reward终局层逃者抵达安全区3.0主 reward惩罚层追者撞墙 / 逃者撞障碍-1.0强化边界意识注意所有 reward 均按智能体独立计算并返回list[float]非 scalar。例如 3 追者 2 逃者环境step()返回的reward是长度为 5 的列表索引 0~2 对应追者3~4 对应逃者。这是 MAPPO、COMA 等算法要求的格式。3. 快速上手三步接入你现有的 RL 训练流程你不需要重写整个训练循环。只要确认你的算法框架支持gym.vector.VectorEnv或能批量处理list[obs]就能无缝集成。以下是基于 Stable-Baselines3 的最小可行接入方案其他框架同理。3.1 环境封装用 VectorEnv 实现高效并行采样Stable-Baselines3 的PPO默认只接受单环境但 MARL 训练必须批量采样以降低 variance。我们用SubprocVecEnv封装多个独立实例from stable_baselines3 import PPO from gym.vector import SubprocVecEnv import numpy as np def make_env(rank, seed0): def _init(): env PursuitEvasionEnv( n_pursuers2, n_evaders1, action_modediscrete, render_modeNone, # 关闭渲染提升速度 max_steps256 # 每 episode 最大步数 ) env.reset(seedseed rank) return env return _init # 创建 8 个并行环境 env SubprocVecEnv([make_env(i) for i in range(8)]) # 注意此处 obs_shape 是 (n_agents, obs_dim)不是 (obs_dim) # SB3 会自动 flatten但你要确保 policy 网络能处理 batched agent dim model PPO( MlpPolicy, env, verbose1, n_steps2048, # 总步数 8 env × 2048 16384 batch_size512, gamma0.99, learning_rate3e-4 )逻辑说明SubprocVecEnv启动 8 个独立进程每个运行一个PursuitEvasionEnv实例n_steps2048指每个进程收集 2048 步数据因此每次 update 实际使用8×204816384个 transitionMlpPolicy会自动将(batch, n_agents, obs_dim)输入 reshape 为(batch×n_agents, obs_dim)但你的自定义 policy 必须显式处理 agent 维度见 4.2 节。3.2 自定义 Policy为多智能体输出设计网络头SB3 的MlpPolicy默认输出单个 action但我们需要为每个智能体输出独立动作。最简方案是继承BasePolicy并重写forward()import torch as th from torch import nn from stable_baselines3.common.policies import BasePolicy class MultiAgentPolicy(BasePolicy): def __init__(self, observation_space, action_space, **kwargs): super().__init__(observation_space, action_space, **kwargs) # 假设 obs_shape (n_agents, obs_dim), action_space.n 5 self.n_agents observation_space.shape[0] self.obs_dim observation_space.shape[1] self.action_dim action_space.n self.feature_extractor nn.Sequential( nn.Linear(self.obs_dim, 128), nn.ReLU(), nn.Linear(128, 128), nn.ReLU() ) self.actor nn.Linear(128, self.action_dim) # 每 agent 一个 head self.critic nn.Linear(128, 1) def forward(self, obs: th.Tensor) - tuple: # obs: (batch, n_agents, obs_dim) → (batch×n_agents, obs_dim) batch_size obs.shape[0] obs_flat obs.view(-1, self.obs_dim) features self.feature_extractor(obs_flat) # (batch×n_agents, 128) logits self.actor(features).view(batch_size, self.n_agents, -1) # (batch, n_agents, 5) values self.critic(features).view(batch_size, self.n_agents, -1).mean(dim1) # (batch, 1) return logits, values参数说明obs.view(-1, self.obs_dim)是关键将多智能体观测展平复用单智能体网络结构logits.view(...)恢复 agent 维度供后续Categorical分布采样values.mean(dim1)对所有 agent 的 critic 输出取均值作为全局 state valueMAPPO 标准做法。3.3 训练启动与日志监控如何验证它真在学别急着跑 1000 万步。先用eval_freq10000和n_eval_episodes5做轻量评估from stable_baselines3.common.callbacks import EvalCallback eval_env SubprocVecEnv([make_env(i100) for i in range(4)]) eval_callback EvalCallback( eval_env, best_model_save_path./logs/best_model/, log_path./logs/, eval_freq10000, n_eval_episodes5, deterministicTrue, renderFalse ) model.learn(total_timesteps500000, callbackeval_callback)关键验证点查看./logs/evaluations.npz中的repisode reward是否从-200随机策略缓慢上升至-50以上用env.render()启动单环境可视化观察追者是否开始形成夹击角而非乱跑检查tensorboard --logdir ./logs/中train/entropy_loss是否稳定下降——熵太高说明策略未收敛。4. 避坑指南五个血泪经验总结出的高频翻车点MARL 环境调试成本远高于单智能体。以下问题均来自某高校实验室在复现该平台时的真实报错日志已定位到根因并验证修复方案。4.1 现象ValueError: Observation not in space但env.observation_space.contains(obs)返回True原因gym的contains()方法对Box空间只检查 shape 和 dtype不校验 NaN 或 inf。而本平台在障碍物遮挡时会将部分观测置零若物理引擎数值不稳定如dt过大可能产生inf速度污染后续观测。解决在step()返回前强制 clip# 在 env.step() 内部末尾添加 obs np.nan_to_num(obs, nan0.0, posinf1.0, neginf-1.0)4.2 现象训练中 reward 突然全为0.0且info[done]始终为False原因max_steps设置过小如128导致 episode 在追逃发生前就强制 truncation。此时reward数组全为即时层小负值但info中无terminal_observation算法误判为“未结束”。解决max_steps必须 ≥2×map_width / min_speed。例如地图宽 1.0最小速度 0.02则max_steps ≥ 100建议设为256或512。4.3 现象多进程训练时OSError: [Errno 24] Too many open files原因SubprocVecEnv为每个子进程创建独立的 PyGame 窗口句柄即使render_modeNonePyGame 初始化仍占用 fd。Linux 默认ulimit -n为 10248 进程 × 每个 128 fd 1024刚好爆满。解决在启动脚本开头添加ulimit -n 4096 python train.py或彻底禁用 PyGame 初始化修改env.__init__()中if render_mode is not None: pygame.init()为if render_mode human: ...。4.4 现象逃者策略学会“贴墙不动”追者永远无法捕获原因reward 设计缺陷——当cohesion_radius过大时追者为获得协作 reward 会优先聚拢反而扩大包围圈空隙同时逃者发现静止时instant_reward损失最小。解决动态调整cohesion_radius或改用距离倒数 reward# 替换原协作 reward 计算 avg_dist np.mean([np.linalg.norm(p1 - p2) for p1 in pursuer_pos for p2 in pursuer_pos if not np.array_equal(p1,p2)]) cohesion_reward 0.05 / (avg_dist 1e-3) # 距离越近 reward 越高4.5 现象PPO训练 loss 波动剧烈clip_range频繁触发原因多智能体动作空间未对齐。例如追者用discrete逃者用continuous但env.action_space只返回一个 space导致 policy 输出维度错乱。解决必须统一所有智能体的动作模式。平台不支持混合动作空间——这是 deliberate design choice有意为之因为混合动作极大增加 credit assignment 复杂度。如需异构应将逃者动作离散化如EVADE_LEFT,EVADE_RIGHT。5. 进阶技巧用 replay buffer 可视化分析策略失败模式训练卡在某个 reward plateau别只盯 tensorboard。我习惯用 replay buffer 抽样回放定位具体失败场景。本平台内置ReplayBuffer导出接口配合 Matplotlib 可生成带轨迹热力图的诊断图。5.1 导出 episode replay 数据在训练循环中每 10 个 episode 保存一次完整轨迹from collections import defaultdict import pickle def save_episode_replay(env, episode_data, path): # episode_data { # observations: list[np.ndarray], # (n_agents, obs_dim) # actions: list[np.ndarray], # (n_agents,) # rewards: list[np.ndarray], # (n_agents,) # dones: list[bool], # } with open(path, wb) as f: pickle.dump(episode_data, f) # 在 callback 中调用 if episode % 10 0: save_episode_replay(env, current_episode_buffer, f./replays/ep_{episode}.pkl)5.2 可视化轨迹与 reward 热力图import matplotlib.pyplot as plt import numpy as np def plot_episode_replay(pkl_path, map_size1.0): with open(pkl_path, rb) as f: data pickle.load(f) # 提取追者前 n_p和逃者后 n_e轨迹 n_p 2 n_e 1 pursuer_trajs [[] for _ in range(n_p)] evader_trajs [[] for _ in range(n_e)] for obs in data[observations]: for i in range(n_p): # obs[i] 是第 i 个追者的观测其中前 2 维是自身 x,y pursuer_trajs[i].append(obs[i, :2]) for i in range(n_e): evader_trajs[i].append(obs[n_pi, :2]) # 绘制 fig, ax plt.subplots(figsize(8, 8)) colors [red, orange, blue] for i, traj in enumerate(pursuer_trajs): traj np.array(traj) ax.plot(traj[:, 0], traj[:, 1], o-, colorcolors[i], markersize2, labelfPursuer {i}) for i, traj in enumerate(evader_trajs): traj np.array(traj) ax.plot(traj[:, 0], traj[:, 1], s-, colorcolors[n_pi], markersize3, labelfEvader {i}) ax.set_xlim(0, map_size) ax.set_ylim(0, map_size) ax.set_aspect(equal) ax.legend() ax.grid(True, alpha0.3) plt.title(fEpisode Replay: {pkl_path.split(_)[-1].split(.)[0]}) plt.savefig(pkl_path.replace(.pkl, _traj.png), dpi300, bbox_inchestight)5.3 分析 reward 分布定位瓶颈对data[rewards]做 agent-level 统计生成 reward 来源分布饼图Agent TypeAvg Reward/Step主要来源问题线索Pursuer 0-0.12即时层-0.01×12移动不足可能卡在角落Pursuer 10.03协作层0.02×1.5有效协同但终局 reward 为 0Evader 0-0.08惩罚层-1.0×0.08频繁撞墙需加强边界 reward从那以后我每次遇到 reward plateau都强制走一遍 replay buffer 抽样 轨迹绘制 reward 拆解三件套。它比调 learning rate 有用十倍——因为你能亲眼看到策略在“做什么”而不是猜它“应该做什么”。希望帮到你。本文还有配套的精品资源点击获取
