强化学习-2-马尔可夫决策过程

强化学习的笔记、理解、感悟及代码实现,仅按个人思维进行精华总结和记录，使用的教程：动手学强化学习

马尔可夫奖励过程(Markov Reward Process, MRP)

不考虑动作策略，一个马尔可夫奖励过程由一个状态空间S、一个转移概率矩阵P、一个奖励函数R、一个折扣因子γ和一个初始状态s0组成。
  • 状态空间S表示马尔可夫过程的状态集合。
  • 转移概率矩阵P是一个S*S的矩阵，其中P(s,s’)表示在状态s下执行动作a后转移到状态s’的概率。
  • 奖励函数R(s)表示转移到状态s可以获得的奖励
  • 折扣因子γ取值在0~1之间，用于衡量短期奖励与长期奖励的折扣系数
alt text

价值函数

从一个状态出发，其期望回报称为价值，由下式计算：

其中，表示在t时刻的奖励。

所有状态的价值组成价值函数，即

进一步得到贝尔曼方程：

矩阵求解析解：

1
2
3
4
5
6
7
8
def compute(P, rewards, gamma, states_num):
    ''' 利用贝尔曼方程的矩阵形式计算解析解,states_num是MRP的状态数 '''
    rewards = np.array(rewards).reshape((-1, 1))  #将rewards写成列向量形式
    value = np.dot(np.linalg.inv(np.eye(states_num, states_num) - gamma * P),
                   rewards)
    return value
V = compute(P, rewards, gamma, 6)
print("MRP中每个状态价值分别为\n", V)

马尔可夫决策过程(Markov Decision Process, MDP)

马尔可夫决策过程（Markov Decision Process，MDP）是指一个由状态、动作、转移概率和奖励组成的随机过程，其中状态的转移仅依赖于当前状态和动作，而与历史无关。
相比MRP，引入了动作决策，奖励函数和状态转移同时取决于状态s和动作a。
  • r(s,a)是奖励函数，此时奖励可以同时取决于状态s和动作a，在奖励函数只取决于状态s时，则退化为r(a)；
  • p(s’|s,a)是状态转移概率，表示在状态s下执行动作a后转移到状态s’的概率；
alt text

状态、动作价值函数及两者关系

定义基于策略的状态价值函数:

定义基于策略的动作价值函数：

得到两个价值函数的贝尔曼期望方程：

状态、动作价值函数的关系

在使用策略中，状态的价值等于在该状态下基于策略采取所有动作的概率与相应的价值相乘再求和的结果：

使用策略时，状态下采取动作的价值等于即时奖励加上经过衰减后的所有可能的下一个状态的状态转移概率与相应的价值的乘积：

FROM MDP TO MRP

根据策略所有动作的概率进行加权，得到无动作的奖励函数(奖励函数的定义由到达状态s的奖励改为到达状态s能够产失的奖励期望即s后所有动作产生奖励的期望)：

将采取动作的概率与使s转移到s’的概率相乘再求和，得到无动作的状态转移概率：

根据求得状态价值函数：

1
2
3
4
5
6
7
8
def compute(P, rewards, gamma, states_num):
    ''' 利用贝尔曼方程的矩阵形式计算解析解,states_num是MRP的状态数 '''
    rewards = np.array(rewards).reshape((-1, 1))  #将rewards写成列向量形式
    value = np.dot(np.linalg.inv(np.eye(states_num, states_num) - gamma * P),
                   rewards)
    return value
V = compute(P, rewards, gamma, 6)
print("MRP中每个状态价值分别为\n", V)

根据状态价值函数和动作价值函数的关系，求得动作价值函数：

使用蒙特卡洛采用计算状态价值函数

随机生成多条状态序列,从后向前统计每个状态出现的次数和回报,计算期望来近似价值函数V:

1.定义MDP过程

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
S = ["s1", "s2", "s3", "s4", "s5"]  # 状态集合
A = ["保持s1", "前往s1", "前往s2", "前往s3", "前往s4", "往s5", "概率前往"]  # 动作集合
P = {
    "s1-保持s1-s1": 1.0,
    "s1-前往s2-s2": 1.0,
    "s2-前往s1-s1": 1.0,
    "s2-前往s3-s3": 1.0,
    "s3-前往s4-s4": 1.0,
    "s3-前往s5-s5": 1.0,
    "s4-前往s5-s5": 1.0,
    "s4-概率前往-s2": 0.2,
    "s4-概率前往-s3": 0.4,
    "s4-概率前往-s4": 0.4,
}# 状态转移函数
R = {
    "s1-保持s1": -1,
    "s1-前往s2": 0,
    "s2-前往s1": -1,
    "s2-前往s3": -2,
    "s3-前往s4": -2,
    "s3-前往s5": 0,
    "s4-前往s5": 10,
    "s4-概率前往": 1,
}# 奖励函数
gamma = 0.5  # 折扣因子
MDP = (S, A, P, R, gamma)

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
34
35
36
37
38
39
40
41
42
43
Pi = {
        "s1-保持s1": 0.5,
        "s1-前往s2": 0.5,
        "s2-前往s1": 0.5,
        "s2-前往s3": 0.5,
        "s3-前往s4": 0.5,
        "s3-前往s5": 0.5,
        "s4-前往s5": 0.5,
        "s4-概率前往": 0.5,
    }  # 策略

def join(str1, str2):
    """把输入的两个字符串通过“-”连接,便于使用定义的P、R变量"""
    return str1 + '-' + str2
def sample(MDP, Pi, timestep_max, number):
    ''' 在MDP中随机采样状态序列,状态序列由多个(s,a,r,s_next)组成,策略Pi,限制最长时间步timestep_max,总共采样序列数number '''
    S, A, P, R, gamma = MDP
    episodes = []#保存所有采样序列
    for _ in range(number):
        episode = []#一条采样序列
        timestep = 0
        s = S[np.random.randint(4)]  # 随机选择一个除s5以外的状态s作为起点
        while s != "s5" and timestep <= timestep_max:# 当前状态为终止状态或者时间步太长时,一次采样结束
            timestep += 1
            # 在状态s下根据策略选择动作
            rand, temp = np.random.rand(), 0
            for a_opt in A:
                temp += Pi.get(join(s, a_opt), 0)
                if temp > rand:
                    a = a_opt
                    r = R.get(join(s, a), 0)
                    break
            # 根据状态转移概率得到下一个状态s_next
            rand, temp = np.random.rand(), 0
            for s_opt in S:
                temp += P.get(join(join(s, a), s_opt), 0)
                if temp > rand:
                    s_next = s_opt
                    break
            episode.append((s, a, r, s_next))  # 把（s,a,r,s_next）元组放入序列中
            s = s_next  # s_next变成当前状态,开始接下来的循环
        episodes.append(episode)
    return episodes

3.从后向前统计每个状态的出现次数和总回报

1
2
3
4
5
6
7
8
9
def MC(episodes, V, N, gamma):
    """对所有采样序列episodes计算所有状态的价值V,采样序列中状态出现的次数N,折扣因子gamma平衡短期和长期利益"""
    for episode in episodes:
        G = 0
        for i in range(len(episode) - 1, -1, -1):  #一个序列从后往前计算，便于计算当前状态至未来的累积奖励即回报G
            (s, a, r, s_next) = episode[i]
            G = r + gamma * G
            N[s] = N[s] + 1
            V[s] = V[s] + (G - V[s]) / N[s]#增量更新

4.总流程

1
2
3
4
5
episodes = sample(MDP, Pi, 20, 1000)
V = {"s1": 0, "s2": 0, "s3": 0, "s4": 0, "s5": 0}
N = {"s1": 0, "s2": 0, "s3": 0, "s4": 0, "s5": 0}
MC(episodes, V, N, gamma)
print("使用蒙特卡洛方法估计MDP的状态价值为\n", V)

占用度量与最优策略

占用度量

占用度量衡量的是一个策略的好坏，即在给定策略下，动作状态对(s,a)被访问到的概率：

占用度量相同等价于策略相同：

通过代码采样，用状态对的频率来估计占用度量：

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
def occupancy(episodes, s, a, timestep_max, gamma):
    ''' 计算状态动作对（s,a）出现的频率,以此来估算策略的占用度量 '''
    rho = 0
    total_times = np.zeros(timestep_max)  # 记录每个时间步t各被经历过几次
    occur_times = np.zeros(timestep_max)  # 记录(s_t,a_t)=(s,a)的次数
    for episode in episodes:
        for i in range(len(episode)):
            (s_opt, a_opt, r, s_next) = episode[i]
            total_times[i] += 1
            if s == s_opt and a == a_opt:
                occur_times[i] += 1
    for i in reversed(range(timestep_max)):
        if total_times[i]:
            rho += gamma**i * occur_times[i] / total_times[i]
    return (1 - gamma) * rho


gamma = 0.5
timestep_max = 1000

episodes_1 = sample(MDP, Pi_1, timestep_max, 1000)
episodes_2 = sample(MDP, Pi_2, timestep_max, 1000)
rho_1 = occupancy(episodes_1, "s4", "概率前往", timestep_max, gamma)
rho_2 = occupancy(episodes_2, "s4", "概率前往", timestep_max, gamma)
print(rho_1, rho_2)

最优策略

贝尔曼最优方程：