企业官网建设流程全解析

简介：本资源是一套基于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框架下的多智能体追逃博弈强化学习平台”的初衷。我希望它不仅仅是一段能跑的代码，更是一个高度模块化、易于扩展的教学与研究平台。你可以用它来：

1. 快速验证MARL算法：无论是经典的独立Q学习（IQL）、多智能体深度确定性策略梯度（MADDPG），还是较新的QMIX、MAPPO，都可以在这个平台上进行基准测试。
2. 深入理解环境设计：完全掌控地图生成、智能体动力学模型、碰撞检测、奖励函数设计等底层细节。
3. 进行课程学习或课程教学：通过调整地图复杂度、智能体数量、传感器范围等参数，构建由易到难的学习序列。

这个平台完全基于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）兼容，我们选择返回字典形式。

我们的平台架构分为四层：

1. 环境层（Environment）：继承自gym.Env，负责实现核心的reset和step方法，管理整个世界的状态。
2. 世界层（World）：这是环境的“引擎”。它包含地图（Grid）、所有智能体（Agent）对象，并负责物理更新、碰撞检测、视距计算等。
3. 智能体层（Agent）：定义追捕者和逃跑者的属性（位置、速度、视野半径等）和行为模型（如何根据动作移动）。
4. 观测与奖励层（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, width=15, height=15, obstacle_density=0.2): self.width = width self.height = height self.grid = np.zeros((height, width), dtype=np.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, replace=False) 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_type='pursuer', sight_range=5, max_speed=1): 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_speed=1意味着智能体每步只能移动到相邻格子。如果你想模拟更真实的连续运动，可以将位置(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*sight+1, 2*sight+1, 4), dtype=np.float32) # 4个通道 for i in range(-sight, sight+1): for j in range(-sight, sight+1): 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[i+sight, j+sight, 0] = 1.0 # 边界视为障碍 elif world.grid[map_y, map_x] == 1: local_grid[i+sight, j+sight, 0] = 1.0 # 通道1: 当前智能体自身 if i == 0 and j == 0: local_grid[i+sight, j+sight, 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[i+sight, j+sight, 2] = 1.0 else: local_grid[i+sight, j+sight, 3] = 1.0 return local_grid

为什么选择这种多通道表示？

• 可分离性：神经网络的不同卷积核可以轻松学习到障碍物、自身、友军、敌军的独立特征，这比用一个通道混合编码所有信息（如用不同数字代表不同物体）更容易学习。
• 扩展性：你可以轻松增加更多通道，例如加入“目标点”、“资源”等信息。
• 兼容性：这种格式天然适合卷积神经网络（CNN）处理，是处理空间信息的标准方式。

4.2 矢量观测与全局信息的权衡

局部网格观测虽然信息丰富，但维度较高（(2*sight+1)^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, dtype=np.float32)

实操心得：在实际项目中，我强烈建议同时提供局部网格观测和矢量观测，并在环境信息info字典中返回。这样，算法开发者可以根据需要选择使用哪一种，或者尝试将两者融合（例如，用CNN处理局部网格，用MLP处理矢量，再拼接起来）。这增加了平台的灵活性。

5. 核心模块三：奖励函数工程——引导智能体学会博弈

奖励函数是强化学习的“指挥棒”，在多智能体追逃博弈中尤其微妙。设计不当会导致智能体学到奇怪的行为，比如追捕者互相挡路，或者逃跑者躲在角落不动。

5.1 基础奖励组件

我们为追捕者和逃跑者分别设计奖励，通常包含以下部分：

对于追捕者（Pursuer）：

1. 捕获奖励：当成功捕获一个逃跑者时，给予一个大的正奖励（如+10）。这是最终目标。
2. 距离奖励：每一步，根据与最近逃跑者距离的缩小程度给予一个小奖励。这鼓励追捕者主动靠近目标，而不是原地等待。公式可以是：reward_distance = (old_distance - new_distance) * scale。
3. 团队协作奖励（可选）：如果多个追捕者同时靠近同一个逃跑者，给予额外奖励，鼓励包围行为。
4. 时间惩罚：每一步给予一个小的负奖励（如-0.01），鼓励快速解决战斗，防止智能体学会拖延。

对于逃跑者（Evader）：

1. 生存奖励：每存活一步，给予一个小的正奖励（如+0.05），鼓励尽可能长时间生存。
2. 距离惩罚：当与最近追捕者的距离过近时，给予惩罚。这鼓励逃跑者保持距离。
3. 被捕惩罚：被捕获时给予一个大的负奖励（如-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, positions=None): """计算势能，例如对于追捕者，是到最近逃跑者距离的负数""" 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_size=15, n_pursuers=2, n_evaders=1, obstacle_density=0.15, sight_range=5): 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({ f'agent_{i}': spaces.Discrete(5) for i in range(self.n_agents) }) # 观测空间：每个智能体获得一个局部网格观测 obs_shape = (2*sight_range+1, 2*sight_range+1, 4) # H, W, C self.observation_space = spaces.Dict({ f'agent_{i}': spaces.Box(low=0, high=1, shape=obs_shape, dtype=np.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, size=2) 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_density=0.15) self._init_agents() observations = self._get_observations() return observations

6.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 = {f'agent_{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[f'agent_{pursuer.id}'] += 10.0 # 追捕者获得捕获奖励 rewards[f'agent_{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[f'agent_{i}'] += (old_dist - min_dist) * 0.1 # 缩放因子 # 时间惩罚（鼓励快速决策） rewards[f'agent_{i}'] -= 0.01 # 5. 检查终止条件 dones = {f'agent_{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[f'agent_{i}'] = True # 如果是步数限制终止，可以给未死亡的逃跑者一个生存奖励 if self.current_step >= self.max_steps: agent = self.agents[i] if agent.type == 'evader' and not agent.captured: rewards[f'agent_{i}'] += 5.0 else: done_global = False # 单个智能体的终止（如被捕获） for i, agent in enumerate(self.agents): if agent.type == 'evader' and agent.captured: dones[f'agent_{i}'] = True # 6. 获取新观测 observations = self._get_observations() # 7. info字典可以包含调试信息，如原始距离、势能值等 infos = {f'agent_{i}': {} for i in range(self.n_agents)} return observations, rewards, dones, infos

6.3 关键调试技巧：可视化与日志

在开发多智能体环境时，可视化是必不可少的调试工具。Gym的render方法可以帮助我们直观地看到智能体的行为。

def render(self, mode='human'): 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_array=mode=='rgb_array') elif mode == 'rgb_array': # 返回一个RGB数组，用于录制视频或远程显示 # 可以简单用matplotlib生成一帧 fig, ax = plt.subplots(figsize=(6,6)) ax.imshow(self.world.grid, cmap='Greys', interpolation='nearest') for agent in self.agents: color = 'red' if agent.type == 'pursuer' else 'blue' ax.scatter(agent.x, agent.y, c=color, s=100, marker='o' if agent.type=='pursuer' 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(), dtype=np.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进行多智能体训练通常需要自定义包装或使用像MAT（Multi-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 = [f"agent_{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_error=True) config = ( PPOConfig() .environment(RllibPursuitEvasion, env_config={"world_size": 10, "n_pursuers": 2, "n_evaders": 1}) .multi_agent( policies={ "pursuer_policy": PolicySpec( # 追捕者策略 config=PPOConfig.overrides(framework_str="torch") ), "evader_policy": PolicySpec( # 逃跑者策略 config=PPOConfig.overrides(framework_str="torch") ), }, policy_mapping_fn=lambda agent_id, episode, worker, **kwargs: "pursuer_policy" if "pursuer" in agent_id else "evader_policy", # 可以设置policies_to_train来指定训练哪些策略 ) .resources(num_gpus=0) # 根据实际情况调整 .rollouts(num_rollout_workers=1) ) algo = config.build() for i in range(100): # 训练100次迭代 result = algo.train() print(f"Iteration {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的研究中，一个稳定、灵活、可解释的环境，往往是成功的一半。

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