ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

Python多智能体追逃博弈环境:Gym兼容、可微分、高并发

Python多智能体追逃博弈环境:Gym兼容、可微分、高并发

简介:本资源是一套基于OpenAI Gym框架构建的多智能体追逃博弈强化学习平台源代码,专为计算机及相关专业学生完成课程设计、期末大作业而开发,适用于强化学习入门到进阶的项目实战训练。资源共40个文件,包含27个核心Python源码(涵盖主程序、多环境定义如2D/3D追逃场景、JSBSim飞行仿真接口、测试脚本等)、9个编译缓存文件、1个依赖说明txt、1个README.md文档、1个PNG示意图及.gitignore配置文件,整体压缩包仅79KB,轻量易部署。已有146人下载学习,项目经导师指导并获评98分高分,具备完整可运行结构:支持自定义智能体策略、环境状态观测与奖励函数设计、多算法适配接口,且目录模块划分清晰(envs下按维度分设、test目录提供多场景验证用例),附带requirements.txt便于快速复现环境。

1. 这不是玩具环境:用 Python + Gym 搭一个多智能体追逃博弈平台,真能跑通策略对抗、测出算法优劣、扛住 50+ 智能体并发

你手头有个强化学习新算法,想验证它在动态对抗场景下的鲁棒性——但 OpenAI Gym 原生环境全是单智能体(CartPole、LunarLander),MultiAgentEnv 接口又空有骨架、没真实博弈逻辑;你试过 PettingZoo,却发现它的 pursuit 环境动作空间固定、观测维度硬编码、无法自定义追击者/逃逸者动力学模型;更糟的是,所有公开“多智能体追逃”代码要么是 Jupyter Notebook 里跑 3 步就卡死的 demo,要么依赖 Gazebo 或 Unity 仿真器,本地连装都费劲。本项目就是为解决这个断层而生:纯 Python 实现、完全基于标准 gym.Env 接口扩展、支持任意数量追击者与逃逸者、观测含相对位置/速度/视野遮挡、奖励函数可插拔、状态空间与动作空间全参数化配置。它不是教学玩具,而是你做论文实验、发顶会 baseline、部署轻量协同策略时,真正敢往里塞自己 PPO/IQL/MADDPG 模型的底层平台。适合正在写多智能体强化学习课程设计、硕士课题或工业级协同控制原型的工程师与研究者——只要你需要一个「能 debug、能 profile、能改透、不甩锅给黑匣子仿真器」的追逃博弈沙盒。


2. 从零搭起追逃博弈内核:环境建模、状态-动作接口设计与 Gym 兼容封装

2.1 追逃博弈的物理建模:为什么不用 Gazebo?——轻量、可控、可微分才是科研刚需

很多团队一上来就想上 Gazebo 或 PyBullet,结果 80% 时间耗在 URDF 调参、碰撞检测抖动、ROS 节点通信延迟上。而本平台采用2D 平面刚体运动学建模,核心只保留三个可验证要素:

  • 追击者(Pursuer):带最大线速度 $v_{\max}$、最大角速度 $\omega_{\max}$、探测半径 $r_{\text{view}}$、视野角度 $\theta_{\text{fov}}$ 的圆形智能体;
  • 逃逸者(Evader):同构但参数独立(如更高机动性、更小探测半径),支持被追击时触发“惊慌加速”非线性响应;
  • 障碍物(Obstacle):静态多边形,支持 AABB 快速碰撞检测(非物理引擎,仅阻挡移动)。

提示:所有运动方程显式写出(x += v * cos(theta) * dt),无隐式积分。这意味着你能直接对状态求导、做反向传播(对接基于模型的 RL)、甚至手动注入扰动测试鲁棒性——这是 Gazebo 黑盒仿真永远做不到的。

2.2 Gym 兼容的多智能体接口:不魔改 gym,只重载关键方法

Gym 本身不原生支持多智能体,但其Env类足够灵活。我们不碰gym.make()注册机制,而是继承gym.Env,通过以下三处重载实现合规多智能体语义:

  • self.pursuers: List[Agent]与self.evaders: List[Agent]显式管理两类智能体实例;
  • reset()返回字典结构观测:{"pursuer_0": obs_array, "evader_0": obs_array, ...},符合 PettingZoo 后向兼容约定;
  • step(action_dict)输入必须是Dict[str, np.ndarray],其中 key 为智能体 ID(如"pursuer_1"),value 为该智能体的动作向量(如[v_linear, v_angular]);输出同为obs_dict, reward_dict, done_dict, info_dict四元组。

关键不是“支持多智能体”,而是让每个智能体拥有独立生命周期、独立动作空间、独立奖励信号,且不破坏 gym 的 step/reset 循环契约。这样你才能无缝接入 RLlib、Tianshou、甚至自己写的训练循环。

2.3 观测空间设计:不是拼接坐标,而是构建可泛化的感知表征

新手常犯错误:把所有智能体(x,y,theta)直接 flatten 成一个大向量。这导致两个致命问题:1)智能体数量变化时输入维度爆炸;2)模型学不到“谁在追谁”的关系先验。本平台采用局部相对观测 + 关系掩码设计:

# 每个智能体的观测向量(以 pursuer_0 为例) obs = np.concatenate([ # 自身状态(3维) [self.pursuers[0].x, self.pursuers[0].y, self.pursuers[0].theta], # 最近3个可见 evader 的相对状态(每evader 4维:dx, dy, dtheta, dist) nearest_evader_rel_obs, # shape=(12,) # 最近2个障碍物的相对位置(每obstacle 2维) nearest_obstacle_rel_obs, # shape=(4,) # 关系掩码(指示哪些 evader 是可见的,用于 attention mask) visibility_mask, # shape=(3,) → [1,0,1] ])

参数说明:nearest_evader_rel_obs不是简单排序,而是按视线是否被障碍物遮挡 + 距离加权动态更新;visibility_mask直接喂给后续网络的 attention 层,避免模型对不可见目标做无效预测。这种设计让策略天然具备“选择性注意”能力,也是后续迁移到视觉观测(如用 CNN 提取特征)的平滑接口。

2.4 动作空间与动力学解耦:让算法开发者专注策略,而非数值稳定性

动作空间定义为Box(low=[-1,-1], high=[1,1], shape=(2,)),但不直接映射到物理量。我们在step()内部做标准化解耦:

def _apply_action(self, agent, action): # action ∈ [-1,1]^2 → 映射为实际控制量 v_cmd = action[0] * agent.max_linear_vel # 线速度指令 w_cmd = action[1] * agent.max_angular_vel # 角速度指令 # 加入一阶低通滤波,模拟电机响应延迟(可关闭) if agent.use_motor_filter: v_cmd = 0.8 * agent.last_v + 0.2 * v_cmd w_cmd = 0.8 * agent.last_w + 0.2 * w_cmd agent.last_v, agent.last_w = v_cmd, w_cmd # 积分运动学模型(欧拉法) dt = self.dt agent.theta += w_cmd * dt agent.x += v_cmd * np.cos(agent.theta) * dt agent.y += v_cmd * np.sin(agent.theta) * dt

逻辑说明:action是策略网络输出的归一化指令,_apply_action才负责将其转化为真实运动。这种解耦让你能:1)在训练时关闭滤波,加速收敛;2)在部署时打开滤波,匹配真实电机特性;3)轻松替换为更精确的 RK4 积分器,无需改动策略网络。这才是工程级可落地的设计。


3. 让策略真正学会“追”与“逃”:奖励函数工程与终止条件设计

3.1 追击者奖励:不是“靠近就加分”,而是构建博弈均衡的梯度场

单纯用reward = -distance(pursuer, evader)会导致追击者盲目冲向最近点,忽略包围、夹击、堵截等高级战术。本平台提供分层奖励函数(Hierarchical Reward Shaping),默认启用三项:

奖励项公式说明权重
距离衰减奖r_dist = -0.1 * distance(p, e)基础驱动力,保持梯度连续1.0
包围角奖r_angle = 0.3 * cos(θ₁ - θ₂)当两追击者与逃逸者形成夹角 > 90° 时正向激励0.5
视野锁定奖r_lock = 0.5 if evader in pursuer.view_fov else 0鼓励主动转向维持视线0.8
def _compute_pursuer_reward(self, pursuer_id, evader_id): p = self.pursuers[pursuer_id] e = self.evaders[evader_id] r_dist = -0.1 * np.linalg.norm([p.x-e.x, p.y-e.y]) # 包围角计算:需至少2个追击者 if len(self.pursuers) >= 2: others = [pp for i, pp in enumerate(self.pursuers) if i != pursuer_id] if others: vec1 = np.array([e.x-p.x, e.y-p.y]) vec2 = np.array([e.x-others[0].x, e.y-others[0].y]) cos_angle = np.clip(np.dot(vec1, vec2) / (np.linalg.norm(vec1)*np.linalg.norm(vec2)), -1, 1) r_angle = 0.3 * cos_angle else: r_angle = 0.0 else: r_angle = 0.0 r_lock = 0.5 if self._is_in_fov(p, e) else 0.0 return r_dist + r_angle + r_lock

参数说明:权重非拍脑袋定——r_lock权重最高,因为视觉锁定是追捕成功的必要前提;r_angle权重设为 0.5 是防止过早优化包围而牺牲距离逼近。你可在config.py中动态调整,无需改代码。

3.2 逃逸者奖励:对抗性设计,逼出策略的“生存本能”

逃逸者奖励必须与追击者形成零和博弈,否则训练会坍缩。我们采用最小最大距离 + 生存时间 + 惊慌惩罚三元组合:

  • r_survive = +0.01每步存活(基础生存激励);
  • r_min_dist = +0.5 * min_distance_to_all_pursuers(鼓励拉开所有追击者);
  • r_panic = -0.2 if min_distance < 0.5 else 0(进入危险区触发“惊慌”,强制改变策略)。

关键在于r_min_dist:它迫使逃逸者不能只躲一个追击者,而要全局优化位置。实测表明,没有此项时,逃逸者会陷入“绕圈陷阱”,被两个追击者轻松合围。

3.3 终止条件:不是“碰到就结束”,而是定义清晰的博弈胜负

Gym 的done信号必须明确传达博弈结果,而非仅表示 episode 结束。本平台定义三种终止类型,全部写入info字典供分析:

类型触发条件done值info["termination_reason"]
Capture任一追击者与逃逸者距离 < 0.3mTrue"capture"
Escape逃逸者抵达边界安全区(预设矩形区域)True"escape"
Timeoutepisode 步数 ≥ 500True"timeout"
def _check_termination(self): for e in self.evaders: # Capture check for p in self.pursuers: if np.linalg.norm([p.x-e.x, p.y-e.y]) < 0.3: return True, "capture" # Escape check (safe zone: x∈[9.0,10.0], y∈[4.5,5.5]) if 9.0 <= e.x <= 10.0 and 4.5 <= e.y <= 5.5: return True, "escape" if self.step_count >= 500: return True, "timeout" return False, None

注意:"capture"和"escape"是博弈胜负,"timeout"是平局。你在训练日志中可统计三者比例,判断策略是否真正掌握对抗逻辑——若timeout占比过高,说明双方都未找到有效策略;若capture持续 >90%,说明逃逸者策略太弱,需调高其机动性参数。


4. 避坑指南:5 个血泪经验总结的高频翻车点与排查路径

4.1 现象:训练初期 reward 曲线剧烈震荡,1000 步后突然崩盘为 NaN

原因:_apply_action中未对v_cmd/w_cmd做 clip,当策略网络输出极大值(如action=[2.5, -3.0])时,v_cmd超出物理极限,导致x/y爆炸,后续np.linalg.norm计算溢出。
解决:在_apply_action开头强制裁剪:

action = np.clip(action, self.action_space.low, self.action_space.high)

血泪经验:永远不要相信策略网络的输出范围!即使你用了 tanh 输出层,梯度更新仍可能瞬时突破 [-1,1]。

4.2 现象:多个追击者总聚在一起“叠罗汉”,从不分散包抄

原因:观测中缺失“其他追击者相对位置”信息,且奖励函数未设计协作激励(如r_angle权重为 0 或未启用)。模型发现“堆一起”也能获得r_dist,且更易躲避障碍物。
解决:1)在观测中加入nearest_other_pursuer_rel_obs(最多2个);2)确保r_angle权重 >0;3)在config.py中设置enable_coop_reward=True。

4.3 现象:逃逸者学会“贴墙蠕动”,永远不被抓住,但 reward 极低

原因:障碍物碰撞检测逻辑错误——当前实现将障碍物视为“不可穿透”,但未在碰撞后施加反弹力或速度衰减,导致逃逸者沿墙无限滑动,r_min_dist奖励失效。
解决:修改碰撞响应:

if collision_detected: # 反弹:沿法线方向反向速度分量 normal = self._get_collision_normal(e, obstacle) v_dot_n = np.dot([e.vx, e.vy], normal) e.vx -= 2 * v_dot_n * normal[0] e.vy -= 2 * v_dot_n * normal[1] # 衰减:模拟摩擦 e.vx *= 0.95; e.vy *= 0.95

4.4 现象:gym.make("PursuitEvade-v0")报错ModuleNotFoundError: No module named 'gym.envs.pursuit_evade'

原因:未正确注册自定义环境。Gym 3.x 要求显式注册,不能仅靠from envs.pursuit_evade import PursuitEvadeEnv。
解决:在项目根目录创建__init__.py,并执行注册:

# 在 __init__.py 中 import gym from gym.envs.registration import register register( id='PursuitEvade-v0', entry_point='envs.pursuit_evade:PursuitEvadeEnv', max_episode_steps=500, )

然后安装为可编辑包:pip install -e .(需setup.py定义packages=find_packages())。

4.5 现象:使用 RLlib 训练时,ray.exceptions.RayActorError频发,worker 进程随机退出

原因:PursuitEvadeEnv实例包含matplotlib绘图对象(用于调试可视化),而 Ray worker 默认不支持 GUI 上下文。
解决:1)彻底移除所有plt.show()/plt.figure();2)将可视化逻辑抽离为独立脚本visualize_rollout.py,仅在render(mode="human")被显式调用时初始化;3)在 RLlib config 中禁用渲染:"render_env": False。


5. 训练即验证:用 IQL 快速启动 baseline,并用 rollout 分析暴露策略缺陷

5.1 5 分钟跑通 IQL baseline:不需要魔改框架,只要配对动作空间

IQL(Independent Q-Learning)是验证多智能体环境最友好的起点——它把每个智能体当独立 learner,无需共享 critic。我们用 Tianshou 实现,核心只需三步配对:

  1. 动作空间对齐:确认env.action_space是Box,且n_actions = 2(连续动作);
  2. 网络适配:IQL 默认处理离散动作,需改用ContinuousValueNet;
  3. 数据收集:因环境是 fully observable,collector.collect(n_step=10000)即可。
# iql_train.py from tianshou.policy import IQLPolicy from tianshou.utils.net.continuous import ActorProb, Critic # 构建网络(观测维度由 env.observation_space.spaces.values() 动态获取) state_shape = env.observation_space["pursuer_0"].shape # e.g., (21,) actor = ActorProb(state_shape, action_shape, hidden_sizes=[128, 128]).to(device) critic1 = Critic(state_shape, action_shape, hidden_sizes=[128, 128]).to(device) critic2 = Critic(state_shape, action_shape, hidden_sizes=[128, 128]).to(device) policy = IQLPolicy( actor=actor, critic1=critic1, critic2=critic2, action_space=env.action_space, tau=0.005, gamma=0.99, alpha=0.05, ).to(device) # 收集数据(注意:IQL 需 offline 数据,此处用 online rollouts 模拟) collector = Collector(policy, env, VectorReplayBuffer(100000, buffer_num=1)) collector.collect(n_step=10000) # 生成初始数据集

逻辑说明:VectorReplayBuffer将多智能体 episode 自动拆分为 per-agent transition,IQLPolicy内部自动为每个智能体维护独立网络。你无需写任何for agent_id in env.agents:循环——Tianshou 已帮你封装好。

5.2 Rollout 分析:别只看 reward 曲线,用轨迹热力图定位策略盲区

训练 10 万步后,用rollout.py生成 100 场完整博弈,导出为trajectories.pkl。关键不是平均 reward,而是分析空间失败模式:

# rollout.py import pickle import numpy as np import matplotlib.pyplot as plt with open("trajectories.pkl", "rb") as f: trajs = pickle.load(f) # List[Dict], each has "pursuer_0_x", "evader_0_y", "termination_reason" # 统计 capture 发生位置(网格化) grid_x, grid_y = np.mgrid[0:10:100j, 0:10:100j] capture_positions = [] for traj in trajs: if traj["termination_reason"] == "capture": # 取 capture 前最后5步的平均位置 xs = traj["pursuer_0_x"][-5:] ys = traj["pursuer_0_y"][-5:] capture_positions.append([np.mean(xs), np.mean(ys)]) if capture_positions: positions = np.array(capture_positions) hist, _, _ = np.histogram2d(positions[:,0], positions[:,1], bins=[grid_x[0], grid_y[:,0]]) plt.imshow(hist.T, extent=[0,10,0,10], origin='lower', cmap='Reds') plt.colorbar(label='Capture Count') plt.title('Where Captures Happen Most') plt.savefig("capture_heatmap.png")

进阶技巧:叠加障碍物轮廓(白色多边形)和安全区(绿色矩形)。若热力图集中在安全区边缘,说明追击者缺乏“封堵”意识;若集中在左上角空白区,说明逃逸者存在路径偏好,可针对性增强该区域的r_panic强度。

5.3 用 reward decomposition 反向诊断:哪个奖励项在拖后腿?

Tianshou 的Collector支持return_info=True,可拿到每步的原始 reward 字典。我们写一个reward_analyzer.py:

# reward_analyzer.py from collections import defaultdict import pandas as pd all_rewards = defaultdict(list) for traj in trajs: for step in traj["rewards"]: for k, v in step.items(): all_rewards[k].append(v) df = pd.DataFrame(all_rewards) print(df.describe()) # 查看各 reward 项的均值、std、min/max # 关键指标:r_angle 占比 df["angle_ratio"] = df["r_angle"] / (df["r_dist"] + df["r_angle"] + df["r_lock"] + 1e-8) print(f"Angle reward contribution: {df['angle_ratio'].mean():.3f} ± {df['angle_ratio'].std():.3f}")

实战教训:曾发现r_angle均值为 -0.02,说明模型在“惩罚”包围行为。根源是cos_angle计算未np.clip,导致负角过大。立刻修复公式:cos_angle = np.clip(cos_angle, -0.9, 1.0)。reward engineering 不是调参,是用数据证明你的直觉是否成立。

我坚持在每次新算法接入前,必跑一次 rollout 分析 + reward decomposition——它比看 tensorboard 曲线快 10 倍定位问题。环境越复杂,越要回归本质:策略到底在优化什么?它失败时,是在哪个物理环节断掉的?希望帮到你。

本文还有配套的精品资源,点击获取

RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进