VmasEnv¶
- torchrl.envs.VmasEnv(*args, **kwargs)[原始碼]¶
Vmas 環境封裝器。
GitHub: https://github.com/proroklab/VectorizedMultiAgentSimulator
論文: https://arxiv.org/abs/2207.03530
- 參數:
scenario (str 或 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
轉換或 torchrl 的StepCounter
不同,此引數不會在 tensordict 中設定"truncated"
項目。categorical_actions (bool, optional) – 如果環境動作是離散的,是否將它們轉換為類別或 one-hot。 預設為
True
。group_map (MarlGroupMapType 或 Dict[str, List[str]], optional) – 在 tensordict 中如何對 agents 進行分組以進行輸入/輸出。預設情況下,如果 agent 名稱遵循
"<name>_<int>"
慣例,它們將按"<name>"
分組。 如果它們不遵循此慣例,它們將全部放入名為"agents"
的群組中。 否則,可以指定群組映射表,或從某些預製選項中選擇。 有關更多信息,請參閱MarlGroupMapType
。**kwargs (Dict, optional) – 這些是可以傳遞給 VMAS scenario 建構子的額外參數(例如,agents 的數量、reward sparsity)。 可用參數將根據選擇的 scenario 而有所不同。 要查看特定 scenario 的可用參數,請參閱來自 scenario 資料夾 中的建構子檔案。
- 變數:
group_map (Dict[str, List[str]]) – 在 tensordict 中如何對 agents 進行分組以進行輸入/輸出。 有關更多信息,請參閱
MarlGroupMapType
。agent_names (list of str) – 環境中 agent 的名稱。
agent_names_to_indices_map (Dict[str, int]) – 字典,將 agent 名稱映射到它們在環境中的索引。
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 specs。
available_envs (List[str]) – 可用於建構的 scenario 列表。
警告
VMAS 返回單個
done
標記,該標記無法區分環境何時達到max_steps
和終止。 如果您認為truncation
信號是必要的,請將max_steps
設定為None
並使用StepCounter
轉換。範例
>>> 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)