移动机器人导航中的深度强化学习教程 — 第 3 部分:培训
在本教程的第 1 部分中,我们克隆并安装了 GitHub 存储库,并首次成功启动了机器人导航训练。在 第 2 部分中,我们详细研究了训练神经网络的实际组成。现在,在这一部分,是时候看看神经网络的训练是如何在移动机器人运动设置中调用的了。再一次,我们将查看 python 文件中的代码
train_velodyne_td3.py
位于:
~<PATH_TO_FOLDER>/DRL-robot-navigation/TD3
机器人与世界
在开始移动机器人的实际运动策略训练之前,我们必须首先了解我们试图解决的任务。用人类的语言来说,我们正试图“找到引导机器人达到给定目标的最佳动作顺序”。但是我们如何将这个问题变成计算机可以理解和执行的东西呢?有两件事需要考虑——动作和动作反应的环境。在移动机器人设置中,很容易用数学形式表达动作。它是施加在每个执行器上的力,以实现可控的自由度。简单地说,就是我们想向任何可控的方向移动多少。
在本教程中,我们将使用 Pioneer P3DX 模拟移动机器人平台。它是一种非整体的差速驱动移动机器人底座,能够来回移动或绕其轴旋转。这意味着,我们有 2 个可以控制的方向——线性运动(向前/向后移动)和角运动(向左/向右旋转)。这两种控制都可以用数值表示为以米/秒为单位的线速度和以弧度/秒为单位的角速度,然后可以很容易地由神经网络进行处理

为了获得对环境状态的反应,需要以逻辑和数字的方式观察它。检测环境(在机器人技术中)的最简单形式之一是使用 LiDAR 或激光传感器。这里的逻辑非常简单——物体离激光传感器越近,发射的光返回光感受器的速度就越快。然后,该速度可以直接用于计算到物体的距离。从逻辑上讲,我们可以推测机器人将根据它与障碍物的距离做出决定,并从中得出运动策略。这给出了激光传感器(以及随后的机器人)周围环境的简单数值表示。

最后要考虑的一件事是,机器人如何知道它需要去哪里。我们心中有一个机器人需要达到的特定目标点,因此需要将这些信息传达给它。要学习自我策略,这些信息应该以机器人的位置为中心。我们将以极坐标的形式表示到目标位置的距离以及机器人航向与朝向目标的航向之间的角度差。

现在我们已经形成了环境和智能体的数字表达式,我们可以描述我们的状态和行动。如前所述,action 是一个元组: a = (v, ω)
我们首先需要表示环境的周围环境。我们将用激光读数来做到这一点。然后目标将以极坐标给出。但我们可能也想把机器人的当前内部状态提供给神经网络,让它知道我们已经移动了多快。因此,我们也将用在上一个时间步采取的行动来表示状态。因此,我们的最终状态表示将变成一个一维向量组合:
s = (laser_state + polar_goal_coordinates + previous_action)
我们将使用 20 个激光读数(我们将在第 4 部分中详细解释如何获得它们)。这意味着我们的最终状态表示将由 20 + 2 + 2 = 24 个特征一维向量组成。
建立培训网络
因此,现在我们已经安装了存储库,创建了 TD3 神经网络,并了解了如何表示状态,让我们开始实际训练移动机器人运动策略。
首先,让我们为训练设置一堆参数。
# 设置参数
device = torch.device("cuda" if torch.cuda.is_available() else "cpu") # cuda or cpu
seed = 0 # 随机种子数,用于确保实验的可重复性
eval_freq = 5e3 # 每隔多少步进行一次评估
max_ep = 500 # 每个episode的最大步数
eval_ep = 10 # 用于评估的episode数量
max_timesteps = 5e6 # 总的最大步数
expl_noise = 1 # 初始探索噪声的起始值,范围在[expl_min, 1]之间
expl_decay_steps = (
500000 # 初始探索噪声将在多少步内衰减
)
expl_min = 0.1 # 探索噪声衰减后的最小值,范围在[0, expl_noise]之间
batch_size = 40 # mini-batch的大小
discount = 0.99999 # 用于计算折扣未来奖励的折扣因子,应接近1
tau = 0.005 # 软更新目标网络的变量,应接近0
policy_noise = 0.2 # 用于探索的添加噪声
noise_clip = 0.5 # 噪声的最大裁剪值
policy_freq = 2 # Actor网络更新的频率
buffer_size = 1e6 # 缓冲区的最大值
file_name = "TD3_velodyne" # 存储策略的文件名
save_model = True # 是否保存模型
load_model = False # 是否加载存储的模型
random_near_obstacle = True # 是否在靠近障碍物时采取随机动作
之后,我们将设置更多的参数和目录。
# 创建网络存储文件夹
if not os.path.exists("./results"):
os.makedirs("./results")
if save_model and not os.path.exists("./pytorch_models"):
os.makedirs("./pytorch_models")
# 建立训练环境
environment_dim = 20
robot_dim = 4
env = GazeboEnv("multi_robot_scenario.launch", environment_dim)
time.sleep(5)
torch.manual_seed(seed)
np.random.seed(seed)
state_dim = environment_dim + robot_dim
action_dim = 2
max_action = 1
如果网络尚不存在,我们将创建一个文件夹来存储网络。然后,我们在environment_dim变量中指定我们需要从模拟中获得多少激光读数。robot_dim将指定“内部机器人参数”的数量,例如极坐标特征 (2) + 先前动作特征 (2) 的组合。在第 10 行中,我们实际上将调用模拟环境的创建。模拟环境创建参数存储在文件“multi_robot_scenario.launch”中。为了创建环境,我们还需要传入我们希望它在每个步骤中返回的激光读数数。创建环境时,可能需要一些时间来启动 Gazebo 模拟中的所有后台进程。因此,我们需要给它一点时间,让我们的程序进入睡眠状态 5 秒钟。(这在大多数系统上应该足够了,但如果出现问题,可能需要更长的睡眠时间。然后设置可重复性的种子,设置上述章节中讨论的state_dim,指定我们的网络需要计算的动作数量(以action_dim为单位),并设置神经网络可以返回的最大可能动作值。
接下来,我们最终创建并初始化 TD3 模型
# 创建网络
network = TD3(state_dim, action_dim, max_action)
# 创建经验回放池
replay_buffer = ReplayBuffer(buffer_size, seed)
if load_model:
try:
network.load(file_name, "./pytorch_models")
except:
print(
"Could not load the stored model parameters, initializing training with random parameters"
)
此外,我们创建了一个回放缓冲区,该缓冲区将存储机器人在模拟中遇到的体验,并将用作神经网络将对其进行优化的动态数据集。如果我们已将 load_model 参数设置为 True,则将为我们的模型加载先前训练的模型参数(如果存在)。如果没有,则将使用随机参数值初始化模型。
在开始训练循环之前,需要设置一些最后的参数。
# 创建评估数据存储
evaluations = [] # 用于存储评估数据的列表
timestep = 0 # 当前的时间步数
timesteps_since_eval = 0 # 距离上次评估经过的时间步数
episode_num = 0 # 当前的episode数
done = True # 标记当前episode是否结束
epoch = 1 # 当前的训练轮数
count_rand_actions = 0 # 随机动作的计数
random_action = [] # 存储随机动作的列表
我们首先创建一个列表来存储评估结果,并创建变量来计算经过的时间、步骤和纪元。最后,创建变量以供以后使用,届时我们需要在障碍物附近执行随机操作。
训练循环
现在一切都设置好了,我们可以开始我们的训练循环了。我们将根据最大训练步骤来计算训练量,但也可以轻松设置其他方法(例如最大纪元数、经过的时间等)。
# 开始训练循环
while timestep < max_timesteps:
以下所有代码(除非另有说明)都将位于此循环中。
注意:
我们将在正文中使用Episode和Epoch的符号。
Episode — 后续步骤的集合,直到达到其中一个终止条件(终止条件是 — 达到目标、与障碍物碰撞或达到max_ep参数中设置的最大步数)
Epoch — 执行评估之间的后续事件或时间步长(在eval_freq参数中设置)
让我们设置将在每一集结束时执行的代码。我们需要在循环的开头设置它,因为我们希望已经对环境进行第一次重置以获得它的第一个观察值(这就是为什么我们之前将 done 参数设置为 True)。
# episode终止
if done:
if timestep != 0:
# 执行网络训练
network.train(
replay_buffer,
episode_timesteps,
batch_size,
discount,
tau,
policy_noise,
noise_clip,
policy_freq,
)
# 进行评估
if timesteps_since_eval >= eval_freq:
print("Validating")
timesteps_since_eval %= eval_freq
evaluations.append(
evaluate(network=network, epoch=epoch, eval_episodes=eval_ep)
)
# 保存模型和评估结果
network.save(file_name, directory="./pytorch_models")
np.save("./results/%s" % (file_name), evaluations)
epoch += 1
# 重置环境和相关变量
state = env.reset()
done = False
episode_reward = 0
episode_timesteps = 0
episode_num += 1
在代码的第一次执行时,时间步长值仍将为 0,因此不会进行网络训练(因为我们还没有任何东西可以训练)。但是,在每完成一集的后续过程中,都会运行神经网络训练。这意味着,我们在每次发作后训练神经网络。训练周期由我们传入的episode_timesteps值决定。理由是,如果剧集中有更多的时间步长,我们就会收集更多的新样本,并且我们有更多的新数据需要优化。如果我们没有很多新样本,则将执行更少的训练周期。在此之后,我们检查是否是时候对当前模型进行评估。如果是这样,我们进行评估,记录结果并保存我们的模型。完成这些检查后,我们需要重置我们的环境。这将使机器人处于新的随机起始位置和方向,将给出一个新的随机放置的目标位置,并开始新的情节。从复位状态,我们将获得对新环境的观察。然后,我们只需将计数变量设置回 0,然后开始执行新剧集。
在 DRL 中,遵循贪婪策略是很常见的。这意味着,神经网络的任务是为任何输入选择最佳计算结果。不幸的是,在训练时,这可能会导致一个恶性循环,即首先计算一个好的结果,然后遵循它,并加强好的结果,而不寻找可能更好的替代方案。这通常被称为利用策略。为了找到最佳结果,我们必须强制神经网络探索替代方案(探索策略)。一种常见的方法是在神经网络输出中添加随机噪声,并在环境中观察结果
# 增加探索噪声
if expl_noise > expl_min:
expl_noise = expl_noise - ((1 - expl_min) / expl_decay_steps)
action = network.get_action(np.array(state))
action = (action + np.random.normal(0, expl_noise, size=action_dim)).clip(
-max_action, max_action
)
我们计算要添加到神经网络计算的动作中的噪声量,并将其保存为expl_noise。噪音水平将在整个expl_decay_steps内降低,直到达到最小expl_min值。我们用神经网络计算动作,并通过添加噪声来更新这个值。然而,机器人在最大和最小速度方面存在物理限制。因此,我们在此范围内裁剪动作值,以获得要在环境中测试的最终动作值。
如果你注意到我们获取随机值以添加到操作中的方式,你会看到我们使用了 numpy.random.normal 函数。这给了我们一个具有高斯分布的随机值。这意味着,如果我们平均噪声的平均值,它将非常接近 0,因为负值与正值一样可能给出。但是,如果机器人已经发现自己处于关键位置,则需要采取果断行动才能摆脱困境,并且每一步的随机高斯噪声是不够的。让我们想象一下,机器人发现自己直接面对一个障碍物。如果它只是继续前进,它就会崩溃。添加高斯噪声无济于事,因为它会在连续的步长中自行平衡。因此,我们实现了一个想法,迫使机器人对随机数量的连续步骤采取单一的决定性行动。这将迫使机器人随机经历一系列动作,使其脱离关键位置。

# 如果机器人面临障碍物,随机强制其采取一致的随机动作。这样做是为了增加在靠近障碍物的情况下的探索能力。训练也可以在没有这个步骤的情况下进行。
if random_near_obstacle:
if (
np.random.uniform(0, 1) > 0.85
and min(state[4:-8]) < 0.6
and count_rand_actions < 1
):
count_rand_actions = np.random.randint(8, 15)
random_action = np.random.uniform(-1, 1, 2)
if count_rand_actions > 0:
count_rand_actions -= 1
action = random_action
action[0] = -1
如果我们选择使用此方法(可以通过将 random_near_obstacle 设置为 False 来关闭它),我们可能不希望在障碍物附近的每次交互中使用它。因此,我们查询一个随机值,仅当随机值大于 0.85 时才继续使用此方法。此外,我们检查机器人前方的障碍物是否小于 0.6 米。请记住,我们的状态由 20 + 2 + 2 个值表示,其中前 20 个是激光传感器状态。为了检查障碍物是否在机器人前面(而不是在侧面),我们检查状态表示中从 4 到 16 的值(状态[4:-8])。我们还检查我们是否已经锁定到此方法的顺序执行中。如果所有这些都通过,我们选择一个介于 8 和 15 之间的随机数来执行此随机操作的步骤数。然后对随机操作进行采样。更新操作值,并对随机操作的计数器进行计数。
注意:
这里我们设置 action[0] = -1 以强制机器人仅进行随机旋转。要设置完全随机的操作,应注释掉或删除此行。
接下来,我们需要“修复”我们的动作值。我们的TD3神经网络模型输出2个取值范围在-1到1之间的动作值。但是我们的激光传感器只记录机器人前方180度的视野。但是我们的激光传感器只能在机器人前方的 180 度视野下进行记录。这意味着,让它向后移动对我们来说是不负责任的,因为机器人无法知道它后面是什么。因此,我们必须确保我们的机器人具有每秒 0 m 的最小线速度值。我们在以下代码中执行此操作。
# 将动作值更新为落在范围[0,1](线速度)和[-1,1](角速度)之间:
a_in = [(action[0] + 1) / 2, action[1]]
#使用更新后的动作值进行环境步进
next_state, reward, done, target = env.step(a_in)
#根据当前时间步数和最大时间步数判断是否达到了一个episode的最后一个时间步:
done_bool = 0 if episode_timesteps + 1 == max_ep else int(done)
done = 1 if episode_timesteps + 1 == max_ep else int(done)
#计算累积回报:
episode_reward += reward
# 将状态、动作、回报、完成标志等信息存储到经验回放缓冲区:
replay_buffer.add(state, action, reward, done_bool, next_state)
# 更新状态和计数器
state = next_state
episode_timesteps += 1
timestep += 1
timesteps_since_eval += 1
作用中的第一个值表示线速度,我们将其压缩在 0,1 范围内。我们可以将此值传递到我们的凉亭模拟器环境,并在其中执行该值。将返回一个新状态,以及状态-操作对奖励和两个布尔值,即状态是否结束(完成)和是否达到目标(在目标中)。接下来,我们检查是否达到最大步数并相应地更新完成并done_bool值(在 Python 中,int 和 bool 值可以互换使用,但在这里我们使用 int,因为我们使用 Bellman 方程中的数值作为状态的终点)。之后,只需更新当前运行状态(我们将在训练循环的下一次迭代中计算动作值)并更新训练计数器。
这个循环将运行,计算每个状态的动作,收集经验并训练神经网络模型,直到它结束......或者我们通过在终端中按 CTRL+C 来明确停止它(这是我最喜欢的方法)。
循环结束后,我们保存评估数据和最终网络参数。
# 在训练完成后,评估网络并保存模型和评估结果
evaluations.append(evaluate(network=network, epoch=epoch, eval_episodes=eval_ep))
if save_model:
network.save("%s" % file_name, directory="./models")
np.save("./results/%s" % file_name, evaluations)
评估
我们还没有研究的最后一件事是如何进行评估。为此,我们创建了一个单独的评估函数。
def evaluate(network, epoch, eval_episodes=10):
avg_reward = 0.0
col = 0
for _ in range(eval_episodes):
count = 0
state = env.reset()
done = False
while not done and count < 501:
action = network.get_action(np.array(state))
a_in = [(action[0] + 1) / 2, action[1]]
state, reward, done, _ = env.step(a_in)
avg_reward += reward
count += 1
if reward < -90:
col += 1
avg_reward /= eval_episodes
avg_col = col / eval_episodes
print("..............................................")
print(
"Average Reward over %i Evaluation Episodes, Epoch %i: %f, %f"
% (eval_episodes, epoch, avg_reward, avg_col)
)
print("..............................................")
return avg_reward
回想一下,这是在每个纪元完成后调用的。让我们设置一些变量来收集我们可能想要查看的数据,例如平均奖励和碰撞率。然后,我们启动评估循环,该循环将执行eval_episodes数量的随机评估运行。在这里,步骤与训练期间相同,只是我们没有对计算的操作应用随机噪声,因为我们有兴趣了解当前模型的性能如何。此外,我们还记录了奖励以及我们是否在运行中坠毁。崩溃的硬编码奖励是 -100,因此我们可以使用奖励来查看是否发生了碰撞(或者我们可以查看 done 是否为 True,而 target 是否为 False)。在执行所有评估运行后,我们计算平均奖励和冲突率。然后在终端中打印出来。理想情况下,我们应该看到接近 0 的高正平均奖励和碰撞率。我们返回平均奖励,以添加到评估结果中。
神经网络模型的训练到此结束,该模型从激光、目标和机器人运动输入训练移动机器人运动策略。从理论上讲,任何模拟环境都可以用于此代码部分。它不需要是我们的 Gazobo 模拟器,也不需要是记录环境的激光传感器输入。变量 env 应该只是具有不同的指定仿真环境,并且设置不同的 environment_dim 值将允许使用具有不同输入数量的传感器。但是我们确实已经有一个工作环境,可以实现 Gazebo 和 ROS,这正是我们将在第 第4 部分中看到的内容。
DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。
更多推荐



所有评论(0)