快捷方式

VmasEnv

torchrl.envs.VmasEnv(*args, **kwargs)[源代码]

Vmas 环境包装器。

GitHub:https://github.com/proroklab/VectorizedMultiAgentSimulator

论文:https://arxiv.org/abs/2207.03530

参数:

scenario (str or vmas.simulator.scenario.BaseScenario) – 要构建的 vmas 场景。必须是 available_envs 之一。有关可用场景的描述和渲染,请参阅自述文件.

关键字参数:
  • num_envs (int) – 矢量化模拟环境的数量。VMAS 使用 PyTorch 执行矢量化模拟。此参数表示在批次中模拟的矢量化环境的数量。它还将确定环境的批次大小。

  • device (torch.device, 可选) – 模拟的设备。默认为默认设备。VMAS 创建的所有张量都将放置在此设备上。

  • continuous_actions (bool, 可选) – 是否使用连续动作。默认为 True。如果为 False,则动作将是离散的。动作的数量及其大小将取决于所选场景。有关更多信息,请参阅 VMAS 存储库。

  • max_steps (int, 可选) – 任务的范围。默认为 None(无限范围)。每个 VMAS 场景都可以终止或不终止。如果指定了 max_steps,则该场景也会在达到此范围时终止(并且会设置 "terminated" 标志)。与 gym 的 TimeLimit 变换或 torchrl 的 StepCounter 不同,此参数不会在 tensordict 中设置 "truncated" 条目。

  • categorical_actions (bool, 可选) – 如果环境动作是离散的,则是否将其转换为分类或独热。默认为 True

  • group_map (MarlGroupMapType or Dict[str, List[str]], 可选) – 如何在 tensordicts 中对输入/输出进行智能体分组。默认情况下,如果智能体名称遵循 "<name>_<int>" 约定,则它们将按 "<name>" 进行分组。如果它们不遵循此约定,则它们将全部放在一个名为 "agents" 的组中。否则,可以从某些预制选项中指定或选择组映射。有关更多信息,请参阅 MarlGroupMapType

  • **kwargs (Dict, 可选) – 这些是可传递给 VMAS 场景构造函数的附加参数。(例如,智能体数量、奖励稀疏性)。可用参数将根据所选场景而有所不同。要查看特定场景的可用参数,请参阅 场景文件夹 中其文件的构造函数。

变量:
  • group_map (Dict[str, List[str]]) – 如何在 tensordicts 中对输入/输出进行智能体分组。有关更多信息,请参阅 MarlGroupMapType

  • agent_names (str 列表) – 环境中智能体的名称

  • agent_names_to_indices_map (Dict[str, int]) – 将智能体名称映射到它们在环境中的索引的字典

  • unbatched_action_spec (TensorSpec) – 没有矢量化维度的规范版本

  • unbatched_observation_spec (TensorSpec) – 没有矢量化维度的规范版本

  • unbatched_reward_spec (TensorSpec) – 没有矢量化维度的规范版本

  • het_specs (bool) – 环境中是否包含任何延迟规范

  • het_specs_map (Dict[str, bool]) – 将每个组映射到一个标志的字典,该标志表示该组是否包含延迟规范

  • available_envs (str 列表) – 可用于构建的场景列表。

警告

VMAS 返回一个单独的 done 标志,该标志不区分环境何时到达 max_steps 和终止。如果您认为 truncation 信号是必要的,请将 max_steps 设置为 None 并使用 StepCounter 变换。

示例

>>>  env = VmasEnv(
...      scenario="flocking",
...      num_envs=32,
...      continuous_actions=True,
...      max_steps=200,
...      device="cpu",
...      seed=None,
...      # Scenario kwargs
...      n_agents=5,
...  )
>>>  print(env.rollout(10))
TensorDict(
    fields={
        agents: TensorDict(
            fields={
                action: Tensor(shape=torch.Size([32, 10, 5, 2]), device=cpu, dtype=torch.float32, is_shared=False),
                info: TensorDict(
                    fields={
                        agent_collision_rew: Tensor(shape=torch.Size([32, 10, 5, 1]), device=cpu, dtype=torch.float32, is_shared=False),
                        agent_distance_rew: Tensor(shape=torch.Size([32, 10, 5, 1]), device=cpu, dtype=torch.float32, is_shared=False)},
                    batch_size=torch.Size([32, 10, 5]),
                    device=cpu,
                    is_shared=False),
                observation: Tensor(shape=torch.Size([32, 10, 5, 18]), device=cpu, dtype=torch.float32, is_shared=False)},
            batch_size=torch.Size([32, 10, 5]),
            device=cpu,
            is_shared=False),
        done: Tensor(shape=torch.Size([32, 10, 1]), device=cpu, dtype=torch.bool, is_shared=False),
        next: TensorDict(
            fields={
                agents: TensorDict(
                    fields={
                        info: TensorDict(
                            fields={
                                agent_collision_rew: Tensor(shape=torch.Size([32, 10, 5, 1]), device=cpu, dtype=torch.float32, is_shared=False),
                                agent_distance_rew: Tensor(shape=torch.Size([32, 10, 5, 1]), device=cpu, dtype=torch.float32, is_shared=False)},
                            batch_size=torch.Size([32, 10, 5]),
                            device=cpu,
                            is_shared=False),
                        observation: Tensor(shape=torch.Size([32, 10, 5, 18]), device=cpu, dtype=torch.float32, is_shared=False),
                        reward: Tensor(shape=torch.Size([32, 10, 5, 1]), device=cpu, dtype=torch.float32, is_shared=False)},
                    batch_size=torch.Size([32, 10, 5]),
                    device=cpu,
                    is_shared=False),
                done: Tensor(shape=torch.Size([32, 10, 1]), device=cpu, dtype=torch.bool, is_shared=False),
                terminated: Tensor(shape=torch.Size([32, 10, 1]), device=cpu, dtype=torch.bool, is_shared=False)},
            batch_size=torch.Size([32, 10]),
            device=cpu,
            is_shared=False),
        terminated: Tensor(shape=torch.Size([32, 10, 1]), device=cpu, dtype=torch.bool, is_shared=False)},
    batch_size=torch.Size([32, 10]),
    device=cpu,
    is_shared=False)

文档

访问 PyTorch 的全面开发者文档

查看文档

教程

获取针对初学者和高级开发人员的深入教程

查看教程

资源

查找开发资源并获得解答

查看资源