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

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 环境的中文描述大概是这样的：

冬天的时候，你和你的朋友们在公园扔飞盘。
你不小心把飞盘扔到了公园的湖中间。
湖面已经结冰，但是有些地方的没有结冰，形成一个冰洞，有人踩上去会掉下去。
这个飞盘对你来说非常宝贵，你觉得非常有必要把飞盘拿回来。
但是冰面很滑，你不能总是想去什么方向就去什么方向，滑滑的冰面可能会带你走向别的方向。
冰面用如下的字符块表示：
    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 运行模式

# 导入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的工作模式。

训练流程

导入依赖

import gym
import numpy as np
from collections import defaultdict
import functools

定义两个主要组件

# 构建 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)

训练循环

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

TabularQAgent 的实现

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

# 获取第一次观察结果
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 问题。

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

# 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的，刚开始训练的时候学习和探索强度比较大，后续慢慢缩小，这样模型就会慢慢收敛。

源代码#

场景#

OpenAI Gym#

Gym 运行模式#

训练流程#

核心代码#

如何更新 Q table#

explore / exploit 问题#

其他没有交代的点#

源代码

场景