OpenAI Gym 入门

OpenAI Gym 介绍

通常训练机器人在现实世界中是困难和缓慢的，增加更多的计算能力不会让机器人移动得更快。通常需要一个模拟环境，用游戏环境来模拟真实环境。OpenAI Gym 是一个工具包，提供了很多游戏模拟环境。安装方式如下

pip install gym

查看所有的可用环境

>>> import gym
>>> print(gym.envs.registry.keys())

选择加载一个环境，这里创建了一个 CartPole 环境。这是一个 2D 模拟，其中推车可以被左右加速，以平衡放置在它上面的平衡杆。

>>> env = gym.make("CartPole-v1")

初始化环境

>>> observation, info = env.reset()

>>> observation
[ 0.0112333  0.03536154 -0.04991545  -0.01171878]

在此例中”observation“为环境初始值小车的状态；info是其它信息，其他环境中提供额外信息用于调试或训练，这里为空。

observation的返回值由四个浮点数组成的数组，四个值分别是：

水平位置：0为中心，右正左负

速度：正数向右

杆角度：0=直立，>0向右倾斜，<0向左

角速度：正为顺时针方向倾倒

查询该环境可以执行的操作空间

>>> env.action_space
Discrete(2)

Discrete(2)的意思是可能的行动是整数 0 和 1，表示向左（0）或向右（1）加速。其它的环境可能有其它离散的行动，或其它种类的行动（例如，连续性行动）。

执行下一步动作，返回状态

执行动作的方法是：env.step(action)，action是要执行的动作。

action = env.action_space.sample()  # 该环境默认动作,在此随机为0或1

# 也可以写成随机数0和1
# action = np.random.randint(2)
observation, reward, terminated, truncated, info = env.step(action)

step()方法执行给定的动作并返回五个值：

observation：观察状态，执行下一步动作后新的状态，返回四个浮点数组成的数组，和上方初始时释义相同。

reward：奖励，在此环境中，无论做什么都会得到1的奖励。所以尽可能让的杆长时间不倒。

terminated：终止，是否到达终止状态(在任务的 MDP 下定义)，在此环境中当平衡杆倾斜太多、或越过屏幕、或超过 200 步时会发生这种情况。

truncated: 截断,是否满足 MDP 范围之外的截断条件。通常是一个时间限制，但也可以用来指示代理在物理上越界。可用于在达到终末状态之前提前结束动作。

info：其它的辅助信息，此环境为空。

渲染

在环境初始化时由render_mode属性指定的渲染模式

env = gym.make("CartPole-v1", render_mode="human")

空：默认值，不渲染

human：渲染在当前终端中连续图像呈现，通常是为了人类观察。此渲染模式不返回任何值。

rgb_array：返回环境当前状态的单个帧，以一个 Numpy 数组格式返回，其（x， y， 3）表示 x×y 像素图像的 RGB 值。

rgb_array_list：返回自上一次重置环境后所有状态的帧列表。每一帧和rgb_array一样。

ansi：返回一个字符串（str），其中包含一个每个时间步长的终端样式，以文本形式表示。文本可以包括换行符和 ANSI 转义序列（例如颜色）

示例，进行1000次的随机动作

import gym

env = gym.make("CartPole-v1", render_mode="human")
observation, info = env.reset()
action = env.action_space.sample()

for _ in range(1000):
    observation, reward, terminated, truncated, info = env.step(action)

    if terminated or truncated:
        observation, info = env.reset()

env.close()  # 关闭环境，释放内存

神经网络训练

这里需要安装tensorflow和keras来搭建神经网络，要注意，tensorflow和keras的版本要相对应，tensorflow又分GPU版本和CPU版本，而GPU要对应你安装的CUDA版本，在tensorflow安装成功后再安装对应版本的keras。

tensorflow官方中文安装指南

使用DQN训练代码如下，代码和学习都来自于开源教程Sklearn 与 TensorFlow 机器学习实用指南第二版 的18章节，很多地方都还是不懂，能看明白的也做了注释，具体学习请参考原书。

import gym
import numpy as np
import tensorflow as tf
from tensorflow import keras
import matplotlib.pyplot as plt

env = gym.make("CartPole-v1")

# 状态空间
input_shape = [4]  # == env.observation_space.shape
# 行为空间
n_outputs = 2  # == env.action_space.n
# 搭建神经网络
model = keras.models.Sequential([
    keras.layers.Dense(24, activation="relu", input_shape=input_shape),
    keras.layers.Dense(24, activation="relu"),
    keras.layers.Dense(n_outputs)
])


# 定义随机选择函数
# 当使用 DQN 选择一个行为时，只需要选择最大的预测 Q-value 即可。然而，为了确保 agent 探索环境，根据概率 epsilon 随机选择行为。
def epsilon_greedy_policy(state, epsilon=0):
    if np.random.rand() < epsilon:
        return np.random.randint(n_outputs)
    else:
        Q_values = model.predict(state[np.newaxis])
        return np.argmax(Q_values[0])


# 创建队列
# 一个重存储器，它包括 agent 的经验，由元组组成:
# (states，actions，rewards，states，dones) 。我们可以利用 deque 类创建队列来进行存储.
from collections import deque
replay_buffer = deque(maxlen=2000)


# 定义经验提取函数
# 创建一个函数用来从重提取器中采集 agent 的经验信息，返回一个5维的 NumPy arrays 数组:
# 5个元素：状态;，智能体选择的动作;奖励;下一个状态;一个知识是否结束的布尔值（done）。需要一个小函数从接力缓存随机采样。
def sample_experiences(batch_size):
    indices = np.random.randint(len(replay_buffer), size=batch_size)
    batch = [replay_buffer[index] for index in indices]
    states, actions, rewards, next_states, dones = [
        np.array([experience[field_index] for experience in batch])
        for field_index in range(5)]
    return states, actions, rewards, next_states, dones


# 定义单步执行函数
# 现在可以定义一个函数使用 DQN 来执行每一步，然后将它们的经验存储在重存储器中
def play_one_step(env, state, epsilon):
    action = epsilon_greedy_policy(state, epsilon)
    next_state, reward, done, info, _ = env.step(action)
    replay_buffer.append((state, action, reward, next_state, done))
    return next_state, reward, done, info


# 定义训练
batch_size = 32  # 批次大小
discount_factor = 0.95  # 折扣系数
optimizer = keras.optimizers.Adam(learning_rate=1e-3)  # 优化器
loss_fn = keras.losses.mean_squared_error  # 损失函数


# 最后创建一个函数来从重存储器中采集经验信息并执行训练步骤
def training_step(batch_size):
    #  提取经验
    experiences = sample_experiences(batch_size)
    #  获取信息
    states, actions, rewards, next_states, dones = experiences
    #  预测Q-value
    next_Q_values = model.predict(next_states)
    #  选择Q-value
    max_next_Q_values = np.max(next_Q_values, axis=1)
    #  目标Q-value
    target_Q_values = (rewards +
                       (1 - dones) * discount_factor * max_next_Q_values)
    target_Q_values = target_Q_values.reshape(-1, 1)
    mask = tf.one_hot(actions, n_outputs)
    # 自动求导
    with tf.GradientTape() as tape:
        all_Q_values = model(states)
        Q_values = tf.reduce_sum(all_Q_values * mask, axis=1, keepdims=True)
        # 计算损失值
        loss = tf.reduce_mean(loss_fn(target_Q_values, Q_values))
    # 计算梯度
    grads = tape.gradient(loss, model.trainable_variables)
    # 优化损失值
    optimizer.apply_gradients(zip(grads, model.trainable_variables))


rewards = []
best_score = 0
for episode in range(4000):
    obs, _ = env.reset()
    for step in range(200):
        epsilon = max(1 - episode / 500, 0.01)
        obs, reward, done, info = play_one_step(env, obs, epsilon)
        if done:
            break
    # 选择一个最优的模型打印
    rewards.append(step)
    if step >= best_score:
        best_weights = model.get_weights()
        best_score = step
    print("\rEpisode: {}, Steps: {}, eps: {:.3f}".format(episode, step + 1, epsilon), end="")
    if episode > 50:
        training_step(batch_size)


# 画图
plt.figure(figsize=(8, 4))
plt.plot(rewards)
plt.xlabel("Episode", fontsize=14)
plt.ylabel("Sun of rewards", fontsize=14)
plt.show()
env.close()

每场200步，一共进行4000场次训练结果如下，可以看到训练3000次左右时可以基本稳定在200步。