image frame

RHODES ISLAND

一个刀客塔的自留地。

pytorch学习

pytorch学习

学习参考：动手学深度学习

1 基础

损失函数（Loss）

定义方法:

  1. L1Loss: $loss(y,y’)= | y’-y |$
  2. L2Loss: $loss(y,y’)=\frac{1}{2}(y’-y)^2$
  3. Huber’s Robust Loss: $loss(y’,y)=\begin{cases}
    | y’-y |-\frac{1}{2},\mathrm{if} | y’-y |>1 \
    \frac{1}{2}(y’-y)^2,\mathrm{otherwise}
    \end{cases}$

网络层

  1. 展平层：（可以reshape输入输出，将其展开成一个向量）nn.Flatten()

激活函数

  1. sigmoid: $\sigma (x)=\frac{1}{1+exp(-x)}$
  2. Tanh: $\sigma (x)=\frac{1-exp(-2x)}{1+exp(-2x)}$
  3. ReLU: $\sigma (x)=max(x,0)$

2 框架

2.1 模型构造：Module&Sequential

Sequential可以用来连接各种Module及基于其写的类（可能是层，也可能是网络）。
对于一个自己写出的层，可以随意定义前向传播函数（即如何进行网络计算），其反向求导都可以进行。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
import torch
from torch import nn
from torch.nn import functional as F

X=torch.rand(2,20)

class MLP(nn.Module):
    def __init__(self):
        super().__init__()
        self.hidden=nn.Linear(20,256)
        self.out=nn.Linear(256,10)

    def forward(self,X):
        return self.out(F.relu(self.hidden(X)))

class MySequential(nn.Module):
    def __init__(self,*args):#此处用指针传了一个列进来
        super().__init__()
        for index,block in enumerate(args):
            #enumerate()对一个列返回其Index和列中元素
            self._modules[index]=block

    def forward(self,X):
        for module in self._modules.values():
            X=module(X)
        return X

class FixedHiddenMLP(nn.Module):
    def __init__(self):
        super().__init__()
        # 不计算梯度的随机权重参数。因此其在训练期间保持不变
        self.rand_weight = torch.rand((20, 20), requires_grad=False)
        self.linear = nn.Linear(20, 20)

    def forward(self, X):#这里主要是想说forward是自由的（
        X = self.linear(X)
        # 使用创建的常量参数以及relu和mm函数
        X = F.relu(torch.mm(X, self.rand_weight) + 1)
        # 复用全连接层。这相当于两个全连接层共享参数
        X = self.linear(X)
        # 控制流
        while X.abs().sum() > 1:
            X /= 2
        return X.sum()

class NestMLP(nn.Module):
    def __init__(self):
        super().__init__()
        self.net = nn.Sequential(nn.Linear(20, 64), nn.ReLU(),
                                 nn.Linear(64, 32), nn.ReLU())
        self.linear = nn.Linear(32, 16)

    def forward(self, X):
        return self.linear(self.net(X))
    #这里主要是说明，需要看forward函数才能知道这个网络具体是怎么个顺序和计算方法）

#sequential可以用来嵌套各种块，层+网络，网络＋网络，层＋层，无论什么都可以（只要基于module）
chimera = nn.Sequential(NestMLP(), nn.Linear(16, 20), FixedHiddenMLP())
print(chimera(X))

2.2 参数管理

2.2.1 参数访问

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

import torch
from torch import nn

net = nn.Sequential(nn.Linear(4, 8), nn.ReLU(), nn.Linear(8, 1))
X = torch.rand(size=(2, 4))
net(X)

#可以直接通过索引得到所有参数（没有参数的不行
print(net[2].state_dict())

#进一步地，对于不同的层，可以直接得到其参数
print(type(net[2].bias))
print(net[2].bias)
print(net[2].bias.data)#还有.bias.grad

#也可以一次性访问许多参数，其中*在print时作为解包存在（即把元素从列表拿出来
print(*[(name, param.shape) for name, param in net[0].named_parameters()])
print(*[(name, param.shape) for name, param in net.named_parameters()])
#.named_parameters()：这是PyTorch中Module类的一个方法，它返回一个迭代器，其中包含模块的所有参数以及它们的名称。
#每次迭代返回一个(name, parameter)对，其中name是参数的名称，parameter是参数的Tensor对象。

net.state_dict()['2.bias'].data
#net.state_dict()：这是PyTorch中Module类的一个方法，它返回一个有序字典（OrderedDict），其中包含了神经网络中所有参数的键值对。
#字典的键是参数的名称，通常是模块的名称和参数的名称组合而成，例如'layer_name.weight'或'layer_name.bias'。字典的值是参数的Tensor对象。

#对于嵌套状态下的块，可以一层一层调用出来
def block1():
    return nn.Sequential(nn.Linear(4, 8), nn.ReLU(),
                         nn.Linear(8, 4), nn.ReLU())

def block2():
    net = nn.Sequential()
    for i in range(4):
        # 在这里嵌套
        net.add_module(f'block {i}', block1())
    return net

rgnet = nn.Sequential(block2(), nn.Linear(4, 1))

print(rgnet[0][1][0].bias.data)

2.2.2 参数初始化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33


def init_normal(m):#此处的m是一个Module
    if type(m) == nn.Linear:
        #都是nn自带的init
        nn.init.normal_(m.weight, mean=0, std=0.01)
        nn.init.zeros_(m.bias)

def init_constant(m):
    if type(m) == nn.Linear:
        nn.init.constant_(m.weight, 1)
        nn.init.zeros_(m.bias)

def init_xavier(m):
    if type(m) == nn.Linear:
        nn.init.xavier_uniform_(m.weight)
def init_42(m):
    if type(m) == nn.Linear:
        nn.init.constant_(m.weight, 42)

def my_init(m):#同样的，我们始终可以自己自定义一些初始化方法。
    if type(m) == nn.Linear:
        print("Init", *[(name, param.shape)
                        for name, param in m.named_parameters()][0])
        nn.init.uniform_(m.weight, -10, 10)
        m.weight.data *= m.weight.data.abs() >= 5

net.apply(init_normal)#只要是Module的函数都可以apply，会自动进行多层的递归操作的
netnet[0].apply(init_xavier)#也可以单独对某个Module进行初始化。
net[0].weight.data[:] += 1
net[0].weight.data[0, 0] = 42
net[0].weight.data[0]
#可以使用暴力重载的方法初始化参数。

2.2.3 参数绑定

有时我们希望在多个层间共享参数。我们可以定义一个层，然后多次使用它。

1
2
3
4
5
6
7
8
9
10
11
12
# 我们需要给共享层一个名称，以便可以引用它的参数
shared = nn.Linear(8, 8)
net = nn.Sequential(nn.Linear(4, 8), nn.ReLU(),
                    shared, nn.ReLU(),
                    shared, nn.ReLU(),#直接将两层重复使用即可，两层会一直绑定的。
                    nn.Linear(8, 1))
net(X)
# 检查参数是否相同
print(net[2].weight.data[0] == net[4].weight.data[0])
net[2].weight.data[0, 0] = 100
# 确保它们实际上是同一个对象，而不只是有相同的值
print(net[2].weight.data[0] == net[4].weight.data[0])

2.3 自定义层

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import torch
import torch.nn.functional as F
from torch import nn

#自定义了一个层，该层的运算是将所有输入的平均值变为0
class CenteredLayer(nn.Module):
    def __init__(self):
        super().__init__()

    def forward(self, X):
        return X - X.mean()

#此处自定义一个带参数的层，同样的，其计算方法定义在forward里面。
class MyLinear(nn.Module):
    def __init__(self, in_units, units):
        super().__init__()
        self.weight = nn.Parameter(torch.randn(in_units, units))
        self.bias = nn.Parameter(torch.randn(units,))
    def forward(self, X):
        linear = torch.matmul(X, self.weight.data) + self.bias.data
        return F.relu(linear)

2.4 读写

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import torch
from torch import nn
from torch.nn import functional as F

class MLP(nn.Module):
    def __init__(self):
        super().__init__()
        self.hidden = nn.Linear(20, 256)
        self.output = nn.Linear(256, 10)

    def forward(self, x):
        return self.output(F.relu(self.hidden(x)))

net = MLP()
X = torch.randn(size=(2, 20))
Y = net(X)

#torch.save和torch.load可以用于存储所有参数
torch.save(net.state_dict(), 'mlp.params')
clone = MLP()
clone.load_state_dict(torch.load('mlp.params'))
clone.eval()

2.5 GPU

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
import torch
from torch import nn

def try_gpu(i=0):  #@save
    """如果存在，则返回gpu(i)，否则返回cpu()"""
    if torch.cuda.device_count() >= i + 1:
        return torch.device(f'cuda:{i}')
    return torch.device('cpu')

def try_all_gpus():  #@save
    """返回所有可用的GPU，如果没有GPU，则返回[cpu(),]"""
    devices = [torch.device(f'cuda:{i}')
             for i in range(torch.cuda.device_count())]
    return devices if devices else [torch.device('cpu')]

X = torch.ones(2, 3, device=try_gpu())#或者直接写'cuda:0/1/2'就行了
Z = X.cuda(1)#可以在GPU之间挪动变量。
net = nn.Sequential(nn.Linear(3, 1))
net = net.to(device=try_gpu())#在GPU之间挪动网络可以使用.to

3 线性回归

对于标准深度学习模型，我们可以使用框架的预定义好的层。这使我们只需关注使用哪些层来构造模型，而不必关注层的实现细节。 我们首先定义一个模型变量net，它是一个Sequential类的实例。 Sequential类将多个层串联在一起。 当给定输入数据时，Sequential实例将数据传入到第一层， 然后将第一层的输出作为第二层的输入，以此类推。 在下面的例子中，我们的模型只包含一个层，因此实际上不需要Sequential。 但是由于以后几乎所有的模型都是多层的，在这里使用Sequential会让你熟悉“标准的流水线”。

PyTorch中，全连接层在Linear类中定义。 值得注意的是，我们将两个参数传递到nn.Linear中。 第一个指定输入特征形状，第二个指定输出特征形状，输出特征形状为单个标量.

此处以回归方程$y=Xw+b$为例，其中$X=[x_1,x_2],w=[w_1,w_2]^T$

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
import numpy as np
import torch
from torch.utils import data
from d2l import torch as d2l

from torch import nn

true_w = torch.tensor([2, -3.4])
true_b = 4.2
features, labels = d2l.synthetic_data(true_w, true_b, 1000)

def load_array(data_arrays, batch_size, is_train=True):  #@save
    """构造一个PyTorch数据迭代器"""
    dataset = data.TensorDataset(*data_arrays)
    return data.DataLoader(dataset, batch_size, shuffle=is_train)

batch_size = 10
data_iter = load_array((features, labels), batch_size)

net=nn.Sequential(nn.Linear(2,1))
net[0].weight.data.normal_(0,0.01)
net[0].bias.data.fill_(0)

loss=nn.MSELoss()

trainer=torch.optim.SGD(net.parameters(),lr=0.03)

num_epochs = 3
for epoch in range(num_epochs):
    for X, y in data_iter:
        l = loss(net(X) ,y)
        trainer.zero_grad()
        l.backward()
        trainer.step()
    l = loss(net(features), labels)
    print(f'epoch {epoch + 1}, loss {l:f}')

此外，还有softmax、多层感知机等内容，此处不再描述，细节在笔记中（因为发现和pytorch关系不大）

4 卷积网络

具体推导见笔记。

4.1 图像卷积

使用了一个简单的图像。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
import torch
from torch import nn
from d2l import torch as d2l
import numpy as np

def corr2d(X,K):
    k_h,k_w=K.shape
    Y=torch.zeros(X.shape[0]-k_h+1,X.shape[1]-k_w+1)
    for i in range(Y.shape[0]):
        for j in range(Y.shape[1]):
            Y[i,j]=(X[i:i+k_h,j:j+k_w]*K).sum()
    return Y

class Cov2d(nn.Module):
    def __init__(self,k_size):
        super().__init__()
        self.weight=nn.parameter(torch.rand(k_size))
        self.bias=nn.parameter(torch.zeros(1))

    def forward(self,X):
        return corr2d(X,self.weight)+self.bias

X=torch.ones(6,8)
X[:,2:6]=0
K = torch.tensor([[1.0, -1.0]])
Y=corr2d(X,K)
print(X,Y)

#前两个1是通道，kernel是核
net=nn.Conv2d(1,1,kernel_size=(1,2),bias=False)
X=X.reshape((1,1,6,8))
Y=Y.reshape((1,1,6,7))
lr=1e-2
episode=20

for i in range(episode):
    Y_=net(X)
    loss=(Y_-Y)**2
    net.zero_grad()
    loss.sum().backward()

    net.weight.data[:]-=lr*net.weight.grad
    if (i + 1) % 2 == 0:
        print(f'epoch {i + 1}, loss {loss.sum():.3f}')

print(net.weight.data)

4.1.1 填充和步幅

由于卷积会使得维度大幅下降，考虑到深度和边缘检测的问题，在边缘加上空白行列。
同样地，对于一些很大维度的图片，考虑增大步幅可以实现较快的维度下降。

1
2
3
conv2d = nn.Conv2D(1, kernel_size=(3, 5), padding=(0, 1), strides=(3, 4))
#padding即在行列上下（左右）分别填充多少，strides指步幅（行，列）
comp_conv2d(conv2d, X).shape

4.1.2 多输入多输出卷积（多通道卷积）

*此外还有核为11的卷积，不再赘述，具体看笔记。

4.1.3 池化层

由于卷积对位置很敏感，用池化层可以虚化对位置的敏感情况。

1
2
3
4
5
6
7
8
9
10
def pool2d(X,pool_size,mode='max'):
    p_h,p_w=pool_size
    Y=np.zeros((X.shape[0]-p_h+1,X.shape[1]-p_w+1))
    for i in range(Y.shape[0]):
        for j in range(Y.shape[1]):
            if mode=='max':
                Y[i,j]=X[i:i+p_h,j:j+p_w].max()
            if mode=='mean':
                Y[i,j]=X[i:i+p_h,j:j+p_w].mean()
    return Y

Python基础学习

Python 基础学习

这里的链接均无授权，如有问题邮件告知我会删除（。

1 数据结构

1.1 队列

1.1.1 双向队列

使用collection模块下的deque:教程指路：CSDN

5 常用函数

  1. zip: 函数用于将可迭代的对象作为参数，将对象中对应的元素打包成一个个元组，然后返回由这些元组组成的列表。教程指路：基础教程
  2. random库，可以用于单次或重复随机抽样，也可以用于生成随机数等。教程指路：基础教程

RL学习-1

RL

RL学习-基础

学习参考：动手学强化学习和强化学习的数学原理

1 多臂老虎机问题

问题基于多臂老虎机开展，其各个拉臂之间相互独立，对老虎机进行类封装，方便进行后续探究。

1.1 epsilon-greedy方法

按照书中内容，这里主要展示的是epsilon逐渐下降的收敛优势。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84

import numpy as np
import matplotlib.pyplot as plt

class Bandit:
    def __init__(self,k):
        self.prob=np.random.uniform(size=k)
        self.max_index=np.argmax(self.prob)
        self.max_prob=np.max(self.prob)
        self.K=k

    def step(self,now):
        if np.random.rand()<self.prob[now]:
            return 1
        else:
            return 0

class Solver:
    def __init__(self,Bandit):
        self.bandit=Bandit
        self.counts=np.zeros(Bandit.K)
        self.regrets=[]
        self.actions=[]
        self.total_regret=0

    def update_regret(self,k):
        self.total_regret+=self.bandit.max_prob-self.bandit.prob[k]
        self.regrets.append(self.total_regret)

    def run_one_step(self):
        print("error?")

    def run_n_times(self,n):
        for i in range(n):
            k=self.run_one_step()
            self.counts[k]+=1
            self.actions.append(k)
            self.update_regret(k)

class EpsilonGreedy(Solver):
    def __init__(self,bandit,prob_init):
    # def __init__(self,bandit,epsilon,prob_init):
        super().__init__(bandit)
        # self.epsilon=epsilon
        self.total_count=0
        self.epsilon = 0
        self.estimate_reward=np.full(bandit.K,prob_init)
    #被注释掉的是不变的epsilon,现在是变epsilon=1/t

    def run_one_step(self):
        self.total_count+=1
        self.epsilon=1/self.total_count
        tmp=np.random.random()
        if tmp<self.epsilon:
            k=np.random.randint(0,self.bandit.K)
        else:
            k=np.argmax(self.estimate_reward)
        reward=self.bandit.step(k)
        self.estimate_reward[k]+=1/(1+self.counts[k])*(reward-self.estimate_reward[k])
        return k

def plot_results(solvers, solver_names):
    """生成累积懊悔随时间变化的图像。输入solvers是一个列表,列表中的每个元素是一种特定的策略。
    而solver_names也是一个列表,存储每个策略的名称"""
    for idx, solver in enumerate(solvers):
        time_list = range(len(solver.regrets))
        plt.plot(time_list, solver.regrets, label=solver_names[idx])
    plt.xlabel('Time steps')
    plt.ylabel('Cumulative regrets')
    plt.title('%d-armed bandit' % solvers[0].bandit.K)
    plt.legend()
    plt.show()

bandit_10_arm=Bandit(10)
np.random.seed(1)
epsilon_greedy_solver = EpsilonGreedy(bandit_10_arm,1.0)
epsilon_greedy_solver.run_n_times(5000)
print('epsilon-贪婪算法的累积懊悔为：', epsilon_greedy_solver.regrets)
plot_results([epsilon_greedy_solver], ["EpsilonGreedy"])

# np.random.seed(1)
# k=10
# test_bandit=Bandit(k)

1.2 上置信界算法（UCB）

引入一个不确定度度量$U(a_t)$，这个不确定度会根据该动作尝试次数的增加而降低。

依据：$X_i$为随机变量，其满足霍夫丁不等式：
$P(E(X)<x(a_t)+u)>1-e^{-2nu^2}$

期望报酬$\hat{Q}(a_t)$为此时的$x(a_t)$，而不确定度度量$\hat{U}(a_t)$则代入其中的$u$，可以得到概率$p=e^{-2N(a_t)(\hat{U}(a_t))^2}$，当其很小时候，大概率期望报酬的上界就是$\hat{Q}(a_t)+\hat{U}(a_t)$。此时可以考虑取期望报酬上界最大的动作，而非取期望报酬最大的动作。
当设定一个$p$，就可以求得对应的不确定度度量$\hat{U}(a_t)=\sqrt{\frac{\log(p)}{-2N(a_t)}}$，进而得到其期望报酬上界。
更直观地说，UCB 算法在每次选择拉杆前，先估计每根拉杆的期望奖励的上界，使得拉动每根拉杆的期望奖励只有一个较小的概率超过这个上界，接着选出期望奖励上界最大的拉杆，从而选择最有可能获得最大期望奖励的拉杆。（原书语，感觉没直观到哪）
事实上，在真正应用时候，可以用一个系数$c$把不确定性压下来，如：$\hat{Q}’(a_t)=\hat{Q}(a_t)+c\times\hat{U}(a_t)$，事实上这个$c$应该可以表示所谓探索和剥削之间的比率，$c$越大越探索，越小越剥削。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

class UCB(Solver):
    def __init__(self,bandit,coef,prob_init):
        super().__init__(bandit)
        self.prob=0 #p=1/t
        self.total_count=0
        self.estimate_reward=np.full(self.bandit.K,prob_init)
        self.coef=coef

    def run_one_step(self):
        self.total_count+=1
        self.prob=1/self.total_count
        Ucb=self.estimate_reward+self.coef*np.sqrt((np.log(self.prob))/(-2*(self.counts+1)))
        k=np.argmax(Ucb)
        reward = self.bandit.step(k)
        self.estimate_reward[k] += 1 / (1 + self.counts[k]) * (reward - self.estimate_reward[k])
        return k

1.3 汤普森采样方法

一种蒙特卡洛（用频率估计概率）的方法，其主要认为根据对某个动作$a_t$的概率可以假设其符合beta分布（一种根据先验知识假设二项分布概率密度函数的分布（个人见解）其参数分别为$(a,b)$，对于拉杆$k$而言，其成功次数$n_k$和失败次数$m_k$构成的Beta分布满足$Beta(n_k+1,m_k+1)$

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
class TompsonSampling(Solver):
    def __init__(self,bandit):
        super().__init__(bandit)
        self.a=np.ones(bandit.K)
        self.b=np.ones(bandit.K)

    def run_one_step(self):
        samples=np.random.beta(self.a,self.b)
        k=np.argmax(samples)
        reward = self.bandit.step(k)
        if reward==1:
            self.a[k]+=1
        else:
            self.b[k]+=1
        return k

对于老虎机问题，还没有引入状态变量$S_t$，其与环境的交互并不会改变环境，即多臂老虎机的每次交互的结果和以往的动作无关。

2 马尔科夫过程

关于马尔可夫过程中重要的几个概念：

概率矩阵/概率函数表示状态转移的情况：
$P=\begin{bmatrix}
p_{1,1} & … & p_{1,n}\
: & & :\
p_{n,1} & … & p_{n,n}
\end{bmatrix}$
回报：
$G_t=R_t+\gamma R_{t+1}+\gamma^2 R_{t+2}+…=\sum_{i=0}^{N}\gamma^{i} R_{t+i} $
价值函数（state value）：
$V(s)=\mathbb{E} [R_t+\gamma G_{t+1}\mid S_{t}=s]=\mathbb{E} [R_t+\gamma V(S_{t+1})\mid S_{t}=s]$
价值函数（action value）：
$Q^{\pi }(s,a)=r(s,a)+\gamma \sum _{s’\in S}P^{\pi}(s’|s,a)V^{\pi}(s’)$

2.1 一个例子


可以简单地给出一个路线，求得其回报，也可以求其价值函数。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

import numpy as np

np.random.seed(0)
P=[
    [0.9,0.1,0,0,0,0],
    [0.5,0,0.5,0,0,0],
    [0,0,0,0.6,0,0.4],
    [0,0,0,0,0.3,0.7],
    [0,0.2,0.3,0.5,0,0],
    [0,0,0,0,0,0]
]
P=np.array(P)
rewards=[-1,-2,-2,10,1,0]

def cal_return(chain,gammar):
    total_reward=0
    for i in range(len(chain)-1,-1,-1):
        print(i,chain[i])
        total_reward=gammar*total_reward+rewards[chain[i]-1]
    return total_reward

def cal_state_value_simple(gammar):
    I=np.eye(6,6)
    reward_=(np.array(rewards)).reshape((-1,1))
    print(I)
    tmp=np.linalg.inv((I-gammar*P))
    print(tmp)
    value_s=np.dot(tmp,reward_)
    print(np.shape(value_s))
    return value_s


Chain=[1,2,3,6]
#print(cal_return(Chain,0.5))
print(cal_state_value_simple(0.5))

此外，对于这部分的一些基础知识推导，参照《强化学习的数学原理》即可。

3 DP

该部分可能与算法中的DP有所不同，但总之思想应该相似。

3.1 值迭代和策略迭代

网格地图，有终点，禁区和普通区域，需要最短化路径同时尽量不能进入禁区。

此处给出了值迭代和策略迭代两种DP方法的代码，其算法思想和伪代码见笔记。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
import copy


class CliffWalkingEnv:
    """ 悬崖漫步环境"""
    def __init__(self, ncol=12, nrow=4):
        self.ncol = ncol  # 定义网格世界的列
        self.nrow = nrow  # 定义网格世界的行
        # 转移矩阵P[state][action] = [(p, next_state, reward, done)]包含下一个状态和奖励
        self.P = self.createP()

    def createP(self):
        # 初始化
        P = [[[] for j in range(4)] for i in range(self.nrow * self.ncol)]
        # 4种动作, change[0]:上,change[1]:下, change[2]:左, change[3]:右。坐标系原点(0,0)
        # 定义在左上角
        change = [[0, -1], [0, 1], [-1, 0], [1, 0]]
        for i in range(self.nrow):
            for j in range(self.ncol):
                for a in range(4):
                    # 位置在悬崖或者目标状态,因为无法继续交互,任何动作奖励都为0
                    if i == self.nrow - 1 and j > 0:
                        P[i * self.ncol + j][a] = [(1, i * self.ncol + j, 0, True)]
                        continue
                    # 其他位置
                    next_x = min(self.ncol - 1, max(0, j + change[a][0]))
                    next_y = min(self.nrow - 1, max(0, i + change[a][1]))
                    #防止超限罢了
                    next_state = next_y * self.ncol + next_x
                    reward = -1
                    done = False
                    # 下一个位置在悬崖或者终点
                    if next_y == self.nrow - 1 and next_x > 0:
                        done = True
                        if next_x != self.ncol - 1:  # 下一个位置在悬崖
                            reward = -100
                    P[i * self.ncol + j][a] = [(1, next_state, reward, done)]
        return P

class PolicyIteration:
    """ 策略迭代算法 """
    def __init__(self, env, theta, gamma):
        self.env = env
        self.v = [0] * self.env.ncol * self.env.nrow  # 初始化价值为0
        self.pi = [[0.25, 0.25, 0.25, 0.25]
                   for i in range(self.env.ncol * self.env.nrow)]  # 初始化为均匀随机策略
        self.theta = theta  # 策略评估收敛阈值
        self.gamma = gamma  # 折扣因子

    def policy_evaluation(self):  # 策略评估
        cnt = 1  # 计数器
        while 1:
            max_diff = 0
            new_v = [0] * self.env.ncol * self.env.nrow
            for s in range(self.env.ncol * self.env.nrow):
                qsa_list = []  # 开始计算状态s下的所有Q(s,a)价值
                for a in range(4):
                    qsa = 0
                    for res in self.env.P[s][a]:
                        p, next_state, r, done = res
                        qsa += p * (r + self.gamma * self.v[next_state] * (1 - done))
                        # 本章环境比较特殊,奖励和下一个状态有关,所以需要和状态转移概率相乘
                    qsa_list.append(self.pi[s][a] * qsa)
                new_v[s] = sum(qsa_list)  # 状态价值函数和动作价值函数之间的关系
                max_diff = max(max_diff, abs(new_v[s] - self.v[s]))
            self.v = new_v
            if max_diff < self.theta: break  # 满足收敛条件,退出评估迭代
            cnt += 1
        print("策略评估进行%d轮后完成" % cnt)

    def policy_improvement(self):  # 策略提升
        for s in range(self.env.nrow * self.env.ncol):
            qsa_list = []
            for a in range(4):
                qsa = 0
                for res in self.env.P[s][a]:
                    p, next_state, r, done = res
                    qsa += p * (r + self.gamma * self.v[next_state] * (1 - done))
                qsa_list.append(qsa)
            maxq = max(qsa_list)
            cntq = qsa_list.count(maxq)  # 计算有几个动作得到了最大的Q值
            # 让这些动作均分概率
            self.pi[s] = [1 / cntq if q == maxq else 0 for q in qsa_list]
        print("策略提升完成")
        return self.pi

    def policy_iteration(self):  # 策略迭代
        while 1:
            self.policy_evaluation()
            old_pi = copy.deepcopy(self.pi)  # 将列表进行深拷贝,方便接下来进行比较
            new_pi = self.policy_improvement()
            if old_pi == new_pi: break

def print_agent(agent, action_meaning, disaster=[], end=[]):
    print("状态价值：")
    for i in range(agent.env.nrow):
        for j in range(agent.env.ncol):
            # 为了输出美观,保持输出6个字符
            print('%6.6s' % ('%.3f' % agent.v[i * agent.env.ncol + j]), end=' ')
        print()

    print("策略：")
    for i in range(agent.env.nrow):
        for j in range(agent.env.ncol):
            # 一些特殊的状态,例如悬崖漫步中的悬崖
            if (i * agent.env.ncol + j) in disaster:
                print('****', end=' ')
            elif (i * agent.env.ncol + j) in end:  # 目标状态
                print('EEEE', end=' ')
            else:
                a = agent.pi[i * agent.env.ncol + j]
                pi_str = ''
                for k in range(len(action_meaning)):
                    pi_str += action_meaning[k] if a[k] > 0 else 'o'
                print(pi_str, end=' ')
        print()

class ValueIteration:
    def __init__(self,env,theta,gamma):
        self.env=env
        self.v=[0]*self.env.ncol*self.env.nrow
        self.theta=theta
        self.gamma=gamma
        self.pi=[None for i in range(self.env.ncol*self.env.nrow)] #最终policy

    def value_iteration(self):
        cnt=1
        while 1:
            max_diff=0
            new_v=[0]*self.env.ncol*self.env.nrow
            for s in range(self.env.ncol*self.env.nrow):
                qsa_list=[]
                for a in range(4):
                    qsa=0
                    for pair_now in self.env.P[s][a]:
                        p,next_s,r,done=pair_now
                        if done==False:
                            qsa += p * (r + self.gamma * self.v[next_s])
                        else:
                            qsa += p * r
                    qsa_list.append(qsa)
                    print(s,qsa_list)
                new_v[s]=max(qsa_list)
                max_diff=max(max_diff,abs(self.v[s]-new_v[s]))
            self.v=new_v
            if max_diff<=theta:break
            cnt+=1
        print("值评估进行%d轮后完成" % cnt)
        self.get_best_policy()

    def get_best_policy(self):
        for s in range(self.env.ncol * self.env.nrow):
            qsa_list = []
            for a in range(4):
                qsa = 0
                for pair_now in self.env.P[s][a]:
                    p, next_s, r, done = pair_now
                    if done == False:
                        qsa += p * (r + self.gamma * self.v[next_s])
                    else:
                        qsa += p * r
                qsa_list.append(qsa)
            qsa_max= max(qsa_list)
            cnt_max=qsa_list.count(qsa_max)
            self.pi[s] = [1 / cnt_max if q == qsa_max else 0 for q in qsa_list]
            # print(s,qsa_list)

env = CliffWalkingEnv()
action_meaning = ['^', 'v', '<', '>']
theta = 0.001
gamma = 0.9
# agent = PolicyIteration(env, theta, gamma)
# agent.policy_iteration()
agent = ValueIteration(env, theta, gamma)
agent.value_iteration()
print_agent(agent, action_meaning, list(range(37, 47)), [47])

*3.2 冰湖环境

冰湖是 OpenAI Gym 库中的一个环境。OpenAI Gym 库中包含了很多有名的环境，例如 Atari 和 MuJoCo，并且支持定制自己的环境。
冰湖环境和悬崖漫步环境相似，也是一个网格世界，大小为4*4。每一个方格是一个状态$S$，智能体起点状态在左上角，目标状态$G$在右下角，中间还有若干冰洞$H$。在每一个状态都可以采取上、下、左、右 4 个动作。由于智能体在冰面行走，因此每次行走都有一定的概率滑行到附近的其它状态，并且到达冰洞或目标状态时行走会提前结束。每一步行走的奖励是 0，到达目标的奖励是 1。
此时

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import gym
import copy
env = gym.make("FrozenLake-v1",render_mode="human")  # 创建环境
env = env.unwrapped  # 解封装才能访问状态转移矩阵P

class PolicyIteration:
...

class ValueIteration:
...

def print_agent(agent, action_meaning, disaster=[], end=[]):
...

# 这个动作意义是Gym库针对冰湖环境事先规定好的
action_meaning = ['<', 'v', '>', '^']
theta = 1e-5
gamma = 0.9
agent = ValueIteration(env, theta, gamma)
agent.value_iteration()
# agent = PolicyIteration(env, theta, gamma)
# agent.policy_iteration()
print_agent(agent, action_meaning, [5, 7, 11, 12], [15])

4 TD（时序差分）

TD的目的是求解一个无模型的贝尔曼公式。

4.1 Sarsa

Sarsa算法，用于估计action value，一般用epsilon-greedy来搜索。伪代码和算法推导见笔记。

基于悬崖漫步的Sarsa代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
import numpy as np
import matplotlib.pyplot as plt
from tqdm import tqdm
class CliffWalking:
    def __init__(self, col, row):
        self.col = col
        self.row = row
        self.x = 0
        self.y = row - 1

    def get_next_step(self, action):
        actions = [[0, -1], [0, 1], [-1, 0], [1, 0]]
        self.x = min(max(self.x + actions[action][0], 0), self.col - 1)
        self.y = min(max(self.y + actions[action][1], 0), self.row - 1)
        next_state = self.y * self.col + self.x
        reward = -1
        done = False
        if (self.y == self.row - 1) and self.x > 0:
            done = True
            if self.x != self.col - 1:
                reward = -100
        return next_state, reward, done

    def reset(self):
        self.x = 0
        self.y = self.row - 1
        next_state = self.y * self.col + self.x
        return next_state


class Sarsa:
    def __init__(self, env, gamma, alpha, epsilon, n_action):
        self.env = env
        self.gamma = gamma
        self.alpha = alpha
        self.epsilon = epsilon
        self.n_action = n_action
        self.Q_table = np.zeros([self.env.col * self.env.row, n_action])

    def take_action(self, state):
        if (np.random.random() < self.epsilon):
            action = np.random.randint(self.n_action)
        else:
            action = np.argmax(self.Q_table[state])
        return action

    def update_q_value(self, at, st, r, at1, st1):
        TD_error =  (r + self.gamma * self.Q_table[st1,at1])-self.Q_table[st,at]
        self.Q_table[st,at] += self.alpha * TD_error

    def get_best_policy(self, state):
        Q_max = np.max(self.Q_table[state])
        a = [0 for i in range(self.n_action)]
        for i in range(self.n_action):
            if self.Q_table[state][i]==Q_max:
                a[i]=1
        return a

ncol = 12
nrow = 4
env = CliffWalking(ncol, nrow)
np.random.seed(0)
epsilon = 0.1
alpha = 0.1
gamma = 0.9
agent = Sarsa(env, gamma, alpha,epsilon,4)
num_episodes = 1000  # 智能体在环境中运行的序列的数量

return_list = []  # 记录每一条序列的回报
for i in range(10):  # 显示10个进度条
    # tqdm的进度条功能
    with tqdm(total=int(num_episodes / 10), desc='Iteration %d' % i) as pbar:
        for i_episode in range(int(num_episodes / 10)):  # 每个进度条的序列数
            episode_return = 0
            state = env.reset()
            action = agent.take_action(state)
            done = False
            while not done:
                next_state, reward, done = env.get_next_step(action)
                next_action = agent.take_action(next_state)
                episode_return += reward  # 这里回报的计算不进行折扣因子衰减
                agent.update_q_value(action, state, reward, next_action, next_state)
                state = next_state
                action = next_action
            return_list.append(episode_return)
            if (i_episode + 1) % 10 == 0:  # 每10条序列打印一下这10条序列的平均回报
                pbar.set_postfix({
                    'episode':
                    '%d' % (num_episodes / 10 * i + i_episode + 1),
                    'return':
                    '%.3f' % np.mean(return_list[-10:])
                })
            pbar.update(1)

episodes_list = list(range(len(return_list)))
plt.plot(episodes_list, return_list)
plt.xlabel('Episodes')
plt.ylabel('Returns')
plt.title('Sarsa on {}'.format('Cliff Walking'))
plt.show()

def print_agent(agent, env, action_meaning, disaster=[], end=[]):
    for i in range(env.row):
        for j in range(env.col):
            if (i * env.col + j) in disaster:
                print('****', end=' ')
            elif (i * env.col + j) in end:
                print('EEEE', end=' ')
            else:
                a = agent.get_best_policy(i * env.col + j)
                pi_str = ''
                for k in range(len(action_meaning)):
                    pi_str += action_meaning[k] if a[k] > 0 else 'o'
                print(pi_str, end=' ')
        print()


action_meaning = ['^', 'v', '<', '>']
print('Sarsa算法最终收敛得到的策略为：')
print_agent(agent, env, action_meaning, list(range(37, 47)), [47])

4.2 n-step Sarsa

Sarsa算法主要会在每一步都对Q表进行更新，但是这样的更新相比于在一串步骤之后再对Q表进行更新的TD算法而言，可能是存在一定偏差的，但TD算法一次更新所需要的episode太长，也不合适。
针对这些问题，可以选择使用n-step Sarsa方法，用episode中的n步进行更新。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
class NStepSarsa:
    def __init__(self, n, env, gamma, alpha, epsilon, n_action=4):
        self.n = n
        self.env = env
        self.gamma = gamma
        self.alpha = alpha
        self.epsilon = epsilon
        self.n_action = n_action
        self.Q_table = np.zeros([self.env.col * self.env.row, n_action])
        self.states = []
        self.actions = []
        self.rewards = []

    def take_action(self, state):
        if (np.random.random() < self.epsilon):
            action = np.random.randint(self.n_action)
        else:
            action = np.argmax(self.Q_table[state])
        return action

    def get_best_policy(self, state):
        Q_max = np.max(self.Q_table[state])
        a = [0 for i in range(self.n_action)]
        for i in range(self.n_action):
            if self.Q_table[state][i] == Q_max:
                a[i] = 1
        return a

    def update_q_value(self, at, st, r, atn, stn, done):
        self.states.append(st)
        self.actions.append(at)
        self.rewards.append(r)
        if self.n==len(self.states):
            G_n=self.Q_table[stn,atn]
            for i in reversed(range(self.n)):
                G_n=self.rewards[i]+self.gamma*G_n
                if done and i>0:
                    s=self.states[i]
                    a=self.actions[i]
                    self.Q_table[s,a]+=self.alpha*(G_n-self.Q_table[s,a])
            s=self.states.pop(0)
            a=self.actions.pop(0)
            self.rewards.pop(0)
            self.Q_table[s,a]+=self.alpha*(G_n-self.Q_table[s,a])
        if done:
            self.states=[]
            self.actions=[]
            self.rewards=[]

 for i_episode in range(int(num_episodes / 10)):  # 每个进度条的序列数
    episode_return = 0
    state = env.reset()
    action = agent.take_action(state)
    done = False
    while not done:
        ext_state, reward, done = env.get_next_step(action)
        next_action = agent.take_action(next_state)
        episode_return += reward  # 这里回报的计算不进行折扣因子衰减
        agent.update_q_value(action, state, reward, next_action, next_state,done)
        tate = next_state
        action = next_action
    return_list.append(episode_return)

4.3 Q-learning

Q-learning主要是在Sarsa的基础上，在更新Q值的时候，采用了最优的action（即使用了最大的action value），而不是给出一个根据策略的action，并在此基础上进行筛选（相当于在Q-learning的过程中，TD target在选择a的时候进行了一定的优化。）
Q-learing中action的选择没有任何策略，但Sarsa则不然。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
import numpy as np
import matplotlib.pyplot as plt
from tqdm import tqdm
class CliffWalking:
    def __init__(self, col, row):
        self.col = col
        self.row = row
        self.x = 0
        self.y = row - 1

    def get_next_step(self, action):
        actions = [[0, -1], [0, 1], [-1, 0], [1, 0]]
        self.x = min(max(self.x + actions[action][0], 0), self.col - 1)
        self.y = min(max(self.y + actions[action][1], 0), self.row - 1)
        next_state = self.y * self.col + self.x
        reward = -1
        done = False
        if (self.y == self.row - 1) and self.x > 0:
            done = True
            if self.x != self.col - 1:
                reward = -100
        return next_state, reward, done

    def reset(self):
        self.x = 0
        self.y = self.row - 1
        next_state = self.y * self.col + self.x
        return next_state


class QLearning:
    def __init__(self, n, env, gamma, alpha, epsilon, n_action=4):
        self.n = n
        self.env = env
        self.gamma = gamma
        self.alpha = alpha
        self.epsilon = epsilon
        self.n_action = n_action
        self.Q_table = np.zeros([self.env.col * self.env.row, n_action])

    def take_action(self, state):
        if (np.random.random() < self.epsilon):
            action = np.random.randint(self.n_action)
        else:
            action = np.argmax(self.Q_table[state])
        return action

    def get_best_policy(self, state):
        Q_max = np.max(self.Q_table[state])
        a = [0 for i in range(self.n_action)]
        for i in range(self.n_action):
            if self.Q_table[state][i] == Q_max:
                a[i] = 1
        return a

    def update_q_value(self, at, st, r, st1):
        TD_error=r+self.gamma*self.Q_table[st1].max()-self.Q_table[st,at]
        self.Q_table[st,at]+=self.alpha*TD_error

ncol=12
nrow=4
env=CliffWalking(ncol,nrow)
np.random.seed(0)
n_step = 5  # 5步Sarsa算法
alpha = 0.1
epsilon = 0.1
gamma = 0.9
agent = QLearning(n_step, env,gamma,alpha,epsilon)
num_episodes = 500  # 智能体在环境中运行的序列的数量

return_list = []  # 记录每一条序列的回报
for i in range(10):  # 显示10个进度条
    #tqdm的进度条功能
    with tqdm(total=int(num_episodes / 10), desc='Iteration %d' % i) as pbar:
        for i_episode in range(int(num_episodes / 10)):  # 每个进度条的序列数
            episode_return = 0
            state = env.reset()
            done = False
            while not done:
                action = agent.take_action(state)
                next_state, reward, done = env.get_next_step(action)
                episode_return += reward  # 这里回报的计算不进行折扣因子衰减
                agent.update_q_value(action, state, reward,next_state)
                state = next_state
            return_list.append(episode_return)
            if (i_episode + 1) % 10 == 0:  # 每10条序列打印一下这10条序列的平均回报
                pbar.set_postfix({
                    'episode':
                    '%d' % (num_episodes / 10 * i + i_episode + 1),
                    'return':
                    '%.3f' % np.mean(return_list[-10:])
                })
            pbar.update(1)

episodes_list = list(range(len(return_list)))
plt.plot(episodes_list, return_list)
plt.xlabel('Episodes')
plt.ylabel('Returns')
plt.title('Q-learning on {}'.format('Cliff Walking'))
plt.show()

def print_agent(agent, env, action_meaning, disaster=[], end=[]):
    for i in range(env.row):
        for j in range(env.col):
            if (i * env.col + j) in disaster:
                print('****', end=' ')
            elif (i * env.col + j) in end:
                print('EEEE', end=' ')
            else:
                a = agent.get_best_policy(i * env.col + j)
                pi_str = ''
                for k in range(len(action_meaning)):
                    pi_str += action_meaning[k] if a[k] > 0 else 'o'
                print(pi_str, end=' ')
        print()

action_meaning = ['^', 'v', '<', '>']
print('Q-learning算法最终收敛得到的策略为：')
print_agent(agent, env, action_meaning, list(range(37, 47)), [47])

* 关于一些区分

*.1 on-policy和off-policy

与环境交互的policy被称作behavior policy，用于生成episode，而达成目标的policy被称为target policy；二者相同则为on-policy，不同则是off-policy
其中，Sarsa是on-policy，Q-learning是off-policy

5 Dyna-Q

这是一种有模型的强化学习方法，通过真实数据去构建模型，并用一种叫做Q-Planning的方法进行强化学习。
这里使用了字典作为模型，事实上可以使用状态方程之类的模型。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
import numpy as np
import matplotlib.pyplot as plt
from tqdm import tqdm
import random
import time
class CliffWalking:
    def __init__(self, col, row):
        self.col = col
        self.row = row
        self.x = 0
        self.y = row - 1

    def get_next_step(self, action):
        actions = [[0, -1], [0, 1], [-1, 0], [1, 0]]
        self.x = min(max(self.x + actions[action][0], 0), self.col - 1)
        self.y = min(max(self.y + actions[action][1], 0), self.row - 1)
        next_state = self.y * self.col + self.x
        reward = -1
        done = False
        if (self.y == self.row - 1) and self.x > 0:
            done = True
            if self.x != self.col - 1:
                reward = -100
        return next_state, reward, done

    def reset(self):
        self.x = 0
        self.y = self.row - 1
        next_state = self.y * self.col + self.x
        return next_state


class DynaQ:
    def __init__(self, env, gamma, alpha, epsilon, n_planning,n_action=4):
        self.env = env
        self.gamma = gamma
        self.alpha = alpha
        self.epsilon = epsilon
        self.n_action = n_action
        self.Q_table = np.zeros([self.env.col * self.env.row, n_action])

        self.n_planning=n_planning
        self.model=dict()

    def take_action(self, state):
        if (np.random.random() < self.epsilon):
            action = np.random.randint(self.n_action)
        else:
            action = np.argmax(self.Q_table[state])
        return action

    def get_best_policy(self, state):
        Q_max = np.max(self.Q_table[state])
        a = [0 for i in range(self.n_action)]
        for i in range(self.n_action):
            if self.Q_table[state][i] == Q_max:
                a[i] = 1
        return a

    def update_q_value(self, at, st, r, st1):
        TD_error=r+self.gamma*self.Q_table[st1].max()-self.Q_table[st,at]
        self.Q_table[st,at]+=self.alpha*TD_error

    def update(self,at,st,r,st1):
        self.update_q_value(at,st,r,st1)
        self.model[(st,at)]=r,st1
        for i in range(self.n_planning):
            (sm,am),(rm,s_m)=random.choice(list(self.model.items()))
            self.update_q_value(am,sm,rm,s_m)

def DynaQ_CliffWalking(n_planning):
    ncol = 12
    nrow = 4
    env = CliffWalking(ncol, nrow)
    epsilon = 0.01
    alpha = 0.1
    gamma = 0.9
    agent = DynaQ(env, gamma, alpha, epsilon, n_planning)
    num_episodes = 300  # 智能体在环境中运行多少条序列

    return_list = []  # 记录每一条序列的回报
    for i in range(10):  # 显示10个进度条
        # tqdm的进度条功能
        with tqdm(total=int(num_episodes / 10),
                  desc='Iteration %d' % i) as pbar:
            for i_episode in range(int(num_episodes / 10)):  # 每个进度条的序列数
                episode_return = 0
                state = env.reset()
                done = False
                while not done:
                    action = agent.take_action(state)
                    next_state, reward, done = env.get_next_step(action)
                    episode_return += reward  # 这里回报的计算不进行折扣因子衰减
                    agent.update(action, state, reward, next_state)
                    state = next_state
                return_list.append(episode_return)
                if (i_episode + 1) % 10 == 0:  # 每10条序列打印一下这10条序列的平均回报
                    pbar.set_postfix({
                        'episode':
                        '%d' % (num_episodes / 10 * i + i_episode + 1),
                        'return':
                        '%.3f' % np.mean(return_list[-10:])
                    })
                pbar.update(1)
    return return_list

np.random.seed(0)
random.seed(0)
n_planning_list = [0, 2, 20]
for n_planning in n_planning_list:
    print('Q-planning步数为：%d' % n_planning)
    time.sleep(0.5)
    return_list = DynaQ_CliffWalking(n_planning)
    episodes_list = list(range(len(return_list)))
    plt.plot(episodes_list,
             return_list,
             label=str(n_planning) + ' planning steps')
plt.legend()
plt.xlabel('Episodes')
plt.ylabel('Returns')
plt.title('Dyna-Q on {}'.format('Cliff Walking'))
plt.show()
  • Copyrights © 2024 suixinhita

请我喝杯咖啡吧~

支付宝
微信