PettingZooWrapper¶
- torchrl.envs.PettingZooWrapper(*args, **kwargs)[原始碼]¶
PettingZoo 環境包裝器。
要安裝 petting zoo,請遵循此處的指南 <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 多代理程式的一項功能,您能夠控制環境中代理程式的分組。您可以將代理程式分組在一起(堆疊它們的張量),以便在透過同一個神經網路傳遞它們時利用向量化。您可以將代理程式分成不同的群組,在這些群組中,它們是異質的,或者應該由不同的神經網路處理。要進行分組,您只需要在 env 建構時間傳遞一個
group_map
。預設情況下,pettingzoo 中的代理程式會依名稱分組。例如,對於代理程式
["agent_0","agent_1","agent_2","adversary_0"]
,tensordict 會看起來像>>> 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 parallel API)。對於資料通過相同神經網路的代理程式來說,分組對於利用向量化非常有用。
- 參數:
env (
pettingzoo.utils.env.ParallelEnv
或pettingzoo.utils.env.AECEnv
) – 要包裝的 pettingzoo 環境。return_state (bool, optional) – 是否從 pettingzoo 返回全域狀態(並非所有環境都可用)。預設為
False
。group_map (MarlGroupMapType 或 Dict[str, List[str]]], optional) – 如何在 tensordict 中分組代理程式以進行輸入/輸出。預設情況下,代理程式將按其名稱分組。否則,可以指定群組對應,或從一些預先製作的選項中選擇。有關更多資訊,請參閱
torchrl.envs.utils.MarlGroupMapType
。use_mask (bool, optional) – 環境是否應輸出
"mask"
。這在包裝的pettingzoo.AECEnv
中是強制性的,用於遮罩非活動代理程式,並且也應在代理程式數量可能變化時用於pettingzoo.ParallelEnv
。預設為False
。categorical_actions (bool, optional) – 如果環境的動作是離散的,是否將它們轉換為分類或 one-hot。
seed (int, optional) – 隨機種子。預設為
None
。done_on_any (bool, optional) – 環境的完成 (done) 鍵是否通過使用
any()
(當True
時)或all()
(當False
時)聚合代理程式鍵來設定。預設值 (None
) 是對 parallel 環境使用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)