如何矢量化强化学习环境?
How does one vectorize reinforcement learning environments?
我有一个符合 OpenAI 环境 API 的 Python class,但它是以非矢量化形式编写的,即它每一步接收一个输入操作并且 returns 每一步一个奖励。如何矢量化环境?我没能在 GitHub.
上找到任何明确的解释
您可以编写一个自定义 class 来迭代内部环境元组,同时保持基本的 Gym API。实际上,会有一些差异,因为底层环境不会在同一时间步终止。因此,将标准 step
和 reset
函数组合起来更容易
一种方法称为 step
。这是一个例子:
class VectorEnv:
def __init__(self, make_env_fn, n):
self.envs = tuple(make_env_fn() for _ in range(n))
# Call this only once at the beginning of training (optional):
def seed(self, seeds):
assert len(self.envs) == len(seeds)
return tuple(env.seed(s) for env, s in zip(self.envs, seeds))
# Call this only once at the beginning of training:
def reset(self):
return tuple(env.reset() for env in self.envs)
# Call this on every timestep:
def step(self, actions):
assert len(self.envs) == len(actions)
return_values = []
for env, a in zip(self.envs, actions):
observation, reward, done, info = env.step(a)
if done:
observation = env.reset()
return_values.append((observation, reward, done, info))
return tuple(return_values)
# Call this at the end of training:
def close(self):
for env in self.envs:
env.close()
然后你可以像这样实例化它:
import gym
make_env_fn = lambda: gym.make('CartPole-v0')
env = VectorEnv(make_env_fn, n=4)
当您调用 step
时,您必须为您的代理做一些簿记工作以处理 return 个值的元组。这也是我更喜欢将函数 make_env_fn
传递给 __init__
的原因,因为很容易添加像 gym.wrappers.Monitor
这样的包装器,它可以单独自动跟踪每个环境的统计信息。
我有一个符合 OpenAI 环境 API 的 Python class,但它是以非矢量化形式编写的,即它每一步接收一个输入操作并且 returns 每一步一个奖励。如何矢量化环境?我没能在 GitHub.
上找到任何明确的解释您可以编写一个自定义 class 来迭代内部环境元组,同时保持基本的 Gym API。实际上,会有一些差异,因为底层环境不会在同一时间步终止。因此,将标准 step
和 reset
函数组合起来更容易
一种方法称为 step
。这是一个例子:
class VectorEnv:
def __init__(self, make_env_fn, n):
self.envs = tuple(make_env_fn() for _ in range(n))
# Call this only once at the beginning of training (optional):
def seed(self, seeds):
assert len(self.envs) == len(seeds)
return tuple(env.seed(s) for env, s in zip(self.envs, seeds))
# Call this only once at the beginning of training:
def reset(self):
return tuple(env.reset() for env in self.envs)
# Call this on every timestep:
def step(self, actions):
assert len(self.envs) == len(actions)
return_values = []
for env, a in zip(self.envs, actions):
observation, reward, done, info = env.step(a)
if done:
observation = env.reset()
return_values.append((observation, reward, done, info))
return tuple(return_values)
# Call this at the end of training:
def close(self):
for env in self.envs:
env.close()
然后你可以像这样实例化它:
import gym
make_env_fn = lambda: gym.make('CartPole-v0')
env = VectorEnv(make_env_fn, n=4)
当您调用 step
时,您必须为您的代理做一些簿记工作以处理 return 个值的元组。这也是我更喜欢将函数 make_env_fn
传递给 __init__
的原因,因为很容易添加像 gym.wrappers.Monitor
这样的包装器,它可以单独自动跟踪每个环境的统计信息。