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
中的一个。有关可用场景的描述和渲染,请参阅 README。- 关键字参数:
num_envs (int) – 向量化模拟环境的数量。VMAS 使用 PyTorch 执行向量化模拟。此参数指示应在批次中模拟的向量化环境数量。它也将决定环境的批次大小。
device (torch.device, optional) – 用于模拟的设备。默认为默认设备。VMAS 创建的所有张量都将放置在此设备上。
continuous_actions (bool, optional) – 是否使用连续动作。默认为
True
。如果为False
,动作将是离散的。动作的数量和大小将取决于所选场景。有关更多信息,请参阅 VMAS 仓库。max_steps (int, optional) – 任务的视野(步数限制)。默认为
None
(无限视野)。每个 VMAS 场景都可以终止或不终止。如果指定了max_steps
,则当达到此视野时,场景也会终止(并设置"terminated"
标志)。与 gym 的TimeLimit
transform 或 torchrl 的StepCounter
不同,此参数不会在 tensordict 中设置"truncated"
条目。categorical_actions (bool, optional) – 如果环境动作为离散类型,是否将其转换为 categorical 或 one-hot。默认为
True
。group_map (MarlGroupMapType or Dict[str, List[str]], optional) – 如何在 tensordicts 中对智能体进行分组以进行输入/输出。默认情况下,如果智能体名称遵循
"<name>_<int>"
的约定,它们将按"<name>"
分组。如果不遵循此约定,它们将全部放在一个名为"agents"
的组中。此外,可以指定分组映射或从一些预设选项中选择。有关更多信息,请参阅MarlGroupMapType
。**kwargs (Dict, optional) – 这些是可以传递给 VMAS 场景构造函数的额外参数。(例如,智能体数量、奖励稀疏度)。可用参数将根据所选场景而异。要查看特定场景的可用参数,请参阅其在场景文件夹中的文件中的构造函数。
- 变量:
group_map (Dict[str, List[str]]) – 如何在 tensordicts 中对智能体进行分组以进行输入/输出。有关更多信息,请参阅
MarlGroupMapType
。agent_names (list of str) – 环境中的智能体名称列表
agent_names_to_indices_map (Dict[str, int]) – 将智能体名称映射到其在环境中的索引的字典
unbatched_action_spec (TensorSpec) – 不含向量化维度的 spec 版本
unbatched_observation_spec (TensorSpec) – 不含向量化维度的 spec 版本
unbatched_reward_spec (TensorSpec) – 不含向量化维度的 spec 版本
het_specs (bool) – 环境是否包含任何 lazy spec
het_specs_map (Dict[str, bool]) – 将每个组映射到一个标志的字典,该标志表示该组是否包含 lazy spec
available_envs (List[str]) – 可用于构建的场景列表。
警告
VMAS 返回一个单独的
done
标志,该标志不区分环境达到max_steps
和终止的情况。如果您认为truncation
信号是必需的,请将max_steps
设置为None
并使用StepCounter
transform。示例
>>> 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)