简介本资源是一套基于Python与OpenAI Gym框架实现的多智能体追逃博弈MAPE-G强化学习仿真平台面向强化学习初学者、多智能体系统研究者及机器人/无人机协同控制方向的工程实践者旨在解决多智能体环境下追捕者与逃避者动态博弈建模、策略训练与交互评估等核心问题。压缩包共41个文件79KB含27个核心Python源码涵盖3D/2D飞行环境、JsbSim仿真接口、dogFight对抗场景等模块、9个编译缓存文件、1个依赖说明requirements.txt、1个README.md文档、1个LICENSE协议及图像与配置文件结构清晰、模块解耦便于二次开发与算法替换。已有243人学习下载读者可直接复现多智能体DQN、策略梯度等主流MARL算法快速构建可扩展的追逃实验环境并深入理解智能体观测空间设计、奖励函数构造及分布式训练流程。1. 项目缘起为什么我们需要一个多智能体追逃博弈平台如果你接触过强化学习大概率是从OpenAI Gym的经典控制任务开始的比如让小车爬坡CartPole或者让钟摆立起来Pendulum。这些单智能体任务为我们理解“状态-动作-奖励”这个核心循环提供了绝佳的入门场景。但现实世界远比这复杂很多问题本质上是多个智能体在共享环境中互动、竞争或协作。比如无人机集群的围捕、自动驾驶车辆的博弈、游戏中的多单位对战甚至是金融市场中多个交易策略的对抗。这时单智能体强化学习的框架就显得捉襟见肘了。多智能体强化学习Multi-Agent Reinforcement Learning, MARL应运而生它研究的就是这种多个智能体共存、相互影响的学习问题。而“追逃博弈”Pursuit-Evasion Game则是MARL领域一个经典且直观的“试金石”。它模拟了一个或多个“追捕者”Pursuer试图捕获一个或多个“逃跑者”Evader的场景。这个看似简单的模型却能衍生出极其丰富的策略空间和动态变化逃跑者需要规划路径、利用障碍物追捕者需要协同配合、预判走位。然而当你兴致勃勃地想用Python和Gym来复现一个追逃博弈环境时往往会发现Gym官方库主要提供单智能体环境虽然社区有一些多智能体环境如PettingZoo但要么封装得太重要么不够灵活难以让我们从零开始理解环境构建、多智能体观测与动作空间设计、奖励函数工程等核心环节。市面上能找到的源码要么过于学术化耦合了特定算法要么过于简陋只是一个静态演示。这正是我动手搭建这个“Gym框架下的多智能体追逃博弈强化学习平台”的初衷。我希望它不仅仅是一段能跑的代码更是一个高度模块化、易于扩展的教学与研究平台。你可以用它来快速验证MARL算法无论是经典的独立Q学习IQL、多智能体深度确定性策略梯度MADDPG还是较新的QMIX、MAPPO都可以在这个平台上进行基准测试。深入理解环境设计完全掌控地图生成、智能体动力学模型、碰撞检测、奖励函数设计等底层细节。进行课程学习或课程教学通过调整地图复杂度、智能体数量、传感器范围等参数构建由易到难的学习序列。这个平台完全基于Python和Gym接口构建依赖清晰结构明了。接下来我将带你从零开始拆解它的每一个核心模块并分享在实现过程中那些文档里不会写的“坑”和技巧。2. 平台架构总览从Gym接口到多智能体世界在开始看代码之前我们必须先理清几个关键概念并确定我们的平台要设计成什么样子。多智能体环境与单智能体环境最大的区别在于**“步进”step的粒度和信息的结构**。核心设计决策采用“协同式”Turn-Based还是“并行式”Simultaneous-Move在追逃博弈中智能体可以轮流行动像下棋也可以同时行动像实时战略游戏。我们选择同时行动模式因为它更符合大多数现实场景如机器人、自动驾驶也是MARL研究的常见设定。这意味着在每个时间步所有智能体基于自己当前的观测同时做出决策并执行动作然后环境更新到下一个状态。Gym接口的适配核心是step和resetGym的标准接口是env.step(action)返回observation, reward, done, info。对于多智能体这个action需要是一个包含所有智能体动作的字典或列表。同样observation和reward也需要是相应的字典或列表。done可以是一个全局终止标志也可以是一个包含每个智能体是否终止的列表。为了保持最大的灵活性并与主流MARL库如RLlib、EPyMARL兼容我们选择返回字典形式。我们的平台架构分为四层环境层Environment继承自gym.Env负责实现核心的reset和step方法管理整个世界的状态。世界层World这是环境的“引擎”。它包含地图Grid、所有智能体Agent对象并负责物理更新、碰撞检测、视距计算等。智能体层Agent定义追捕者和逃跑者的属性位置、速度、视野半径等和行为模型如何根据动作移动。观测与奖励层Observation Reward定义每个智能体如何感知世界例如基于网格的局部观察、相对位置向量以及如何计算其奖励。下面是一个简化的模块关系图我们用文字描述其数据流Agent 1 Action ───┐ Agent 2 Action ───┤ ... ├─→ Environment.step() → World.update() → 计算新状态、碰撞、奖励 Agent N Action ───┘ ↓ 返回给算法: {‘agent_1’: (obs1, reward1, done1, info1), ...}这种设计将环境逻辑Gym接口与物理模拟逻辑世界更新分离使得我们未来可以轻松替换地图表示从2D网格到连续空间或智能体动力学模型而无需重写接口代码。3. 核心模块一二维网格世界与智能体建模我们首先从最底层的世界表示开始。为了简化并聚焦于MARL算法本身我们使用一个二维离散网格世界。每个格子可以是空地可通行、障碍物不可通行或智能体所在位置。3.1 地图生成可控的随机性与课程学习基础一个固定的简单地图很快会让算法过拟合。因此我们实现一个随机的、参数化的地图生成器。import numpy as np class GridWorld: def __init__(self, width15, height15, obstacle_density0.2): self.width width self.height height self.grid np.zeros((height, width), dtypenp.int8) # 0: 空地 1: 障碍物 self._generate_obstacles(obstacle_density) def _generate_obstacles(self, density): # 随机放置障碍物但确保起点区域和中心区域相对开阔 num_obstacles int(self.width * self.height * density) indices np.random.choice(self.width * self.height, num_obstacles, replaceFalse) for idx in indices: x, y idx % self.width, idx // self.width # 避免在边界和中心区域例如地图中心5x5区域放置障碍物以保证游戏可玩性 if not (self.width//2 - 2 x self.width//2 2 and self.height//2 - 2 y self.height//2 2): self.grid[y, x] 1 def is_valid_position(self, x, y): 检查位置是否在地图范围内且不是障碍物 return 0 x self.width and 0 y self.height and self.grid[y, x] 0注意完全随机的障碍物可能生成无法通行的孤岛或包围圈导致智能体一开始就被困死。上述代码中避免在中心区域放置障碍物是一种简单的启发式方法。更健壮的做法是使用连通性检查算法如BFS确保所有可通行格子是连通的。3.2 智能体类追捕者与逃跑者的异同追捕者和逃跑者在本质上都是智能体拥有位置、速度、视野等共同属性。但它们的目标不同这主要通过奖励函数来体现而非物理属性。因此我们可以用一个基类Agent来定义共同属性并通过一个agent_type字段来区分。class Agent: def __init__(self, agent_id, x, y, agent_typepursuer, sight_range5, max_speed1): self.id agent_id self.x x self.y y self.type agent_type # pursuer 或 evader self.sight_range sight_range # 观测范围曼哈顿距离或欧氏距离 self.max_speed max_speed # 每步最大移动距离网格数 self.captured False # 是否被捕获对逃跑者或已完成任务对追捕者 self.last_action None def move(self, action, world): 根据动作更新位置。action是一个离散值或连续向量。 # 示例离散动作空间0-3代表上下左右4代表不动 dx, dy 0, 0 if action 0: dy -1 # 上 elif action 1: dy 1 # 下 elif action 2: dx -1 # 左 elif action 3: dx 1 # 右 # action 4 代表不动 # 计算新位置 new_x self.x dx * self.max_speed new_y self.y dy * self.max_speed # 边界和障碍物检查 new_x np.clip(new_x, 0, world.width - 1) new_y np.clip(new_y, 0, world.height - 1) if world.is_valid_position(int(new_x), int(new_y)): self.x, self.y new_x, new_y # 如果新位置无效可以选择停留在原地如上或者进行滑动处理沿障碍物边缘 self.last_action action这里有一个关键细节max_speed1意味着智能体每步只能移动到相邻格子。如果你想模拟更真实的连续运动可以将位置(x, y)改为浮点数并在move方法中处理连续坐标到网格的映射。但离散网格能极大简化碰撞检测和观测生成非常适合算法验证初期。4. 核心模块二多智能体观测空间的设计哲学观测空间的设计直接决定了智能体能获取多少信息以及算法的学习难度。在部分可观测Partially Observable的多智能体环境中每个智能体只能看到世界的一部分。4.1 局部网格观测一种通用且强大的表示对于基于网格的世界一个直观的观测是为每个智能体返回一个以其为中心的局部网格“快照”。这个快照是一个3D张量[H, W, C]其中H和W是局部网格的高度和宽度由sight_range决定C是通道数表示不同语义信息。def get_local_obs(self, agent, world): 获取智能体周围的局部网格观测 sight agent.sight_range local_grid np.zeros((2*sight1, 2*sight1, 4), dtypenp.float32) # 4个通道 for i in range(-sight, sight1): for j in range(-sight, sight1): map_x, map_y int(agent.x j), int(agent.y i) # 通道0: 障碍物 if not (0 map_x world.width and 0 map_y world.height): local_grid[isight, jsight, 0] 1.0 # 边界视为障碍 elif world.grid[map_y, map_x] 1: local_grid[isight, jsight, 0] 1.0 # 通道1: 当前智能体自身 if i 0 and j 0: local_grid[isight, jsight, 1] 1.0 # 通道2: 友方智能体同类型 # 通道3: 敌方智能体不同类型 for other in world.agents: if other.id agent.id: continue if int(other.x) map_x and int(other.y) map_y: if other.type agent.type: local_grid[isight, jsight, 2] 1.0 else: local_grid[isight, jsight, 3] 1.0 return local_grid为什么选择这种多通道表示可分离性神经网络的不同卷积核可以轻松学习到障碍物、自身、友军、敌军的独立特征这比用一个通道混合编码所有信息如用不同数字代表不同物体更容易学习。扩展性你可以轻松增加更多通道例如加入“目标点”、“资源”等信息。兼容性这种格式天然适合卷积神经网络CNN处理是处理空间信息的标准方式。4.2 矢量观测与全局信息的权衡局部网格观测虽然信息丰富但维度较高(2*sight1)^2 * channels。对于需要快速迭代的算法研究或者当智能体需要知道全局队友位置以进行协同如集中围捕时我们可以提供矢量观测作为补充或替代。矢量观测可以包括自身绝对坐标归一化后。所有友方智能体的相对位置和状态。所有敌方智能体的相对位置和状态如果在视野内。自身与最近障碍物的距离和方向。def get_vector_obs(self, agent, world): obs [agent.x / world.width, agent.y / world.height] # 归一化自身位置 for other in world.agents: if other.id agent.id: continue dx (other.x - agent.x) / world.width dy (other.y - agent.y) / world.height dist np.sqrt(dx**2 dy**2) in_sight dist agent.sight_range obs.extend([dx, dy, 1.0 if other.type agent.type else 0.0, 1.0 if in_sight else 0.0]) # 如果智能体数量不固定这里需要padding或masking这是一个常见的工程难点。 return np.array(obs, dtypenp.float32)实操心得在实际项目中我强烈建议同时提供局部网格观测和矢量观测并在环境信息info字典中返回。这样算法开发者可以根据需要选择使用哪一种或者尝试将两者融合例如用CNN处理局部网格用MLP处理矢量再拼接起来。这增加了平台的灵活性。5. 核心模块三奖励函数工程——引导智能体学会博弈奖励函数是强化学习的“指挥棒”在多智能体追逃博弈中尤其微妙。设计不当会导致智能体学到奇怪的行为比如追捕者互相挡路或者逃跑者躲在角落不动。5.1 基础奖励组件我们为追捕者和逃跑者分别设计奖励通常包含以下部分对于追捕者Pursuer捕获奖励当成功捕获一个逃跑者时给予一个大的正奖励如10。这是最终目标。距离奖励每一步根据与最近逃跑者距离的缩小程度给予一个小奖励。这鼓励追捕者主动靠近目标而不是原地等待。公式可以是reward_distance (old_distance - new_distance) * scale。团队协作奖励可选如果多个追捕者同时靠近同一个逃跑者给予额外奖励鼓励包围行为。时间惩罚每一步给予一个小的负奖励如-0.01鼓励快速解决战斗防止智能体学会拖延。对于逃跑者Evader生存奖励每存活一步给予一个小的正奖励如0.05鼓励尽可能长时间生存。距离惩罚当与最近追捕者的距离过近时给予惩罚。这鼓励逃跑者保持距离。被捕惩罚被捕获时给予一个大的负奖励如-10。5.2 实现细节与“稀疏奖励”问题直接实现上述距离奖励可能会遇到“稀疏奖励”问题在早期智能体动作随机很难偶然缩小距离因此几乎得不到正向反馈学习缓慢。解决方案使用“势能函数”Potential-Based Reward Shaping势能函数是一种在不改变最优策略的前提下提供更密集学习信号的技术。其核心思想是将奖励定义为新状态势能与旧状态势能之差r_shaped r γ * Φ(s) - Φ(s)其中Φ(s)是状态s的势能函数γ是折扣因子。在我们的场景中可以定义势能Φ(s)为追捕者到逃跑者的负距离或逃跑者到追捕者的距离。这样即使没有发生捕获只要距离缩小r_shaped就会为正。这能显著加速初期学习。def get_shaped_reward(self, agent, world, old_positions): 计算基于势能的整形奖励 base_reward self._get_base_reward(agent, world) # 基础奖励如时间惩罚 potential_now self._compute_potential(agent, world) potential_prev self._compute_potential(agent, world, old_positions) shaped_reward base_reward self.gamma * potential_now - potential_prev return shaped_reward def _compute_potential(self, agent, world, positionsNone): 计算势能例如对于追捕者是到最近逃跑者距离的负数 if agent.type pursuer: min_dist float(inf) for other in world.agents: if other.type evader and not other.captured: dist np.sqrt((agent.x - other.x)**2 (agent.y - other.y)**2) min_dist min(min_dist, dist) return -min_dist if min_dist ! float(inf) else 0.0 else: # 对于逃跑者势能可以是到最近追捕者距离的相反数或者一个常数 # 逃跑者更复杂因为它的目标是最大化生存时间而非最小化某个距离 # 一种简单做法是使用生存时间作为势能 return 0.0 # 这里简化处理踩坑记录奖励整形是一把双刃剑。如果势能函数设计得不好例如与最终目标不一致可能会引入“局部最优”导致智能体学会利用势能函数“刷分”而非真正完成任务。例如如果只奖励追捕者靠近它们可能会紧紧跟着逃跑者但永远不完成捕获如果捕获需要特殊动作。因此捕获奖励必须足够大以覆盖整形奖励可能带来的偏差。6. 核心模块四Gym环境接口的完整实现与关键调试技巧现在我们将所有模块组装成一个标准的Gym环境。这是算法库如Stable-Baselines3, RLlib与我们的自定义环境交互的桥梁。6.1 环境初始化与reset方法import gym from gym import spaces import numpy as np class PursuitEvasionEnv(gym.Env): metadata {render.modes: [human, rgb_array]} def __init__(self, world_size15, n_pursuers2, n_evaders1, obstacle_density0.15, sight_range5): super(PursuitEvasionEnv, self).__init__() self.world_size world_size self.n_pursuers n_pursuers self.n_evaders n_evaders self.n_agents n_pursuers n_evaders # 初始化世界和智能体 self.world GridWorld(world_size, world_size, obstacle_density) self.agents [] self._init_agents() # 定义多智能体的动作和观测空间 # 动作空间每个智能体有5个离散动作上下左右停 self.action_space spaces.Dict({ fagent_{i}: spaces.Discrete(5) for i in range(self.n_agents) }) # 观测空间每个智能体获得一个局部网格观测 obs_shape (2*sight_range1, 2*sight_range1, 4) # H, W, C self.observation_space spaces.Dict({ fagent_{i}: spaces.Box(low0, high1, shapeobs_shape, dtypenp.float32) for i in range(self.n_agents) }) # 渲染相关 self.viewer None def _init_agents(self): # 在地图上随机、不重叠地放置智能体 positions [] for i in range(self.n_agents): while True: x, y np.random.randint(0, self.world_size, size2) if self.world.is_valid_position(x, y) and all((x,y) ! p for p in positions): positions.append((x, y)) break agent_type pursuer if i self.n_pursuers else evader self.agents.append(Agent(i, x, y, agent_type)) def reset(self): 重置环境到初始状态 self.world GridWorld(self.world_size, self.world_size, obstacle_density0.15) self._init_agents() observations self._get_observations() return observations6.2step方法多智能体交互的核心这是整个环境最复杂也最重要的方法。它需要处理1并行执行所有智能体动作2更新世界状态3检测碰撞与捕获4计算奖励和终止条件5收集观测。def step(self, actions): actions: 字典键为agent_i值为动作索引0-4 返回: observations, rewards, dones, infos 其中observations, rewards, dones都是字典infos可以是包含额外信息的字典。 # 1. 保存旧位置用于势能计算 old_positions [(a.x, a.y) for a in self.agents] # 2. 并行应用所有动作实际是顺序执行但在同一时间步内 for agent_id, action in actions.items(): agent_idx int(agent_id.split(_)[1]) self.agents[agent_idx].move(action, self.world) # 3. 处理捕获逻辑 rewards {fagent_{i}: 0.0 for i in range(self.n_agents)} for evader in [a for a in self.agents if a.type evader and not a.captured]: for pursuer in [a for a in self.agents if a.type pursuer]: # 简单捕获条件位置重合 if int(evader.x) int(pursuer.x) and int(evader.y) int(pursuer.y): evader.captured True # 奖励分配 rewards[fagent_{pursuer.id}] 10.0 # 追捕者获得捕获奖励 rewards[fagent_{evader.id}] - 10.0 # 逃跑者获得被捕惩罚 break # 一个逃跑者只能被捕获一次 # 4. 计算基于势能的距离奖励/惩罚和时间惩罚 for i, agent in enumerate(self.agents): if agent.type pursuer and not agent.captured: # 找到最近的存活逃跑者 min_dist float(inf) for evader in [a for a in self.agents if a.type evader and not a.captured]: dist np.sqrt((agent.x - evader.x)**2 (agent.y - evader.y)**2) min_dist min(min_dist, dist) if min_dist ! float(inf): # 简单距离奖励距离缩小给予小奖励 old_dist np.sqrt((old_positions[i][0] - evader.x)**2 (old_positions[i][1] - evader.y)**2) rewards[fagent_{i}] (old_dist - min_dist) * 0.1 # 缩放因子 # 时间惩罚鼓励快速决策 rewards[fagent_{i}] - 0.01 # 5. 检查终止条件 dones {fagent_{i}: False for i in range(self.n_agents)} # 全局终止条件所有逃跑者都被捕获或达到最大步数需要维护一个step计数器 self.current_step 1 all_evaders_captured all(a.captured for a in self.agents if a.type evader) if all_evaders_captured or self.current_step self.max_steps: done_global True for i in range(self.n_agents): dones[fagent_{i}] True # 如果是步数限制终止可以给未死亡的逃跑者一个生存奖励 if self.current_step self.max_steps: agent self.agents[i] if agent.type evader and not agent.captured: rewards[fagent_{i}] 5.0 else: done_global False # 单个智能体的终止如被捕获 for i, agent in enumerate(self.agents): if agent.type evader and agent.captured: dones[fagent_{i}] True # 6. 获取新观测 observations self._get_observations() # 7. info字典可以包含调试信息如原始距离、势能值等 infos {fagent_{i}: {} for i in range(self.n_agents)} return observations, rewards, dones, infos6.3 关键调试技巧可视化与日志在开发多智能体环境时可视化是必不可少的调试工具。Gym的render方法可以帮助我们直观地看到智能体的行为。def render(self, modehuman): if mode human: if self.viewer is None: from gym.envs.classic_control import rendering # 或其他渲染后端 # 初始化viewer绘制网格、障碍物、智能体 # ... 具体绘图代码取决于使用的渲染库如pygame, matplotlib pass # 更新所有元素的位置 # ... return self.viewer.render(return_rgb_arraymodergb_array) elif mode rgb_array: # 返回一个RGB数组用于录制视频或远程显示 # 可以简单用matplotlib生成一帧 fig, ax plt.subplots(figsize(6,6)) ax.imshow(self.world.grid, cmapGreys, interpolationnearest) for agent in self.agents: color red if agent.type pursuer else blue ax.scatter(agent.x, agent.y, ccolor, s100, markero if agent.typepursuer else s) ax.set_xlim(-0.5, self.world_size-0.5) ax.set_ylim(-0.5, self.world_size-0.5) ax.invert_yaxis() # 让y轴向下增长符合数组索引 fig.canvas.draw() image np.frombuffer(fig.canvas.tostring_rgb(), dtypenp.uint8) image image.reshape(fig.canvas.get_width_height()[::-1] (3,)) plt.close(fig) return image除了可视化详细的日志记录也至关重要。我建议在step函数的关键位置加入日志记录每个智能体的动作、奖励、距离等信息。特别是在奖励函数复杂时打印出每一步各分项奖励的值能帮你快速定位是哪个奖励组件导致了异常行为比如奖励爆炸或始终为零。一个常见的坑是dones字典的处理。有些算法框架如早期的RLlib要求dones字典里必须有一个__all__键来表示全局终止。为了兼容性最好在infos或单独设置一个全局done标志。最稳妥的方式是查阅你计划使用的算法库的文档看它们对多智能体环境的done信号有何要求。7. 平台使用示例连接主流MARL算法库环境搭建好后我们就可以用它来训练算法了。这里以两个流行的库为例展示如何接入。7.1 使用Stable-Baselines3和PettingZoo适配器Stable-Baselines3 (SB3) 本身主要支持单智能体但可以通过PettingZoo的ParallelEnv接口进行适配。我们的环境本身是并行式Simultaneous-Move的与ParallelEnv兼容。首先需要让我们的环境遵循PettingZoo的ParallelEnvAPI主要是agents属性、possible_agents和last方法。或者更简单的方法是使用PettingZoo提供的ss超级环境工具进行包装。# 假设我们已经将环境改造成了符合PettingZoo ParallelEnv的类 PursuitEvasionParallelEnv from pettingzoo import AECEnv, ParallelEnv from stable_baselines3 import PPO from stable_baselines3.common.env_util import make_vec_env from sb3_contrib import RecurrentPPO # 如果需要RNN # 由于SB3直接支持的是gym.Env我们需要一个包装器将多智能体环境转化为单智能体环境通过智能体ID拼接观测和动作 # 一种常见做法是使用“集中式训练分散式执行”CTDE框架如MADDPG但这需要自定义算法。 # 对于SB3一个简单的基线是将其视为一个“大”的单智能体其动作空间是所有智能体动作的笛卡尔积。 # 但这在智能体较多时动作空间会爆炸。因此更实用的方法是使用支持多智能体的库如RLlib。 print(使用SB3进行多智能体训练通常需要自定义包装或使用像MATMulti-Agent Transformer这样的第三方扩展。)7.2 使用RLlib进行分布式训练推荐RLlib原生支持多智能体强化学习是我们的理想选择。它允许我们为不同类型的智能体配置不同的策略网络甚至共享参数。import ray from ray import tune from ray.rllib.algorithms.ppo import PPOConfig from ray.rllib.env.multi_agent_env import MultiAgentEnv from ray.rllib.policy.policy import PolicySpec # 1. 将我们的环境包装成RLlib的MultiAgentEnv class RllibPursuitEvasion(MultiAgentEnv): def __init__(self, config): self.env PursuitEvasionEnv(**config.get(env_config, {})) self.agents [fagent_{i} for i in range(self.env.n_agents)] self._agent_ids set(self.agents) self.observation_space self.env.observation_space self.action_space self.env.action_space # ... 实现reset和step将返回的字典键改为RLlib期望的格式 def reset(self): obs self.env.reset() # 确保obs是字典且key是agent id return obs def step(self, action_dict): # action_dict的key是agent id obs, rewards, dones, infos self.env.step(action_dict) # RLlib期望dones字典里有一个__all__表示全局终止 dones[__all__] all(dones.values()) return obs, rewards, dones, infos # 2. 配置和运行训练 ray.init(ignore_reinit_errorTrue) config ( PPOConfig() .environment(RllibPursuitEvasion, env_config{world_size: 10, n_pursuers: 2, n_evaders: 1}) .multi_agent( policies{ pursuer_policy: PolicySpec( # 追捕者策略 configPPOConfig.overrides(framework_strtorch) ), evader_policy: PolicySpec( # 逃跑者策略 configPPOConfig.overrides(framework_strtorch) ), }, policy_mapping_fnlambda agent_id, episode, worker, **kwargs: pursuer_policy if pursuer in agent_id else evader_policy, # 可以设置policies_to_train来指定训练哪些策略 ) .resources(num_gpus0) # 根据实际情况调整 .rollouts(num_rollout_workers1) ) algo config.build() for i in range(100): # 训练100次迭代 result algo.train() print(fIteration {i}: reward{result[episode_reward_mean]}) ray.shutdown()在RLlib中的关键配置policy_mapping_fn这个函数决定了哪个智能体使用哪个策略。在我们的例子中根据agent_id是否包含“pursuer”来分配。策略共享我们可以让所有追捕者共享同一个策略网络pursuer_policy所有逃跑者共享另一个。这能有效减少参数并促进协作。训练模式可以设置policies_to_train[pursuer_policy]来只训练追捕者而让逃跑者使用固定策略如随机策略这在课程学习中很有用。7.3 训练过程中的监控与评估训练多智能体系统时监控指标比单智能体更复杂。除了总回报你还需要关注各策略的平均回报追捕者和逃跑者的回报应该呈现此消彼长的竞争关系。捕获成功率/平均存活时间这是最直接的性能指标。智能体的探索情况通过渲染视频观察智能体是学会了有效的策略还是在做无意义的随机运动。我通常会在训练循环中定期比如每10次迭代调用env.render()并将画面保存下来生成一个训练过程的GIF动画这是最直观的评估方式。同时将上述指标记录到TensorBoard中方便分析学习曲线的趋势。8. 进阶扩展与未来研究方向这个基础平台已经可以用于许多有趣的实验。但它的真正价值在于其可扩展性。以下是一些可以尝试的进阶方向8.1 增加环境复杂性连续动作空间将智能体的移动从离散网格改为连续的速度和方向控制。这需要修改动作空间Box和Agent.move方法并可能引入更复杂的物理引擎如PyBullet进行碰撞检测。部分可观测与通信限制每个智能体的视野FOV而非全向视野并引入通信信道让智能体可以传递有限的信息。这可以研究涌现的通信协议。动态环境与地形效应加入可移动的障碍物、奖励物品如加速道具或不同摩擦系数的地形如沼泽地减速。异构智能体让追捕者拥有不同的速度、视野或抓捕能力让逃跑者拥有隐身、挖地道等特殊技能。8.2 算法实验与对比集中式 vs 分散式对比像MADDPG集中式批评家和IQL完全分散式这类算法的性能差异。合作与竞争将奖励函数从零和博弈追捕者得分即逃跑者失分改为混合动机博弈研究合作与竞争的平衡。课程学习与课程教学从简单的空旷地图开始训练逐步增加障碍物密度或智能体数量。也可以让一个已经训练好的策略作为老师来指导新策略的学习。8.3 工程优化与部署向量化环境使用SubprocVecEnv或Ray进行环境并行化大幅提升数据采集速度。自定义神经网络架构为局部网格观测设计专用的CNN为矢量观测设计MLP并研究如何有效地融合两者例如通过注意力机制。部署到真实机器人虽然从仿真到现实存在巨大的鸿沟Sim2Real Gap但这个轻量化的平台可以作为算法在真实机器人如小型差分轮式机器人上部署前的快速验证沙盒。搭建这个平台的过程本身就是一个深入理解多智能体强化学习核心概念的过程。从定义观测空间、设计奖励函数到处理多智能体交互的逻辑每一步都需要仔细权衡。我希望这份详细的拆解和附带的思考能帮助你不仅复现这个平台更能理解其背后的设计哲学并在此基础上开展属于你自己的创新研究。代码的最终版本我会整理后开源其中包含了更完善的错误处理、更丰富的配置选项以及更多的示例脚本。记住在MARL的研究中一个稳定、灵活、可解释的环境往往是成功的一半。本文还有配套的精品资源点击获取