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, 可选) – 任务的范围。默认为
None
(无限范围)。每个 VMAS 场景都可以是终止的或非终止的。如果指定了max_steps
,则只要达到此范围,场景也将终止(并设置"terminated"
标志)。与 gym 的TimeLimit
转换或 torchrl 的StepCounter
不同,此参数不会在 tensordict 中设置"truncated"
条目。categorical_actions (bool, 可选) – 如果环境动作是离散的,则是否将其转换为分类或独热编码。默认为
True
。group_map (MarlGroupMapType 或 Dict[str, List[str]], 可选) – 如何在 tensordicts 中对输入/输出的智能体进行分组。默认情况下,如果智能体名称遵循
"<name>_<int>"
约定,则它们将按"<name>"
进行分组。如果它们不遵循此约定,则它们将全部放在一个名为"agents"
的组中。否则,可以指定组映射或从某些预制选项中进行选择。有关更多信息,请参阅MarlGroupMapType
。
- 变量:
group_map (Dict[str, List[str]]) – 如何在 tensordicts 中对输入/输出的智能体进行分组。有关更多信息,请参阅
MarlGroupMapType
。agent_names (字符串列表) – 环境中智能体的名称
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 (字符串列表) – 可用于构建的场景列表。
警告
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)