PettingZooWrapper¶
- torchrl.envs.PettingZooWrapper(*args, **kwargs)[source]¶
PettingZoo 环境封装器。
要安装 PettingZoo,请遵循此处的指南 <https://github.com/Farama-Foundation/PettingZoo#installation>__。
此类是适用于所有 PettingZoo 环境的通用 torchrl 封装器。它能够封装
pettingzoo.AECEnv
和pettingzoo.ParallelEnv
。让我们更详细地了解
在封装的
pettingzoo.ParallelEnv
中,所有智能体将在环境的每一步都执行动作。如果在任务期间智能体的数量发生变化,请将use_mask
设置为True
。每个分组中将提供"mask"
作为输出,应使用它来屏蔽已结束的智能体。一旦有智能体完成,环境将立即重置(除非done_on_any
为False
)。在封装的
pettingzoo.AECEnv
中,每一步只有一个智能体执行动作。因此,对于此类环境,强制要求将use_mask
设置为True
。每个分组中将提供"mask"
作为输出,可用于屏蔽未执行动作的智能体。仅当所有智能体都完成后,环境才会重置(除非done_on_any
为True
)。如果智能体有任何不可用的动作,环境还将自动更新其
action_spec
的掩码,并为每个分组输出一个"action_mask"
以反映最新的可用动作。在训练期间应将其传递给带掩码的分布。作为 torchrl 多智能体的一个特性,您能够控制环境中智能体的分组。您可以将智能体分组在一起(堆叠它们的张量),以便在通过相同的神经网络时利用向量化。您也可以将不同质或需要由不同神经网络处理的智能体分到不同的组中。要进行分组,您只需在构建环境时传递一个
group_map
。默认情况下,PettingZoo 中的智能体将按名称分组。例如,对于智能体
["agent_0","agent_1","agent_2","adversary_0"]
,tensordicts 将如下所示>>> print(env.rand_action(env.reset())) TensorDict( fields={ agent: TensorDict( fields={ action: Tensor(shape=torch.Size([3, 9]), device=cpu, dtype=torch.int64, is_shared=False), action_mask: Tensor(shape=torch.Size([3, 9]), device=cpu, dtype=torch.bool, is_shared=False), done: Tensor(shape=torch.Size([3, 1]), device=cpu, dtype=torch.bool, is_shared=False), observation: Tensor(shape=torch.Size([3, 3, 3, 2]), device=cpu, dtype=torch.int8, is_shared=False), terminated: Tensor(shape=torch.Size([3, 1]), device=cpu, dtype=torch.bool, is_shared=False), truncated: Tensor(shape=torch.Size([3, 1]), device=cpu, dtype=torch.bool, is_shared=False)}, batch_size=torch.Size([3]))}, adversary: TensorDict( fields={ action: Tensor(shape=torch.Size([1, 9]), device=cpu, dtype=torch.int64, is_shared=False), action_mask: Tensor(shape=torch.Size([1, 9]), device=cpu, dtype=torch.bool, is_shared=False), done: Tensor(shape=torch.Size([1, 1]), device=cpu, dtype=torch.bool, is_shared=False), observation: Tensor(shape=torch.Size([1, 3, 3, 2]), device=cpu, dtype=torch.int8, is_shared=False), terminated: Tensor(shape=torch.Size([1, 1]), device=cpu, dtype=torch.bool, is_shared=False), truncated: Tensor(shape=torch.Size([1, 1]), device=cpu, dtype=torch.bool, is_shared=False)}, batch_size=torch.Size([1]))}, batch_size=torch.Size([])) >>> print(env.group_map) {"agent": ["agent_0", "agent_1", "agent_2"], "adversary": ["adversary_0"]}
否则,可以指定分组映射或从一些预设选项中选择。有关更多信息,请参阅
torchrl.envs.utils.MarlGroupMapType
。例如,您可以提供MarlGroupMapType.ONE_GROUP_PER_AGENT
,表示每个智能体都应该有自己的 tensordict(类似于 pettingzoo 的并行 API)。分组对于利用数据通过同一神经网络的智能体之间的向量化非常有用。
- 参数:
env (
pettingzoo.utils.env.ParallelEnv
或pettingzoo.utils.env.AECEnv
) – 要封装的 PettingZoo 环境。return_state (bool, 可选) – 是否从 PettingZoo 返回全局状态(并非所有环境都可用)。默认为
False
。group_map (MarlGroupMapType 或 Dict[str, List[str]]], 可选) – 如何在 tensordicts 中对智能体进行输入/输出分组。默认情况下,智能体将按名称分组。否则,可以指定分组映射或从一些预设选项中选择。有关更多信息,请参阅
torchrl.envs.utils.MarlGroupMapType
。use_mask (bool, 可选) – 环境是否应输出
"mask"
。这对于封装的pettingzoo.AECEnv
是强制性的,用于屏蔽非执行动作的智能体;当智能体数量可能变化时,也应将其用于pettingzoo.ParallelEnv
。默认为False
。categorical_actions (bool, 可选) – 如果环境的动作是离散的,是否将其转换为类别型或独热编码。
seed (int, 可选) – 种子。默认为
None
。done_on_any (bool, 可选) – 环境的完成(done)键是否通过使用
any()
(当为True
时)或all()
(当为False
时)聚合智能体键来设置。默认值(None
)是对于并行环境使用any() 马,对于 AEC 环境使用
all()
。
示例
>>> # Parallel env >>> from torchrl.envs.libs.pettingzoo import PettingZooWrapper >>> from pettingzoo.butterfly import pistonball_v6 >>> kwargs = {"n_pistons": 21, "continuous": True} >>> env = PettingZooWrapper( ... env=pistonball_v6.parallel_env(**kwargs), ... return_state=True, ... group_map=None, # Use default for parallel (all pistons grouped together) ... ) >>> print(env.group_map) ... {'piston': ['piston_0', 'piston_1', ..., 'piston_20']} >>> env.rollout(10) >>> # AEC env >>> from pettingzoo.classic import tictactoe_v3 >>> from torchrl.envs.libs.pettingzoo import PettingZooWrapper >>> from torchrl.envs.utils import MarlGroupMapType >>> env = PettingZooWrapper( ... env=tictactoe_v3.env(), ... use_mask=True, # Must use it since one player plays at a time ... group_map=None # # Use default for AEC (one group per player) ... ) >>> print(env.group_map) ... {'player_1': ['player_1'], 'player_2': ['player_2']} >>> env.rollout(10)