4. Cross-Entropy交叉熵方法的RL

这应该是最简单的Reinforcement Learning方法之一了，方法实现的过程也很符合我们直观理解RL到底在干什么。只是这个方法有一些试用场景的限制，就像回头会给大家介绍的另一种简单方法“表格法”，也有其适用场景的限制。CE方法主要还是更加适合一些短回合的小场景。

后面基本开始正式介绍强化学习 了，所以有必要在这里交代下通常强化学习的分类。

4.1 RL方法的分类

这里有个直观的认识即可。后续遇到具体的场景再具体说。按照分类标准不同，大致公认的有三种分法。

• Model-free 和 Model-based：Model-free主要是不需要对环境建模。Agent观察到什么“内容(observations)”直接做行为(Action)选择。也就是说，Agent拿到它观察到的信息，做些计算（当然是在寻找过去的经验），然后选择最合适的行为。而Model-based方法则试图对环境建模，预测未来可能观察到的状态以及可能获取的回报(reward)。这样的模型通常预测多步之后的情况，似乎看得更远。但其实两者各有优劣。Model-based通常用在确定性更强的环境，原因很简单更长远的预测可能有效，当然可以让Agent更快地找到该环境的适应规则。而Model-free则适用于更加不确定的环境，很难刻画和预测的未来。比如：金融市场。

• Policy-based 和 Value-based：前者顾名思义是直接近似迭代出Agent的最佳Action选择。而Value-based则在Agent和Action之间用Value建立起一座桥梁。就好比我们人在做某个选择前先对不同选择有个“价值评判”，选我们认为可能更好的那一个一样。当然人经常会价值评判失误，机器也不例外。但RL提供了机制不断地去调整机器的价值评判标准，期望它的评价可能成功率更高。这个评价标准也就是这类方法要找的Value Function。

• On-Policy 和 Off-Policy：这个留在后面说。这里直接说会比较抽象。

4.2 补充一点小知识

通常实践中我们会将一些非线性 关系用监督学习的网络结构来做映射。比如：Agent从环境获取了一些观察信息，需要做出行为选择。那这个时候观测信息作为input，却并不清楚这些观测信息跟行为是怎么样一种关系，这时候我们就可以用一个待训练的网络结构来映射观测信息到行为选择概率分布。也即是：

就是在状态下选择不同行为的一个概率分布。当Agent已经训练到一个比较好的状态，给出的状态下较大概率选择的行为也就是对于Agent在没有更新信息之前的一个较佳行为选择方案。如果我们用这种方法直接去寻找行动方案，其实也就是一个Policy-based方法。当然上面那个监督学习方案也像是一个分类问题。有了分布，Agent的行为选择就是一个在分布中随机抽样的结果了。

假想一个多回合（episodes）的场景，比如Agent玩某个游戏等等。每一局游戏Agent有一连串的行动方案直至本局游戏结束。当Agent玩了很多局之后：

每一局都会玩一定步数直至这局游戏结束，所以每个episode的实际经历步数大概率是不等的。但是每局游戏结束都会有一个总的反馈。当不考虑得分的时间价值，不设多步之间的折现比率的情况下，每局游戏的回报自然就是：

对每局游戏下来的得分进行比对，自然就知道哪局游戏玩得好，哪局游戏玩得相对差。那么这时候我们只需要定位表现好的那些episodes（可以设定一定比例，如前），总结在这些episodes过程中，面对不同的观察值所选择的行为，即可找到分布。当然，个episodes不够，还可以多来几轮，使得分布相对稳定和收敛，就找到了这个游戏场景的优化行动方案了。

其实这也就是Cross-Entropy的RL学习思路了。有点遗传算法中的思想，找到表现好的留下，分析表现好的原因，用表现好的实践信息去逼近优化的行动方案。

4.3 CartPole的例子

倒立锤场景：场景很简单，小车上有一个杆。小车以一定速度在水平方向左右移动，而杆以一定角速度转动。目标是不要让杆倒下，倒下则episode结束。这个系统用于观察的四个状态变量分别是：小车的速度，小车的位置，杆与垂直线的夹角，杆转动的角速度。而Agent可以采取的行动是左右移动小车，使得小车上的杆不倒下来，倒下来即这个回合episode结束。例子将会运用gym已经创建好的测试场景。

例子中将会用一个简单线性隐含层的网络结构将观察值映射到Agent的行动选择方案（分布），即：

首先设置实验例子的全局参数：

HIDDEN_SIZE = 128    # 用来映射观察值到行动方案（概率分布）的网络隐含层神经元个数
BATCH_SIZE = 16      # 每个epoch迭代使用一个batch中回合次数，即16个episodes
PERCENTILE = 70      # 运用Cross-entropy方法择优选择表现最好的30%episodes训练网络

然后定义网络结构:

class Net(nn.Module):
    def __init__(self, obs_size, hidden_size, n_actions):
        super(Net, self).__init__()
        self.net = nn.Sequential(
            nn.Linear(obs_size, hidden_size),
            nn.ReLU(),
            nn.Linear(hidden_size, n_actions)    # 单隐含层结构，用ReLU激活
        )

    def forward(self, x):       # 定义net的前向计算forward()方法
        return self.net(x)

网络结构的定义基本上是大家遇到的网络结构中的极简版本了，即使这样简单的一个网络结构，在这个例子中也能取得极好的效果。大家注意obs_size、n_actions的维度即可。不清楚的可以去看看gym的源码。这里obs_size = 4, n_actions = 2。由于pytorch中接收batch数据传入net，所以我们的观察数据通常处理成1×4的shape传入，第一个维度表示batch，第二个维度是观测变量。

Episode = namedtuple('Episode', field_names=['reward', 'steps'])
EpisodeStep = namedtuple('EpisodeStep', field_names=['observation', 'action'])

随后定义Episode和EpisodeStep来存储游戏每回合的一些信息。就像上面4.2部分提到的那样，每个回合episode中，直到游戏结束，都经历了些什么样的观察值的序列，针对不同的观察值，Agent都选择了什么样的行为，最终episode的得分如何，用reward来记录。

这里会用到一个namedtuple的数据结构。其实它是个类，Episode和EpisodeStep都是用namedtuple来实例化的东东。后续大家可以看到，例子代码用EpisodeStep来记录每一个episode中，Agent经历的观察值序列和行动序列。而Episode记录的是每个episode的reward和这个episode对应的EpisodeStep。层次结构上，Episode在EpisodeStep之上。

而后就是根据多个episode游戏的结果，生成用于训练观察值到行动分布的网络Net的数据。这里数据生成采用batch模式，每个batch生成16个episode的数据。而函数iterate_batches调用是个生成器（这种写法的好处是节省内存，不明白的找python的书来看吧），方便后续迭代调用的时候，每次由yield返回batch中的一个episode的信息。

def iterate_batches(env, net, batch_size):
    batch = []       # 初始化一个空列表，后续装填不同的episode的信息
    episode_reward = 0.0   # 每回合初始化reward=0.0
    episode_steps = []     # 初始化一个空列表，用来记录一个episode中观察值和行动值序列
    obs = env.reset()      # gym的cartpole环境重置，生成初始化的观察值
    sm = nn.Softmax(dim=1)  # 这个softmax用来将网络的输出转化为行动概率分布
    while True:
        obs_v = torch.FloatTensor([obs])   # 将环境传来的观察值obs转换为Tensor
        act_probs_v = sm(net(obs_v))       # 用当前网络输出行动概率分布
        act_probs = act_probs_v.data.numpy()[0]  # 将tensor的行动概率分布转换为np.ndarray
        action = np.random.choice(len(act_probs), p=act_probs) # 按照行动概率分布抽样选择行为
        next_obs, reward, is_done, _ = env.step(action)  
        # Agent采取行动后，env返回该行动的reward，行动后系统的观察值以备下一步行动选择，is_done用来标识该回合游戏是否结束
        episode_reward += reward   # 这里累积这个episode中Agent每步行动获取的reward
        episode_steps.append(EpisodeStep(observation=obs, action=action))
        # 把这一步的观察值obs和采取的行动action记录到episode_steps中
        if is_done:  # 下面是env如果返回episode回合结束的逻辑
            batch.append(Episode(reward=episode_reward, steps=episode_steps))
            # 首先把这个回合总的reward返回
            # 再次把这个回合经历过的obs和action以episode_steps的形式返回
            # 一起作为Episode装填到batch列表中
            episode_reward = 0.0   # 重置episode_reward， 用于记录下一个episode的reward的起点
            episode_steps = []     # 同理要重置episode_steps，用于记录下一个episode的观察值和行动序列
            next_obs = env.reset() # 重置游戏环境，生成下一回合游戏起点的观察值
            if len(batch) == batch_size: # 判断batch中是否生成了足够多的episode，例子里是16，如果是，那么一个batch的数据就算是准备好了。
            # 用yield返回batch。注意不是return返回。所以iterate_batches()是生成器
                yield batch
                batch = []  # 重置batch，用于装填下一个batch
        obs = next_obs      
        # 如果该回合没有结束，肯定要将env返回来的next_obs作为下一步行动的obs，所以最后这句也必不可少。

batch中的数据准备好后，是直接把所有的Agent在每个episode中尝试过的行为路径都用来训练网络Net吗？显然不是。Cross-entropy的方法是要每次用那些在每个episode中得分更高的拿来训练，不断迭代找到越来越好的行动方案。所以，要通过每次episode中的得分对每个batch中的数据进行过滤。

def filter_batch(batch, percentile):
    rewards = list(map(lambda s: s.reward, batch))  # 将batch中每个episode的reward转换为列表
    reward_bound = np.percentile(rewards, percentile) # 运用分位数找到入选表现好的episode的分值
    reward_mean = float(np.mean(rewards))  # 计算这个batch的平均分，用于跟踪Agent是不是每个batch迭代完了表现越来越好

    train_obs = []    # 建立训练用的观察值列表
    train_act = []    # 建立训练用的行为值列表
    for example in batch:  # 把符合优秀标准的episode数据装入train_obs和train_act
        if example.reward < reward_bound:
            continue
        train_obs.extend(map(lambda step: step.observation, example.steps))
        train_act.extend(map(lambda step: step.action, example.steps))

    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

功能模块完成之后，就是主程序部分了。主程序部分要实现的主要逻辑就是：Agent不断地从环境拿到obs，选择action，得到next_obs，再选择action直到episode结束，再循环这个过程，直到一个batch（16个episode）结束。构造这个batch的训练数据，调整优化网络Net的参数。然后再从头循环下一个batch，这时候根据obs选择行为action的时候用第一个batch调整优化过的Net进行。如此循环下去，多个batch之后，就找到一个针对倒立锤的一个“合理”的行动方案。

if __name__ == "__main__":
    env = gym.make("CartPole-v1")   # 利用gym建立CartPole环境env
    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)  # 设定优化器和学习率
    writer = SummaryWriter(comment="-cartpole")   # 设定tensorboard的记录
    
    # 下面实现迭代每个batch并训练网络的逻辑
    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()      # 每次优化参数前先将net中参数梯度清零
        action_scores_v = net(obs_v)   # 用第一个batch中训练观察值经过net生成预测行为
        loss_v = objective(action_scores_v, acts_v) # 跟实际采取的行为比对计算损失
        loss_v.backward()   # 反向传播计算梯度
        optimizer.step()    # 优化调整参数
        
        # 下面就是记录训练结果和加载数据到tensorboard进行观察
        print("%d: loss=%.3f, reward_mean=%.1f, reward_bound=%.1f" % (
            iter_no, loss_v.item(), reward_m, reward_b))
        writer.add_scalar("loss", loss_v.item(), iter_no)
        writer.add_scalar("reward_bound", reward_b, iter_no)
        writer.add_scalar("reward_mean", reward_m, iter_no)
        if reward_m > 180:
            print("Solved!")
            break
    writer.close()

最后，我们就可以尝试跑一跑上面的代码。我的电脑上大概30个batch不到就达到我们设定的训练目标了。可以从tensorboard监控中的图可以看到，运用Cross-Entropy的方法，在这个场景中还是具有比较好的收敛性特征的。即使我们用来映射观察值obs和行动action的网络Net非常简单。并且我们也可以看到我们每个batch挑选得分高的episode条件也越来越高，每个batch优化之后的net输出的行动方案也使得Agent在每个episode中平均得分也越来越高，我们的目的也就达到了。