LongLong's Blog

分享IT技术,分享生活感悟,热爱摄影,热爱航天。

使用PyTorch实现强化学习

一直以为强化学习在各种神经网络框架中会有单独的模块,然而试了才发现其用的就是一般的神经网络,只是在使用上有一套独特的逻辑——强化学习的核心是通过神经网络对行动价值函数Q(s,a)进行预测,进而选择价值函数最高的行动从而实现行动的最优化。

1. 强化学习的基本概念

强化学习相对一般类型的机器学习,其特点是通过和环境的交互中收集数据作为输入,并据此获得实现奖励最大化。

收益:t时刻后的奖励总和,其中每经理一次行动都会将之前每次的奖励都乘以折现率(0~1之间),避免出现溢出问题,同时也使得近期能够获得的奖励权重更高。

\[G_t = R_t + \gamma R_{t+1} + \gamma^2 R_{t+2} + …\ = R_t + \gamma G_{t+1}\]

状态价值函数V:在某状态S下以策略P行动获得收益的期望值。

行动价值函数Q:在某状态S下采取采取A行动,此后以策略P行动获得收益的期望。

强化学习的过程为通过蒙特卡洛方法对行动价值函数Q进行估计,从而能够在选择行动是选取使Q最大的行动A。

2. 构建强化学习的数据结构

首先需要使用一个队列来对每一次行动进行记录,从而以此来训练对应的神经网络,可以借助Python中的dqueue,对应的功能代码如下

class ReplayMemory():
    def __init__(self, capacity):
        self.memory = collections.deque(maxlen=capacity);

    def add(self, data):
        self.memory.append(data);

    def sample(self, batch_size):
        return random.sample(self.memory, batch_size);

    def __len__(self):
        return len(self.memory);

提供了增加一项数据和从中抽取一定量数据两个方法,同时还实现了计算加入的数据总量的方法。

然后将设计几个进行数据预测的神经网络结构,这里使用三层全连接网络,其结构如下

class QNet(torch.nn.Module):
    def __init__(self, n_observations, n_action):
        super(QNet, self).__init__();
        self.layer1 = torch.nn.Linear(n_observations, 128);
        self.layer2 = torch.nn.Linear(128, 128);
        self.layer3 = torch.nn.Linear(128, n_action);

    def forward(self, x):
        x = torch.nn.functional.relu(self.layer1(x));
        x = torch.nn.functional.relu(self.layer2(x));
        return self.layer3(x);

其输入数据的维度为状态参数的维度,对应相关的状态参数,输出数据维度为行动种类的维度,对应每种行动的行动价值函数值。相当于拟合函数Q(S,A)

一般在进行强化学习时会使用两个神经网络,一个通过数据进行参数的优化,并使用改神经网络决定行动,同时将参数定期将参数同步给另一个(不太明白问什么需要这样操作,感觉可能是为了提高收敛性)。

#当前神经网络
policy_net = QNet(n_observations, n_actions).to(device);
#目标神经网络
target_net = QNet(n_observations, n_actions).to(device);

3. 强化学习的流程

进行episodes轮的行动,并在每轮中以一定的概率根据预测到的Q最大的行动进行行动,一定的概率随机行动,并记录每次行动的状态和对应的奖励为一组数据,并放置到队列中,当队列有足够的长度后,每次行动后抽取一定的数据样本,通过计算和神经网络分别得到行动价值函数,并通过优化神经网络当前神经网络减两者之间的误差。在完成一次当前神经网络的训练后,将当前神经网络和目标神经网络进行加权平均作为新的目标神经网络,如此完成一轮行动。

for i in range(0, n_episodes):
    state, info = env.reset();
    state = torch.tensor(state, dtype=torch.float32, device=device).unsqueeze(0);
    for t in itertools.count():
        with torch.no_grad():
        #根据当前网络得到行动最大价值的行动,并以一定概率选择该行动,一定概率随机行动
            action = policy_net(state).max(1).indices[0].item() \
                if random.random() > EPS_END else numpy.random.choice(n_actions);
        #获取行动后的需结果
        observation, reward, terminated, truncated, _ = env.step(action);
        next_state = None if terminated \
            else torch.tensor(observation, dtype=torch.float32, device=device).unsqueeze(0);

        #将结果记录到记忆中,并更新当前状态
        memory.add((state, action, next_state, reward));
        state = next_state;

        #当记忆中的数据大于一次训练所需要的数据,进行一次神经网络的训练
        if len(memory) >= BATCH_SIZE:
            #抽取数据,并进行封装
            batch = memory.sample(BATCH_SIZE);
            state_batch = [];
            action_batch = [];
            reward_batch = [];
            next_state_batch = [];
          
            for item in batch:
                state_batch.append(item[0]);
                next_state_batch.append(item[2]);
                action_batch.append(torch.tensor([[item[1]]], device=device));
                reward_batch.append(torch.tensor([[item[3]]], device=device));
            state_batch = torch.cat(state_batch);
            action_batch = torch.cat(action_batch);
            reward_batch = torch.cat(reward_batch);

            #剔除为终止位置的数据(终止位置行动价值函数为0)
            non_final_mask = torch.tensor(tuple(map(lambda s: s is not None, next_state_batch)), \
                device=device, dtype=torch.bool);
            non_final_next_states = torch.cat([s for s in next_state_batch if s is not None]);

            #使用当前神经网络预测行动价值函数
            state_action_values = policy_net(state_batch).gather(1, action_batch);
            next_state_values = torch.zeros(BATCH_SIZE, device=device);

            with torch.no_grad():
                #使用目标神经网络预测下一状态的价值函数
                next_state_values[non_final_mask] = target_net(non_final_next_states).max(1).values;
            expected_state_action_values = (next_state_values.unsqueeze(1) * GAMMA) + reward_batch;

            #对当前网络的预测进行优化
            loss_fn = torch.nn.SmoothL1Loss();
            loss = loss_fn(state_action_values, expected_state_action_values);
            optim.zero_grad();
            loss.backward();
            torch.nn.utils.clip_grad_value_(policy_net.parameters(), 100);
            optim.step();
      
        #将当前网络同步到目标网络,这里使用了当前网络和目标网络的加权平均进行同步,一定程度提高收敛性 
        target_net_state_dict = target_net.state_dict();
        policy_net_state_dict = policy_net.state_dict();
        for key in policy_net_state_dict:
            target_net_state_dict[key] = policy_net_state_dict[key] * \
                TAU + target_net_state_dict[key] * (1 - TAU);
        target_net.load_state_dict(target_net_state_dict);

        if terminated or truncated == True:
            print(i, t);
            break;

4. 总结

强化学习的过程是通过蒙特卡洛方法对于行动价值函数Q(s,a)进行拟合的过程,因为一旦能够准确预测行动价值函数,就很容易选择行动价值最大化的行为进行行动,同时由于对于行动价值函数的预估可能存在偏差,可始终保留一定概率继续通过蒙特卡洛方法修正行动价值函数,使得得到的模型可以不断的提升性能。

使用Python构建图形界面

自从习惯了Web程序后,自己的PC图形界面程序开发基本还停留在Java上,使用Netbeans的图形界面构建工具搭配Swing实现图形界面程序的构建,并且一直感觉Java的图形界面编程的逻辑是非常清晰的——各种操作行为对应相应的Action类或是Listener接口。

之后也尝试过Qt或是其他的图形界面库,感觉都不是特别好用。如今已经好多年不用Java,而且Java目前的生态环境相比现在Python的流行也要差很多了,估计主要还是用在Android开发和比较老的电商系统上了吧。

目前的需求是借助Python的Numpy和Matplotlib实现一些类型Matlab的功能,并且要封装成为一个可执行文件。

1. Tk图形界面库

Tkinter 模块(Tk 接口)是Python的标准Tk GUI工具包的接口 .Tkinter可以在大多数的Unix平台下使用,同样可以应用在Windows 和macOS系统里。Tk8.0的后续版本可以实现本地窗口风格,并良好地运行在绝大多数平台中。

其最大的优点就是作为了Python的常用标准库,直接就可以使用,而且使用上也相对简单,对于开发较为简单的图形界面程序比较友好,使用难度也相对较低。其运行的核心代码如下

import tkinter;

if __name__ == '__main__':
    tk = tkinter.Tk(className='Main');
    tk.mainloop();

2. Tk控件

Tkinter的提供各种控件,如按钮,标签和文本框等共计15种。在使用时需要关联到控件的上层控件,实现整个控件的树形结构。如下面的代码实现了一个完整的菜单控件。控件使用add_command方法实现绑定操作时执行的代码。

menu = tkinter.Menu(tk);
m1 = tkinter.Menu(menu, tearoff=False);
m1.add_command(label='打开', command=load_data);
m1.add_command(label='保存', command=save_data);
m1.add_command(label='退出', command=on_exit);
menu.add_cascade(label='文件', menu=m1);
tk.config(menu=menu);

除了基本控件,Tkinter还提供了一些常用的组合控件,如各种类型的提示框,实现消息提示,文件选择等功能。

#提示信息
tkinter.messagebox.showerror('警告', '数据格式错误!');
#选择要打开的文件
path = tkinter.filedialog.askopenfilename();

3. 界面布局

Tkinter控件有特定的几何状态管理方法,管理整个控件区域组织,以下是Tkinter公开的几何管理类:包、网格、位置。比较类似Java中的Layout。使用包(pack)可是实现控件大方位的指定(上下左右)。

#放置在顶部
canvas.get_tk_widget().pack(side=tkinter.TOP);
#放置在底部
s2.pack(side=tkinter.BOTTOM);

4. 动态调整控件属性

在图形界面程序中经常需要进行一定操作后,根据执行的结果动态调整某些控件的某些属性,Tk中则是通过控件的config方法,该方法接受一个Python字典结构,字典中的键值为需要调整的属性,值则为调整后的属性值。如以下代码将两个范围控件的最大值进行更新。

fm = data.maxFreq() + 2;
s1.config({'to' : fm});
s2.config({'to' : fm});

5. 在Tk中集成Matplotlib图表

由于需要使用Matplotlib绘制图表,并继承在图形界面中,Tk也支持将Matplotlib图表放置在界面中,可通过以下方法实现

#创建一个Matplotlib绘图
figure = matplotlib.figure.Figure();
#利用绘图创建一个画布
canvas = matplotlib.backends.backend_tkagg.FigureCanvasTkAgg(figure, master=tk);
#将画布放置到界面中
canvas.get_tk_widget().pack(side=tkinter.TOP);
#在完成绘图后调用更新显示
canvas.draw();

6. 打包程序

如今Python可以借助pyinstaller实现对源代码进行打包,形成一个可执行文件,如此就可以实现程序的分发,而不需要运行的计算机安装配置对应的Python环境,同时也能一定程度上隐藏源代码。打包的命令格式如下

pyinstaller -F -w -i <程序图标文件> [源代码文件]

其中-F表示生成一个exe文件,-w表示运行时不弹出命令提示符窗口,-i可以指定可执行文件的图标。