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

AI 学习强化学习马尔可夫决策过程贝尔曼方程蒙特卡洛采样

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

2024-08-152024-08-15

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

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

不考虑动作策略，一个马尔可夫奖励过程由一个状态空间S、一个转移概率矩阵P、一个奖励函数R、一个折扣因子γ和一个初始状态s0组成。

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

价值函数

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

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

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

进一步得到贝尔曼方程：

矩阵求解析解：

def compute(P, rewards, gamma, states_num):
    ''' 利用贝尔曼方程的矩阵形式计算解析解,states_num是MRP的状态数 '''
    rewards = np.array(rewards).reshape((-1, 1))  
    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’的概率；

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

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

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

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

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

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

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

FROM MDP TO MRP

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

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

根据求得状态价值函数：

def compute(P, rewards, gamma, states_num):
    ''' 利用贝尔曼方程的矩阵形式计算解析解,states_num是MRP的状态数 '''
    rewards = np.array(rewards).reshape((-1, 1))  
    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过程

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.根据动作策略采样

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)]  
        while s != "s5" and timestep <= timestep_max:
            timestep += 1
            
            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
            
            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 = s_next  
        episodes.append(episode)
    return episodes

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

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):  
            (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.总流程

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)被访问到的概率：

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

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

def occupancy(episodes, s, a, timestep_max, gamma):
    ''' 计算状态动作对（s,a）出现的频率,以此来估算策略的占用度量 '''
    rho = 0
    total_times = np.zeros(timestep_max)  
    occur_times = np.zeros(timestep_max)  
    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)

最优策略

贝尔曼最优方程：

推荐订阅源

FrWalker Blog