AsyncVectorEnv

class gymnasium.vector.AsyncVectorEnv(env_fns: Sequence[Callable[[], Env]], shared_memory: bool = True, copy: bool = True, context: str | None = None, daemon: bool = True, worker: Callable[[int, Callable[[], Env], Connection, Connection, bool, Queue], None] | None = None)

并行运行多个环境的矢量化环境。

它使用 multiprocessing 进程，以及用于通信的管道。

示例

>>> import gymnasium as gym
>>> envs = gym.make_vec("Pendulum-v1", num_envs=2, vectorization_mode="async")
>>> envs
AsyncVectorEnv(Pendulum-v1, num_envs=2)
>>> envs = gym.vector.AsyncVectorEnv([
...     lambda: gym.make("Pendulum-v1", g=9.81),
...     lambda: gym.make("Pendulum-v1", g=1.62)
... ])
>>> envs
AsyncVectorEnv(num_envs=2)
>>> observations, infos = envs.reset(seed=42)
>>> observations
array([[-0.14995256,  0.9886932 , -0.12224312],
       [ 0.5760367 ,  0.8174238 , -0.91244936]], dtype=float32)
>>> infos
{}
>>> _ = envs.action_space.seed(123)
>>> observations, rewards, terminations, truncations, infos = envs.step(envs.action_space.sample())
>>> observations
array([[-0.1851753 ,  0.98270553,  0.714599  ],
       [ 0.6193494 ,  0.7851154 , -1.0808398 ]], dtype=float32)
>>> rewards
array([-2.96495728, -1.00214607])
>>> terminations
array([False, False])
>>> truncations
array([False, False])
>>> infos
{}

参数:

• env_fns – 创建环境的函数。

• shared_memory – 如果 True ，那么来自工作进程的观察结果将通过共享变量进行通信。如果观察结果很大（例如图像），这可以提高效率。

• copy – 如果 True ，那么 :meth:AsyncVectorEnv.reset 和 :meth:AsyncVectorEnv.step 方法将返回观察结果的副本。

• context – multiprocessing 的上下文。如果为 None ，则使用默认上下文。

• daemon – 如果 True，那么子进程的 daemon 标志将被打开；也就是说，如果主进程退出，它们也会退出。然而，daemon=True 会阻止子进程生成子进程，因此在某些环境中，您可能希望将其设置为 False。

• worker – 如果设置，则在子进程中使用该工作线程，而不是默认的工作线程。这对于覆盖某些内部向量环境逻辑可能很有用，例如，如何处理终止或截断时的重置。

警告

worker 是一个高级模式选项。它提供了高度的灵活性和高概率的自我伤害；因此，如果你正在编写自己的 worker，建议从 _worker（或 _worker_shared_memory）方法的代码开始，并添加更改。

抛出:

• RuntimeError – 如果某些子环境的观测空间与 observation_space 不匹配（或者，默认情况下，与第一个子环境的观测空间不匹配）。

• ValueError – 如果 observation_space 是一个自定义空间（即不是 Gym 中的默认空间，例如 gymnasium.spaces.Box、gymnasium.spaces.Discrete 或 gymnasium.spaces.Dict）且 shared_memory 为 True。

reset(*, seed: int | list[int] | None = None, options: dict[str, Any] | None = None) → tuple[ObsType, dict[str, Any]]

重置所有并行子环境并返回连接的观察和信息批次。

参数:

• seed – 环境重置种子

• options – 如果返回选项

返回:

一批来自向量化环境的观察和信息。

step(actions: ActType) → tuple[ObsType, ArrayType, ArrayType, ArrayType, dict[str, Any]]

为每个并行环境采取行动。

参数:

actions – 来自 :attr:action_space 的元素 动作批次。

返回:

一批 (观察值, 奖励, 终止, 截断, 信息)

close(**kwargs: Any)

关闭所有并行环境并释放资源。

它还会关闭所有现有的图像查看器，然后调用 :meth:close_extras 并将 :attr:closed 设置为 True。

警告

这个函数本身不会关闭环境，它应该在 :meth:close_extras 中处理。这对于同步和异步的向量化环境都是通用的。

备注

这将在垃圾回收或程序退出时自动调用。

参数:

**kwargs – 传递给 :meth:close_extras 的关键字参数

call(name: str, *args: Any, **kwargs: Any) → tuple[Any, ...]

使用参数和关键字参数从每个并行环境中调用一个方法。

参数:

• name (str) – 要调用的方法或属性的名称。

• *args – 应用于方法调用的位置参数。

• **kwargs – 应用于方法调用的关键字参数。

返回:

每个环境中对方法或属性的单独调用的结果列表。

get_attr(name: str) → tuple[Any, ...]

从每个并行环境中获取一个属性。

参数:

name (str) – 要从每个单独环境中获取的属性名称。

返回:

具有名称的属性

set_attr(name: str, values: list[Any] | tuple[Any] | object)

设置子环境的属性。

参数:

• name – 在每个单独环境中要设置的属性名称。

• values – 要设置的属性值。如果 values 是一个列表或元组，那么它对应于每个单独环境的值，否则为所有环境设置一个单一值。

抛出:

• ValueError – 值必须是一个列表或元组，其长度等于环境数量。

• AlreadyPendingCallError – 在等待挂起的调用完成时调用 :meth:set_attr。

附加方法

property AsyncVectorEnv.np_random: tuple[Generator, ...]

返回包装环境中的 numpy 随机数生成器元组。

property AsyncVectorEnv.np_random_seed: tuple[int, ...]

返回所有包装环境的 np_random 种子的元组。