GymLikeEnv

class torchrl.envs.GymLikeEnv(*args, **kwargs)

類似 gym 的環境是一種環境。

它的行為與 gym 環境相似，在於期望常見的方法（特別是 reset 和 step）所做的事情。

GymLikeEnv 有一個 step() 方法，其簽章如下

env.step(action: np.ndarray) -> Tuple[Union[np.ndarray, dict], double, bool, *info]

其中輸出分別是觀察、獎勵和完成狀態。在此實作中，info 輸出會被捨棄（但可以透過更新 info_dict_reader 來讀取特定金鑰，請參閱 set_info_dict_reader() 方法）。

預設情況下，第一個輸出會寫入輸出 tensordict 中的 "observation" 鍵值對，除非第一個輸出是字典。在這種情況下，每個觀察輸出都會放入每個字典的對應 f"{key}" 位置。

也預期 env.reset() 會傳回類似於完成步驟後觀察到的觀察結果。

property action_key: NestedKey

環境的動作金鑰。

預設情況下，這將是 "action"。

如果環境中有多個動作金鑰，此函數將引發例外狀況。

property action_keys: List[NestedKey]

環境的動作金鑰。

預設情況下，只會有一個名為 "action" 的金鑰。

金鑰依資料樹中的深度排序。

property action_spec: TensorSpec

action 規格。

action_spec 始終儲存為複合規格。

如果動作規格以簡單規格提供，則會傳回此規格。

>>> env.action_spec = Unbounded(1)
>>> env.action_spec
UnboundedContinuous(
    shape=torch.Size([1]),
    space=ContinuousBox(
        low=Tensor(shape=torch.Size([]), device=cpu, dtype=torch.float32, contiguous=True),
        high=Tensor(shape=torch.Size([]), device=cpu, dtype=torch.float32, contiguous=True)),
    device=cpu,
    dtype=torch.float32,
    domain=continuous)

如果動作規格以複合規格提供，且僅包含一個葉節點，則此函數將僅傳回葉節點。

>>> env.action_spec = Composite({"nested": {"action": Unbounded(1)}})
>>> env.action_spec
UnboundedContinuous(
    shape=torch.Size([1]),
    space=ContinuousBox(
        low=Tensor(shape=torch.Size([]), device=cpu, dtype=torch.float32, contiguous=True),
        high=Tensor(shape=torch.Size([]), device=cpu, dtype=torch.float32, contiguous=True)),
    device=cpu,
    dtype=torch.float32,
    domain=continuous)

如果動作規格以複合規格提供，且有多個葉節點，則此函數將傳回整個規格。

>>> env.action_spec = Composite({"nested": {"action": Unbounded(1), "another_action": Categorical(1)}})
>>> env.action_spec
Composite(
    nested: Composite(
        action: UnboundedContinuous(
            shape=torch.Size([1]),
            space=ContinuousBox(
                low=Tensor(shape=torch.Size([]), device=cpu, dtype=torch.float32, contiguous=True),
                high=Tensor(shape=torch.Size([]), device=cpu, dtype=torch.float32, contiguous=True)),
            device=cpu,
            dtype=torch.float32,
            domain=continuous),
        another_action: Categorical(
            shape=torch.Size([]),
            space=DiscreteBox(n=1),
            device=cpu,
            dtype=torch.int64,
            domain=discrete), device=cpu, shape=torch.Size([])), device=cpu, shape=torch.Size([]))

若要擷取傳遞的完整規格，請使用

>>> env.input_spec["full_action_spec"]

此屬性是可變的。

範例

>>> from torchrl.envs.libs.gym import GymEnv
>>> env = GymEnv("Pendulum-v1")
>>> env.action_spec
BoundedContinuous(
    shape=torch.Size([1]),
    space=ContinuousBox(
        low=Tensor(shape=torch.Size([1]), device=cpu, dtype=torch.float32, contiguous=True),
        high=Tensor(shape=torch.Size([1]), device=cpu, dtype=torch.float32, contiguous=True)),
    device=cpu,
    dtype=torch.float32,
    domain=continuous)

add_module(name: str, module: Optional[Module]) → None

將一個子模組新增至目前的模組。

可以使用給定的名稱作為屬性存取該模組。

參數:

• name (str) – 子模組的名稱。 可以使用給定的名稱從此模組存取子模組

• module (Module) – 要新增至模組的子模組。

add_truncated_keys() → EnvBase

將 truncated keys 新增至環境。

append_transform(transform: 'Transform' | Callable[[TensorDictBase], TensorDictBase]) → None

傳回一個轉換後的環境，其中應用了傳遞的可呼叫物件/轉換。

參數:

transform (Transform 或 Callable[[TensorDictBase], TensorDictBase]) – 要應用於環境的轉換。

範例

>>> from torchrl.envs import GymEnv
>>> import torch
>>> env = GymEnv("CartPole-v1")
>>> loc = 0.5
>>> scale = 1.0
>>> transform = lambda data: data.set("observation", (data.get("observation") - loc)/scale)
>>> env = env.append_transform(transform=transform)
>>> print(env)
TransformedEnv(
    env=GymEnv(env=CartPole-v1, batch_size=torch.Size([]), device=cpu),
    transform=_CallableTransform(keys=[]))

apply(fn: Callable[[Module], None]) → T

將 fn 遞迴地應用於每個子模組（如 .children() 傳回的）以及 self。

典型的使用案例包括初始化模型的參數 (另請參閱 torch.nn.init)。

參數:

fn (Module -> None) – 要應用於每個子模組的函式

傳回:

self

傳回類型:

Module

範例

>>> @torch.no_grad()
>>> def init_weights(m):
>>>     print(m)
>>>     if type(m) == nn.Linear:
>>>         m.weight.fill_(1.0)
>>>         print(m.weight)
>>> net = nn.Sequential(nn.Linear(2, 2), nn.Linear(2, 2))
>>> net.apply(init_weights)
Linear(in_features=2, out_features=2, bias=True)
Parameter containing:
tensor([[1., 1.],
        [1., 1.]], requires_grad=True)
Linear(in_features=2, out_features=2, bias=True)
Parameter containing:
tensor([[1., 1.],
        [1., 1.]], requires_grad=True)
Sequential(
  (0): Linear(in_features=2, out_features=2, bias=True)
  (1): Linear(in_features=2, out_features=2, bias=True)
)

auto_register_info_dict(ignore_private: bool = True, *, info_dict_reader: Optional[BaseInfoDictReader] = None) → EnvBase

自動註冊 info dict 並在需要時附加 TensorDictPrimer 實例。

如果沒有提供 info_dict_reader，則假定 info dict 中包含的所有資訊都可以註冊為 tensordict 中的數值。

這個方法會傳回一個（可能經過轉換的）環境，我們會確保 torchrl.envs.utils.check_env_specs() 成功，無論 info 是否在重置時填寫。

注意

此方法需要在環境中執行幾次疊代，以手動檢查行為是否符合預期。

參數:

ignore_private (bool, optional) – 如果 True，則將忽略私有 info（以下底線開頭）。預設為 True。

關鍵字參數:

info_dict_reader (BaseInfoDictReader, optional) – info_dict_reader（如果事先已知）。與 set_info_dict_reader() 不同，此方法將建立必要的 primers 來使 check_env_specs() 執行。

範例

>>> from torchrl.envs import GymEnv
>>> env = GymEnv("HalfCheetah-v4")
>>> # registers the info dict reader
>>> env.auto_register_info_dict()
GymEnv(env=HalfCheetah-v4, batch_size=torch.Size([]), device=cpu)
>>> env.rollout(3)
TensorDict(
    fields={
        action: Tensor(shape=torch.Size([3, 6]), device=cpu, dtype=torch.float32, is_shared=False),
        done: Tensor(shape=torch.Size([3, 1]), device=cpu, dtype=torch.bool, is_shared=False),
        next: TensorDict(
            fields={
                done: Tensor(shape=torch.Size([3, 1]), device=cpu, dtype=torch.bool, is_shared=False),
                observation: Tensor(shape=torch.Size([3, 17]), device=cpu, dtype=torch.float64, is_shared=False),
                reward: Tensor(shape=torch.Size([3, 1]), device=cpu, dtype=torch.float32, is_shared=False),
                reward_ctrl: Tensor(shape=torch.Size([3]), device=cpu, dtype=torch.float64, is_shared=False),
                reward_run: Tensor(shape=torch.Size([3]), device=cpu, dtype=torch.float64, 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),
                x_position: Tensor(shape=torch.Size([3]), device=cpu, dtype=torch.float64, is_shared=False),
                x_velocity: Tensor(shape=torch.Size([3]), device=cpu, dtype=torch.float64, is_shared=False)},
            batch_size=torch.Size([3]),
            device=cpu,
            is_shared=False),
        observation: Tensor(shape=torch.Size([3, 17]), device=cpu, dtype=torch.float64, is_shared=False),
        reward_ctrl: Tensor(shape=torch.Size([3]), device=cpu, dtype=torch.float64, is_shared=False),
        reward_run: Tensor(shape=torch.Size([3]), device=cpu, dtype=torch.float64, 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),
        x_position: Tensor(shape=torch.Size([3]), device=cpu, dtype=torch.float64, is_shared=False),
        x_velocity: Tensor(shape=torch.Size([3]), device=cpu, dtype=torch.float64, is_shared=False)},
    batch_size=torch.Size([3]),
    device=cpu,
    is_shared=False)

property batch_locked: bool

環境是否可以使用與初始化時不同的批次大小。

如果為 True，則環境需要與具有與環境相同批次大小的 tensordict 一起使用。 batch_locked 是一個不可變的屬性。

property batch_size: Size

在此環境實例中批次處理的環境數量，組織在 torch.Size() 物件中。

環境可能相似或不同，但假定它們之間幾乎沒有互動（例如，平行多任務或批次執行）。

bfloat16() → T

將所有浮點參數和緩衝區轉換為 bfloat16 資料類型。

注意

此方法會就地修改模組。

傳回:

self

傳回類型:

Module

buffers(recurse: bool = True) → Iterator[Tensor]

傳回模組緩衝區的迭代器。

參數:

recurse (bool) – 如果為 True，則產生此模組和所有子模組的緩衝區。 否則，僅產生作為此模組直接成員的緩衝區。

產生:

torch.Tensor – 模組緩衝區

範例

>>> # xdoctest: +SKIP("undefined vars")
>>> for buf in model.buffers():
>>>     print(type(buf), buf.size())
<class 'torch.Tensor'> (20L,)
<class 'torch.Tensor'> (20L, 1L, 5L, 5L)

children() → Iterator[Module]

傳回立即子模組的迭代器。

產生:

Module – 子模組

close() → None

如果可能，關閉包含的環境。

compile(*args, **kwargs)

使用 torch.compile() 編譯此模組的 forward 函數。

此模組的 __call__ 方法會被編譯，所有參數會原封不動地傳遞給 torch.compile()。

有關此函數的引數的詳細資訊，請參閱 torch.compile()。

cpu() → T

將所有模型參數和緩衝區移動到 CPU。

注意

此方法會就地修改模組。

傳回:

self

傳回類型:

Module

cuda(device: Optional[Union[int, device]] = None) → T

將所有模型參數和緩衝區移動到 GPU。

這也會使相關的參數和緩衝區成為不同的物件。 因此，如果模組在被優化時將存在於 GPU 上，則應在建構優化器之前呼叫它。

注意

此方法會就地修改模組。

參數:

device (int, optional) – 如果指定，所有參數將被複製到該裝置

傳回:

self

傳回類型:

Module

property done_key

環境的 done 鍵。

預設情況下，這將是 “done”。

如果環境中有多個 done 鍵，此函數將引發例外。

property done_keys: List[NestedKey]

環境的 done 鍵。

預設情況下，只會有一個名為 “done” 的鍵。

金鑰依資料樹中的深度排序。

property done_keys_groups

一個 done 鍵的列表，與重置鍵分組。

這是一個列表的列表。 外部列表的長度為重置鍵的長度，內部列表包含 done 鍵（例如，done 和 truncated），這些 done 鍵可以讀取以在缺少重置鍵時確定重置。

property done_spec: TensorSpec

done 的 spec。

done_spec 總是儲存為複合 spec。

如果 done spec 是以簡單 spec 的形式提供，這將會被返回。

>>> env.done_spec = Categorical(2, dtype=torch.bool)
>>> env.done_spec
Categorical(
    shape=torch.Size([]),
    space=DiscreteBox(n=2),
    device=cpu,
    dtype=torch.bool,
    domain=discrete)

如果 done spec 是以複合 spec 的形式提供，並且只包含一個葉子，這個函數將只返回這個葉子。

>>> env.done_spec = Composite({"nested": {"done": Categorical(2, dtype=torch.bool)}})
>>> env.done_spec
Categorical(
    shape=torch.Size([]),
    space=DiscreteBox(n=2),
    device=cpu,
    dtype=torch.bool,
    domain=discrete)

如果 done spec 是以複合 spec 的形式提供，並且有多個葉子，這個函數將返回整個 spec。

>>> env.done_spec = Composite({"nested": {"done": Categorical(2, dtype=torch.bool), "another_done": Categorical(2, dtype=torch.bool)}})
>>> env.done_spec
Composite(
    nested: Composite(
        done: Categorical(
            shape=torch.Size([]),
            space=DiscreteBox(n=2),
            device=cpu,
            dtype=torch.bool,
            domain=discrete),
        another_done: Categorical(
            shape=torch.Size([]),
            space=DiscreteBox(n=2),
            device=cpu,
            dtype=torch.bool,
            domain=discrete), device=cpu, shape=torch.Size([])), device=cpu, shape=torch.Size([]))

要始終檢索傳遞的完整 spec，請使用

>>> env.output_spec["full_done_spec"]

此屬性是可變的。

範例

>>> from torchrl.envs.libs.gym import GymEnv
>>> env = GymEnv("Pendulum-v1")
>>> env.done_spec
Categorical(
    shape=torch.Size([1]),
    space=DiscreteBox(n=2),
    device=cpu,
    dtype=torch.bool,
    domain=discrete)

double() → T

將所有浮點參數和緩衝區轉換為 double 資料類型。

注意

此方法會就地修改模組。

傳回:

self

傳回類型:

Module

empty_cache()

清除所有快取的值。

對於常規的 envs，鍵列表（reward、done 等）被快取，但在某些情況下，它們可能會在程式碼執行期間發生變化（例如，當添加一個 transform 時）。

eval() → T

將模組設定為評估模式。

這僅對某些模組有效。有關訓練/評估模式下特定模組行為的詳細資訊，請參閱特定模組的文件（如果它們受到影響），例如 Dropout、BatchNorm 等。

這等同於 self.train(False)。

請參閱 局部禁用梯度計算，以比較 .eval() 與幾個可能與之混淆的類似機制。

傳回:

self

傳回類型:

Module

extra_repr() → str

設定模組的額外表示法。

要列印自訂的額外資訊，您應該在自己的模組中重新實作此方法。單行和多行字串皆可接受。

fake_tensordict() → TensorDictBase

返回一個虛擬 tensordict，其鍵值對在形狀、裝置和資料類型上與環境 rollout 期間預期的相符。

float() → T

將所有浮點參數和緩衝區轉換為 float 資料類型。

注意

此方法會就地修改模組。

傳回:

self

傳回類型:

Module

forward(tensordict: TensorDictBase) → TensorDictBase

定義每次呼叫時執行的計算。

應該被所有子類別覆寫。

注意

雖然 forward pass 的配方需要在這個函數中定義，但之後應該呼叫 Module 實例，而不是這個函數，因為前者會處理執行已註冊的 hooks，而後者會靜默地忽略它們。

property full_action_spec: Composite

完整的動作規格。

full_action_spec 是一個 Composite` 實例，包含所有動作條目。

範例

>>> from torchrl.envs import BraxEnv
>>> for envname in BraxEnv.available_envs:
...     break
>>> env = BraxEnv(envname)
>>> env.full_action_spec

Composite(

action: BoundedContinuous(

shape=torch.Size([8]), space=ContinuousBox(

low=Tensor(shape=torch.Size([8]), device=cpu, dtype=torch.float32, contiguous=True), high=Tensor(shape=torch.Size([8]), device=cpu, dtype=torch.float32, contiguous=True)),

device=cpu, dtype=torch.float32, domain=continuous), device=cpu, shape=torch.Size([]))

property full_done_spec: Composite

完整的完成規格。

full_done_spec 是一個 Composite` 實例，包含所有完成條目。它可用於生成虛擬資料，其結構模仿在運行時獲得的結構。

範例

>>> import gymnasium
>>> from torchrl.envs import GymWrapper
>>> env = GymWrapper(gymnasium.make("Pendulum-v1"))
>>> env.full_done_spec
Composite(
    done: Categorical(
        shape=torch.Size([1]),
        space=DiscreteBox(n=2),
        device=cpu,
        dtype=torch.bool,
        domain=discrete),
    truncated: Categorical(
        shape=torch.Size([1]),
        space=DiscreteBox(n=2),
        device=cpu,
        dtype=torch.bool,
        domain=discrete), device=cpu, shape=torch.Size([]))

property full_reward_spec: Composite

完整的獎勵規格。

full_reward_spec 是一個 Composite` 實例，包含所有獎勵條目。

範例

>>> import gymnasium
>>> from torchrl.envs import GymWrapper, TransformedEnv, RenameTransform
>>> base_env = GymWrapper(gymnasium.make("Pendulum-v1"))
>>> env = TransformedEnv(base_env, RenameTransform("reward", ("nested", "reward")))
>>> env.full_reward_spec
Composite(
    nested: Composite(
        reward: UnboundedContinuous(
            shape=torch.Size([1]),
            space=ContinuousBox(
                low=Tensor(shape=torch.Size([]), device=cpu, dtype=torch.float32, contiguous=True),
                high=Tensor(shape=torch.Size([]), device=cpu, dtype=torch.float32, contiguous=True)),
            device=cpu,
            dtype=torch.float32,
            domain=continuous), device=None, shape=torch.Size([])), device=cpu, shape=torch.Size([]))

property full_state_spec: Composite

完整的狀態規格。

full_state_spec 是一個 Composite` 實例，包含所有狀態條目（即，不是動作的輸入資料）。

範例

>>> from torchrl.envs import BraxEnv
>>> for envname in BraxEnv.available_envs:
...     break
>>> env = BraxEnv(envname)
>>> env.full_state_spec
Composite(
    state: Composite(
        pipeline_state: Composite(
            q: UnboundedContinuous(
                shape=torch.Size([15]),
                space=None,
                device=cpu,
                dtype=torch.float32,
                domain=continuous),
    [...], device=cpu, shape=torch.Size([])), device=cpu, shape=torch.Size([])), device=cpu, shape=torch.Size([]))

get_buffer(target: str) → Tensor

如果 target 存在，則返回它指定的緩衝區，否則拋出錯誤。

有關此方法的功能以及如何正確指定 target 的更詳細說明，請參閱 get_submodule 的 docstring。

參數:

target – 要尋找的緩衝區的完整字串名稱。（請參閱 get_submodule 以了解如何指定完整字串。）

傳回:

由 target 引用的緩衝區

傳回類型:

torch.Tensor

Raises:

AttributeError – 如果目標字串引用了無效的路徑或解析為非緩衝區的內容

get_extra_state() → Any

返回要包含在模組 state_dict 中的任何額外狀態。

如果您需要儲存額外狀態，請為您的模組實作此方法和相應的 set_extra_state()。在建構模組的 state_dict() 時，會呼叫此函數。

請注意，額外的狀態應該是可序列化的，以確保 state_dict 的序列化能正常運作。我們只保證 Tensors 序列化的向後相容性；如果其他物件的序列化 pickled 形式發生變化，則可能會破壞向後相容性。

傳回:

任何要儲存在模組 state_dict 中的額外狀態

傳回類型:

物件 (object)

get_parameter(target: str) → Parameter

如果存在，則返回由 target 指定的參數，否則拋出錯誤。

有關此方法的功能以及如何正確指定 target 的更詳細說明，請參閱 get_submodule 的 docstring。

參數:

target – 要查找的 Parameter 的完整字串名稱。（有關如何指定完整字串，請參閱 get_submodule。）

傳回:

由 target 引用的 Parameter

傳回類型:

torch.nn.Parameter

Raises:

AttributeError – 如果目標字串引用了無效路徑或解析為非 nn.Parameter 的內容

get_submodule(target: str) → Module

如果存在，則返回由 target 指定的子模組，否則拋出錯誤。

例如，假設您有一個 nn.Module A，如下所示

A(
    (net_b): Module(
        (net_c): Module(
            (conv): Conv2d(16, 33, kernel_size=(3, 3), stride=(2, 2))
        )
        (linear): Linear(in_features=100, out_features=200, bias=True)
    )
)

（該圖顯示了一個 nn.Module A。A 有一個巢狀子模組 net_b，它本身有兩個子模組 net_c 和 linear。net_c 然後有一個子模組 conv。）

要檢查我們是否具有 linear 子模組，我們將調用 get_submodule("net_b.linear")。要檢查我們是否具有 conv 子模組，我們將調用 get_submodule("net_b.net_c.conv")。

get_submodule 的運行時間受 target 中模組巢狀結構的程度限制。對 named_modules 的查詢可以達到相同的結果，但它是 O(N)，其中 N 是傳遞模組的數量。因此，對於檢查某些子模組是否存在的簡單檢查，應始終使用 get_submodule。

參數:

target – 要查找的子模組的完整字串名稱。（有關如何指定完整字串，請參閱上面的範例。）

傳回:

由 target 引用的子模組

傳回類型:

torch.nn.Module

Raises:

AttributeError – 如果目標字串引用了無效路徑或解析為非 nn.Module 的內容

half() → T

將所有浮點參數和緩衝區轉換為 half 資料類型。

注意

此方法會就地修改模組。

傳回:

self

傳回類型:

Module

property input_spec: TensorSpec

輸入規格 (Input spec)。

包含環境資料輸入的所有規格的複合規格。

它包含

• “full_action_spec”：輸入動作的規格

• “full_state_spec”：所有其他環境輸入的規格

此屬性已鎖定，應為唯讀。相反，要設定其中包含的規格，請使用相應的屬性。

範例

>>> from torchrl.envs.libs.gym import GymEnv
>>> env = GymEnv("Pendulum-v1")
>>> env.input_spec
Composite(
    full_state_spec: None,
    full_action_spec: Composite(
        action: BoundedContinuous(
            shape=torch.Size([1]),
            space=ContinuousBox(
                low=Tensor(shape=torch.Size([1]), device=cpu, dtype=torch.float32, contiguous=True),
                high=Tensor(shape=torch.Size([1]), device=cpu, dtype=torch.float32, contiguous=True)),
            device=cpu,
            dtype=torch.float32,
            domain=continuous), device=cpu, shape=torch.Size([])), device=cpu, shape=torch.Size([]))

ipu(device: Optional[Union[int, device]] = None) → T

將所有模型參數和緩衝區移動到 IPU。

這也會使相關的參數和緩衝區成為不同的物件。因此，如果模組將在 IPU 上運行並進行最佳化，則應在建構最佳化器之前呼叫它。

注意

此方法會就地修改模組。

參數:

device (int, optional) – 如果指定，所有參數將被複製到該裝置

傳回:

self

傳回類型:

Module

load_state_dict(state_dict: Mapping[str, Any], strict: bool = True, assign: bool = False)

將 state_dict 中的參數和緩衝區複製到此模組及其子模組中。

如果 strict 為 True，則 state_dict 的鍵必須與此模組的 state_dict() 函數傳回的鍵完全匹配。

警告

如果 assign 為 True，則必須在呼叫 load_state_dict 之後建立最佳化器，除非 get_swap_module_params_on_conversion() 為 True。

參數:

• state_dict (dict) – 包含參數和持久緩衝區的字典。

• strict (bool, optional) – 是否嚴格強制要求 state_dict 中的鍵與此模組的 state_dict() 函數傳回的鍵匹配。預設值：True

• assign (bool, optional) – 當 False 時，保留目前模組中 tensors 的屬性，而當 True 時，保留 state dict 中 Tensors 的屬性。唯一的例外是 requires_grad 欄位。 預設值： False

傳回:

• missing_keys 是一個 str 的列表，包含此模組預期但提供的 state_dict 中缺失的任何鍵。

  由這個模組，但在提供的 state_dict 中遺失。

• unexpected_keys 是一個 str 的列表，包含此模組不預期但在提供的 state_dict 中存在的鍵。

  預期的，但在提供的 state_dict 中存在。

傳回類型:

具有 missing_keys 和 unexpected_keys 欄位的 NamedTuple

注意

如果參數或緩衝區註冊為 None 並且其對應的鍵存在於 state_dict 中，則 load_state_dict() 將引發 RuntimeError。

maybe_reset(tensordict: TensorDictBase) → TensorDictBase

檢查輸入 tensordict 的完成鍵，如果需要，重置已完成的環境。

參數:

tensordict (TensorDictBase) – 來自 step_mdp() 輸出的 tensordict。

傳回:

一個與輸入相同的 tensordict，其中環境未重置，並且包含環境已重置的新重置資料。

modules() → Iterator[Module]

傳回網路中所有模組的疊代器。

產生:

Module – 網路中的模組

注意

重複的模組只會傳回一次。在以下範例中，l 只會傳回一次。

範例

>>> l = nn.Linear(2, 2)
>>> net = nn.Sequential(l, l)
>>> for idx, m in enumerate(net.modules()):
...     print(idx, '->', m)

0 -> Sequential(
  (0): Linear(in_features=2, out_features=2, bias=True)
  (1): Linear(in_features=2, out_features=2, bias=True)
)
1 -> Linear(in_features=2, out_features=2, bias=True)

mtia(device: Optional[Union[int, device]] = None) → T

將所有模型參數和緩衝區移動到 MTIA。

這也會使相關的參數和緩衝區成為不同的物件。因此，如果模組將在 MTIA 上運行並進行優化，則應在建構優化器之前呼叫此函式。

注意

此方法會就地修改模組。

參數:

device (int, optional) – 如果指定，所有參數將被複製到該裝置

傳回:

self

傳回類型:

Module

named_buffers(prefix: str = '', recurse: bool = True, remove_duplicate: bool = True) → Iterator[Tuple[str, Tensor]]

傳回模組緩衝區的迭代器，產生緩衝區的名稱和緩衝區本身。

參數:

• prefix (str) – 要加在所有緩衝區名稱前面的前綴。

• recurse (bool, optional) – 如果為 True，則產生此模組和所有子模組的緩衝區。 否則，僅產生作為此模組直接成員的緩衝區。 預設值為 True。

• remove_duplicate (bool, optional) – 是否刪除結果中的重複緩衝區。 預設值為 True。

產生:

(str, torch.Tensor) – 包含名稱和緩衝區的 Tuple

範例

>>> # xdoctest: +SKIP("undefined vars")
>>> for name, buf in self.named_buffers():
>>>     if name in ['running_var']:
>>>         print(buf.size())

named_children() → Iterator[Tuple[str, Module]]

傳回直接子模組的迭代器，產生模組的名稱和模組本身。

產生:

(str, Module) – 包含名稱和子模組的 Tuple

範例

>>> # xdoctest: +SKIP("undefined vars")
>>> for name, module in model.named_children():
>>>     if name in ['conv4', 'conv5']:
>>>         print(module)

named_modules(memo: Optional[Set[Module]] = None, prefix: str = '', remove_duplicate: bool = True)

傳回網路中所有模組的迭代器，產生模組的名稱和模組本身。

參數:

• memo – 用於儲存已新增到結果中的模組集合的 memo

• prefix – 將新增到模組名稱的前綴

• remove_duplicate – 是否刪除結果中重複的模組實例

產生:

(str, Module) – 名稱和模組的 Tuple

注意

重複的模組只會傳回一次。在以下範例中，l 只會傳回一次。

範例

>>> l = nn.Linear(2, 2)
>>> net = nn.Sequential(l, l)
>>> for idx, m in enumerate(net.named_modules()):
...     print(idx, '->', m)

0 -> ('', Sequential(
  (0): Linear(in_features=2, out_features=2, bias=True)
  (1): Linear(in_features=2, out_features=2, bias=True)
))
1 -> ('0', Linear(in_features=2, out_features=2, bias=True))

named_parameters(prefix: str = '', recurse: bool = True, remove_duplicate: bool = True) → Iterator[Tuple[str, Parameter]]

傳回模組參數的迭代器，產生參數的名稱和參數本身。

參數:

• prefix (str) – 要加在所有參數名稱前面的前綴。

• recurse (bool) – 如果為 True，則產生此模組和所有子模組的參數。 否則，僅產生作為此模組直接成員的參數。

• remove_duplicate (bool, optional) – 是否刪除結果中的重複參數。 預設值為 True。

產生:

(str, Parameter) – 包含名稱和參數的 Tuple

範例

>>> # xdoctest: +SKIP("undefined vars")
>>> for name, param in self.named_parameters():
>>>     if name in ['bias']:
>>>         print(param.size())

property observation_spec: Composite

Observation spec (觀測規格)。

必須是一個 torchrl.data.Composite 實例。在重置 (reset) 和步進 (step) 後，可以立即存取規格中列出的鍵 (keys)。

在 TorchRL 中，即使它們嚴格來說並非「觀測 (observations)」，所有來自環境的資訊 (info)、狀態 (states)、轉換 (transforms) 的結果等輸出，都會儲存在 observation_spec 中。

因此，"observation_spec" 應該被視為環境輸出的一般資料容器，這些輸出不是 done 或 reward 資料。

範例

>>> from torchrl.envs.libs.gym import GymEnv
>>> env = GymEnv("Pendulum-v1")
>>> env.observation_spec
Composite(
    observation: BoundedContinuous(
        shape=torch.Size([3]),
        space=ContinuousBox(
            low=Tensor(shape=torch.Size([3]), device=cpu, dtype=torch.float32, contiguous=True),
            high=Tensor(shape=torch.Size([3]), device=cpu, dtype=torch.float32, contiguous=True)),
        device=cpu,
        dtype=torch.float32,
        domain=continuous), device=cpu, shape=torch.Size([]))

property output_spec: TensorSpec

Output spec (輸出規格)。

包含環境輸出資料所有規格的 composite spec。

它包含

• “full_reward_spec”: reward 的規格

• “full_done_spec”: done 的規格

• “full_observation_spec”: 所有其他環境輸出的規格

此屬性已鎖定，應為唯讀。相反，要設定其中包含的規格，請使用相應的屬性。

範例

>>> from torchrl.envs.libs.gym import GymEnv
>>> env = GymEnv("Pendulum-v1")
>>> env.output_spec
Composite(
    full_reward_spec: Composite(
        reward: UnboundedContinuous(
            shape=torch.Size([1]),
            space=None,
            device=cpu,
            dtype=torch.float32,
            domain=continuous), device=cpu, shape=torch.Size([])),
    full_observation_spec: Composite(
        observation: BoundedContinuous(
            shape=torch.Size([3]),
            space=ContinuousBox(
                low=Tensor(shape=torch.Size([3]), device=cpu, dtype=torch.float32, contiguous=True),
                high=Tensor(shape=torch.Size([3]), device=cpu, dtype=torch.float32, contiguous=True)),
            device=cpu,
            dtype=torch.float32,
            domain=continuous), device=cpu, shape=torch.Size([])),
    full_done_spec: Composite(
        done: Categorical(
            shape=torch.Size([1]),
            space=DiscreteBox(n=2),
            device=cpu,
            dtype=torch.bool,
            domain=discrete), device=cpu, shape=torch.Size([])), device=cpu, shape=torch.Size([]))

parameters(recurse: bool = True) → Iterator[Parameter]

Return an iterator over module parameters. (回傳模組參數的迭代器。)

This is typically passed to an optimizer. (這通常會傳遞給最佳化器。)

參數:

recurse (bool) – 如果為 True，則產生此模組和所有子模組的參數。 否則，僅產生作為此模組直接成員的參數。

產生:

Parameter – module parameter (模組參數)

範例

>>> # xdoctest: +SKIP("undefined vars")
>>> for param in model.parameters():
>>>     print(type(param), param.size())
<class 'torch.Tensor'> (20L,)
<class 'torch.Tensor'> (20L, 1L, 5L, 5L)

rand_action(tensordict: Optional[TensorDictBase] = None)

Performs a random action given the action_spec attribute. (根據 action_spec 屬性執行隨機動作。)

參數:

tensordict (TensorDictBase, optional) – tensordict where the resulting action should be written. (應寫入結果動作的 tensordict，可選。)

傳回:

a tensordict object with the “action” entry updated with a random sample from the action-spec. (一個 tensordict 物件，其中的 “action” 條目已使用來自 action-spec 的隨機樣本更新。)

rand_step(tensordict: Optional[TensorDictBase] = None) → TensorDictBase

Performs a random step in the environment given the action_spec attribute. (根據 action_spec 屬性在環境中執行隨機步進。)

參數:

tensordict (TensorDictBase, optional) – tensordict where the resulting info should be written. (應寫入結果資訊的 tensordict，可選。)

傳回:

a tensordict object with the new observation after a random step in the environment. The action will be stored with the “action” key. (一個 tensordict 物件，其中包含在環境中隨機步進後的新觀測。該動作將使用 “action” 鍵儲存。)

read_action(action)

Reads the action obtained from the input TensorDict and transforms it in the format expected by the contained environment. (讀取從輸入 TensorDict 取得的動作，並將其轉換為包含的環境所期望的格式。)

參數:

action (Tensor or TensorDict) – an action to be taken in the environment (要在環境中採取的動作)

Returns: an action in a format compatible with the contained environment. (回傳：與包含的環境相容的動作格式。)

read_done(terminated: Optional[bool] = None, truncated: Optional[bool] = None, done: Optional[bool] = None) → Tuple[bool | numpy.ndarray, bool | numpy.ndarray, bool | numpy.ndarray, bool]

Done state reader. (Done 狀態讀取器。)

In torchrl, a “done” signal means that a trajectory has reach its end, either because it has been interrupted or because it is terminated. Truncated means the episode has been interrupted early. Terminated means the task is finished, the episode is completed. (在 torchrl 中，“done” 訊號表示軌跡已到達其終點，可能是因為它已被中斷或已終止。 Truncated 表示 episode 已提早中斷。 Terminated 表示任務已完成，episode 已完成。)

參數:

• terminated (np.ndarray, boolean or other format) – completion state obtained from the environment. "terminated" equates to "termination" in gymnasium: the signal that the environment has reached the end of the episode, any data coming after this should be considered as nonsensical. Defaults to None. (從環境取得的完成狀態。"terminated" 等同於 gymnasium 中的 "termination"：環境已達到 episode 終點的訊號，此後的所有資料都應視為無意義。 預設值為 None。)

• truncated (bool 或 None) – 提早截斷的訊號。預設值為 None。

• done (bool 或 None) – 軌跡結束的訊號。這應該是環境的備用值，如果環境沒有明確指定 "done" 指向 "terminated" 還是 "truncated" 的話。預設值為 None。

Returns: 一個包含 4 個布林/張量值的元組，

• 一個終止狀態，

• 一個截斷狀態，

• 一個完成狀態，

• 一個布林值，指示是否應該中斷 frame_skip 迴圈。

read_obs(observations: Union[Dict[str, Any], Tensor, ndarray]) → Dict[str, Any]

從環境讀取觀測值，並回傳與輸出 TensorDict 相容的觀測值。

參數:

observations (observation 以內部環境規定的格式呈現) – 要讀取的觀測值。

read_reward(reward)

讀取獎勵並將其映射到獎勵空間。

參數:

reward (torch.Tensor 或 TensorDict) – 要映射的獎勵。

register_backward_hook(hook: Callable[[Module, Union[Tuple[Tensor, ...], Tensor], Union[Tuple[Tensor, ...], Tensor]], Union[None, Tuple[Tensor, ...], Tensor]]) → RemovableHandle

在模組上註冊一個向後鉤子。

此函式已被棄用，建議使用 register_full_backward_hook()，並且此函式的行為將在未來的版本中更改。

傳回:

一個控制代碼，可用於通過呼叫 handle.remove() 來移除添加的鉤子。

傳回類型:

torch.utils.hooks.RemovableHandle

register_buffer(name: str, tensor: Optional[Tensor], persistent: bool = True) → None

將緩衝區添加到模組。

這通常用於註冊不應被視為模型參數的緩衝區。例如，BatchNorm 的 running_mean 不是參數，但卻是模組狀態的一部分。緩衝區預設是持久性的，並會與參數一起儲存。此行為可以透過將 persistent 設定為 False 來變更。持久性緩衝區與非持久性緩衝區之間的唯一區別在於，後者不會是此模組的 state_dict 的一部分。

可以使用給定的名稱作為屬性來存取緩衝區。

參數:

• name (str) – 緩衝區的名稱。可以使用給定的名稱從此模組存取緩衝區

• tensor (Tensor 或 None) – 要註冊的緩衝區。如果 None，則會忽略在緩衝區上執行的操作，例如 cuda。如果 None，則緩衝區不包含在模組的 state_dict 中。

• persistent (bool) – 緩衝區是否為此模組的 state_dict 的一部分。

範例

>>> # xdoctest: +SKIP("undefined vars")
>>> self.register_buffer('running_mean', torch.zeros(num_features))

register_forward_hook(hook: Union[Callable[[T, Tuple[Any, ...], Any], Optional[Any]], Callable[[T, Tuple[Any, ...], Dict[str, Any], Any], Optional[Any]]], *, prepend: bool = False, with_kwargs: bool = False, always_call: bool = False) → RemovableHandle

在此模組上註冊一個 forward hook。

每次在 forward() 計算出輸出後，都會呼叫此 hook。

如果 with_kwargs 為 False 或未指定，則輸入僅包含傳遞給模組的位置引數。關鍵字引數將不會傳遞給 hook，僅傳遞給 forward。 hook 可以修改輸出。它可以就地修改輸入，但由於在呼叫 forward() 之後，這不會對 forward 產生影響。 hook 應具有以下簽章

hook(module, args, output) -> None or modified output

如果 with_kwargs 為 True，則 forward hook 將會傳遞給 forward 函式的 kwargs，並預期返回可能經過修改的輸出。hook 應具有以下簽章

hook(module, args, kwargs, output) -> None or modified output

參數:

• hook (Callable) – 要註冊的使用者定義 hook。

• prepend (bool) – 如果 True，則提供的 hook 將在此 torch.nn.modules.Module 上的所有現有 forward hook 之前觸發。否則，提供的 hook 將在此 torch.nn.modules.Module 上的所有現有 forward hook 之後觸發。請注意，使用 register_module_forward_hook() 註冊的全域 forward hook 將在以這個方法註冊的所有 hook 之前觸發。預設值：False

• with_kwargs (bool) – 如果 True，則 hook 將會傳遞給 forward 函式的 kwargs。預設值：False

• always_call (bool) – 如果 True，則無論在呼叫 Module 時是否引發例外，都會執行 hook。預設值：False

傳回:

一個控制代碼，可用於通過呼叫 handle.remove() 來移除添加的鉤子。

傳回類型:

torch.utils.hooks.RemovableHandle

register_forward_pre_hook(hook: Union[Callable[[T, Tuple[Any, ...]], Optional[Any]], Callable[[T, Tuple[Any, ...], Dict[str, Any]], Optional[Tuple[Any, Dict[str, Any]]]], *, prepend: bool = False, with_kwargs: bool = False) → RemovableHandle

在模組上註冊一個前向預先hook (forward pre-hook)。

這個hook將會在每次呼叫 forward() 之前被調用。

如果 with_kwargs 為 false 或未指定，輸入僅包含傳遞給模組的位置參數。關鍵字參數將不會傳遞給 hook，而只會傳遞給 forward。這個 hook 可以修改輸入。使用者可以在 hook 中返回一個 tuple 或一個單一的修改值。如果返回的是單一值，我們會將該值包裝到一個 tuple 中 (除非該值已經是一個 tuple)。hook 應該具有以下簽名：

hook(module, args) -> None or modified input

如果 with_kwargs 為 true，前向預先 hook 將會被傳遞給 forward 函數的 kwargs。並且如果 hook 修改了輸入，args 和 kwargs 都應該被返回。hook 應該具有以下簽名：

hook(module, args, kwargs) -> None or a tuple of modified input and kwargs

參數:

• hook (Callable) – 要註冊的使用者定義 hook。

• prepend (bool) – 如果為 true，提供的 hook 將會在所有現有的 forward_pre hooks 之前觸發。否則，提供的 hook 將會在這個 torch.nn.modules.Module 上的所有現有的 forward_pre hooks 之後觸發。請注意，使用 register_module_forward_pre_hook() 註冊的全域 forward_pre hooks 將會在通過此方法註冊的所有 hooks 之前觸發。預設值： False

• with_kwargs (bool) – 如果為 true，hook 將會被傳遞給 forward 函數的 kwargs。預設值： False

傳回:

一個控制代碼，可用於通過呼叫 handle.remove() 來移除添加的鉤子。

傳回類型:

torch.utils.hooks.RemovableHandle

register_full_backward_hook(hook: Callable[[Module, Union[Tuple[Tensor, ...], Tensor], Union[Tuple[Tensor, ...], Tensor]], Union[None, Tuple[Tensor, ...], Tensor]], prepend: bool = False) → RemovableHandle

在模組上註冊一個向後鉤子。

每次計算模組的梯度時，都會呼叫此 hook，也就是說，只有在計算模組輸出的梯度時，才會執行此 hook。此 hook 應具有以下簽名：

hook(module, grad_input, grad_output) -> tuple(Tensor) or None

grad_input 和 grad_output 是包含相對於輸入和輸出的梯度的元組。 Hook 不應修改其引數，但它可以選擇返回相對於輸入的新梯度，該梯度將代替 grad_input 用於後續計算。 grad_input 將僅對應於作為位置引數給出的輸入，並且所有 kwarg 引數都將被忽略。 對於所有非 Tensor 引數，grad_input 和 grad_output 中的條目將為 None。

由於技術原因，當此 hook 應用於模組時，其 forward 函式將接收傳遞給該模組的每個 Tensor 的視圖。 同樣，呼叫者將接收由模組的 forward 函式返回的每個 Tensor 的視圖。

警告

使用反向 hook 時，不允許就地修改輸入或輸出，否則會引發錯誤。

參數:

• hook (Callable) – 要註冊的使用者定義的 hook。

• prepend (bool) – 如果為 true，則提供的 hook 將在此 torch.nn.modules.Module 上的所有現有 backward hook 之前觸發。 否則，提供的 hook 將在此 torch.nn.modules.Module 上的所有現有 backward hook 之後觸發。 請注意，使用 register_module_full_backward_hook() 註冊的全域 backward hook 將在此方法註冊的所有 hook 之前觸發。

傳回:

一個控制代碼，可用於通過呼叫 handle.remove() 來移除添加的鉤子。

傳回類型:

torch.utils.hooks.RemovableHandle

register_full_backward_pre_hook(hook: Callable[[Module, Union[Tuple[Tensor, ...], Tensor]], Union[None, Tuple[Tensor, ...], Tensor]], prepend: bool = False) → RemovableHandle

在模組上註冊一個 backward pre-hook。

每次計算模組的梯度時，都會呼叫這個 hook。這個 hook 應該具有以下簽名：

hook(module, grad_output) -> tuple[Tensor] or None

grad_output 是一個元組。這個 hook 不應該修改它的參數，但它可以選擇性地返回一個關於輸出的新梯度，該梯度將被用來代替後續計算中的 grad_output。grad_output 中的條目對於所有非 Tensor 參數將為 None。

由於技術原因，當此 hook 應用於模組時，其 forward 函式將接收傳遞給該模組的每個 Tensor 的視圖。 同樣，呼叫者將接收由模組的 forward 函式返回的每個 Tensor 的視圖。

警告

使用 backward hooks 時，不允許就地修改輸入，否則會引發錯誤。

參數:

• hook (Callable) – 要註冊的使用者定義的 hook。

• prepend (bool) – 如果為 true，提供的 hook 將在這個 torch.nn.modules.Module 上所有現有的 backward_pre hooks 之前觸發。否則，提供的 hook 將在這個 torch.nn.modules.Module 上所有現有的 backward_pre hooks 之後觸發。 請注意，使用 register_module_full_backward_pre_hook() 註冊的全域 backward_pre hooks 將在通過此方法註冊的所有 hooks 之前觸發。

傳回:

一個控制代碼，可用於通過呼叫 handle.remove() 來移除添加的鉤子。

傳回類型:

torch.utils.hooks.RemovableHandle

classmethod register_gym(id: str, *, entry_point: Callable | None = None, transform: 'Transform' | None = None, info_keys: List[NestedKey] | None = None, backend: str = None, to_numpy: bool = False, reward_threshold: float | None = None, nondeterministic: bool = False, max_episode_steps: int | None = None, order_enforce: bool = True, autoreset: bool = False, disable_env_checker: bool = False, apply_api_compatibility: bool = False, **kwargs)

在 gym(nasium) 中註冊一個環境。

這個方法的設計考慮了以下範圍：

• 將 TorchRL-first 環境整合到使用 Gym 的框架中；

• 將另一個環境（例如，DeepMind Control、Brax、Jumanji 等）整合到使用 Gym 的框架中。

參數:

id (str) – 環境的名稱。應遵循 gym 命名慣例。

關鍵字參數:

• entry_point (callable, optional) –

  建立環境的進入點。如果未傳入任何值，則將使用父類別作為進入點。通常，這用於註冊一個不一定繼承自所使用基礎的環境

  >>> from torchrl.envs import DMControlEnv
  >>> DMControlEnv.register_gym("DMC-cheetah-v0", env_name="cheetah", task="run")
  >>> # equivalently
  >>> EnvBase.register_gym("DMC-cheetah-v0", entry_point=DMControlEnv, env_name="cheetah", task="run")
  

• transform (torchrl.envs.Transform) – 與環境一起使用的轉換（或 torchrl.envs.Compose 實例中的轉換列表）。可以在呼叫 make() 時傳入此參數（請參閱以下範例）。

• info_keys (List[NestedKey], optional) –

  如果提供，這些鍵將用於建立 info 字典，並將從 observation 鍵中排除。可以在呼叫 make() 時傳入此參數（請參閱以下範例）。

  警告

  可能發生的情況是，使用 info_keys 會使 spec 為空，因為內容已移至 info 字典。Gym 不喜歡 spec 中有空的 Dict，因此應使用 RemoveEmptySpecs 移除此空內容。

• backend (str, optional) – 後端。可以是 “gym” 或 “gymnasium” 或任何其他與 set_gym_backend 相容的後端。

• to_numpy (bool, optional) – 如果 True，則對 step 和 reset 的呼叫結果將被映射到 numpy 陣列。預設為 False（結果為張量）。可以在呼叫 make() 時傳入此參數（請參閱以下範例）。

• reward_threshold (float, optional) – [Gym kwarg] 認為已學習環境的獎勵閾值。

• nondeterministic (bool, optional) – [Gym kwarg] 環境是否為非確定性的（即使知道初始種子和所有動作）。預設為 False。

• max_episode_steps (int, optional) – [Gym kwarg] 截斷之前的最大 episode 步數。由 Time Limit wrapper 使用。

• order_enforce (bool, optional) – [Gym >= 0.14] 是否應應用 order enforcer wrapper 以確保使用者以正確的順序執行函數。預設為 True。

• autoreset (bool, optional) – [Gym >= 0.14] 是否應添加 autoreset wrapper，以便不需要呼叫 reset。預設為 False。

• disable_env_checker – [Gym >= 0.14] 是否應針對環境停用環境檢查器。預設為 False。

• apply_api_compatibility – [Gym >= 0.26] 是否應用 StepAPICompatibility wrapper。預設為 False。

• **kwargs – 傳遞給環境建構子的任意關鍵字參數。

注意

TorchRL 的環境沒有 "info" 字典的概念，因為 TensorDict 提供了大多數訓練設定中認為必要的所有儲存需求。儘管如此，您可以使用 info_keys 參數來精細控制哪些被認為是觀察，哪些應該被視為資訊。

範例

>>> # Register the "cheetah" env from DMControl with the "run" task
>>> from torchrl.envs import DMControlEnv
>>> import torch
>>> DMControlEnv.register_gym("DMC-cheetah-v0", to_numpy=False, backend="gym", env_name="cheetah", task_name="run")
>>> import gym
>>> envgym = gym.make("DMC-cheetah-v0")
>>> envgym.seed(0)
>>> torch.manual_seed(0)
>>> envgym.reset()
({'position': tensor([-0.0855,  0.0215, -0.0881, -0.0412, -0.1101,  0.0080,  0.0254,  0.0424],
       dtype=torch.float64), 'velocity': tensor([ 1.9609e-02, -1.9776e-04, -1.6347e-03,  3.3842e-02,  2.5338e-02,
         3.3064e-02,  1.0381e-04,  7.6656e-05,  1.0204e-02],
       dtype=torch.float64)}, {})
>>> envgym.step(envgym.action_space.sample())
({'position': tensor([-0.0833,  0.0275, -0.0612, -0.0770, -0.1256,  0.0082,  0.0186,  0.0476],
       dtype=torch.float64), 'velocity': tensor([ 0.2221,  0.2256,  0.5930,  2.6937, -3.5865, -1.5479,  0.0187, -0.6825,
         0.5224], dtype=torch.float64)}, tensor([0.0018], dtype=torch.float64), tensor([False]), tensor([False]), {})
>>> # same environment with observation stacked
>>> from torchrl.envs import CatTensors
>>> envgym = gym.make("DMC-cheetah-v0", transform=CatTensors(in_keys=["position", "velocity"], out_key="observation"))
>>> envgym.reset()
({'observation': tensor([-0.1005,  0.0335, -0.0268,  0.0133, -0.0627,  0.0074, -0.0488, -0.0353,
        -0.0075, -0.0069,  0.0098, -0.0058,  0.0033, -0.0157, -0.0004, -0.0381,
        -0.0452], dtype=torch.float64)}, {})
>>> # same environment with numpy observations
>>> envgym = gym.make("DMC-cheetah-v0", transform=CatTensors(in_keys=["position", "velocity"], out_key="observation"), to_numpy=True)
>>> envgym.reset()
({'observation': array([-0.11355747,  0.04257728,  0.00408397,  0.04155852, -0.0389733 ,
       -0.01409826, -0.0978704 , -0.08808327,  0.03970837,  0.00535434,
       -0.02353762,  0.05116226,  0.02788907,  0.06848346,  0.05154399,
        0.0371798 ,  0.05128025])}, {})
>>> # If gymnasium is installed, we can register the environment there too.
>>> DMControlEnv.register_gym("DMC-cheetah-v0", to_numpy=False, backend="gymnasium", env_name="cheetah", task_name="run")
>>> import gymnasium
>>> envgym = gymnasium.make("DMC-cheetah-v0")
>>> envgym.seed(0)
>>> torch.manual_seed(0)
>>> envgym.reset()
({'position': tensor([-0.0855,  0.0215, -0.0881, -0.0412, -0.1101,  0.0080,  0.0254,  0.0424],
       dtype=torch.float64), 'velocity': tensor([ 1.9609e-02, -1.9776e-04, -1.6347e-03,  3.3842e-02,  2.5338e-02,
         3.3064e-02,  1.0381e-04,  7.6656e-05,  1.0204e-02],
       dtype=torch.float64)}, {})

注意

此功能也適用於無狀態環境（例如，BraxEnv）。

>>> import gymnasium
>>> import torch
>>> from tensordict import TensorDict
>>> from torchrl.envs import BraxEnv, SelectTransform
>>>
>>> # get action for dydactic purposes
>>> env = BraxEnv("ant", batch_size=[2])
>>> env.set_seed(0)
>>> torch.manual_seed(0)
>>> td = env.rollout(10)
>>>
>>> actions = td.get("action")
>>>
>>> # register env
>>> env.register_gym("Brax-Ant-v0", env_name="ant", batch_size=[2], info_keys=["state"])
>>> gym_env = gymnasium.make("Brax-Ant-v0")
>>> gym_env.seed(0)
>>> torch.manual_seed(0)
>>>
>>> gym_env.reset()
>>> obs = []
>>> for i in range(10):
...     obs, reward, terminated, truncated, info = gym_env.step(td[..., i].get("action"))

register_load_state_dict_post_hook(hook)

註冊一個在模組的 load_state_dict() 呼叫後執行的 post-hook。

它應該具有以下簽名：

hook(module, incompatible_keys) -> None

module 參數是此 hook 註冊的目前模組，而 incompatible_keys 參數是一個 NamedTuple，包含屬性 missing_keys 和 unexpected_keys。missing_keys 是一個 list 的 str，包含遺失的鍵，而 unexpected_keys 是一個 list 的 str，包含未預期的鍵。

如果需要，可以就地修改給定的 incompatible_keys。

請注意，使用 strict=True 呼叫 load_state_dict() 時執行的檢查會受到 hook 對 missing_keys 或 unexpected_keys 所做的修改的影響，正如預期的那樣。如果 strict=True，則新增到任一組鍵將導致拋出錯誤，並且清除所有遺失和未預期的鍵將避免錯誤。

傳回:

一個控制代碼，可用於通過呼叫 handle.remove() 來移除添加的鉤子。

傳回類型:

torch.utils.hooks.RemovableHandle

register_load_state_dict_pre_hook(hook)

註冊一個在模組的 load_state_dict() 呼叫前執行的 pre-hook。

它應該具有以下簽名：

hook(module, state_dict, prefix, local_metadata, strict, missing_keys, unexpected_keys, error_msgs) -> None # noqa: B950

參數:

hook (Callable) – 將在載入 state dict 之前呼叫的可呼叫 hook。

register_module(name: str, module: Optional[Module]) → None

等同於 add_module()。

register_parameter(name: str, param: Optional[Parameter]) → None

將參數新增到模組。

可以使用給定的名稱作為屬性存取參數。

參數:

• name (str) – 參數的名稱。可以使用給定的名稱從此模組存取參數

• param (Parameter 或 None) – 要新增到模組的參數。如果 None，則會忽略在參數上執行的操作，例如 cuda。如果 None，則參數不會包含在模組的 state_dict 中。

register_state_dict_post_hook(hook)

為 state_dict() 方法註冊一個後置 hook。

它應該具有以下簽名：

hook(module, state_dict, prefix, local_metadata) -> None

已註冊的 hook 可以在原地修改 state_dict。

register_state_dict_pre_hook(hook)

為 state_dict() 方法註冊一個前置 hook。

它應該具有以下簽名：

hook(module, prefix, keep_vars) -> None

已註冊的 hook 可用於在進行 state_dict 呼叫之前執行預處理。

requires_grad_(requires_grad: bool = True) → T

變更 autograd 是否應記錄此模組中參數上的操作。

此方法會就地設定參數的 requires_grad 屬性。

此方法有助於凍結部分模組以進行微調，或單獨訓練模型的部分（例如，GAN 訓練）。

請參閱 本機停用梯度計算，以比較 .requires_grad_() 和幾個可能與其混淆的類似機制。

參數:

requires_grad (bool) – autograd 是否應記錄此模組中參數上的操作。預設值： True。

傳回:

self

傳回類型:

Module

reset(tensordict: Optional[TensorDictBase] = None, **kwargs) → TensorDictBase

重設環境。

至於 step 和 _step，只有私有方法 _reset 應由 EnvBase 子類別覆寫。

參數:

• tensordict (TensorDictBase, optional) – 用於包含產生的新觀察的 tensordict。在某些情況下，此輸入也可用於將引數傳遞給 reset 函數。

• kwargs (optional) – 要傳遞給原生 reset 函數的其他引數。

傳回:

一個 tensordict（或任何輸入的 tensordict），使用產生的觀察結果就地修改。

property reset_keys: List[NestedKey]

傳回重設鍵的清單。

重置鍵是指示部分重置的鍵，用於批次、多任務或多代理設定中。它們的結構為 (*prefix, "_reset")，其中 prefix 是一個 (可能為空的) 字串元組，指向可以找到完成狀態 (done state) 的 tensordict 位置。

金鑰依資料樹中的深度排序。

property reward_key

環境的回饋 (reward) 鍵。

預設情況下，這將是 “reward”。

如果環境中有多個回饋鍵，此函式將引發例外。

property reward_keys: List[NestedKey]

環境的回饋鍵。

預設情況下，只會有一個名為 “reward” 的鍵。

金鑰依資料樹中的深度排序。

property reward_spec: TensorSpec

reward 的規格 (spec)。

reward_spec 總是儲存為複合規格 (composite spec)。

如果回饋規格以簡單規格 (simple spec) 的形式提供，則將返回此規格。

>>> env.reward_spec = Unbounded(1)
>>> env.reward_spec
UnboundedContinuous(
    shape=torch.Size([1]),
    space=ContinuousBox(
        low=Tensor(shape=torch.Size([]), device=cpu, dtype=torch.float32, contiguous=True),
        high=Tensor(shape=torch.Size([]), device=cpu, dtype=torch.float32, contiguous=True)),
    device=cpu,
    dtype=torch.float32,
    domain=continuous)

如果回饋規格以複合規格的形式提供，且僅包含一個葉節點 (leaf)，則此函式將僅返回該葉節點。

>>> env.reward_spec = Composite({"nested": {"reward": Unbounded(1)}})
>>> env.reward_spec
UnboundedContinuous(
    shape=torch.Size([1]),
    space=ContinuousBox(
        low=Tensor(shape=torch.Size([]), device=cpu, dtype=torch.float32, contiguous=True),
        high=Tensor(shape=torch.Size([]), device=cpu, dtype=torch.float32, contiguous=True)),
    device=cpu,
    dtype=torch.float32,
    domain=continuous)

如果回饋規格以複合規格的形式提供，且有多個葉節點，則此函式將返回整個規格。

>>> env.reward_spec = Composite({"nested": {"reward": Unbounded(1), "another_reward": Categorical(1)}})
>>> env.reward_spec
Composite(
    nested: Composite(
        reward: UnboundedContinuous(
            shape=torch.Size([1]),
            space=ContinuousBox(
                low=Tensor(shape=torch.Size([]), device=cpu, dtype=torch.float32, contiguous=True),
                high=Tensor(shape=torch.Size([]), device=cpu, dtype=torch.float32, contiguous=True)),
            device=cpu,
            dtype=torch.float32,
            domain=continuous),
        another_reward: Categorical(
            shape=torch.Size([]),
            space=DiscreteBox(n=1),
            device=cpu,
            dtype=torch.int64,
            domain=discrete), device=cpu, shape=torch.Size([])), device=cpu, shape=torch.Size([]))

若要擷取傳遞的完整規格，請使用

>>> env.output_spec["full_reward_spec"]

此屬性是可變的。

範例

>>> from torchrl.envs.libs.gym import GymEnv
>>> env = GymEnv("Pendulum-v1")
>>> env.reward_spec
UnboundedContinuous(
    shape=torch.Size([1]),
    space=None,
    device=cpu,
    dtype=torch.float32,
    domain=continuous)

rollout(max_steps: int, policy: Optional[Callable[[TensorDictBase], TensorDictBase]] = None, callback: Optional[Callable[[TensorDictBase, ...], Any]] = None, *, auto_reset: bool = True, auto_cast_to_device: bool = False, break_when_any_done: Optional[bool] = None, break_when_all_done: Optional[bool] = None, return_contiguous: bool = True, tensordict: Optional[TensorDictBase] = None, set_truncated: bool = False, out=None, trust_policy: bool = False)

在環境中執行 rollout。

只要其中一個包含的環境返回 done=True，函式就會停止。

參數:

• max_steps (int) – 要執行的最大步數。如果環境在執行 max_steps 之前達到完成狀態，則實際步數可能會更少。

• policy (callable, optional) – 可呼叫物件，用於計算所需的動作。如果未提供策略，則將使用 env.rand_step() 呼叫動作。該策略可以是任何可讀取 tensordict 或整個觀察條目序列的可呼叫物件，__sorted as__ the env.observation_spec.keys()。預設值為 None。

• callback (Callable[[TensorDict], Any], optional) – 一個函數，在每次迭代時會以給定的 TensorDict 呼叫。預設值為 None。 callback 的輸出將不會被收集，使用者有責任在 callback 呼叫中儲存任何結果，如果資料需要在呼叫 rollout 之外保留。

關鍵字參數:

• auto_reset (bool, optional) – 如果 True，當 rollout 開始時，如果環境處於 done 狀態，則自動重置環境。預設值為 True。

• auto_cast_to_device (bool, optional) – 如果 True，在使用 policy 之前，tensordict 的裝置會自動轉換為 policy 的裝置。預設值為 False。

• break_when_any_done (bool) – 如果任何 done 狀態為 True，則中斷。如果為 False，則對已完成的子環境呼叫 reset()。預設值為 True。

• break_when_all_done (bool) – TODO

• return_contiguous (bool) – 如果為 False，則會傳回 LazyStackedTensorDict。預設值為 True。

• tensordict (TensorDict, optional) – 如果 auto_reset 為 False，則必須提供一個初始的 tensordict。Rollout 將檢查此 tensordict 是否具有 done 標誌，並在這些維度中重置環境（如果需要）。如果 tensordict 是 reset 的輸出，則通常不應發生這種情況，但如果 tensordict 是先前 rollout 的最後一步，則可能會發生。當 auto_reset=True 時，也可以提供 tensordict，如果元資料需要傳遞給 reset 方法，例如無狀態環境的 batch-size 或裝置。

• set_truncated (bool, optional) – 如果 True，在 rollout 完成後，"truncated" 和 "done" 鍵將設定為 True。如果在 done_spec 中找不到 "truncated"，則會引發例外。Truncated 鍵可以透過 env.add_truncated_keys 設定。預設值為 False。

• trust_policy (bool, optional) – 如果 True，則會信任非 TensorDictModule policy 與收集器相容。對於 CudaGraphModules，此預設值為 True，否則為 False。

傳回:

包含結果軌跡的 TensorDict 物件。

傳回的資料將在 tensordict 的最後一個維度（在 env.ndim 索引處）標記為“time”維度名稱。

rollout 非常方便地顯示環境的資料結構。

範例

>>> # Using rollout without a policy
>>> from torchrl.envs.libs.gym import GymEnv
>>> from torchrl.envs.transforms import TransformedEnv, StepCounter
>>> env = TransformedEnv(GymEnv("Pendulum-v1"), StepCounter(max_steps=20))
>>> rollout = env.rollout(max_steps=1000)
>>> print(rollout)
TensorDict(
    fields={
        action: Tensor(shape=torch.Size([20, 1]), device=cpu, dtype=torch.float32, is_shared=False),
        done: Tensor(shape=torch.Size([20, 1]), device=cpu, dtype=torch.bool, is_shared=False),
        next: TensorDict(
            fields={
                done: Tensor(shape=torch.Size([20, 1]), device=cpu, dtype=torch.bool, is_shared=False),
                observation: Tensor(shape=torch.Size([20, 3]), device=cpu, dtype=torch.float32, is_shared=False),
                reward: Tensor(shape=torch.Size([20, 1]), device=cpu, dtype=torch.float32, is_shared=False),
                step_count: Tensor(shape=torch.Size([20, 1]), device=cpu, dtype=torch.int64, is_shared=False),
                truncated: Tensor(shape=torch.Size([20, 1]), device=cpu, dtype=torch.bool, is_shared=False)},
            batch_size=torch.Size([20]),
            device=cpu,
            is_shared=False),
        observation: Tensor(shape=torch.Size([20, 3]), device=cpu, dtype=torch.float32, is_shared=False),
        step_count: Tensor(shape=torch.Size([20, 1]), device=cpu, dtype=torch.int64, is_shared=False),
        truncated: Tensor(shape=torch.Size([20, 1]), device=cpu, dtype=torch.bool, is_shared=False)},
    batch_size=torch.Size([20]),
    device=cpu,
    is_shared=False)
>>> print(rollout.names)
['time']
>>> # with envs that contain more dimensions
>>> from torchrl.envs import SerialEnv
>>> env = SerialEnv(3, lambda: TransformedEnv(GymEnv("Pendulum-v1"), StepCounter(max_steps=20)))
>>> rollout = env.rollout(max_steps=1000)
>>> print(rollout)
TensorDict(
    fields={
        action: Tensor(shape=torch.Size([3, 20, 1]), device=cpu, dtype=torch.float32, is_shared=False),
        done: Tensor(shape=torch.Size([3, 20, 1]), device=cpu, dtype=torch.bool, is_shared=False),
        next: TensorDict(
            fields={
                done: Tensor(shape=torch.Size([3, 20, 1]), device=cpu, dtype=torch.bool, is_shared=False),
                observation: Tensor(shape=torch.Size([3, 20, 3]), device=cpu, dtype=torch.float32, is_shared=False),
                reward: Tensor(shape=torch.Size([3, 20, 1]), device=cpu, dtype=torch.float32, is_shared=False),
                step_count: Tensor(shape=torch.Size([3, 20, 1]), device=cpu, dtype=torch.int64, is_shared=False),
                truncated: Tensor(shape=torch.Size([3, 20, 1]), device=cpu, dtype=torch.bool, is_shared=False)},
            batch_size=torch.Size([3, 20]),
            device=cpu,
            is_shared=False),
        observation: Tensor(shape=torch.Size([3, 20, 3]), device=cpu, dtype=torch.float32, is_shared=False),
        step_count: Tensor(shape=torch.Size([3, 20, 1]), device=cpu, dtype=torch.int64, is_shared=False),
        truncated: Tensor(shape=torch.Size([3, 20, 1]), device=cpu, dtype=torch.bool, is_shared=False)},
    batch_size=torch.Size([3, 20]),
    device=cpu,
    is_shared=False)
>>> print(rollout.names)
[None, 'time']

使用 policy（一個常規的 Module 或一個 TensorDictModule）也很容易

範例

>>> from torch import nn
>>> env = GymEnv("CartPole-v1", categorical_action_encoding=True)
>>> class ArgMaxModule(nn.Module):
...     def forward(self, values):
...         return values.argmax(-1)
>>> n_obs = env.observation_spec["observation"].shape[-1]
>>> n_act = env.action_spec.n
>>> # A deterministic policy
>>> policy = nn.Sequential(
...     nn.Linear(n_obs, n_act),
...     ArgMaxModule())
>>> env.rollout(max_steps=10, policy=policy)
TensorDict(
    fields={
        action: Tensor(shape=torch.Size([10]), device=cpu, dtype=torch.int64, is_shared=False),
        done: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.bool, is_shared=False),
        next: TensorDict(
            fields={
                done: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.bool, is_shared=False),
                observation: Tensor(shape=torch.Size([10, 4]), device=cpu, dtype=torch.float32, is_shared=False),
                reward: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.float32, is_shared=False),
                terminated: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.bool, is_shared=False),
                truncated: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.bool, is_shared=False)},
            batch_size=torch.Size([10]),
            device=cpu,
            is_shared=False),
        observation: Tensor(shape=torch.Size([10, 4]), device=cpu, dtype=torch.float32, is_shared=False),
        terminated: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.bool, is_shared=False),
        truncated: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.bool, is_shared=False)},
    batch_size=torch.Size([10]),
    device=cpu,
    is_shared=False)
>>> # Under the hood, rollout will wrap the policy in a TensorDictModule
>>> # To speed things up we can do that ourselves
>>> from tensordict.nn import TensorDictModule
>>> policy = TensorDictModule(policy, in_keys=list(env.observation_spec.keys()), out_keys=["action"])
>>> env.rollout(max_steps=10, policy=policy)
TensorDict(
    fields={
        action: Tensor(shape=torch.Size([10]), device=cpu, dtype=torch.int64, is_shared=False),
        done: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.bool, is_shared=False),
        next: TensorDict(
            fields={
                done: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.bool, is_shared=False),
                observation: Tensor(shape=torch.Size([10, 4]), device=cpu, dtype=torch.float32, is_shared=False),
                reward: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.float32, is_shared=False),
                terminated: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.bool, is_shared=False),
                truncated: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.bool, is_shared=False)},
            batch_size=torch.Size([10]),
            device=cpu,
            is_shared=False),
        observation: Tensor(shape=torch.Size([10, 4]), device=cpu, dtype=torch.float32, is_shared=False),
        terminated: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.bool, is_shared=False),
        truncated: Tensor(shape=torch.Size([10, 1]), device=cpu, dtype=torch.bool, is_shared=False)},
    batch_size=torch.Size([10]),
    device=cpu,
    is_shared=False)

在某些情況下，無法獲得連續的 tensordict，因為它們無法堆疊。當每一步傳回的資料可能具有不同的形狀，或者當不同的環境一起執行時，可能會發生這種情況。在這種情況下，return_contiguous=False 會導致傳回的 tensordict 成為 tensordict 的延遲堆疊

非連續 rollout 的範例

>>> rollout = env.rollout(4, return_contiguous=False)
>>> print(rollout)
LazyStackedTensorDict(
    fields={
        action: Tensor(shape=torch.Size([3, 4, 1]), device=cpu, dtype=torch.float32, is_shared=False),
        done: Tensor(shape=torch.Size([3, 4, 1]), device=cpu, dtype=torch.bool, is_shared=False),
        next: LazyStackedTensorDict(
            fields={
                done: Tensor(shape=torch.Size([3, 4, 1]), device=cpu, dtype=torch.bool, is_shared=False),
                observation: Tensor(shape=torch.Size([3, 4, 3]), device=cpu, dtype=torch.float32, is_shared=False),
                reward: Tensor(shape=torch.Size([3, 4, 1]), device=cpu, dtype=torch.float32, is_shared=False),
                step_count: Tensor(shape=torch.Size([3, 4, 1]), device=cpu, dtype=torch.int64, is_shared=False),
                truncated: Tensor(shape=torch.Size([3, 4, 1]), device=cpu, dtype=torch.bool, is_shared=False)},
            batch_size=torch.Size([3, 4]),
            device=cpu,
            is_shared=False),
        observation: Tensor(shape=torch.Size([3, 4, 3]), device=cpu, dtype=torch.float32, is_shared=False),
        step_count: Tensor(shape=torch.Size([3, 4, 1]), device=cpu, dtype=torch.int64, is_shared=False),
        truncated: Tensor(shape=torch.Size([3, 4, 1]), device=cpu, dtype=torch.bool, is_shared=False)},
    batch_size=torch.Size([3, 4]),
    device=cpu,
    is_shared=False)
    >>> print(rollout.names)
    [None, 'time']

Rollout 可以在迴圈中使用，以模擬資料收集。為此，您需要在呼叫 step_mdp() 後，將前一個 rollout 的最後一個 tensordict 作為輸入傳遞。

資料收集 rollout 的範例

>>> from torchrl.envs import GymEnv, step_mdp
>>> env = GymEnv("CartPole-v1")
>>> epochs = 10
>>> input_td = env.reset()
>>> for i in range(epochs):
...     rollout_td = env.rollout(
...         max_steps=100,
...         break_when_any_done=False,
...         auto_reset=False,
...         tensordict=input_td,
...     )
...     input_td = step_mdp(
...         rollout_td[..., -1],
...     )

set_extra_state(state: Any) → None

設定載入的 state_dict 中包含的額外狀態。

此函數從 load_state_dict() 呼叫，以處理在 state_dict 中找到的任何額外狀態。如果您需要在 state_dict 中儲存額外狀態，請為您的模組實作此函數和相應的 get_extra_state()。

參數:

state (dict) – 來自 state_dict 的額外狀態

set_info_dict_reader(info_dict_reader: Optional[BaseInfoDictReader] = None, ignore_private: bool = True) → GymLikeEnv

設定 info_dict_reader 函數。

此函數應將 info_dict 字典和 step 函數傳回的 tensordict 作為輸入，並以特定方式將值從一個寫入另一個。

參數:

• info_dict_reader (Callable[[Dict], TensorDict], optional) – 一個可調用物件，接受輸入字典並輸出 TensorDict 作為引數。此函式應就地修改 tensordict。如果未提供，將使用 default_info_dict_reader。

• ignore_private (bool, optional) – 如果 True，則將忽略私有 info（以下底線開頭）。預設為 True。

Returns: 註冊了 dict_reader 的相同環境。

注意

自動註冊 info_dict reader 應透過 auto_register_info_dict() 完成，這將確保 env specs 被正確構建。

範例

>>> from torchrl.envs import default_info_dict_reader
>>> from torchrl.envs.libs.gym import GymWrapper
>>> reader = default_info_dict_reader(["my_info_key"])
>>> # assuming "some_env-v0" returns a dict with a key "my_info_key"
>>> env = GymWrapper(gym.make("some_env-v0")).set_info_dict_reader(info_dict_reader=reader)
>>> tensordict = env.reset()
>>> tensordict = env.rand_step(tensordict)
>>> assert "my_info_key" in tensordict.keys()

set_seed(seed: Optional[int] = None, static_seed: bool = False) → Optional[int]

設定環境的種子，並返回下一個要使用的種子（如果只有單一環境存在，則為輸入的種子）。

參數:

• seed (int) – 要設定的種子。種子僅在環境中本地設定。要處理全域種子，請參閱 manual_seed()。

• static_seed (bool, optional) – 如果 True，則種子不會遞增。預設為 False。

傳回:

也就是說，如果與此環境同時建立另一個環境，則該環境應使用的種子。

傳回類型:

表示「下一個種子」的整數。

set_submodule(target: str, module: Module) → None

如果存在，則設定由 target 給定的子模組，否則拋出錯誤。

例如，假設您有一個 nn.Module A，如下所示

A(
    (net_b): Module(
        (net_c): Module(
            (conv): Conv2d(16, 33, kernel_size=(3, 3), stride=(2, 2))
        )
        (linear): Linear(in_features=100, out_features=200, bias=True)
    )
)

（該圖顯示了一個 nn.Module A。A 有一個巢狀子模組 net_b，它本身有兩個子模組 net_c 和 linear。net_c 然後有一個子模組 conv。）

要使用新的子模組 Linear 覆寫 Conv2d，您可以呼叫 set_submodule("net_b.net_c.conv", nn.Linear(33, 16))。

參數:

• target – 要查找的子模組的完整字串名稱。（有關如何指定完整字串，請參閱上面的範例。）

• module – 要將子模組設定為的模組。

Raises:

• ValueError – 如果目標字串為空。

• AttributeError – 如果目標字串引用了無效路徑或解析為非 nn.Module 的內容

property shape

等同於 batch_size。

share_memory() → T

請參閱 torch.Tensor.share_memory_()。

property specs: Composite

返回一個 Composite 容器，其中包含所有環境。

此功能允許建立環境，在單一資料容器中檢索所有規格，然後從工作空間中刪除環境。

state_dict(*args, destination=None, prefix='', keep_vars=False)

返回一個字典，其中包含對模組整個狀態的引用。

包括參數和持久緩衝區（例如，運行平均值）。鍵是對應的參數和緩衝區名稱。設定為 None 的參數和緩衝區不包括在內。

注意

返回的物件是淺拷貝。它包含對模組的參數和緩衝區的引用。

警告

目前，state_dict() 也接受 destination、prefix 和 keep_vars 的位置引數。但是，這已被棄用，並且將在未來的版本中強制執行關鍵字引數。

警告

請避免使用引數 destination，因為它不是為最終使用者設計的。

參數:

• destination (dict, optional) – 如果提供，模組的狀態將更新到 dict 中，並返回相同的物件。否則，將建立並返回 OrderedDict。預設值：None。

• prefix (str, optional) – 新增到參數和緩衝區名稱的前綴，以組成 state_dict 中的鍵。預設值：''。

• keep_vars (bool, optional) – 預設情況下，state dict 中返回的 Tensor 與 autograd 分離。如果設定為 True，則不會執行分離。預設值：False。

傳回:

一個包含模組整個狀態的字典。

傳回類型:

dict

範例

>>> # xdoctest: +SKIP("undefined vars")
>>> module.state_dict().keys()
['bias', 'weight']

property state_keys: List[NestedKey]

環境的狀態鍵。

預設情況下，只會有一個名為「state」的鍵。

金鑰依資料樹中的深度排序。

property state_spec: Composite

狀態規格 (State spec)。

必須是 torchrl.data.Composite 的實例。此處列出的鍵應與動作一起作為環境的輸入提供。

在 TorchRL 中，即使它們嚴格來說不是「狀態 (state)」，所有不是動作的環境輸入都會儲存在 state_spec 中。

因此，"state_spec" 應該被視為環境輸入（非動作資料）的通用資料容器。

範例

>>> from torchrl.envs import BraxEnv
>>> for envname in BraxEnv.available_envs:
...     break
>>> env = BraxEnv(envname)
>>> env.state_spec
Composite(
    state: Composite(
        pipeline_state: Composite(
            q: UnboundedContinuous(
                shape=torch.Size([15]),
                space=None,
                device=cpu,
                dtype=torch.float32,
                domain=continuous),
    [...], device=cpu, shape=torch.Size([])), device=cpu, shape=torch.Size([])), device=cpu, shape=torch.Size([]))

step(tensordict: TensorDictBase) → TensorDictBase

在環境中執行一個步驟 (step)。

Step 接受一個單一參數 tensordict，通常包含一個 'action' 鍵，指示要採取的動作。Step 將呼叫一個異地 (out-place) 的私有方法 _step，此方法將由 EnvBase 子類別重新編寫。

參數:

tensordict (TensorDictBase) – 包含要採取之動作的 TensorDict。如果輸入的 tensordict 包含 "next" 條目，則其中包含的值將優先於新計算的值。這提供了一種覆蓋底層計算的機制。

傳回:

輸入的 tensordict，就地修改 (modified in place) 以包含產生的觀測結果、完成狀態和獎勵（以及其他如果需要）。

step_and_maybe_reset(tensordict: TensorDictBase) → Tuple[TensorDictBase, TensorDictBase]

在環境中執行一個步驟，並在需要時（部分）重置它。

參數:

tensordict (TensorDictBase) – step() 方法的輸入資料結構。

此方法允許輕鬆編寫非停止的 rollout 函數。

範例

>>> from torchrl.envs import ParallelEnv, GymEnv
>>> def rollout(env, n):
...     data_ = env.reset()
...     result = []
...     for i in range(n):
...         data, data_ = env.step_and_maybe_reset(data_)
...         result.append(data)
...     return torch.stack(result)
>>> env = ParallelEnv(2, lambda: GymEnv("CartPole-v1"))
>>> print(rollout(env, 2))
TensorDict(
    fields={
        done: Tensor(shape=torch.Size([2, 2, 1]), device=cpu, dtype=torch.bool, is_shared=False),
        next: TensorDict(
            fields={
                done: Tensor(shape=torch.Size([2, 2, 1]), device=cpu, dtype=torch.bool, is_shared=False),
                observation: Tensor(shape=torch.Size([2, 2, 4]), device=cpu, dtype=torch.float32, is_shared=False),
                reward: Tensor(shape=torch.Size([2, 2, 1]), device=cpu, dtype=torch.float32, is_shared=False),
                terminated: Tensor(shape=torch.Size([2, 2, 1]), device=cpu, dtype=torch.bool, is_shared=False),
                truncated: Tensor(shape=torch.Size([2, 2, 1]), device=cpu, dtype=torch.bool, is_shared=False)},
            batch_size=torch.Size([2, 2]),
            device=cpu,
            is_shared=False),
        observation: Tensor(shape=torch.Size([2, 2, 4]), device=cpu, dtype=torch.float32, is_shared=False),
        terminated: Tensor(shape=torch.Size([2, 2, 1]), device=cpu, dtype=torch.bool, is_shared=False),
        truncated: Tensor(shape=torch.Size([2, 2, 1]), device=cpu, dtype=torch.bool, is_shared=False)},
    batch_size=torch.Size([2, 2]),
    device=cpu,
    is_shared=False)

to(device: Union[device, str, int]) → EnvBase

移動和/或轉換參數和緩衝區。

可以如下呼叫：

to(device=None, dtype=None, non_blocking=False)

to(dtype, non_blocking=False)

to(tensor, non_blocking=False)

to(memory_format=torch.channels_last)

它的簽名 (signature) 類似於 torch.Tensor.to()，但僅接受浮點數 (floating point) 或複數 (complex) dtype。此外，此方法僅將浮點數或複數參數和緩衝區轉換為 dtype（如果已給定）。整數 (integral) 參數和緩衝區將被移動到 device（如果已給定），但 dtype 不變。當設定 non_blocking 時，如果可能，它會嘗試相對於主機 (host) 進行非同步轉換/移動，例如，將具有釘住記憶體 (pinned memory) 的 CPU Tensors 移動到 CUDA 裝置。

請參閱下面的範例。

注意

此方法會就地修改模組。

參數:

• device (torch.device) – 此模組中參數和緩衝區的目標裝置

• dtype (torch.dtype) – 此模組中參數和緩衝區的目標浮點數或複數 dtype

• tensor (torch.Tensor) – Tensor，其 dtype 和 device 是此模組中所有參數和緩衝區的目標 dtype 和 device

• memory_format (torch.memory_format) – 此模組中 4D 參數和緩衝區的目標記憶體格式 (僅關鍵字參數)

傳回:

self

傳回類型:

Module

範例

>>> # xdoctest: +IGNORE_WANT("non-deterministic")
>>> linear = nn.Linear(2, 2)
>>> linear.weight
Parameter containing:
tensor([[ 0.1913, -0.3420],
        [-0.5113, -0.2325]])
>>> linear.to(torch.double)
Linear(in_features=2, out_features=2, bias=True)
>>> linear.weight
Parameter containing:
tensor([[ 0.1913, -0.3420],
        [-0.5113, -0.2325]], dtype=torch.float64)
>>> # xdoctest: +REQUIRES(env:TORCH_DOCTEST_CUDA1)
>>> gpu1 = torch.device("cuda:1")
>>> linear.to(gpu1, dtype=torch.half, non_blocking=True)
Linear(in_features=2, out_features=2, bias=True)
>>> linear.weight
Parameter containing:
tensor([[ 0.1914, -0.3420],
        [-0.5112, -0.2324]], dtype=torch.float16, device='cuda:1')
>>> cpu = torch.device("cpu")
>>> linear.to(cpu)
Linear(in_features=2, out_features=2, bias=True)
>>> linear.weight
Parameter containing:
tensor([[ 0.1914, -0.3420],
        [-0.5112, -0.2324]], dtype=torch.float16)

>>> linear = nn.Linear(2, 2, bias=None).to(torch.cdouble)
>>> linear.weight
Parameter containing:
tensor([[ 0.3741+0.j,  0.2382+0.j],
        [ 0.5593+0.j, -0.4443+0.j]], dtype=torch.complex128)
>>> linear(torch.ones(3, 2, dtype=torch.cdouble))
tensor([[0.6122+0.j, 0.1150+0.j],
        [0.6122+0.j, 0.1150+0.j],
        [0.6122+0.j, 0.1150+0.j]], dtype=torch.complex128)

to_empty(*, device: Optional[Union[int, str, device]], recurse: bool = True) → T

將參數和緩衝區移動到指定的設備，而不複製儲存空間。

參數:

• device (torch.device) – 此模組中參數和緩衝區的目標設備。

• recurse (bool) – 是否應將子模組的參數和緩衝區遞迴地移動到指定的設備。

傳回:

self

傳回類型:

Module

train(mode: bool = True) → T

將模組設定為訓練模式。

這僅對某些模組有效。有關訓練/評估模式下特定模組行為的詳細資訊，請參閱特定模組的文件（如果它們受到影響），例如 Dropout、BatchNorm 等。

參數:

mode (bool) – 是否設定為訓練模式 (True) 或評估模式 (False)。預設值：True。

傳回:

self

傳回類型:

Module

type(dst_type: Union[dtype, str]) → T

將所有參數和緩衝區轉換為 dst_type。

注意

此方法會就地修改模組。

參數:

dst_type (type 或 string) – 目標類型

傳回:

self

傳回類型:

Module

xpu(device: Optional[Union[int, device]] = None) → T

將所有模型參數和緩衝區移動到 XPU。

這也會使相關的參數和緩衝區成為不同的物件。因此，如果模組將在 XPU 上運行並進行最佳化，則應在建構最佳化器之前呼叫此方法。

注意

此方法會就地修改模組。

參數:

device (int, optional) – 如果指定，所有參數將被複製到該裝置

傳回:

self

傳回類型:

Module

zero_grad(set_to_none: bool = True) → None

重置所有模型參數的梯度。

請參閱 torch.optim.Optimizer 下的類似函數以獲取更多上下文。

參數:

set_to_none (bool) – 將梯度設定為 None 而不是設定為零。請參閱 torch.optim.Optimizer.zero_grad() 以獲取詳細資訊。