深度强化Q学习-基于微软AirSim仿真环境的自动驾驶案例（原理代码详解简易可行）

本文主要是介绍深度强化Q学习-基于微软AirSim仿真环境的自动驾驶案例（原理代码详解简易可行），希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

最近在学习《强化学习原理与Python实现》，肖智清著。参考第十二章的自动驾驶案例，写一篇总结，尽量做到简单可行，让有志于自动驾驶的小伙伴可以通过此篇博客可以管中窥豹，初步了解自动驾驶的强化学习训练过程。
自动驾驶任务存在于连续的时间环境中，并且没有公认的奖励定义，也没有公认的回合划分。本篇将自动驾驶问题转化为回合制的强化学习任务，设计以车辆观察图像和运行状态为输入的自动驾驶算法，并在AirSim仿真环境中进行训练和测试。本篇使用带经验回放和目标网络的基于深度Q网络算法设计并实现智能体，对经验回放、目标网络和深度Q学习算法进行简单介绍，并用代码实现。
希望你能耐心看完这篇文章，干货满满，必有收获。另外码字不易，博主现在博客等级还是2级，不能自主创建索引标签，意味着博主的文章不能被更多人看到。如果你可以学到新知识，请来个一键三连（点赞关注收藏）吧，你的关注是我更新的动力，原理代码视频都有了，你就说该不该三联。

1. 安装和运行AirSim

本篇的自动驾驶算法基于AirSim仿真环境。AirSim是Microsoft发布的开源仿真软件，绿色免安装，请访问AirSim网页，并选择对应操作系统的最新版本的压缩包并解压。本篇选择城市模拟环境，Windows对应的压缩包为Neighborhood.zip，Linux对应的压缩包为Neighborhood.zip。值得注意的是，其中Windows版本为1.3.0，Linux版本为1.3.1。可能微软还没来得及更新Windows版的1.3.1版本。解压后得到可执行文件。
这里需要说明一下：原文中提到，AirSim需要依赖Unreal引擎，首次运行时可能会提示安装Unreal引擎及其依赖的软件。博主没碰到，直接解压后就可以运行了，有碰到的小伙伴请留言博主。
Windows系统下双击AirSimNH.exe，Linux系统下运行其中的sh文件，启动时会出现如图1-1提示界面：

图1-1 启动提示界面

提示选择要驾驶的设备，选择是为汽车，否为四旋翼无人机，选择是（汽车）即可，出现如图1-2运行界面：

图1-2 车辆运行界面

现在已经启动了AirSim界面。默认情况下，AirSim处于键盘控制模式。在键盘模式下，可以按照表1-1用方向键控制汽车的移动，按照表1-2控制窗口的观察场景，还可以按照表1-3显示或隐藏和前景有关的子视图，如图1-3。按退格键（backspace）可以将汽车重置到起始位置。按功能键F1可以显示帮助。

表1-1 键盘模式下方向键控制汽车移动

方向键	控制效果
向上	加油门前进
向下	倒挡加速后退
向左	逆时针打方向盘
向右	顺时针打方向盘

表1-2 切换主视图的键值

键值	窗口主视图
F	第一人称视图（坐在驾驶位置上的驾驶员看到的图像）
B	飞行跟随视图（在高空中有个自动跟随的摄像头看到的图像）
\	陆地观察者视图
/	刚载入时的默认视图
M	键盘控制视图（可用方向键，上一页、下一页、WSAD进一步控制）

表1-3 显示和隐藏子视图的键值

键值	切换主视图的键值
0	开关所有子视图
1	开关深度视图（对前景的深度信息用灰度图像表示）
2	开关划分视图（对前景的观测进行简单的物体划分）
3	开关场景子视图（前景的彩色图像）

图1-3 窗口下3个子视图，从左到右分别是深度视图、划分视图和场景视图

接下来我们讲解驾驶环境量化为强化学习。

2. 为自动驾驶设计强化学习环境

自动驾驶问题没有定义奖励，并不是一个天然的强化学习问题。为了将自动驾驶问题转化为强化学习问题，需要设计奖励函数。
我们希望自动驾驶算法能够安全而快速的驾驶汽车，我们不希望撞到其它汽车或冲出路面，同时行驶也要讲究效率，最好不要行驶过慢甚至一直静止不动。在本节中，我们使用以下奖励函数：

• 如果汽车撞到其它东西，则奖励为0；
• 如果汽车速度小于2，则奖励为0；
• 在其他情况下，计算汽车和路面中心的最小距离distance，奖励值为exp(-1.2 * distance)。

最小距离distance的计算方式会单独讲解。这样的奖励函数鼓励汽车以大于2的速度在道路中心行驶。这个奖励函数离真实的需求还有很大差距。不过，训练和测试结果表明，这个奖励函数已经可以引导汽车在环境中巡航行驶了。
有了奖励函数，自动驾驶问题就变成了强化学习问题。为了训练方便，我们进一步将自动驾驶问题建模为回合制的问题。回合的定义参考了奖励函数的定义。当出现下面任意一个状况时，回合结束：

• 汽车撞到其它东西；
• 汽车速度小于2；
• 汽车和路面中心的最小距离大于3.5时；
• 在设置回合最长时间的情况下，运行时间超过设置时间。

在训练过程中有必要限制回合的最长时间。基于同样的道理，在训练过程中最好选择不同的起点以避免陷入局部行为。同时，在回合开始应该先对汽车进行加速，使得其速度超过2，以免出现因为启动速度过小而是回合立即结束的情况。
至此，我们已经初步设计了回合制的强化学习任务。接下来，我们进入本篇最重要的实验环节。

3. 运行环境类实现

前面讲解了如何运行自动驾驶的实验环境AirSim，以及设计了带奖励回合机制的强化学习任务。本节将从准备运行环境及Python语言控制、奖励函数实现、获取坐标信息等方面讲解运行环境类的实现，同时在附件中给出所有代码运行的jupyter文件，感兴趣的小伙伴可下载后自行运行训练。
代码清单3-1给出了环境类AirSimCarEnv的设计框架。它的成员方法reset()可以开始新的回合，成员方法get_image()和get_car_state()可以获得观测图片和车辆信息。成员方法control()可以控制车辆的驾驶，成员方法get_reward()可以返回最近的奖励并判断回合是否结束。成员方法control()运行一段时间后，然后再调用get_reward()，这个过程联合起来相当于Gym库中的env.step()函数。

代码清单3-1 环境类AirSimCarEnv的框架

class AirSimCarEnv:def __init__(self): self.connect()def connect(self): ...def reset(self, explore_start=False, brake_confirm=True,start_accelerate=True, max_epoch_time=None, verbose=True): ... # 开始新回合def get_image(self): ... # 获取图像def get_car_state(self): ... # 获取车辆状态def control(self, throttle=0, steering=0, brake=0, handbrake=False): ...def get_reward(self): ... # 计算奖励，判断回合是否结束

下面逐一进行介绍。

3.1 准备运行环境，用python访问AirSim

本节介绍准备实验运行环境，及如何通过Python扩展库访问和控制AirSim。
首先开始之前，你需要安装好Anaconda3中的jupyter，同时创建并激活代码运行环境gym，在环境gym中已安装库TensorFlow 2.0和gym。
鉴于篇幅和聚焦重点，请网友自行找教程安装，这里就不扩展了。关于jupyter的使用，也请自行查找教程。需要注意的是，在jupyter中运行代码时，需要将环境切换到gym，对此部分有疑问的小伙伴请私信博主。jupyter截图如图3-1 ：

图3-1 jupyter截图

在环境gym下，安装Python扩展库msgpack-rpc-python和airsim。其中msgpack-rpc-python是远程过程调用的库，airsim库使用这个库和上一节中的仿真窗口通信。切换到gym环境，安装命令及代码块中导入扩展库代码为：

代码清单3-2 安装命令及导入库代码

# 安装命令
pip install msgpack-rpc-python
pip install airsim
# 导入扩展库
import msgpackrpc
import airsim

接下来，要让Python程序连接仿真窗口，并读取仿真环境的信息。返回的结果是一个airsim.CarState对象。可以进一步用car_state.speed读取汽车速度，用car_state.kenematis_estimated读取其动力学估计值。实现如代码清单3-3：

代码清单3-3 连接客户端并获取状态

car_client = airsim.CarClient()
car_client.confirmConnection()
# 获取车辆状态（速度等）
def get_car_state(self):return self.car_client.getCarState()

除此之外，AirSim还提供了一些API读取仿真环境中的信息。这样的方法一般被命名为airsim.Client.simXXX()，例如simGetImages()可以读取图像，simGetCollisionInfo()可以读取碰撞信息。

代码清单3-4 读取图像和碰撞信息

# 读取图像信息
def get_image(self):while True: # 有的时候会失败，这时候重试就好try:image_request = airsim.ImageRequest(0, airsim.ImageType.Scene, False, False)image_response = self.car_client.simGetImages([image_request,])[0]image1d = np.frombuffer(image_response.image_data_uint8, dtype=np.uint8)image_rgba = image1d.reshape(image_response.height, image_response.width, -1)breakexcept:print('获取图像失败，重试')return image_rgba[76:135,0:255,0:3].astype(float)# 判断是否发生碰撞
collision_info = car_client.simGetCollisionInfo()
collision_info.has_collided

接下来介绍如何控制汽车的运行。
首先，将AirSim从键盘模式切换到API控制模式，使得汽车的控制由Python API来接管。
然后，在API模式下，可以通过airsim.CarClient类的setCarControls()方法控制汽车的运行。方法的参数是一个airsim.CarControls对象，其构造方法有以下参数：

• throttle：float类型，表示油门
• steering：float类型，表示方向盘转动。负数是逆时针转方向盘，正数是顺时针。
• brake：float类型，表示刹车
• handbrake：bool类型，表示是否拉手刹

代码清单4给出了API接管和控制车辆运行的代码片段：

代码清单3-5 Python API接管及控制车辆运行

# Python API接管
def connect(self):# 连接正在运行的 AirSimself.car_client = airsim.CarClient()self.car_client.confirmConnection()self.car_client.enableApiControl(True)
# API模式下控制车辆运行
def control(self, throttle=0, steering=0, brake=0, handbrake=False):# 设置油门、方向和刹车car_controls = airsim.CarControls(throttle=throttle,steering=steering, brake=brake, handbrake=handbrake)self.car_client.setCarControls(car_controls)

基础应用代码就讲到这里。接下来看一下比较重要的新回合设置及位置的获取方法。

3.2 启动新回合和获取起始位置

现在看一下启动新回合的reset()函数，如代码清单5。启动新回合的第一步是要在地图上随机选择一个起始点，这个逻辑有后面的函数get_start_pose()实现。设置起始位置时并没有设置速度，所以汽车可能会沿着设置前的速度继续行驶，甚至会翻车、碰撞。因此当brake_confirm为真，在设置位置前先预设置一次，此时刹车一段时间，然后再正式设置，这样可以让汽车在设置的位置速度为0。然后start_accelerate选项使得汽车从速度0开始加速，使得开始时汽车有一些速度，不至于一开始就判断成回合结束。最后还要记录回合的起始时间，以便后续判断回合是否结束。

代码清单3-6 启动新的回合

def reset(self, explore_start=False, brake_confirm=True,start_accelerate=True, max_epoch_time=None, verbose=True):if verbose:print('开始新回合')# 起始探索start_pose = self.get_start_pose(random=explore_start)if brake_confirm:# 预设值起始位置, 刹车，并等待车变的稳定# 因为 simSetVehiclePose() 不能设置速度，只能选择刹车再等待self.car_client.simSetVehiclePose(start_pose, True)env.control(brake=1, handbrake=True)time.sleep(4)# 再次设置初始位置if verbose:print('设置初始位置')self.car_client.simSetVehiclePose(start_pose, True)if start_accelerate:# 让车加速一段时间，否则如果车的速度太小，后面会认为回合结束if verbose:print('直行一段时间')env.control(throttle=1)time.sleep(4)# 回合开始时间和预期结束时间self.start_time = dt.datetime.now()self.end_time = Noneif max_epoch_time:self.expected_end_time = self.start_time + \dt.timedelta(seconds=max_epoch_time)else:self.expected_end_time = None

然后来看一下如何确定起始位置并将汽车放在环境中任意的位置。代码清单5中函数get_start_pose()确定起始位置，当其参数random=True时，随机选择起始位置，不过要选在路上，并且车的起始朝向需要顺着路的方向。实现代码是先构造代表位置坐标的airsim.Vector3r对象，再用airsim.to_quaternion()函数确定汽车的朝向，用弧度值yaw表示。最后构造airsim.Pose对象，将其作为参数传给airsim.Client类的simSetVehiclePose()方法。关于四元数的理解，可参考博主的文章《
三维空间刚体运动4：四元数表示变换》。

代码清单3-7 确定回合起始位置

def get_start_pose(self, random=True, verbose=True):# 在路上选择一个起始位置和方向if not random: # 固定选择默认的起始位置position = np.array([0., 0.])yaw = 0.else: # 随机选择一个位置if not hasattr(self, 'roads_without_corners'):self.roads_without_corners = self.get_roads(include_corners=False)# 计算位置road_index = np.random.choice(len(self.roads_without_corners))p, q = self.roads_without_corners[road_index]t = np.random.uniform(0.3, 0.7)position = t * p + (1. - t) * q# 计算朝向if np.isclose(p[0], q[0]): # 与 Y 轴平行yaws = [0.5 * math.pi, -0.5 * math.pi]elif np.isclose(p[1], q[1]): # 与 X 轴平行yaws = [0., math.pi]yaw = np.random.choice(yaws)if verbose:print('起始位置 = {}, 方向 = {}'.format(position, yaw))position = airsim.Vector3r(position[0], position[1], -0.6)orientation = airsim.to_quaternion(pitch=0., roll=0., yaw=yaw)pose = airsim.Pose(position, orientation)return pose
pose = get_start_pose()
car_client.simSetVehiclePose(pose, True)

接下来，我们看一下本实验的核心：奖励函数的具体实现和distance的计算方法。

3.3 奖励函数的实现

代码清单6给出了奖励函数get_reward()的实现。首先，它判断是否撞车了，如果撞车了则返回奖励0并指示回合结束。然后，判断是否停止了，如果停止则返回奖励0并指示回合结束。最后，计算当前车到路的最小距离，也就是当前车到任意一段路的距离的最小值，返回基于距离的奖励。
这里用到了点到线段距离的计算方法，如图5所示：

图3-2 点C到线段PQ的距离

为了求点C到线段PQ上任意一点的最小距离，过点C做直线PQ的垂线并与直线PQ交于点S。考虑到：

$$\cos \angle CPQ=\frac{\overrightarrow{PC}\cdot \overrightarrow{PQ}}{\Vert \overrightarrow{PC}\Vert \cdot \Vert \overrightarrow{PQ}\Vert }$$

所以：

$$\overrightarrow{PS}=\Vert \overrightarrow{PC}\Vert \cos \angle CPQ\frac{\overrightarrow{PQ}}{\Vert \overrightarrow{PQ}\Vert }=\frac{\overrightarrow{PC}\cdot \overrightarrow{PQ}}{\Vert \overrightarrow{PC}\Vert \cdot \Vert \overrightarrow{PQ}\Vert }\overrightarrow{PQ}$$

设点T是线段PQ上离点C最近的点。如果点S在线段PQ上，则S=T；否则点T为P或Q。分类讨论可得：
$$\overrightarrow{PT}=clip\left(\frac{\overrightarrow{PC}\cdot \overrightarrow{PQ}}{\Vert \overrightarrow{PQ}\Vert \cdot \Vert \overrightarrow{PQ}\Vert ,0,1}\right)\overrightarrow{PQ}$$
其中clip为阶段函数。获得点T的坐标后，$\overrightarrow{CT}$就是待求的点到线段的最小距离。

代码清单3-8 计算回合奖励并判断回合是否结束

def get_reward(self):# 计算奖励，并评估回合是否结束collision_info = self.car_client.simGetCollisionInfo() # 碰撞信息if collision_info.has_collided: # 如果撞车了，没有奖励，回合结束self.end_time = dt.datetime.now()return 0.0, True, {'message' : 'collided'}car_state = self.car_client.getCarState() # 获取车辆速度信息if car_state.speed < 2: # 如果停车了，没有奖励，回合结束self.end_time = dt.datetime.now()return 0.0, True, {'speed' : car_state.speed}car_point = car_state.kinematics_estimated.position. \to_numpy_array() # 获取车辆位置信息if not hasattr(self, 'roads'):self.roads = self.get_roads()# 计算位置到各条路的最小距离distance = float('+inf')for p, q in self.roads:# 点到线段的最小距离frac = np.dot(car_point[:2] - p, q - p) / np.dot(q - p, q - p)clipped_frac = np.clip(frac, 0., 1.)closest = p + clipped_frac * (q - p)dist = np.linalg.norm(car_point[:2] - closest)distance = min(dist, distance) # 更新最小距离reward = math.exp(-1.2 * distance) # 基于距离的奖励函数if distance > 3.5: # 偏离路面太远，回合结束self.end_time = dt.datetime.now()return reward, True, {'distance' : distance}# 判断是否超时now = dt.datetime.now()if self.expected_end_time is not None and now > \self.expected_end_time:self.end_time = nowinfo = {'start_time' : self.start_time,'end_time' : self.end_time}return reward, True, info # 回合超时结束return reward, False, {}

接下来，看另外一个知识点，如何获取坐标信息，帮助读者更好地理解自动驾驶汽车的运行。

3.4 获取坐标信息

无论是在回合开始时在路上选择回合起始点，还是在回合过程中计算车到路的距离，都需要获取地图上路的坐标信息。代码清单7中的get_roads()函数，根据街道地图（见图3-3）返回各街道起始点坐标。函数get_roads()有个参数include_corners，当它为True时，返回的坐标包括在道路拐弯和交叉处的小斜线的坐标，这些小斜线会参与距离的计算；当它为False时，返回的坐标不包括那些小斜线的坐标，只包括较长的道路线段，这可用于回合起始位置确定时的道路选择。

代码清单3-9 获取街道起始坐标

def get_roads(self, include_corners=True):# 获得地图上的路的信息lines = [[[-128, -121], [-128, 119]],[[-120, -129], [120, -129]],[[-120, 127], [120, 127]],[[128, -121], [128, 119]],[[0, -121], [0, 119]],[[-120, 0], [120, 0]],[[80, -124], [80, -5]],]if include_corners: # 路的拐弯for x0, x1 in [[-128, -120], [0, -8], [0, 8], [120, 128]]:corners = [[[x0, -121], [x1, -129]],[[x0, -8], [x1, 0]],[[x0, 8], [x1, 0]],[[x0, 119], [x1, 127]],]lines += cornersfor x0, x1 in [[80, 75], [80, 85]]:corners = [[[x0, -124], [x1, -129]],[[x0, -5], [x1, 0]],]lines += cornersroads = [(np.array(p), np.array(q)) for p, q in lines]return roads

街道简图信息见图7：

图3-3 AirSIMNH的街道地图

至此，我们已经实现了AirSimCarEnv类，完成了“智能体/环境接口”中的环境接口部分。下面看智能体的实现。

4. 强化学习智能体

本节使用带经验回放和目标网络的基于深度Q网络算法设计并实现智能体。本节分为两部分讲解：带经验回放和目标网络的深度Q学习算法和深度Q网络的设计。智能体的整体框架如代码清单所示：

代码清单4-1 深度Q学习智能体整体框架

class DQNAgent():def __init__(self, gamma=0.99, batch_size=32, replayer_capacity=2000, random_initial_steps=50, weight_path=None, train_conv=True,epsilon=1., min_epsilon=0.1, epsilon_decrease_rate=0.003): ... # 初始化智能体def build_network(self, activation='relu', weight_path=None, train_conv=True, verbose=True): ... # 构建网络def decide(self, observation, random=False): ... # 决策算法def action2control(self, action, car_state): ... # 将动作转换为控制信号def learn(self, observation, action, reward, next_observation, done): ... # 网络学习

4.1 带经验回放和目标网络的深度Q学习算法

本节介绍一种目前非常热门的函数近似方法----深度Q学习。深度Q学习将深度学习和强化学习相结合，是第一个深度强化学习算法。深度Q学习的核心就是用一个人工神经网络$q(s,a;w),sϵS,aϵA$来代替动作价值函数$q(s,a)$。由于神经网络具有强大的表达能力，能够自动寻找特征，所以采用神经网络的潜力比传统人工特征强大得多。
本小节将从三方面进行介绍：Q学习、经验回放和目标网络，首先简要介绍下Q学习。

4.1.1 Q学习

期望SARSA算法将时序查分目标从SARSA算法的$$U_t=R_{t+1}+\gamma q(S_{t+1},A_{t+1})$$改为$$U_t=R_{t+1}+\gamma E[q(S_{t+1},A_{t+1})]$$从而避免了偶尔出现的不当行为给整体结果带来的负面影响。其中$U_t,R_t,S_t,A_t$分别表示$t$时刻的时序差分目标函数，奖励，状态和动作。
Q学习则是从改进后的策略行为出发，将时序差分目标改为：$$U_t=R_{t+1}+\gamma \underset{a\in A(S_{t+1})}{\max }q(S_{t+1},a)$$Q学习算法认为，在根据$S_{t+1}$估计$U_t$时，与其使用$q(S_{t+1},A_{t+1})$或$v(S_{t+1})$，还不如使用根据$q(S_{t+1},\cdot )$改进来的策略来更新，毕竟这样更接近最优价值。因此Q学习的更新式不是基于当前的策略，而是基于另外一个并不一定要使用的确定性策略来更新动作价值。从这个意义上看，Q学习是一个异策算法。
Q学习算法中更新动作价值策略时，更新$q(S,A)$以减小$[U-q(S,A){]}^2$，比如：$$q(S,A)=q(S,A)+\alpha [U-q(S,A)]$$
其中$\alpha$为学习率。

当Q学习同时出现异策、自益和函数近似时，无法保证收敛性，会出现训练不稳定或训练困难等问题。针对出现的各种问题，研究人员主要从以下两方面进行改进：

• 经验回放（experience replay）：将经验（即历史的状态、动作、奖励等）存储起来，再在存储的经验中按一定的规则采样。
• 目标网络（target network）：修改网络的更新方式，例如不把刚学习到的网络权重马上用于后续的自益过程

本节后续内容将从这两条主线出发，介绍基于深度Q网络的强化学习算法。

4.1.2 经验回放

对于样本来说，采用批处理的模式能够提供稳定性。经验回放就是一种让经验的概率分布变得稳定的技术，它能提高训练的稳定性。经验回放主要有“存储”和“采样回放”两大关键步骤：

• 存储：将轨迹以$(S_t,A_t,R_t,S_{t+1})$等形式存储起来；
• 采样回放：使用某种规则从存储的$(S_t,A_t,R_t,S_{t+1})$中随机取出一条或多条经验。

经验回放有以下好处：

• 在训练Q网络时，可以消除数据间的关联，使得数据更像是独立同分布的。这样可以减少参数更新的方差，加快收敛；
• 能够重复使用经验，对于数据获取困难的情况尤其有用。

从存储的角度，经验回放可以分为集中式回放和分布式回放：

• 集中式回放：智能体在一个环境中运行，把经验统一存储在经验池中；
• 分布式回放：智能体的多份拷贝（worker）同时在多个环境中运行，并将经验统一存储于经验池中。由于多个智能体拷贝同时生成经验，所以能够在使用更多资源的同时更快地收集经验。

从采样的角度，经验回放可以分为均匀回放和优先回放：

• 均匀回放：等概率从经验集中取经验，并用取得的经验更新最优价值函数；
• 优先回放：（Prioritized Experience Replay, PER）:为经验池的每个经验指定一个优先级，在选择经验时更倾向于选择优先级高的经验。

一般做法是，如果某个经验i的优先级为$p_i$，那么选取该经验的概率为：$$P_i=\frac{p_i}{{\sum }_{k=0}^n{p}_k}$$
优先回放的经验概率值有许多不同的选取方法，最常见的选取方法有成比例优先和基于排序优先：

• 成比例优先（rank-based priority）：第$i$个经验的优先级为：$$p_i=({\delta }_i+\epsilon {)}^{\alpha }$$其中${\delta }_i$是时序查分误差，$\epsilon$是预先选择的一个小正数，$\alpha$是正参数。
• 基于排序优先（rank-based priority）：第$i$个经验的优先级为：$$p_i={\left(\frac{1}{ran{k}_i}\right)}^{\alpha }$$其中$ran{k}_i$是第$i$个经验从大到小排序的排名，排名从1开始。

经验回放也不是完全没有缺点。例如，它也会导致回合更新和多步学习算法无法使用。一般情况下，如果我们将经验回放用于Q学习，就规避了这个缺点。经验回放类的实现见代码清单4-2。

代码清单4-2 经验回放类

class DQNReplayer:def __init__(self, capacity):self.memory = pd.DataFrame(index=range(capacity),columns=['observation', 'action', 'reward','next_observation', 'done'])self.i = 0self.count = 0self.capacity = capacitydef store(self, *args):self.memory.loc[self.i] = argsself.i = (self.i + 1) % self.capacityself.count = min(self.count + 1, self.capacity)def sample(self, size):indices = np.random.choice(self.count, size=size)return (np.stack(self.memory.loc[indices, field]) for field in \self.memory.columns)

4.1.3 带目标网络的深度Q学习

对于基于自益（即自己评估自己）的Q学习，其回报的估计和动作价值的估计都和权重w有关。当权重值变化时，回报的估计和动作价值的估计都会变化。在学习的过程中，动作价值试图追逐一个变化的回报，也容易出现不稳定的情况。半梯度下降算法可以有效解决这个问题。在半梯度下降中，更新价值参数$w$时，不对基于自益得到的回报估计$U_t$求梯度。其中一种阻止对$U_t$求梯度的方法就是将$w$复制一份得到$w_{目标}$，在计算$U_t$时用$w_{目标}$计算，由此，V.Mnih等在2015年发表了论文《Human-level control through deep reinforcement learning》，提出了目标网络（target network）这一概念。
目标网络是在原有的神经网络之外再搭建一份结构完全相同的网络，原有的神经网络被称为评估网络（evaluation network）。在学习的过程中，使用目标网络来进行自益得到回报的评估值，作为学习的目标。在权重更新的过程中，只更新评估网络的权重，而不更新目标网络的权重，这样，更新权重时针对的目标不会在每次迭代都变化，是一个固定的目标。在完成一定次数的更新后，再将评估网络的权重复制给目标网络，进而进行下一批更新，这样，目标网络也得到更新。由于在目标网络没有变化的一段时间内回报的估计是相对固定的，因此目标网络的引入增加了学习的稳定性。所以，目标网络目前已成为深度Q学习的主流做法。
带经验回放和目标网络的深度Q学习最优策略求解步骤：

1. （初始化）初始化评估网络$q(\cdot ,\cdot ;w)$的参数$w$；目标网络$q(\cdot ,\cdot ;w_{目标})$的参数$w_{目标}$=$w$；
2. 逐回合执行以下操作：
   2.1 （初始化状态动作对）选择状态$S$；
   2.2 如果回合未结束，执行以下操作：
   （1）（采样）根据$q(S,\cdot ;w)$选择最优动作$A$并执行，观测得到奖励$R$和新状态$S^{{}^{\prime }}$；
   （2）（经验存储）将经验$(S,A,R,S^{{}^{\prime }})$存入经验库$D$中；
   （3）（经验回放）从经验库$D$中选取一批经验$(S_i,A_i,R_i,S_i^{{}^{\prime }}),i\in \beta$；
   （4）（计算回报的估计值）$U_i=R_i+\gamma {\max }_{a}q(S_i^{{}^{\prime }},a;w_{目标}),i\in \beta$；
   （5）（更新动作价值函数）更新$w$以减小$\frac{1}{|\beta |}{\sum }_{i\in \beta }[U_i-q(S_i^{{}^{\prime }},a;w){]}^2$，比如 $w=w+\alpha \frac{1}{|\beta |}{\sum }_{i\in \beta }[U_i-q(S_i^{{}^{\prime }},a;w)]▽q(S_i,A_i,w)$；
   （6）（更新状态）$S=S^{{}^{\prime }}$；
   （7）（更新目标网络）在一定条件下，例如访问本步若干次后，更新目标网络的权重$w_{目标}=w$。

在更新目标网络时，可以简单地把评估网络的参数直接赋值给目标网络，也可以引入一个学习率${\alpha }_{目标}$，把旧的目标网络参数和新的评估网络参数直接做加权平均后赋值给目标网络，即：$w_{目标}=(1-{\alpha }_{目标})w_{目标}+{\alpha }_{目标}w$。
算法在试验中的实现代码如下：

代码清单4-3 带目标网络的深度Q学习算法实现代码

def learn(self, observation, action, reward, next_observation, done):agent.replayer.store(observation, action, reward, next_observation, done) # 存储经验if self.replayer.count < self.random_initial_steps:return # 还没到存足够多的经验，先不训练神经网络observations, actions, rewards, next_observations, dones = \self.replayer.sample(self.batch_size) # 经验回放next_qs = self.target_net.predict(next_observations)next_max_qs = next_qs.max(axis=-1)us = rewards + self.gamma * next_max_qs * (1. - dones)targets = self.evaluate_net.predict(observations)targets[np.arange(us.shape[0]), actions] = usself.evaluate_net.fit(observations, targets, verbose=0)if done:self.target_net.set_weights(self.evaluate_net.get_weights())# 减小 epsilon 的值self.epsilon -= self.epsilon_decrease_rateself.epsilon = max(self.epsilon, self.min_epsilon)

4.2 深度Q网络的设计

深度Q网络的设计见图4-1。

图4-1 深度Q网络的设计

网络输入是一张形状为（59,255,3）的彩色图像，它由前景图像裁剪而来。网络的设计是典型的卷积神经网络的设计，先有3层带有最大池化的卷积层，再有2层全连接层。网络的输出是5个动作价值函数估计，分别代表转向steering为-1，-0.5,0，+0.5，+1时的动作价值函数估计。也就是说，这个深度Q网络只能对这5个转向进行区别。代码清单4-3实现了这个深度Q网络。

代码清单4-4 构造Q网络的代码

def build_network(self, activation='relu', weight_path=None,  train_conv=True, verbose=True):inputs = keras.Input(shape=(59, 255, 3))x = inputs# 卷积层for filte in [16, 32, 32]:z = keras.layers.Conv2D(filte, 3, padding='same',activation=activation, trainable=train_conv)(x)x = keras.layers.MaxPooling2D(pool_size=2)(z)y = keras.layers.Flatten()(x)# 全连接层x = keras.layers.Dropout(0.2)(y)z = keras.layers.Dense(128, activation=tf.nn.relu,kernel_initializer=RandomNormal(stddev=0.01))(x)y = keras.layers.Dropout(0.2)(z)outputs = keras.layers.Dense(self.action_n,kernel_initializer=RandomNormal(stddev=0.01))(y)net = keras.Model(inputs=inputs, outputs=outputs)net.compile(optimizer='adam', loss='mse')if verbose:net.summary()if weight_path:net.load_weights(weight_path)if verbose:print('载入网络权重 {}'.format(weight_path))return net

智能体在根据输入图像做决策时，先根据深度Q网络输出5个动作价值估计，然后选择价值估计最大的动作编号（取值范围为{0,1,2,3,4}）。这个逻辑由代码清单4-2中的decide()函数实现。这个动作编号可以一一对应到方向值{-1,-0.5,0,0.5,1}。另外，还根据车辆的速度决定油门和刹车。速度小了则加油门，速度大了则踩刹车，这样使得车辆的速度基本保持稳定。这部分的逻辑由action2control()函数实现。

代码清单4-5 智能体根据观测图像和车辆速度决定车辆控制信息

def decide(self, observation, random=False):if random or np.random.rand() < self.epsilon:return np.random.randint(self.action_n)observations = observation[np.newaxis]qs = self.evaluate_net.predict(observations)return np.argmax(qs)def action2control(self, action, car_state):# 将动作转换为控制信号steering = 0.5 * action - 1. # 方向，可取 -1, -0.5, 0, 0.5, 1if car_state.speed > 9:return 0, steering, 1else:return 1, steering, 0

将上述网络结构和控制信号设计整合进深度Q网络智能体，就完成了本节需要的智能体代码。

5. 智能体的训练和测试

在上面两个小节就已经得到了环境和智能体。本节讲述如何训练和测试。

5.1 环境和智能体的交互

在训练和测试时，我们可以用代码清单5-1中的play_once()函数，让智能体和环境交互一个回合。这个函数有个参数random，表示智能体是否在回合里随机选择动作。一般在刚开始收集经验时，网络并没有训练好，通过神经网络运算得到的动作与随机动作相比并没有太大优势。这时候用随机动作填充经验，能加快经验收集的速度。

代码清单5-1 智能体与环境的交互一个回合

def play_once(env, agent, explore_start=False, random=False, train=False,max_epoch_time=None, wait_delta_sec=0.01, verbose=True):# 启动新回合，在地图上选择一个地方，并让汽车前进一段env.reset(explore_start=explore_start, max_epoch_time=max_epoch_time)# 正式开始学习for step in itertools.count():image = env.get_image()car_state = env.get_car_state()action = agent.decide(image, random=random)# 根据动作影响环境throttle, steering, brake = agent.action2control(action, car_state)if verbose:print('动作 = {}, 速度 = {}, 油门 = {}, 方向 = {}, 刹车 = {}' \.format(action, car_state.speed, throttle, steering, brake))env.control(throttle, steering, brake)# 等待一段时间time.sleep(wait_delta_sec)# 获得更新后的观测、奖励和回合结束指示next_image = env.get_image()reward, done, info = env.get_reward()# 如果回合刚开始就结束了，就不是靠谱的回合if step == 0 and done:if verbose:print('不成功的回合，放弃保存')breakif train: # 根据经验学习agent.learn(image, action, reward, next_image, done)# 回合结束if done:if verbose:print('回合 从 {} 到 {} 结束. {}'.format(env.start_time, env.end_time, info))break

5.2 训练和测试

代码清单5-2给出了设置参数。本节准备了三套参数，需要根据不同的场景进行选择，选择一种时，将其他两种注释掉即可。

代码清单5-2 训练和测试参数设置

"""
从头开始训练，需要训练几周
"""
weight_path = None # 载入权重数据的位置
train_conv = True # 是否训练卷积层
max_epoch_time = 30. # 最长回合时间
random_initial_steps = 1000 # 随机运行的初始步数
train = True # 是否训练"""
预训练的结果，只需要再训练全连接层，再训练数小时即可
"""
weight_path = 'pretrain_weights.h5'
train_conv = False
max_epoch_time = 30.
random_initial_steps = 50
train = True"""
测试，训练好的结果，直接能用
"""
weight_path = 'weights.h5'
train_conv = False
max_epoch_time = None
random_initial_steps = 0
train = False

代码清单5-3给出了训练智能体的代码。从头开始训练智能体需要非常长的时间（例如一周）。在训练过程中，可以通过AirSim图形界面观察训练的情况。当训练到满意的结果时，可以手动中断程序运行或加入终止条件，通过model.save_weights(path)存储训练得到的权重。
训练结束后，可以测试算法。首先设置参数，然后调用play_once()函数运行一个回合。我们可以在图形界面中查看自动驾驶的效果。
注意：在运行过程中，在创建环境类中连接环境时，可能会出现卡死的情况，重新加载jupyter文件并重新打开AirSim，还没有解决的话就重启电脑吧。

代码清单5-3 训练和测试智能体

env = AirSimCarEnv()
agent = DQNAgent(weight_path=weight_path, train_conv=train_conv,random_initial_steps=random_initial_steps)
if train:print('开始训练')while True: # 无限循环，永不停止。需要手动中断try:# 判断是否用随机动作填充经验库random = agent.replayer.count < random_initial_stepsplay_once(env, agent, explore_start=True, random=random,train=True, max_epoch_time=max_epoch_time)# 极少数情况下 AirSim 会停止工作，需要重新启动并连接except msgpackrpc.error.TimeoutError:print('与 AirSim 连接中断。开始重新链接')env.connect()
else:print('开始测试')agent.epsilon = 0. # 取消探索play_once(env, agent, max_epoch_time=max_epoch_time)

5.3 训练结果

训练过程日志如下图所示：

代码清单5-4 训练结果展示

Connected!
Client Ver:1 (Min Req: 1), Server Ver:1 (Min Req: 1)_________________________________________________________________
Layer (type)                 Output Shape              Param #   
=================================================================
input_1 (InputLayer)         (None, 59, 255, 3)        0         
_________________________________________________________________
conv2d (Conv2D)              (None, 59, 255, 16)       448       
_________________________________________________________________
max_pooling2d (MaxPooling2D) (None, 29, 127, 16)       0         
_________________________________________________________________
conv2d_1 (Conv2D)            (None, 29, 127, 32)       4640      
_________________________________________________________________
max_pooling2d_1 (MaxPooling2 (None, 14, 63, 32)        0         
_________________________________________________________________
conv2d_2 (Conv2D)            (None, 14, 63, 32)        9248      
_________________________________________________________________
max_pooling2d_2 (MaxPooling2 (None, 7, 31, 32)         0         
_________________________________________________________________
flatten (Flatten)            (None, 6944)              0         
_________________________________________________________________
dropout (Dropout)            (None, 6944)              0         
_________________________________________________________________
dense (Dense)                (None, 128)               888960    
_________________________________________________________________
dropout_1 (Dropout)          (None, 128)               0         
_________________________________________________________________
dense_1 (Dense)              (None, 5)                 645       
=================================================================
Total params: 903,941
Trainable params: 889,605
Non-trainable params: 14,336
_________________________________________________________________
载入网络权重 weights.h5
开始测试
开始新回合
起始位置 = [0. 0.], 方向 = 0.0
设置初始位置
直行一段时间
动作 = 3, 速度 = 3.1895177364349365, 油门 = 1, 方向 = 0.5, 刹车 = 0
动作 = 1, 速度 = 4.957728862762451, 油门 = 1, 方向 = -0.5, 刹车 = 0
动作 = 3, 速度 = 5.633646011352539, 油门 = 1, 方向 = 0.5, 刹车 = 0
动作 = 3, 速度 = 6.329768657684326, 油门 = 1, 方向 = 0.5, 刹车 = 0
动作 = 1, 速度 = 7.1464715003967285, 油门 = 1, 方向 = -0.5, 刹车 = 0
动作 = 1, 速度 = 7.820223808288574, 油门 = 1, 方向 = -0.5, 刹车 = 0
动作 = 3, 速度 = 8.458905220031738, 油门 = 1, 方向 = 0.5, 刹车 = 0
动作 = 3, 速度 = 9.062920570373535, 油门 = 0, 方向 = 0.5, 刹车 = 1
获取图像失败，重试
动作 = 1, 速度 = 7.294376850128174, 油门 = 1, 方向 = -0.5, 刹车 = 0
动作 = 1, 速度 = 8.06904411315918, 油门 = 1, 方向 = -0.5, 刹车 = 0
动作 = 1, 速度 = 8.723608016967773, 油门 = 1, 方向 = -0.5, 刹车 = 0
动作 = 1, 速度 = 9.375661849975586, 油门 = 0, 方向 = -0.5, 刹车 = 1
动作 = 1, 速度 = 9.36580753326416, 油门 = 0, 方向 = -0.5, 刹车 = 1
动作 = 3, 速度 = 7.376546859741211, 油门 = 1, 方向 = 0.5, 刹车 = 0
动作 = 3, 速度 = 8.023926734924316, 油门 = 1, 方向 = 0.5, 刹车 = 0
回合 从 2019-01-01 08:01:47.925642 到 2019-01-01 08:06:59.343107 结束. {'distance': 4.6839447021484375}

运行视频，上传在腾讯视频，请打开链接观看：

深度强化Q学习-基于微软AirSim仿真环境的自动驾驶案例

当然也少不了翻车:(

好了，这篇博客就到这里。有问题可以写下评论或私信博主，博主会一一回复。

附件1：AirSimNH_tf.ipynb
附件2：pretrain_weights.h5
附件3：weights.h5

这篇关于深度强化Q学习-基于微软AirSim仿真环境的自动驾驶案例（原理代码详解简易可行）的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---