OpenAI gym: 'Copy-v0'

最近对增强学习领域很感兴趣，刚好OpenAI推出了测试RL算法的gym（https://gym.openai.com/），上面有很多预设的测试环境，并且提供可视化的支持，于是决定用它小试牛刀。

众所周知，在增强学习算法中有观察（observation）、回报（reward）、价值（value）、策略（policy）四个核心元素。使用python可以安装并加载OpenAI Gym提供的预设环境。这些环境在每一轮动作执行后以新的观察和即时回报作为输出，接受下一轮执行的动作作为输入（输入输出的格式见官方文档）。

以下为gym的基本使用方法。

import gym

# load Copy-v0 environment
env = gym.make('Copy-v0')

# the action space of the environment
print(env.action_space)

# the observation space of the environment
print(env.observation_space)

for i_episode in range(100):
# reset environment
observation = env.reset()

for t in range(1000):
# display training process
env.render()

# choose an action randomly
action = env.action_space.sample()

observation, reward, done, info = env.step(action)
if done:
print('episode done in %d episodes'%i_episode)
break

之前我在金溆林老师的博客（http://www.cnblogs.com/jinxulin/）上学习了MDP、Monte Carlo等方法。手动实现算法之后，发现缺少一个好的问题来进行测试。为此我在gym中选择了一个入门级别的环境：Copy-v0。

Copy-v0的题目要求实现一个拷贝机器人，屏幕上方给出一行为待拷贝的标准串，标准串上有一个可以移动的光标；下面一行是输入行，仅能向后输入，不能改变光标位置。

每一轮，环境输出的观察值是标准串上光标指着的字符序号（标准串仅由英文字母A，B，C，D，E构成，对应输出0，1，2，3，4），如果指向的是标准串结尾的空格则返回5。而要求的输入为一个三元组(x, y, z)，其中x取0或1；y取0或1；z取0~5——x代表下一轮向左(0)或向右(1)移动光标；y代表下一轮是否进行一次输入；z代表下一轮要输入的字符（5代表F）。

举例：

第n轮模式串

ABC**D**DA（粗体为光标位置）

输入串：ABC

观察值：3

下一轮动作：（0,1,4）

第n+1轮模式串：AB**C**DDA

输入串：ABCE（因为输入有误，实际上环境会返回done=True）

观察值：2

下一轮动作：……

虽然知道了输入与输出的意义之后问题的最优解（1, 1, observation）看起来非常明确，但难点在于我们的目标是设计一个agent来自学这个系统的规则——也就是说，一个好的算法其实不需要知道输入与输出代表的物理意义。

我的解决方案使用了金溆林老师第四篇博客讲授的蒙特卡罗控制算法。

收敛速度不算很快（solved in 4282 episodes），当然这个收敛速度有一定运气成分，但是和175步就训练完成的算法相比一定还是有差距的。

在这个任务中我得到的最重要的经验是，在必要的时候，要有意识地调整“回报”来“激励”算法的学习——这样说比较模糊。以这个问题为例，最开始我实现的时候直接使用了环境给出的reward值，始终没有办法训练出可用的agent。确认算法本身无误后，我又手动输入action运行了几轮，发现了一个很重要的问题：如果仅仅移动光标而不进行输入，这一轮的回报值是零。回报值是零就意味着，虽然这个动作没有什么好处，但agent也没有必要学着去规避它。但这个问题是非常严重的，因为

1.观察值不包含光标位置信息，一旦非法移动光标，之后的方案将是完全随机的.

2.在所有可能的动作中，存在非法移动光标问题的占绝大多数，而在第一轮就采用正确答案(1, 1, observation)的概率微乎其微，几乎不可能用1-epsilon方法来补救（epsilon等于1也不行）.

所以最后我将所有的reward都再扣除0.5分，让agent感受到这种动作的危害，它们就可以自学成优秀的agent了。

结果评估

https://gym.openai.com/evaluations/eval_iK0iJksORiyq8QjBjx9rTw

代码实现

import numpy as np
import gym
import random

env = gym.make('Copy-v0')

# observation_space: Discrete(6)
# action_space: (Discrete(2), Discrete(2), Discrete(5))

# states: 0, 1, 2, 3, 4, 5

# MARK: problem specific functions

def action_to_index(action):
return action[0]*10+action[1]*5+action[2]

def index_to_action(index):
action = [index/10, (index/5) % 2, index % 5]
return tuple(action)

# MARK: Monte Carlo ES method

STATE_COUNT = 6
ACTION_COUNT = 20

def init_mces(state_count, action_count):
q = np.random.rand(state_count, action_count)
rets = np.zeros((state_count, action_count), dtype=np.double)
policy = [random.randint(0, action_count-1) for _ in range(state_count)]
return q, rets, policy

def learning(env):
q, rets, policy = init_mces(STATE_COUNT, ACTION_COUNT)
gamma = 0.7
epsilon = 1
total_score = 0.0
i_episode = 0

for i_episode in range(5000):
total_reward = 0
observation = env.reset()
g = np.zeros((STATE_COUNT, ACTION_COUNT), dtype=np.double)
passed = np.zeros((STATE_COUNT, ACTION_COUNT), dtype=np.double)

for t in range(100
a672
):
raw_action = policy[observation]
# 1 - epsilon greedy
if random.random() < epsilon:
raw_action = action_to_index((random.randint(0, 1),
random.randint(0, 1),
random.randint(0, 4)))
if passed[observation, raw_action] == 0.0:
passed[observation, raw_action] = 1.0

action = index_to_action(raw_action)
env.render()
observation, reward, done, info = env.step(action)
reward -= 0.5  # IMPORTANT: punish useless (even dangerous) actions whose environment reward is 0.0
total_reward += reward

for i in range(STATE_COUNT):
for j in range(ACTION_COUNT):
passed[i][j] *= gamma
g[i][j] += reward * passed[i][j]
if done:
break

# reduce exploration chance
if i_episode % 100 == 0:
epsilon *= 0.9

rets += g
q = rets / (i_episode+1)
policy = np.argmax(q, axis=1).tolist()

total_score += total_reward

return policy, i_episode

def test_policy(env, policy):
total_reward = 0.0
obs = env.reset()
for t in range(1000):
action = index_to_action(policy[obs])
obs, reward, done, info = env.step(action)
total_reward += reward
if done:
break
print('total reward: %f'%total_reward)

env.monitor.start('Copyv0-experiment-0')
policy, n_episode = learning(env)
env.monitor.close()

print('final policy: '+str(policy))
print('episodes trained: '+str(n_episode))
test_policy(env, policy)