强化学习

在强化学习（RL）中，智能体学习在环境中操作以获得奖励。代理的目标是随着时间的推移提高其性能并获得更高的奖励。

在我们本周将要使用的示例中，我们将使用体育馆python包来提供代理将在其中工作的环境。体育馆包允许我们设置模拟环境，代理将在其中参与一系列回合。在每一轮中，环境将提供代理状态信息，然后要求代理为当前回合选择一个动作。一旦代理选择了一个要执行的动作，环境将通过计算代理的新状态来响应，并可能为代理最近的动作提供奖励。

代理在强化学习场景中面临的主要挑战是，奖励可能与任何个人行为选择在时间上相去甚远。例如，如果我们正在训练一个智能体下棋，那么唯一可用的奖励可能是当智能体获胜时获得1分的奖励。在这几节课中，我们将学习两个例子，其中奖励与智能体的行为相差不大。在下面的第一个例子中，代理将在每一轮中获得一分，而在第二个例子中，我们将训练代理玩Atari Breakout游戏：在该游戏中，玩家必须移动桨将球击中墙壁以击倒砖块。代理人每敲出一块砖就得一分。

下面两个例子的代码改编自Maxim Lapin的《Deep Reinforcement Learning Hands-On， 2nd Edition》一书。

Cartpole的例子

在Cartpole游戏中，我们训练一个代理在一辆可以左右移动的车上保持一根柱子直立。

杆子通过杆子底部的铰链连接到小车上。在每一轮游戏中，代理可以选择向左或向右推车。为了帮助智能体决定该做什么，环境将以以下信息的形式提供状态信息：小车的速度、杆子与垂直方向的夹角以及杆子的角速度。特工的目标是在尽可能多的回合中保持杆直立。

随机策略

对于我们的第一个解决方案，我们将使用纯随机策略：在游戏的每一轮中，程序将随机向左或向右推车。这一策略并不是很有效，并且通常只能在大约10个时间步内保持杆子不倒下。

import gymnasium as gym如果__name__ == "__main__": env = gym.make("CartPole-v1“) total_reward = 0.0 total_steps = 0 obs = env.reset（）而True: action = env.action_space.sample（）观察，奖励，终止，截断，info = env.step(action) total_reward += reward total_steps += 1如果终止或截断：break print(”情节在%d步骤中完成，总奖励%。% (total_steps, total_reward))

代码首先设置一个玩游戏的环境。为了开始游戏，我们调用环境的reset（）方法，该方法返回关于游戏开始时状态的状态信息。每个环境都包含一个action_space成员变量，其中包含关于玩家可用的动作的信息。动作空间对象提供了一个sample（）方法，我们可以调用它来随机选择一个动作。

要在环境中执行单个步骤，我们调用环境的step（）方法，将我们想要执行的操作传递给它。环境将使用包含新状态信息、此步骤的奖励和其他信息的元组进行响应。在侧翻环境中，每走一步，我们就会得到1分的奖励。

训练网络来玩这个游戏

在这个问题的第二个版本中，我们将使用神经网络来决定在每一步采取什么行动。

具体地说,

我们在这里注意到几个关键点：

下面是实现这个更高级策略的代码：

从collection中导入体育馆作为gym导入namedtuple导入numpy作为np导入torch导入torch。Nn为Nn进口火炬。optim as optim HIDDEN_SIZE = 128 BATCH_SIZE = 16 PERCENTILE = 70 class Net(nn. module): def __init__(self, obs_size, HIDDEN_SIZE, n_actions): super(Net, self).__init__() self.net = nn。顺序(nn。线性（obs_size, hidden_size）， nn. relu (), nn.Linear(hidden_size, n_actions)) def forward(self, x): return self.net(x) Episode = namedtuple('Episode', field_names=['reward', 'steps']) EpisodeStep = namedtuple('EpisodeStep', field_names=['observation', 'action']) def iterate_batch (env, net, batch_size): batch =[] episode_reward = 0.0 episode_steps =[] obs = env.reset()[0] sm = nn.Softmax(dim=0) while True：obs_v = torch.FloatTensor(obs) act_probs_v = sm(net(obs_v)) act_probs = act_probs_v.data.numpy() action= np.random.choice(len(act_probs), p=act_probs) next_obs, reward, terminated, truncated, info = env.step(action) episode_reward += reward step = EpisodeStep(observation=obs, action=action) episode_steps.append（step）如果被终止或截断：e = Episode(reward=episode_reward, steps=episode_steps) batch.append(e) episode_reward = 0.0 episode_steps = [] next_obs = env.reset()[0] if len(batch) == batch_size: yield batch = [] obs = next_obs def filter_batch(batch, percentile): rewards = list(map(lambda s: s.reward, batch)) reward_bound = np。百分位（奖励，百分位）reward_mean = float（np.mean（奖励））train_obs = [] train_act =[]对于奖励，分批步骤：如果奖励< reward_bound：继续train_obs。扩展(map（lambda）步骤：步骤。观察，步骤))扩展(map（lambda）步骤：步骤。) train_obs_v = torch.FloatTensor(train_obs) train_act_v = torch.LongTensor(train_act) return train_obs_v, train_act_v, reward_bound, reward_mean if __name__ == "__main__": env =健身房.make("CartPole-v1") obs_size = env.observation_space. .Shape [0] n_actions = env.action_space。n net =net (obs_size, HIDDEN_SIZE, n_actions) objective = nn.CrossEntropyLoss() optimizer = optim.Adam(params=net.parameters(), lr=0.01) for iter_no, batch in enumerate(iterate_batches(env, net, BATCH_SIZE)): obs_v, acts_v, reward_b, reward_m = \ filter_batch(batch， PERCENTILE) optimizer.zero_grad() action_scores_v =net (obs_v) loss_v = objective(action_scores_v, acts_v) loss_v.backward() optimizer.step() print（"%d: loss=% d "）3 f, reward_mean = %。1 f, rw_bound = %。if " % (iter_no, loss_v.item(), reward_m, reward_b)) if reward_m > 199: print("Solved!") break

以下是对这段代码的一些观察：

  1. 的iterate_batches ()函数运行游戏章节，直到收集到足够的数据形成一批游戏章节。
  2. iterate_batches ()是Python生成器函数的一个示例。它用yield代替return。当你点击yield语句时，你将返回一个结果。下次调用生成器函数时，它将返回yield语句所在的位置。
  3. 吐出的批次iterate_batches ()包含成功运行和不成功运行。的filter_batch ()函数只保留较长时间运行的数据，并返回一个元组，其中包含从这些成功运行中收集的赢博体育步骤。

编程任务

上面的代码为网络使用了PyTorch模型。删除PyTorch模型并将其替换为具有相同功能的Keras模型。