基于Python与Gym构建多智能体追逃博弈平台:从MARL原理到实战

发布时间:2026/9/5 12:55:42
基于Python与Gym构建多智能体追逃博弈平台:从MARL原理到实战 简介本资源是一套基于Python与OpenAI Gym框架构建的多智能体追逃博弈强化学习平台源代码面向强化学习初学者、多智能体系统研究者及算法工程师旨在解决追捕者与逃避者动态博弈建模、环境定制与策略训练等核心问题适用于机器人协同、无人机对抗、网络安全攻防等典型MARL应用场景。压缩包共41个文件含27个核心Python模块涵盖3D/2D仿真环境、JSBSim飞行模型接口、测试脚本及环境封装、9个编译缓存文件、1份依赖说明requirements.txt、1份README文档及许可证等整体仅79KB轻量易部署。已有243人学习下载代码结构清晰模块化envs目录实现多维追逃环境抽象test目录提供可运行验证用例支持快速接入DQN、PPO等主流算法进行策略训练与对比实验。读者可直接复现博弈场景、自定义智能体角色逻辑、调整观测空间与奖励函数并基于现有架构扩展合作型或混合博弈任务。1. 项目概述从单智能体到多智能体的博弈世界如果你已经玩过 OpenAI Gym 里的 CartPole 或者 MountainCar对强化学习RL的基本流程——智能体观察环境、采取行动、获得奖励、更新策略——感到熟悉那么是时候进入一个更复杂、也更有趣的领域了多智能体强化学习MARL。今天要聊的这个项目就是基于 Python 和 Gym 框架亲手搭建一个“追逃博弈”的仿真平台。简单来说你可以把它想象成一个数字化的“猫捉老鼠”游戏场只不过这里的“猫”和“老鼠”都是由强化学习算法驱动的智能体它们在这个场地里学习如何追逐、如何躲避。为什么这个项目值得一做因为在现实世界中纯粹的单一智能体决策场景越来越少。从自动驾驶汽车在复杂路况下的交互到多机器人协同搬运再到金融市场的多空博弈本质上都是多个智能体在共享环境中相互竞争或合作。追逃博弈Pursuit-Evasion Game是研究这类交互的一个经典范式它剥离了复杂的环境细节聚焦于最核心的对抗性策略学习。通过构建这个平台你不仅能深入理解 MARL 的核心挑战如非平稳性、信用分配等还能获得一套可扩展的代码框架用于验证各种前沿算法从经典的 MADDPG 到最新的 MAPPO。这个平台适合所有希望从单智能体 RL 迈向多智能体 RL 的实践者。无论你是想为自己的研究课题找一个快速验证想法的沙盒还是想通过一个完整的项目来深化对 MARL 的理解甚至是为未来的游戏 AI 或机器人集群控制打基础从这里开始都是一个绝佳的选择。接下来我会带你从设计思路到代码实现完整地走一遍构建过程并分享那些在官方教程里不会写的“踩坑”经验。2. 平台核心设计思路与架构拆解在动手写代码之前我们必须把设计思路理清楚。一个稳健的多智能体平台其架构决定了后续算法实验的便捷性和代码的可维护性。2.1 环境设计定义追逃世界的规则追逃环境是整个平台的基础。我们的设计目标是高度可配置、易于观察、奖励函数清晰。首先我们定义世界的核心参数。一个二维的网格世界Grid World是个不错的起点因为它直观且易于可视化。我们需要定义网格大小如 10x10追捕者Pursuer和逃跑者Evader的初始位置、数量例如 2 个追捕者对 1 个逃跑者以及智能体的移动方式通常为四向或八向移动。障碍物是可选项加入后能显著增加环境的复杂性和策略的丰富性。注意在初始设计时建议先实现无障碍物的“空旷场地”版本。这能确保你的智能体学习基础策略如围堵时不受干扰待核心逻辑跑通后再加入障碍物作为环境复杂度的扩展。状态State的设计至关重要。对于每个智能体其观察Observation可以包括自身绝对坐标智能体知道自己在哪里。相对位置信息这是关键。例如对于追捕者它需要知道目标逃跑者的相对方向Δx, Δy和距离。对于逃跑者它需要知道所有追捕者的相对位置。这种设计符合“局部观察”的设定更贴近现实。全局信息可选为了简化早期学习难度你也可以在训练初期提供全局地图信息如所有智能体的位置但在评估时切换为局部观察以测试算法的泛化能力。动作Action空间通常是离散的。例如[0: 上 1: 下 2: 左 3: 右 4: 静止]。连续动作空间虽然更精细但会大幅增加算法设计和训练的复杂度建议在离散动作版本稳定后再考虑。奖励函数Reward是引导智能体学习的“指挥棒”。设计时需要兼顾稀疏性与密集性。追捕者奖励捕获奖励当追捕者与逃跑者处于同一格时给予一个大的正奖励如 10。距离奖励密集奖励每一步给予一个与到逃跑者距离缩短成比例的微小正奖励如 -0.01 * 距离变化。这能缓解稀疏奖励问题帮助智能体在初期找到学习方向。时间惩罚每一步给予一个小的负奖励如 -0.05鼓励快速结束回合。逃跑者奖励生存奖励每存活一步给予一个小的正奖励如 0.1。被捕获惩罚被抓住时给予一个大的负奖励如 -10。距离奖励每一步给予一个与到最近追捕者距离增加成比例的微小正奖励。这样的设计使得奖励信号既包含最终目标的稀疏信号也包含引导策略优化的密集信号。2.2 智能体架构独立学习与集中训练多智能体系统的核心架构决策是智能体是独立学习还是协同学习在追逃博弈中双方利益完全对立属于竞争关系。一种简单的方法是让每个智能体独立运行一个 DQNDeep Q-Network算法将其它智能体视为环境的一部分。这种方法实现简单但存在严重问题环境对于每个独立学习者来说是非平稳的因为其他智能体也在学习变化导致训练不稳定难以收敛。因此更先进的思路是采用“集中式训练分布式执行”Centralized Training with Decentralized Execution, CTDE的框架。这也是 MADDPG、QMIX 等知名 MARL 算法的核心思想。在我们的平台中可以这样实现分布式执行每个智能体在行动时只根据自己的局部观察如相对位置做出决策。集中式训练在训练时我们可以为智能体提供额外的信息。例如在训练追捕者时可以把所有追捕者的观察和动作甚至是逃跑者的观察作为输入去学习一个联合的批评家网络Critic这个批评家网络能够评估在全局视角下某个联合动作的好坏。而每个智能体自身的演员网络Actor则根据局部观察做出决策并依靠全局批评家的指导来更新策略。这种架构既能应对环境非平稳性又保证了执行时的分布式特性是构建高性能多智能体系统的关键。在我们的平台中我们会预留出这样的接口便于后续集成 MADDPG 等算法。2.3 Gym 接口封装标准化与可复用性为了能让我们的环境无缝接入现有的 RL 算法库如 Stable-Baselines3, Ray RLLib必须遵循 GymnasiumGym 的维护分支的接口规范。这主要意味着实现两个核心方法reset(seedNone): 重置环境到初始状态并返回初始观察。需要支持随机种子。step(action): 接收一个包含所有智能体动作的字典或列表执行一步环境更新返回(observations, rewards, terminated, truncated, info)。observations: 所有智能体的新观察。rewards: 所有智能体的即时奖励。terminated: 布尔值表示回合是否正常结束如逃跑者被抓住。truncated: 布尔值表示回合是否因步数限制而被截断。info: 包含额外信息的字典如调试数据。此外还需要定义observation_space和action_space。对于多智能体环境通常使用gym.spaces.Dict来为每个智能体定义其独立的空间。例如self.observation_spaces { f“pursuer_{i}”: gym.spaces.Box(low0, highgrid_size, shape(4,), dtypenp.float32) for i in range(self.n_pursuers) } self.observation_spaces[“evader_0”] gym.spaces.Box(...)这样封装后我们的环境就可以像任何标准 Gym 环境一样被使用了。3. 核心模块代码实现与解析理论说再多不如一行代码。接下来我们深入到几个核心模块的实现细节。我会以模块化的方式展示关键代码并解释其中的设计考量。3.1 环境类PursuitEvasionEnv的实现这是整个平台的心脏。我们继承gym.Env类来构建。import numpy as np import gymnasium as gym from gymnasium import spaces import matplotlib.pyplot as plt class PursuitEvasionEnv(gym.Env): metadata {render_modes: [human, rgb_array], render_fps: 4} def __init__(self, grid_size10, n_pursuers2, n_evaders1, max_steps100, obstacle_density0.0, render_modeNone): super().__init__() self.grid_size grid_size self.n_pursuers n_pursuers self.n_evaders n_evaders self.max_steps max_steps self.render_mode render_mode self.steps 0 # 生成障碍物 self.obstacles self._generate_obstacles(obstacle_density) # 定义动作空间离散5个动作 [上下左右停] self.action_space spaces.Discrete(5) # 为每个智能体创建独立的动作空间字典 self.action_spaces { **{f“pursuer_{i}”: self.action_space for i in range(n_pursuers)}, **{f“evader_{i}”: self.action_space for i in range(n_evaders)} } # 定义观察空间假设每个智能体观察自身坐标和所有对手的相对坐标 # 例如追捕者观察[自身x, 自身y, 目标Δx, 目标Δy] obs_low [0, 0, -grid_size, -grid_size] obs_high [grid_size, grid_size, grid_size, grid_size] self.observation_space spaces.Box(lownp.array(obs_low), highnp.array(obs_high), dtypenp.float32) self.observation_spaces { **{f“pursuer_{i}”: self.observation_space for i in range(n_pursuers)}, **{f“evader_{i}”: self.observation_space for i in range(n_evaders)} } # 初始化智能体位置 self.agent_positions {} def _generate_obstacles(self, density): obstacles [] if density 0: num_obstacles int(self.grid_size * self.grid_size * density) for _ in range(num_obstacles): pos (np.random.randint(0, self.grid_size), np.random.randint(0, self.grid_size)) # 确保障碍物不重叠简单实现 if pos not in obstacles: obstacles.append(pos) return obstacles def reset(self, seedNone, optionsNone): super().reset(seedseed) self.steps 0 # 随机初始化智能体位置确保不重叠且不在障碍物上 all_positions set(self.obstacles) self.agent_positions {} agent_ids [f“pursuer_{i}” for i in range(self.n_pursuers)] [f“evader_{i}” for i in range(self.n_evaders)] for agent_id in agent_ids: while True: pos (self.np_random.integers(0, self.grid_size), self.np_random.integers(0, self.grid_size)) if pos not in all_positions: self.agent_positions[agent_id] pos all_positions.add(pos) break observations self._get_observations() info {} return observations, info def _get_observations(self): obs {} evader_pos self.agent_positions[“evader_0”] # 假设只有一个逃跑者 for agent_id, pos in self.agent_positions.items(): if agent_id.startswith(‘pursuer’): # 追捕者观察自身坐标 相对于逃跑者的向量 rel_x evader_pos[0] - pos[0] rel_y evader_pos[1] - pos[1] obs[agent_id] np.array([pos[0], pos[1], rel_x, rel_y], dtypenp.float32) elif agent_id.startswith(‘evader’): # 逃跑者观察自身坐标 相对于最近追捕者的向量简化 pursuer_positions [self.agent_positions[pid] for pid in self.agent_positions if pid.startswith(‘pursuer’)] if pursuer_positions: # 计算到所有追捕者的距离取最近的一个 distances [np.linalg.norm(np.array(pos)-np.array(p_pos)) for p_pos in pursuer_positions] nearest_idx np.argmin(distances) nearest_pursuer pursuer_positions[nearest_idx] rel_x nearest_pursuer[0] - pos[0] rel_y nearest_pursuer[1] - pos[1] obs[agent_id] np.array([pos[0], pos[1], rel_x, rel_y], dtypenp.float32) else: obs[agent_id] np.array([pos[0], pos[1], 0, 0], dtypenp.float32) return obs def step(self, actions): # actions 是一个字典键为 agent_id值为动作索引 self.steps 1 terminated False truncated False # 1. 处理移动 new_positions {} for agent_id, action in actions.items(): pos self.agent_positions[agent_id] 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 np.clip(pos[0] dx, 0, self.grid_size - 1) new_y np.clip(pos[1] dy, 0, self.grid_size - 1) new_pos (new_x, new_y) # 碰撞检测不能移动到障碍物上 if new_pos not in self.obstacles: # 简易处理允许智能体重叠后续可改为阻挡 new_positions[agent_id] new_pos else: new_positions[agent_id] pos # 撞墙则留在原地 self.agent_positions.update(new_positions) # 2. 计算奖励和终止条件 rewards {agent_id: 0.0 for agent_id in self.agent_positions.keys()} evader_pos self.agent_positions[“evader_0”] pursuer_ids [pid for pid in self.agent_positions if pid.startswith(‘pursuer’)] # 检查是否捕获任一追捕者与逃跑者同格 for pid in pursuer_ids: if self.agent_positions[pid] evader_pos: terminated True rewards[pid] 10.0 # 捕获奖励 rewards[“evader_0”] - 10.0 # 被捕获惩罚 break # 假设一次捕获即结束 # 密集奖励基于距离变化 for pid in pursuer_ids: old_dist np.linalg.norm(np.array(self._prev_positions[pid]) - np.array(self._prev_positions[“evader_0”])) new_dist np.linalg.norm(np.array(self.agent_positions[pid]) - np.array(evader_pos)) rewards[pid] (old_dist - new_dist) * 0.01 # 距离缩短给正奖励 # 逃跑者的密集奖励与最近追捕者距离增加 if pursuer_ids: old_dists [np.linalg.norm(np.array(self._prev_positions[pid]) - np.array(self._prev_positions[“evader_0”])) for pid in pursuer_ids] new_dists [np.linalg.norm(np.array(self.agent_positions[pid]) - np.array(evader_pos)) for pid in pursuer_ids] rewards[“evader_0”] (min(new_dists) - min(old_dists)) * 0.01 # 距离增加给正奖励 # 时间惩罚/生存奖励 for pid in pursuer_ids: rewards[pid] - 0.05 rewards[“evader_0”] 0.1 # 步数限制 if self.steps self.max_steps: truncated True # 3. 获取新观察 observations self._get_observations() # 为下一轮计算保存当前位置 self._prev_positions self.agent_positions.copy() # 4. 信息字典 info {‘steps’: self.steps, ‘terminated’: terminated} return observations, rewards, terminated, truncated, info def render(self): if self.render_mode ‘human’: if not hasattr(self, ‘fig’): self.fig, self.ax plt.subplots(figsize(6,6)) plt.ion() # 交互模式 self.ax.clear() self.ax.set_xlim(-0.5, self.grid_size-0.5) self.ax.set_ylim(-0.5, self.grid_size-0.5) self.ax.set_xticks(range(self.grid_size)) self.ax.set_yticks(range(self.grid_size)) self.ax.grid(True) # 画障碍物 for obs in self.obstacles: self.ax.add_patch(plt.Rectangle((obs[0]-0.5, obs[1]-0.5), 1, 1, color‘gray’)) # 画智能体 for agent_id, pos in self.agent_positions.items(): color ‘red’ if agent_id.startswith(‘pursuer’) else ‘blue’ marker ‘s’ if agent_id.startswith(‘pursuer’) else ‘o’ # 追捕者方形逃跑者圆形 self.ax.plot(pos[0], pos[1], markermarker, colorcolor, markersize12, labelagent_id) plt.pause(0.5) # 控制渲染速度代码解析与心得位置初始化与碰撞reset方法中确保智能体和障碍物不重叠是关键。这里用了简单的循环重试在网格密度大时可能效率低但对于中小型网格足够。生产环境可能需要更高效的算法如 Poisson Disk Sampling。观察设计_get_observations函数体现了局部观察的思想。追捕者只关心逃跑者的相对位置逃跑者只关心最近追捕者的相对位置。这种设计降低了观察维度也迫使智能体学习基于相对关系的策略泛化性更好。奖励计算step方法中的奖励计算是核心逻辑。注意我们同时计算了稀疏奖励捕获/被捕获和密集奖励距离变化。密集奖励的系数如 0.01需要仔细调参太小则引导作用弱太大可能让智能体过于“短视”忽视最终目标。我的经验是从一个较小的值开始如 0.001根据训练曲线调整。渲染render方法使用 Matplotlib 实现简单可视化。plt.ion()和plt.pause()的组合可以实现动态更新。对于更复杂的可视化或需要录制视频的情况可以考虑使用pygame或gymnasium的RecordVideo包装器。3.2 智能体基类与独立 DQN 智能体实现为了支持多种算法我们先定义一个智能体基类然后实现一个最简单的独立 DQN 智能体。import torch import torch.nn as nn import torch.optim as optim import numpy as np from collections import deque import random class BaseAgent: def __init__(self, agent_id, observation_space, action_space): self.agent_id agent_id self.observation_space observation_space self.action_space action_space def act(self, observation, exploreTrue): raise NotImplementedError def store_transition(self, obs, action, reward, next_obs, done): raise NotImplementedError def learn(self): raise NotImplementedError class IndependentDQNAgent(BaseAgent): def __init__(self, agent_id, observation_space, action_space, learning_rate1e-3, gamma0.99, epsilon_start1.0, epsilon_end0.01, epsilon_decay0.995, buffer_size10000, batch_size64): super().__init__(agent_id, observation_space, action_space) self.action_dim action_space.n self.obs_dim observation_space.shape[0] self.gamma gamma self.epsilon epsilon_start self.epsilon_end epsilon_end self.epsilon_decay epsilon_decay self.batch_size batch_size # Q-Network 和目标网络 self.q_net self._build_network() self.target_q_net self._build_network() self.target_q_net.load_state_dict(self.q_net.state_dict()) self.target_update_freq 100 # 每100步更新一次目标网络 self.learn_step_counter 0 self.optimizer optim.Adam(self.q_net.parameters(), lrlearning_rate) self.loss_fn nn.MSELoss() # 经验回放缓冲区 self.buffer deque(maxlenbuffer_size) def _build_network(self): # 一个简单的三层全连接网络 return nn.Sequential( nn.Linear(self.obs_dim, 64), nn.ReLU(), nn.Linear(64, 64), nn.ReLU(), nn.Linear(64, self.action_dim) ) def act(self, observation, exploreTrue): if explore and np.random.random() self.epsilon: return np.random.randint(self.action_dim) else: with torch.no_grad(): obs_tensor torch.FloatTensor(observation).unsqueeze(0) q_values self.q_net(obs_tensor) return q_values.argmax().item() def store_transition(self, obs, action, reward, next_obs, done): self.buffer.append((obs, action, reward, next_obs, done)) def learn(self): if len(self.buffer) self.batch_size: return # 采样 batch random.sample(self.buffer, self.batch_size) obs_batch, action_batch, reward_batch, next_obs_batch, done_batch zip(*batch) obs_tensor torch.FloatTensor(np.array(obs_batch)) action_tensor torch.LongTensor(action_batch).unsqueeze(1) # 用于 gather reward_tensor torch.FloatTensor(reward_batch) next_obs_tensor torch.FloatTensor(np.array(next_obs_batch)) done_tensor torch.FloatTensor(done_batch) # 计算当前 Q 值 current_q_values self.q_net(obs_tensor).gather(1, action_tensor).squeeze() # 计算目标 Q 值 with torch.no_grad(): next_q_values self.target_q_net(next_obs_tensor).max(1)[0] target_q_values reward_tensor self.gamma * next_q_values * (1 - done_tensor) # 计算损失并更新 loss self.loss_fn(current_q_values, target_q_values) self.optimizer.zero_grad() loss.backward() # 梯度裁剪防止爆炸 torch.nn.utils.clip_grad_norm_(self.q_net.parameters(), max_norm1.0) self.optimizer.step() # 更新探索率 self.epsilon max(self.epsilon_end, self.epsilon * self.epsilon_decay) # 定期更新目标网络 self.learn_step_counter 1 if self.learn_step_counter % self.target_update_freq 0: self.target_q_net.load_state_dict(self.q_net.state_dict()) return loss.item()实现要点网络结构_build_network是一个简单的多层感知机。对于更复杂的观察如图像需要替换为 CNN。探索与利用act方法使用 ε-greedy 策略。在训练初期高 ε 值鼓励探索随着训练进行ε 衰减智能体逐渐依赖学到的 Q 网络进行利用。经验回放store_transition和learn方法实现了 DQN 的核心——经验回放和目标网络。这能打破数据间的相关性稳定训练。独立性每个IndependentDQNAgent实例完全独立只根据自己的经验学习。这正是前文提到的“非平稳性”问题的根源。这个实现主要用于对比和教学让你直观感受独立学习的局限性。3.3 训练循环与多智能体协同有了环境和智能体我们需要一个训练循环将它们组织起来。这个循环要处理多智能体的动作收集、经验存储和协同学习。def train_episode(env, agents, episode_num, renderFalse): obs, info env.reset() total_rewards {agent_id: 0.0 for agent_id in agents.keys()} done False while not done: if render: env.render() # 1. 收集所有智能体的动作 actions {} for agent_id, agent in agents.items(): action agent.act(obs[agent_id], exploreTrue) # 训练时探索 actions[agent_id] action # 2. 环境执行一步 next_obs, rewards, terminated, truncated, info env.step(actions) done terminated or truncated # 3. 存储经验并学习 for agent_id, agent in agents.items(): agent.store_transition(obs[agent_id], actions[agent_id], rewards[agent_id], next_obs[agent_id], done) loss agent.learn() # 每个智能体独立学习 if loss: pass # 可以记录损失 # 4. 累积奖励并更新状态 for agent_id in agents.keys(): total_rewards[agent_id] rewards[agent_id] obs next_obs return total_rewards, info[‘steps’] # 主训练流程 if __name__ “__main__”: env PursuitEvasionEnv(grid_size8, n_pursuers2, n_evaders1, max_steps50, render_modeNone) # 初始化智能体 agents {} for i in range(env.n_pursuers): agent_id f“pursuer_{i}” agents[agent_id] IndependentDQNAgent(agent_id, env.observation_spaces[agent_id], env.action_spaces[agent_id]) agents[“evader_0”] IndependentDQNAgent(“evader_0”, env.observation_spaces[“evader_0”], env.action_spaces[“evader_0”]) num_episodes 5000 log_interval 100 for episode in range(num_episodes): rewards, steps train_episode(env, agents, episode, render(episode % log_interval 0)) if episode % log_interval 0: avg_pursuer_reward sum([rewards[pid] for pid in agents if pid.startswith(‘pursuer’)]) / env.n_pursuers avg_evader_reward rewards[“evader_0”] print(f“Episode {episode}, Steps: {steps}, Avg Pursuer Reward: {avg_pursuer_reward:.2f}, Avg Evader Reward: {avg_evader_reward:.2f}”) # 可以在这里保存模型或记录指标训练循环解析动作收集循环遍历所有智能体调用其act方法。注意exploreTrue仅在训练时开启。经验存储与学习环境执行后每个智能体将自己的(s, a, r, s’, done)元组存入自己的经验缓冲区并立即调用learn方法从缓冲区采样进行学习。这是在线学习模式。性能评估定期如每 100 轮打印平均奖励和步数。奖励是衡量智能体表现的核心指标。理想情况下追捕者的平均奖励应逐渐上升捕获次数增多逃跑者的平均奖励可能先升后降因为追捕者变强了但存活步数可能增加。可视化通过render参数控制是否渲染可以定期查看智能体的实际运动策略非常直观。实操心得在训练初期你可能会看到智能体进行完全随机的运动奖励曲线波动剧烈。这是正常的。大约在 1000-2000 轮后如果奖励函数设计合理你应该能看到追捕者开始表现出“围堵”的倾向逃跑者开始尝试“绕圈”或“躲藏”。独立 DQN 的收敛性并不好但这正是我们引入更高级算法如 MADDPG的动机。4. 高级扩展迈向集中式训练与实战调优基础平台搭建完成后我们可以进行高级扩展以解决独立学习的问题并提升性能。4.1 实现集中式批评家MADDPG 思路我们可以修改智能体结构引入一个共享的集中式批评家网络。这个批评家在训练时能观察到所有智能体的状态和动作从而做出更准确的全局价值评估。首先定义集中式批评家网络class CentralizedCritic(nn.Module): def __init__(self, total_obs_dim, total_action_dim, hidden_dim128): super().__init__() # total_obs_dim: 所有智能体观察拼接后的维度 # total_action_dim: 所有智能体动作拼接后的维度如果是离散动作需要先做嵌入或 one-hot self.net nn.Sequential( nn.Linear(total_obs_dim total_action_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, 1) # 输出一个标量 Q 值 ) def forward(self, global_obs, global_actions): x torch.cat([global_obs, global_actions], dim-1) return self.net(x)然后修改智能体类使其演员网络Actor保持不变基于局部观察输出动作但在学习时使用集中式批评家来计算损失。智能体的经验缓冲区也需要存储全局状态和全局动作。训练时从缓冲区采样一批全局数据输入给批评家网络计算目标 Q 值然后同时更新所有智能体的演员网络和共享的批评家网络。这种实现较为复杂但它是理解 MADDPG、MADQN 等先进算法的基础。你可以先运行独立 DQN 作为基线然后再尝试实现集中式批评家对比两者的训练稳定性和最终性能体会 CTDE 框架的优势。4.2 超参数调优与训练技巧多智能体强化学习的训练 notoriously tricky出了名的棘手。以下是一些实战调优技巧学习率Learning Rate这是最重要的超参数之一。对于 DQN通常从 1e-3 或 1e-4 开始尝试。如果奖励曲线震荡剧烈或不上升尝试调低学习率。可以使用学习率调度器如torch.optim.lr_scheduler.StepLR在训练后期衰减学习率。折扣因子Gamma控制未来奖励的重要性。在追逃任务中因为目标是尽快结束捕获或超时Gamma 可以设得稍低如 0.9 到 0.99。太高的 Gamma 可能导致智能体过于“远视”忽视即时收益。探索率Epsilonε-greedy 策略中的 ε 衰减速度很重要。epsilon_decay参数控制衰减速度。如果智能体过早陷入局部最优如追捕者只会直线冲可能是 ε 衰减太快探索不足。可以尝试更慢的衰减如 0.998或使用更复杂的探索策略如噪声注入。批次大小Batch Size从经验回放缓冲区采样学习的数据量。太小如 32可能导致更新噪声大不稳定太大如 512可能降低学习速度并需要更多内存。64 或 128 是常见的起点。目标网络更新频率DQN 中目标网络每隔多少步更新一次。更新太频繁如每步会破坏训练的稳定性更新太慢如每 1000 步会减慢学习速度。通常设置在 100 到 500 步之间。网络结构如果智能体学习缓慢可以尝试增加网络层数或每层的神经元数量如从 64 增加到 128。但也要警惕过拟合。奖励塑形Reward Shaping这是调优的“艺术”。除了距离奖励你还可以尝试团队奖励给所有追捕者一个共享的捕获奖励鼓励协作。区域惩罚如果逃跑者长时间躲在角落给予追捕者一个负奖励鼓励它们主动搜索。探索奖励给访问过较少区域的智能体微小正奖励鼓励探索地图。我的经验是保持耐心一次只调整一个超参数并仔细观察几十到几百个 episode 内的奖励曲线变化。使用 TensorBoard 或 Weights Biases 等工具记录训练过程至关重要。4.3 评估与可视化分析训练完成后我们需要评估智能体的真实水平并可视化其策略。评估模式在测试时关闭智能体的探索exploreFalse让它们完全根据学到的策略行动。运行一定数量的 episode如 100 次统计平均捕获时间、捕获成功率、逃跑者平均存活时间等指标。策略可视化除了动态渲染还可以绘制价值函数图或策略图。例如固定逃跑者的位置在网格的每个点上计算追捕者 Q 网络认为的最优动作用箭头表示可以直观看到追捕者认为的“最佳移动方向”。轨迹分析录制一些 episode 的完整轨迹分析成功捕获或成功逃脱的典型模式。追捕者是否学会了“分头包抄”逃跑者是否学会了“秦王绕柱”这能帮你理解智能体学到了什么以及奖励函数是否按预期引导了行为。5. 常见问题排查与实战心得在开发和训练过程中你几乎一定会遇到下面这些问题。这里我把自己踩过的坑和解决方案总结一下。5.1 训练不收敛奖励曲线乱跳可能原因1学习率过高。这是最常见的原因。RL 对学习率非常敏感。解决方案将学习率降低一个数量级如从 1e-3 降到 1e-4再试。可能原因2奖励尺度问题。如果密集奖励距离奖励相对于稀疏奖励捕获奖励过大或过小都会干扰学习。解决方案调整密集奖励的系数。确保智能体做出正确决策如捕获时获得的奖励远大于单步距离变化带来的奖励。可以尝试将捕获奖励设为 100距离奖励系数设为 0.01。可能原因3探索不足。智能体过早地陷入一个次优策略循环。解决方案提高初始 ε降低 ε 衰减速度或者尝试在演员网络的输出上添加奥恩斯坦-乌伦贝克Ornstein-Uhlenbeck过程噪声对于连续动作或使用熵正则化。可能原因4网络结构或优化器问题。解决方案检查网络是否有梯度消失/爆炸。使用梯度裁剪clip_grad_norm_。尝试不同的优化器如 RMSprop 有时比 Adam 更稳定。5.2 智能体表现“愚蠢”行为不符合预期可能原因1观察空间设计不合理。智能体没有得到做出明智决策所需的信息。解决方案重新审视_get_observations函数。逃跑者是否需要知道所有追捕者的位置而不仅仅是最近的那个追捕者是否需要知道队友的位置以进行协作可以尝试增加观察维度。可能原因2动作空间限制。也许 5 个动作上下左右停不足以实现复杂的围堵策略。解决方案可以考虑增加“斜向移动”动作或者引入“加速”、“转向”等更精细的控制这会使问题从离散变为连续。可能原因3环境动力学过于简单或复杂。解决方案如果环境太简单如无障碍物策略可能缺乏深度如果太复杂如密集障碍物智能体可能难以学习。从简单环境开始逐步增加复杂度课程学习。5.3 代码调试与性能优化问题渲染导致训练极慢。解决方案训练时务必关闭渲染render_modeNone。仅在评估和演示时开启。Matplotlib 的渲染在循环中开销很大。问题经验回放缓冲区内存占用过大。解决方案如果缓冲区存储的是图像等高维观察内存会迅速耗尽。可以考虑使用torch.utils.data.TensorDataset或更高效的数据结构或者定期清理旧数据。问题训练过程无法复现。解决方案设置所有随机种子Python, NumPy, PyTorch。在代码开头添加import random import numpy as np import torch seed 42 random.seed(seed) np.random.seed(seed) torch.manual_seed(seed) if torch.cuda.is_available(): torch.cuda.manual_seed_all(seed)问题多智能体环境中某个智能体一直学不会。解决方案这可能是多智能体信用分配问题。独立学习时一个智能体的好行为可能被队友的差行为或对手的强大所掩盖导致其奖励信号始终很差。这就是为什么需要集中式批评家或专门的多智能体算法来区分每个智能体的贡献。构建这个多智能体追逃博弈平台的过程就像在搭建一个微观的 AI 社会实验场。从最初智能体像无头苍蝇一样乱撞到后来逐渐涌现出围捕、躲闪甚至简单的诱敌策略每一次策略的进化都让人兴奋。这个项目的价值不仅在于最终的代码更在于你亲手将 MARL 的理论概念——非平稳性、信用分配、集中训练分布式执行——通过一个具体的游戏场景具象化并攻克的过程。当你看到两个红色方块终于学会一左一右地夹击那个蓝色圆圈时你会对“智能”与“协作”产生新的理解。这只是一个起点你可以在此基础上轻松地更换算法、增加智能体数量、引入更复杂的地形和规则去探索多智能体系统那无穷的可能性。本文还有配套的精品资源点击获取