AI模型强化学习

强化学习环境框架-从一个迷宫环境看MDP的要点

字数统计: 3.4k字   |   阅读时长: 13分

摘要: 强化学习环境的框架，MDP 的要点

【对算法，数学，计算机感兴趣的同学，欢迎关注我哈，阅读更多原创文章】
我的网站：潮汐朝夕的生活实验室
我的公众号：算法题刷刷
我的知乎：潮汐朝夕
我的github：FennelDumplings
我的leetcode：FennelDumplings

在此前我们已经用 OpenAI Gym 进行过一些强化学习的实践。例如在文章中 OpenAI-Gym入门 中，我们初步学习了 gym，并跑了一个示例程序；在文章 OpenAI-Gym神经网络策略及其训练(策略梯度) 中，我们初步学习了策略梯度，并用 gym 中的一个环境实现了策略梯度算法；在文章 OpenAI-Gym-render画基本形状 中，我们初步了解了如何在 gym 框架下使用 render 画基本形状；在文章 OpenAI-Gym-自定义环境的要点 中，我们初步了解了如何在 gym 框架下根据业务自己定义环境。

在文章 马尔科夫决策过程 中，我们初步学习了 MDP 的机制。包括无记忆随机过程和马尔科夫决策过程的概念，贝尔曼方程，Q 值，以及 Q 值迭代算法。

本文我们将强化学习的基础概念，以及 MDP 的机制回顾一下，并实现一个符合 MDP 的环境，在此基础上实践一下 MDP 的机制。参考资料是《用Python动手学强化学习》。

强化学习及相关概念回顾

如图，机器学习是实现人工智能的一种技术，深度学习是机器学习中的一种模型，模型需要通过某些学习方法进行学习，调整模型参数。学习方法有三种

  • 监督学习：事先给定数据和标签，然后对模型的参数进行调整，使得输出与给定的数据一致
  • 无监督学习：事先给定数据，然后对模型的参数进行调整，以提取数据的特征(结构或表征)
  • 强化学习：事先给定一个可以根据行动得到奖励的环境(任务)，然后对模型的参数进行调整，以便让不同状态下的行动与奖励联系起来。

对于无监督学习，学习结构的一个典型例子是聚类，例如生物学中的分类框架，是根据观察，积累经验总结出来的。学习表征的一个典型例子是数据压缩，自编码器是一种获取压缩表征的方法。

强化学习给定的是环境，而不是数据。环境可以理解为到达某种状态即可获取奖励的空间，其中定义了行动，以及与行动对应的状态的变化。

强化学习需要对模型参数进行调整，以便从环境中获得奖励。这里的模型是一个接收状态并输出行动的函数。

强化学习在根据行动给予奖励这一点上与监督学习相似，但不同点在于，强化学习不是根据单次的立即奖励进行优化的，其优化的目的是使得一个回合的整体奖励最大化。这里的回合的意思是指从环境开始到结束的整个期间。

强化学习根据能否让整体奖励最大化来评价行动的，具体如何评价，需要模型自己去学习。也就是说，强化学习模型(接收状态并输出行动的函数，也称为策略)需要学习两项内容：

(1) 行动的评价方法
(2) 基于评价方法对行动进行选择的方法(策略)

能对行动的评价方法进行学习是强化学习的优点。但是这也意味着行动的评价方法完全交给了模型，因为我们没有提供标签这样的正确答案。这个弱点与无监督学习一样。

强化学习的问题设定 — 马尔科夫决策过程(MDP)

(1) 马尔可夫性

强化学习的环境(到达某种状态即可获取奖励的空间)是要遵循一定规则的，这个规则就是马尔可夫性，如下

  • 迁移后的状态由迁移前的状态和行动决定。
  • 奖励由迁移前的状态和迁移后的状态决定。

具有马尔可夫性的环境称为马尔可夫决策过程。

(2) MDP 的 4 元素

MDP 包含以下四个元素

  • s: 状态
  • a: 行动
  • T: 迁移函数 (以状态和行动为参数，输出迁移后的状态和迁移概率的函数)
  • R: 奖励函数 (以状态和行动为参数，输出奖励的函数)

如图，机器人是输入状态，输出行动的函数，称为策略，记为 $\pi$，强化学习的过程就是调整策略的参数，使得模型能根据状态输出合适的行动，根据策略执行行动的主体(本图中为机器人)，称为智能体。

(3) 立即奖励与整体奖励

MDP 中的奖励 r 由迁移前后迁移后的状态决定，称为立即奖励。立即奖励与监督学习中的标签类似，但是只关注立即奖励，就无法使得回合内整体奖励最大化。

MDP 的整体奖励 $G_{t}$ 定义如下

(4) 折扣率，价值

但由于在 t 时刻的时候，是不知道后续时刻 t+1, t+2, …, T 的立即奖励的，因此在一个回合结束之前，是无法计算 $G_{t}$ 的。

对智能体来说，在 t 时刻要选择合适行动的时候，是需要知道整体奖励的，这就需要对整体奖励进行估计。这个估计值是不确定的，因此需要

对其打折，打折系数是折扣率，$\gamma$，使用折扣率定义的整体奖励如下

下图展示了使用折扣率定义的整体奖励的计算方法

上面的计算方式可以用递归的方式进行

使用折扣率定义的整体奖励 $G_{t}$ 是整体奖励的估计值，称为期望奖励(有的书上也叫动作回报)，也称为价值。

计算这个价值的过程称为价值近似，这个价值近似是强化学习要学习的两项内容之一，即行动的评价方法。

实现 MDP 的框架

前面我们回顾了强化学习的基本概念，DMP 的机制，最大化的对象(价值)。这里我们通过一个迷宫环境的实现，形成 MDP 环境的代码框架。以后在实现基于其它业务场景的环境的时候可以套用。

我们实现的环境是下图这样的迷宫环境。

该环境的四要素如下

  • 行动 a: 上下左右移动
  • 状态 s: 当前所在位置
  • 迁移函数: 接收状态和行动，返回智能体可以移动到的各自的位置以及智能体向其移动的概率(迁移概率)
  • 奖励函数: 接收状态，如果是绿色格子，返回 1，如果是红色格子，返回 -1

注意这里人为设定了由迁移概率决定的行动以外的行动的发生概率。比如在向上前进的情况下，虽然无法向下前进，但是可以左右移动。

注意这里的奖励函数只取决于迁移后的状态，因此参数只有状态。根据实际业务场景奖励函数的参数也可能需要加上行动。

(1) 状态 State, 行动 Action

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
class State():
    def __init__(self, row=-1, column=-1):
        self.row = row
        self.column = column

    def __repr__(self):
        return "<State: [{}, {}]>".format(self.row, self.column)

    def clone(self):
        return State(self.row, self.column)

    def __hash__(self):
        return hash((self.row, self.column))

    def __eq__(self, other):
        return self.row == other.row and self.column == other.column

状态 State 定义了格子的位置 row, column

1
2
3
4
5
6
7
from enum import Enum

class Action(Enum):
    UP = 1
    DOWN = -1
    LEFT = 2
    RIGHT = -2

行动 Action 定义了上下左右这四种行动。

(2) 环境实体 Environment

四要素相关的属性和方法

Environment 接收迷宫的定义(grid 属性)，把迷宫内的格子当做状态(states property 方法)。环境内可以执行的行动只有上下左右四中(actions property 方法)。

transit_func 方法是迁移函数，move_prob 属性是以所选方向移动的概率，can_action_at 方法判定某状态是否可以达到，_move 方法负责具体移动。

reward_func 方法是奖励函数，default_rewatd 属性是默认奖励，也就是每多一步就会施加的一个奖励。

便于外部调用的属性和方法

reset，step，transit 方法是几个便于外部使用该环境的函数。其中 step 方法接收智能体行动，通过迁移函数和奖励函数返回迁移后的状态和立即奖励(在 transit 方法中调用 transit_func 和 reward_func 方法进行，step 调用 transit 方法即可)。reset 方法初始化智能体位置，返回状态 State。

下面是该迷宫的 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
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
import numpy as np

class Environment():
    def __init__(self, grid, move_prob=0.8):
        # grid是一个二维数组，它的值可以看作属性
        # 一些属性的情况如下
        #  0: 普通格子
        #  -1: 有危险的格子 (游戏结束)
        #  1: 有奖励的格子 (游戏结束)
        #  9: 被屏蔽的格子 (无法放置智能体)
        self.grid = grid
        self.agent_state = State()

        # 默认的奖励是负数，就像施加了初始位置惩罚
        # 这意味着智能体必须快速到达终点
        self.default_reward = -0.04

        # 智能体能够以 move_prob 的概率向所选方向移动
        # 如果概率值在(1 - move_prob)内
        # 则意味着智能体将移动到不同的方向
        self.move_prob = move_prob
        self.reset()

    @property
    def row_length(self):
        return len(self.grid)

    @property
    def column_length(self):
        return len(self.grid[0])

    @property
    def actions(self):
        return [Action.UP, Action.DOWN,
                Action.LEFT, Action.RIGHT]

    @property
    def states(self):
        states = []
        for row in range(self.row_length):
            for column in range(self.column_length):
                # state中不包含被屏蔽的格子
                if self.grid[row][column] != 9:
                    states.append(State(row, column))
        return states

    def transit_func(self, state, action):
        transition_probs = {}
        if not self.can_action_at(state):
            # 已经到达游戏结束的格子
            return transition_probs

        opposite_direction = Action(action.value * -1)

        for a in self.actions:
            prob = 0
            if a == action:
                prob = self.move_prob
            elif a != opposite_direction:
                prob = (1 - self.move_prob) / 2

            next_state = self._move(state, a)
            if next_state not in transition_probs:
                transition_probs[next_state] = prob
            else:
                transition_probs[next_state] += prob

        return transition_probs

    def can_action_at(self, state):
        if self.grid[state.row][state.column] == 0:
            return True
        else:
            return False

    def _move(self, state, action):
        if not self.can_action_at(state):
            raise Exception("Can't move from here!")

        next_state = state.clone()

        # 执行行动（移动）
        if action == Action.UP:
            next_state.row -= 1
        elif action == Action.DOWN:
            next_state.row += 1
        elif action == Action.LEFT:
            next_state.column -= 1
        elif action == Action.RIGHT:
            next_state.column += 1

        # 检查状态是否在grid外
        if not (0 <= next_state.row < self.row_length):
            next_state = state
        if not (0 <= next_state.column < self.column_length):
            next_state = state

        # 检查智能体是否到达了被屏蔽的格子
        if self.grid[next_state.row][next_state.column] == 9:
            next_state = state

        return next_state

    def reward_func(self, state):
        reward = self.default_reward
        done = False

        # 检查下一种状态的属性
        attribute = self.grid[state.row][state.column]
        if attribute == 1:
            # 获取奖励，游戏结束
            reward = 1
            done = True
        elif attribute == -1:
            # 遇到危险，游戏结束
            reward = -1
            done = True

        return reward, done

    def reset(self):
        # 将智能体放置到左下角
        self.agent_state = State(self.row_length - 1, 0)
        return self.agent_state

    def step(self, action):
        next_state, reward, done = self.transit(self.agent_state, action)
        if next_state is not None:
            self.agent_state = next_state

        return next_state, reward, done

    def transit(self, state, action):
        transition_probs = self.transit_func(state, action)
        if len(transition_probs) == 0:
            return None, None, True

        next_states = []
        probs = []
        for s in transition_probs:
            next_states.append(s)
            probs.append(transition_probs[s])

        next_state = np.random.choice(next_states, p=probs)
        reward, done = self.reward_func(next_state)
        return next_state, reward, done

(3) 智能体 Agent

环境内的智能体具体执行行动的代码如下。policy 是接收状态返回行动的函数。这里的代码中只是随机选择行动。

1
2
3
4
5
6
7
8
import random

class Agent():
    def __init__(self, env):
        self.actions = env.actions

    def policy(self, state):
        return random.choice(self.actions)

(4) 创建实例并进行推理

main 中描述迷宫的定义，以及如何在迷宫内探索。

创建环境实例 env 和智能体实例 agent 后，在 for 循环内进行 10 次探索，在探索前，需要 env.reset() 初始化智能体位置。智能体会一直移动直至 done 变为 True。

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
def main():
    # 生成grid环境
    grid = [
        [0, 0, 0, 1],
        [0, 9, 0, -1],
        [0, 0, 0, 0]
    ]
    env = Environment(grid)
    agent = Agent(env)

    # 尝试10次游戏
    for i in range(10):
        # 初始化智能体的位置
        state = env.reset()
        total_reward = 0
        done = False

        while not done:
            action = agent.policy(state)
            next_state, reward, done = env.step(action)
            total_reward += reward
            state = next_state

        print("Episode {}: Agent gets {} reward.".format(i, total_reward))


if __name__ == "__main__":
    main()

整体流程是根据 agent.policy 选择 action，然后将 action 传入 env.step，得到迁移后的状态 next_state 和立即奖励 reward。

改变 reward_func 和 transit_func 可以观察到智能体获得奖励的变化情况。

总结

本文我们解决了强化学习的基本概念，以及作为强化学习机制的 MDP，并且以 grid 为例实现了 MDP 环境，这个环境的实现方式可以迁移到其它业务场景中。

现在有了 MDP 环境，我们需要知道如何在其中选择最优行动，也就是本文代码框架中 agent.policy 方法的实现，具体来说就是价值的计算方法(价值近似)和基于价值近似的行动选择方法(策略)。

Share