基于 OpenAI Gym 的 Q-Learning 算法演示

Binder

TL;DR 从零开始实现 Q-learning 算法,在 OpenAI Gym 的环境中演示:如何一步步实现增强学习。

前面的博文里已经介绍过 Q-learning 的一些基本情况了,如果你没见过前面的博文或者已经忘记的差不多了,那么可以使用这个 Reinforcement Learning: 初次交手,多多指教 访问。

但是总的来说,如果没有实际代码跑一番,估计你对这个算法的正确性还是有疑虑的。本文将从头构建一个 Q-learning 算法,来解决一个 toy 级别的强化学习场景的学习工作。希望能加深你对 Q-learning 的理解和对强化学习的认知。

源代码

  • 比较精美的,但是做了一定扩展的实现在 q_learning_demo
  • 和本文代码相对应的,稍有改动的 Jupyter Notebook 在 proof-of-concept

场景

我们要用 Q-learning 解决什么问题呢?我们使用 OpenAI Gym 里提供的一个环境:FrozenLake-v0.

FrozenLake-v0 环境的中文描述大概是这样的:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
冬天的时候,你和你的朋友们在公园扔飞盘。
你不小心把飞盘扔到了公园的湖中间。
湖面已经结冰,但是有些地方的没有结冰,形成一个冰洞,有人踩上去会掉下去。
这个飞盘对你来说非常宝贵,你觉得非常有必要把飞盘拿回来。
但是冰面很滑,你不能总是想去什么方向就去什么方向,滑滑的冰面可能会带你走向别的方向。
冰面用如下的字符块表示:
SFFF
FHFH
FFFH
HFFG
S : Safe,开始点,安全
F : frozen surface, 冻结的表面,安全
H : hole, 掉下去就死定了
G : goal, 飞盘所在的地方

每个轮回,以你拿回飞盘或者掉进洞里而结束。
只有当你拿到飞盘才能获得1个奖励,其他情况都为0

OpenAI Gym

OpenAI 是 Elon Musk 创建的一家致力于非盈利的通用人工智能的公司。 其开源产品 Gym 是提供了一种增强学习的实现框架,主要用于提供一些模拟器供研究使用。

之前的博客提到过,增强学习是 Agent 和 Environment 直接的交互构成的。Gym 提供了很多常见的 Environment 对象。利用这些 Environment,研究者可以很快构建增强学习的应用。

Gym 运行模式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# 导入gym
import gym

# 构建环境
env = gym.make("Taxi-v1")

# 获取第一次的观察结果
observation = env.reset()

# 开始探索环境
for _ in range(1000):
env.render() # 渲染观察结果

# 你的 Agent 应该会根据观察结果,选择最合适的动作,但这里我们使用随机选择的动作
action = env.action_space.sample() # your agent here (this takes random actions)

# 将动作发送给环境,获取新的观察结果、奖励和是否结束的标志等
observation, reward, done, info = env.step(action)

if done: # 游戏结束
break

通过上面的示例,你应该了解OpenAI gym的工作模式。

训练流程

导入依赖

1
2
3
4
import gym
import numpy as np
from collections import defaultdict
import functools

定义两个主要组件

1
2
3
4
5
6
7
8
9
10
11
# 构建 Environment
env = gym.make('FrozenLake-v0')
env.seed(0) # 确保结果具有可重现性

# 构建 Agent
tabular_q_agent = TabularQAgent(env.observation_space, env.action_space)

# 开始训练
train(tabular_q_agent, env)

tabular_q_agent.test(env)

训练循环

1
2
3
def train(tabular_q_agent, env):
for episode in range(100000): # 训练 100000 次
all_reward, step_count = tabular_q_agent.learn(env)

TabularQAgent 的实现

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
class TabularQAgent(object):
def __init__(self, observation_space, action_space):
self.observation_space = observation_space
self.action_space = action_space
self.action_n = action_space.n
self.config = {
"learning_rate": 0.5,
"eps": 0.05, # Epsilon in epsilon greedy policies
"discount": 0.99,
"n_iter": 10000} # Number of iterations

self.q = defaultdict(functools.partial(generate_zeros, n=self.action_n))

def act(self, observation, eps=None):
if eps is None:
eps = self.config["eps"]
# epsilon greedy.
action = np.argmax(self.q[observation]) if np.random.random() > eps else self.action_space.sample()
return action

def learn(self, env):
obs = env.reset()

rAll = 0
step_count = 0

for t in range(self.config["n_iter"]):
action = self.act(obs)
obs2, reward, done, _ = env.step(action)

future = 0.0
if not done:
future = np.max(self.q[obs2])
self.q[obs][action] = (1 - self.config["learning_rate"]) * self.q[obs][action] + self.config["learning_rate"] * (reward + self.config["discount"] * future)

obs = obs2

rAll += reward
step_count += 1

if done:
break

return rAll, step_count

def test(self, env):
obs = env.reset()
env.render(mode='human')

for t in range(self.config["n_iter"]):
env.render(mode='human')

action = self.act(obs, eps=0)
obs2, reward, done, _ = env.step(action)
env.render(mode='human')

if done:
break

obs = obs2

核心代码

我们重点关注核心代码,Q-learning 是如何学习的,相关代码简化后得到:

如何更新 Q table

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# 获取第一次观察结果
obs = env.reset()

while True: # 一直循环,直到游戏结束
action = self.act(obs) # 根据策略,选择 action
obs2, reward, done, _ = env.step(action)

future = 0.0
if not done:
future = np.max(self.q[obs2]) # 获取后一步期望的最大奖励

# 更新 Q 表格,保留部分当前值 加上 部分当前奖励和未来一步的最大奖励
self.q[obs][action] = (1 - self.config["learning_rate"]) * self.q[obs][action] + self.config["learning_rate"] * (reward + self.config["discount"] * future)

# 更新
obs = obs2

# 游戏结束,退出循环
if done:
break

explore / exploit 问题

上面的代码我只提到了 self.act 会根据策略选择 action,那么该如何选择呢?这里就涉及到了 explore exploit tradeoff 的问题了。我们理想中的 action 选择策略是既能充分利用现有学习到的知识,每次都去最大化的最终的reward,这就是 exploit。但是同时,我们也希望我们的选择策略能适当的去探索一下其他路径,不能固定在已经知道的最优选择,避免局部最优解,适当时候也去探索其他路径,可能能发现更加优秀的路径,也就是全局最优解,这就是 explore 问题。

我们采取了一个概率方案,有一定概率去通过随机选择的方式,探索新路径。

1
2
3
4
5
6
# eps 数值在 [0, 1] ,控制探索的力度,越大探索的越多
if eps is None:
eps = self.config["eps"]
# epsilon greedy.
action = np.argmax(self.q[observation]) if np.random.random() > eps else self.action_space.sample()
return action

其他没有交代的点

由于本篇是科普性质,所以没有cover很多其他的问题点,比如学习和探索的因子可以是decay的,刚开始训练的时候学习和探索强度比较大,后续慢慢缩小,这样模型就会慢慢收敛。