VmasWrapper¶
- torchrl.envs.VmasWrapper(*args, **kwargs)[源代码]¶
Vmas 环境包装器。
GitHub: https://github.com/proroklab/VectorizedMultiAgentSimulator
论文: https://arxiv.org/abs/2207.03530
- 参数:
env (
vmas.simulator.environment.environment.Environment
) – 要包装的 vmas 环境。- 关键字参数:
num_envs (int) – 向量化模拟环境的数量。VMAS 使用 PyTorch 执行向量化模拟。此参数指示应在一个批次中模拟的向量化环境的数量。它还将决定环境的批次大小。
device (torch.device, 可选) – 用于模拟的设备。默认为默认设备。VMAS 创建的所有张量都将放置在此设备上。
continuous_actions (bool, 可选) – 是否使用连续动作。默认为
True
。如果为False
,则动作将是离散的。动作的数量及其大小将取决于所选的场景。有关更多信息,请参阅 VMAS 存储库。max_steps (int, 可选) – 任务的 horizon(时限)。默认为
None
(无限 horizon)。每个 VMAS 场景可以是终止的或不终止的。如果指定了max_steps
,则当达到此 horizon 时,场景也会终止(并且设置"terminated"
标志)。与 gym 的TimeLimit
转换或 torchrl 的StepCounter
不同,此参数不会在 tensordict 中设置"truncated"
条目。categorical_actions (bool, 可选) – 如果环境动作是离散的,是否将其转换为类别型或 one-hot 编码。默认为
True
。group_map (MarlGroupMapType 或 Dict[str, List[str]], 可选) – 如何在 tensordict 中对智能体进行分组以进行输入/输出。默认情况下,如果智能体名称遵循
"<name>_<int>"
约定,它们将按"<name>"
分组。如果它们不遵循此约定,它们将全部放在名为"agents"
的一个组中。否则,可以指定组映射或从一些预制选项中选择。有关更多信息,请参阅MarlGroupMapType
。
- 变量:
group_map (Dict[str, List[str]]) – 如何在 tensordict 中对智能体进行分组以进行输入/输出。有关更多信息,请参阅
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) – 环境是否具有任何惰性 spec
het_specs_map (Dict[str, bool]) – 将每个组映射到一个标志的字典,该标志表示该组是否具有惰性 spec
available_envs (List[str]) – 可用于构建的场景列表。
警告
VMAS 返回一个
done
标志,该标志不区分环境何时达到max_steps
和终止。如果您认为truncation
信号是必要的,请将max_steps
设置为None
并使用StepCounter
转换。示例
>>> env = VmasWrapper( ... vmas.make_env( ... 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)