捷徑

TicTacToeEnv

class torchrl.envs.TicTacToeEnv(*args, **kwargs)[原始碼]

井字遊戲的實作。

關鍵字引數:
  • single_player (bool, optional) – 是否需要考慮一個或兩個玩家。single_player=True 表示 "player1" 正在隨機玩。如果 False (預設),則在每個回合中,兩位玩家之一必須玩。

  • device (torch.device, optional) – 將張量放置的裝置。預設為 None (預設裝置)。

環境是無狀態的。若要跨多個批次執行,請呼叫

>>> env.reset(TensorDict(batch_size=desired_batch_size))

如果存在 "mask" 條目,則 rand_action 會將其納入考量以產生下一個動作。在此 env 上執行的任何策略都應將此遮罩以及玩家的回合 (儲存在 "turn" 輸出條目中) 納入考量。

規格
Composite(
output_spec: Composite(
full_observation_spec: Composite(
board: Categorical(

shape=torch.Size([3, 3]), space=DiscreteBox(n=2), dtype=torch.int32, domain=discrete),

turn: Categorical(

shape=torch.Size([1]), space=DiscreteBox(n=2), dtype=torch.int32, domain=discrete),

mask: Categorical(

shape=torch.Size([9]), space=DiscreteBox(n=2), dtype=torch.bool, domain=discrete),

shape=torch.Size([])),

full_reward_spec: Composite(
player0: Composite(
reward: UnboundedContinuous(

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)),

dtype=torch.float32, domain=continuous),

shape=torch.Size([])),

player1: Composite(
reward: UnboundedContinuous(

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)),

dtype=torch.float32, domain=continuous),

shape=torch.Size([])),

shape=torch.Size([])),

full_done_spec: Composite(
done: Categorical(

shape=torch.Size([1]), space=DiscreteBox(n=2), dtype=torch.bool, domain=discrete),

terminated: Categorical(

shape=torch.Size([1]), space=DiscreteBox(n=2), dtype=torch.bool, domain=discrete),

truncated: Categorical(

shape=torch.Size([1]), space=DiscreteBox(n=2), dtype=torch.bool, domain=discrete),

shape=torch.Size([])),

shape=torch.Size([])),

input_spec: Composite(
full_state_spec: Composite(
board: Categorical(

shape=torch.Size([3, 3]), space=DiscreteBox(n=2), dtype=torch.int32, domain=discrete),

turn: Categorical(

shape=torch.Size([1]), space=DiscreteBox(n=2), dtype=torch.int32, domain=discrete),

mask: Categorical(

shape=torch.Size([9]), space=DiscreteBox(n=2), dtype=torch.bool, domain=discrete), shape=torch.Size([])),

full_action_spec: Composite(
action: Categorical(

shape=torch.Size([1]), space=DiscreteBox(n=9), dtype=torch.int64, domain=discrete),

shape=torch.Size([])),

shape=torch.Size([])),

shape=torch.Size([]))

若要執行虛擬 rollout,請執行下列命令

範例

>>> env = TicTacToeEnv()
>>> env.rollout(10)
TensorDict(
    fields={
        action: Tensor(shape=torch.Size([9, 1]), device=cpu, dtype=torch.int64, is_shared=False),
        board: Tensor(shape=torch.Size([9, 3, 3]), device=cpu, dtype=torch.int32, is_shared=False),
        done: Tensor(shape=torch.Size([9, 1]), device=cpu, dtype=torch.bool, is_shared=False),
        mask: Tensor(shape=torch.Size([9, 9]), device=cpu, dtype=torch.bool, is_shared=False),
        next: TensorDict(
            fields={
                board: Tensor(shape=torch.Size([9, 3, 3]), device=cpu, dtype=torch.int32, is_shared=False),
                done: Tensor(shape=torch.Size([9, 1]), device=cpu, dtype=torch.bool, is_shared=False),
                mask: Tensor(shape=torch.Size([9, 9]), device=cpu, dtype=torch.bool, is_shared=False),
                player0: TensorDict(
                    fields={
                        reward: Tensor(shape=torch.Size([9, 1]), device=cpu, dtype=torch.float32, is_shared=False)},
                    batch_size=torch.Size([9]),
                    device=None,
                    is_shared=False),
                player1: TensorDict(
                    fields={
                        reward: Tensor(shape=torch.Size([9, 1]), device=cpu, dtype=torch.float32, is_shared=False)},
                    batch_size=torch.Size([9]),
                    device=None,
                    is_shared=False),
                terminated: Tensor(shape=torch.Size([9, 1]), device=cpu, dtype=torch.bool, is_shared=False),
                truncated: Tensor(shape=torch.Size([9, 1]), device=cpu, dtype=torch.bool, is_shared=False),
                turn: Tensor(shape=torch.Size([9, 1]), device=cpu, dtype=torch.int32, is_shared=False)},
            batch_size=torch.Size([9]),
            device=None,
            is_shared=False),
        terminated: Tensor(shape=torch.Size([9, 1]), device=cpu, dtype=torch.bool, is_shared=False),
        truncated: Tensor(shape=torch.Size([9, 1]), device=cpu, dtype=torch.bool, is_shared=False),
        turn: Tensor(shape=torch.Size([9, 1]), device=cpu, dtype=torch.int32, is_shared=False)},
    batch_size=torch.Size([9]),
    device=None,
    is_shared=False)
property action_key: NestedKey

環境的動作鍵。

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

如果環境中有多個動作鍵,則此函式會引發例外。

property action_keys: List[NestedKey]

環境的動作鍵。

預設情況下,只會有一個名為 "action" 的鍵。

鍵會依照資料樹中的深度排序。

property action_spec: TensorSpec

action 的規格。

action_spec 總是會被儲存為 composite spec。

如果 action spec 是以簡單 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)

如果 action spec 是以 composite spec 的形式提供,且只包含一個 leaf,此函數將只回傳該 leaf。

>>> 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)

如果 action spec 是以 composite spec 的形式提供,且包含多個 leaf,此函數將回傳完整的 spec。

>>> 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([]))

要取得完整的 spec,請使用

>>> env.input_spec["full_action_spec"]

這個 property 是可變的。

範例

>>> 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 鍵添加到環境。

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

回傳一個轉換後的環境,其中會應用傳入的 callable/transform。

參數:

transform (Transform or 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() 回傳)以及自身。

典型的用法包括初始化模型的參數(另請參閱 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)
)
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 – 子模組

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 鍵的列表,分組為 reset 鍵。

這是一個列表的列表。外層列表的長度等於 reset 鍵的數量,內層列表包含 done 鍵 (例如,done 和 truncated),可以讀取這些鍵以確定缺少 reset 時的重置。

property done_spec: TensorSpec

done 的規範。

done_spec 始終儲存為複合規範。

如果 done 規範以簡單規範提供,則將傳回此規範。

>>> 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 規範以複合規範提供,並且只包含一個 leaf,則此函數將只傳回該 leaf。

>>> 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 規範以複合規範提供,並且有多個 leaf,則此函數將傳回整個規範。

>>> 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([]))

若要始終檢索傳遞的完整規範,請使用

>>> env.output_spec["full_done_spec"]

這個 property 是可變的。

範例

>>> 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()

清除所有快取的值。

對於常規 env,鍵列表 (reward、done 等) 會被快取,但在某些情況下,它們可能會在程式碼執行期間發生變化 (例如,新增轉換時)。

eval() T

將模組設定為評估模式。

這只對某些模組有任何影響。請參閱特定模組的文件,了解它們在訓練/評估模式下的行為詳情 (如果受到影響),例如 DropoutBatchNorm 等。

這等同於 self.train(False)

如需 .eval() 與可能與之混淆的幾種類似機制之間的比較,請參閱 在本機停用梯度計算

回傳:

self

回傳類型:

Module

extra_repr() str

設定模組的額外表示。

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

fake_tensordict() TensorDictBase

傳回一個假的 tensordict,其鍵值對在形狀、裝置和 dtype 方面與環境 rollout 期間預期的內容相符。

float() T

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

注意

這個方法會就地修改模組。

回傳:

self

回傳類型:

Module

forward(tensordict: TensorDictBase) TensorDictBase

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

應該由所有子類別覆寫。

注意

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

property full_action_spec: Composite

完整的 action spec。

full_action_spec 是一個 Composite` 實例,其中包含所有 action 條目。

範例

>>> 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

完整的 done spec。

full_done_spec 是一個 Composite` 實例,其中包含所有 done 條目。它可以用於生成模擬運行時獲得的結構的假數據。

範例

>>> 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

完整的 reward spec。

full_reward_spec 是一個 Composite` 實例,其中包含所有 reward 條目。

範例

>>> 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

完整的 state spec。

full_state_spec 是一個 Composite` 實例,其中包含所有 state 條目 (即,不是 action 的輸入數據)。

範例

>>> 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 給定的 buffer,否則拋出錯誤。

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

參數:

target – 要查找的 buffer 的完整字串名稱。(有關如何指定完整字串,請參閱 get_submodule。)

回傳:

target 引用的 buffer

回傳類型:

torch.Tensor

Raises:

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

get_extra_state() Any

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

如果需要儲存額外狀態,請為您的模組實現此方法和對應的 set_extra_state()。此函數在構建模組的 state_dict() 時調用。

請注意,額外狀態應該是可 pickle 的,以確保 state_dict 的工作序列化。我們僅為序列化 Tensors 提供向後兼容性保證;如果其他物件的序列化 pickle 形式發生變化,則可能會破壞向後兼容性。

回傳:

要儲存在模組的 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 AA 有一個巢狀子模組 net_b,它本身有兩個子模組 net_clinearnet_c 然後有一個子模組 conv。)

要檢查我們是否具有 linear 子模組,我們將呼叫 get_submodule("net_b.linear")。要檢查我們是否具有 conv 子模組,我們將呼叫 get_submodule("net_b.net_c.conv")

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

參數:

target – 要尋找的子模組的完整限定字串名稱。(有關如何指定完整限定字串,請參閱以上範例。)

回傳:

target 參考的子模組

回傳類型:

torch.nn.Module

Raises:

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

half() T

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

注意

這個方法會就地修改模組。

回傳:

self

回傳類型:

Module

property input_spec: TensorSpec

輸入規範。

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

它包含

  • “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 複製到此模組及其後代。

如果 strictTrue,則 state_dict 的鍵必須與此模組的 state_dict() 函數傳回的鍵完全匹配。

警告

如果 assignTrue,則必須在呼叫 load_state_dict 之後建立最佳化器,除非 get_swap_module_params_on_conversion()True

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

  • strict (bool, optional) – 是否嚴格要求 state_dict 中的鍵值 (keys) 必須與此模組的 state_dict() 函式回傳的鍵值相符。預設值:True

  • assign (bool, optional) – 當 False 時,將保留當前模組中 tensors 的屬性;而當 True 時,將保留 state dict 中 Tensors 的屬性。 唯一的例外是 requires_grad 欄位。 Default: ``False`

回傳:

  • missing_keys 是一個字串列表,其中包含預期在此模組中但遺失的鍵值。

    由這個模組所預期,但在提供的 state_dict 中卻遺失的鍵值。

  • unexpected_keys 是一個字串列表,其中包含這個模組

    所不預期,但在提供的 state_dict 中卻存在的鍵值。

回傳類型:

包含 missing_keysunexpected_keys 欄位的 NamedTuple

注意

如果一個參數或緩衝區 (buffer) 被註冊為 None,且其對應的鍵值存在於 state_dict 中,則 load_state_dict() 會引發 RuntimeError

maybe_reset(tensordict: TensorDictBase) TensorDictBase

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

參數:

tensordict (TensorDictBase) – 從 step_mdp() 輸出的 tensordict。

回傳:

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

modules() Iterator[Module]

回傳一個遍歷網路中所有模組的迭代器 (iterator)。

產生:

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 – 用於儲存已新增至結果的模組集合的備忘錄。

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

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

產生:

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

注意

重複的模組只會回傳一次。 在下面的範例中,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) – 包含名稱和參數的元組。

範例

>>> # 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 實例。 在重置和步進後,可以直接存取規格中列出的鍵。

在 TorchRL 中,即使它們不嚴格地說是「觀察」,環境的所有資訊、狀態、轉換結果等輸出都儲存在 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)。

包含來自環境之資料輸出的所有規格的複合規格。

它包含

  • “full_reward_spec”: 獎勵的規格

  • “full_done_spec”: 完成的規格

  • “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]

回傳一個遍歷模組參數的迭代器。

通常會將此傳遞給最佳化器。

參數:

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

產生:

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)[原始碼]

根據 action_spec 屬性執行隨機動作。

參數:

tensordict (TensorDictBase, optional) – 應該在其中寫入結果動作的 tensordict。

回傳:

一個 tensordict 物件,其中的 “action” 條目已使用來自 action-spec 的隨機樣本更新。

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

根據 action_spec 屬性,在環境中執行一個隨機步驟。

參數:

tensordict (TensorDictBase, optional) – 儲存結果資訊的 tensordict,為可選參數。

回傳:

一個 tensordict 物件,包含環境中隨機步驟後的新觀測值。動作將以 “action” 鍵儲存。

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

在模組上註冊一個反向 hook。

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

回傳:

一個可移除的控制代碼,可用於呼叫 handle.remove() 來移除已添加的 hook。

回傳類型:

torch.utils.hooks.RemovableHandle

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

向模組添加一個 buffer。

這通常用於註冊不應被視為模型參數的 buffer。例如,BatchNorm 的 running_mean 不是一個參數,但它是模組狀態的一部分。預設情況下,Buffers 是持久性的,將與參數一起保存。 可以通過將 persistent 設置為 False 來更改此行為。 持久性 buffer 和非持久性 buffer 之間的唯一區別是後者不會成為此模組的 state_dict 的一部分。

可以使用給定的名稱作為屬性訪問 Buffers。

參數:
  • name (str) – buffer 的名稱。 可以使用給定的名稱從此模組訪問 buffer。

  • tensor (TensorNone) – 要註冊的 buffer。 如果 None,則忽略在 buffers 上運行的操作,例如 cuda。 如果 None,則 將 buffer 包含在模組的 state_dict 中。

  • persistent (bool) – buffer 是否為此模組的 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。

這個 hook 將會在每次 forward() 計算出輸出後被呼叫。

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

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

如果 with_kwargsTrue,則 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) – 如果 Truehook 將被傳遞給 forward 函數的 kwargs。 預設值: False

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

回傳:

一個可移除的控制代碼,可用於呼叫 handle.remove() 來移除已添加的 hook。

回傳類型:

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

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

每次調用forward()之前都會呼叫該 hook。

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

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

如果 with_kwargs 為 true,則 forward pre-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 將在這個 torch.nn.modules.Module 上所有現有的 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() 來移除已添加的 hook。

回傳類型:

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 應該具有以下簽章:

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

grad_inputgrad_output 是包含關於輸入和輸出梯度的元組。這個 hook 不應該修改它的參數,但它可以選擇性地傳回一個關於輸入的新梯度,這個梯度將會取代 grad_input 用於後續的計算。grad_input 只會對應到以位置參數形式給定的輸入,所有關鍵字參數都會被忽略。grad_inputgrad_output 中的條目對於所有非 Tensor 的參數都會是 None

基於技術原因,當這個 hook 應用到一個 Module 時,它的 forward 函數將會接收到傳遞給 Module 的每個 Tensor 的一個視圖(view)。同樣地,呼叫者將會接收到 Module 的 forward 函數所傳回的每個 Tensor 的一個視圖。

警告

當使用 backward hooks 時,不允許就地 (inplace) 修改輸入或輸出,否則會引發錯誤。

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

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

回傳:

一個可移除的控制代碼,可用於呼叫 handle.remove() 來移除已添加的 hook。

回傳類型:

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 應用到一個 Module 時,它的 forward 函數將會接收到傳遞給 Module 的每個 Tensor 的一個視圖(view)。同樣地,呼叫者將會接收到 Module 的 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() 來移除已添加的 hook。

回傳類型:

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 優先的環境整合到使用 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,則對 stepreset 的呼叫結果將會對應到 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 引數來精細控制哪些內容應被視為觀察,以及哪些內容應被視為 info。

範例

>>> # 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)

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

它應該具有以下簽名:

hook(module, incompatible_keys) -> None

module 引數是此 hook 註冊到的目前模組,而 incompatible_keys 引數是一個 NamedTuple,包含屬性 missing_keysunexpected_keysmissing_keys 是一個 liststr,包含遺失的鍵,而 unexpected_keys 是一個 liststr,包含非預期的鍵。

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

請注意,使用 strict=True 呼叫 load_state_dict() 時執行的檢查會受到 hook 對 missing_keysunexpected_keys 進行的修改的影響,正如預期的那樣。 向任一組鍵添加內容將導致在 strict=True 時擲出錯誤,而清除遺失和非預期的鍵將避免錯誤。

回傳:

一個可移除的控制代碼,可用於呼叫 handle.remove() 來移除已添加的 hook。

回傳類型:

torch.utils.hooks.RemovableHandle

register_load_state_dict_pre_hook(hook)

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

它應該具有以下簽名:

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

參數:

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

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

add_module() 的別名。

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

新增一個參數到模組。

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

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

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

register_state_dict_post_hook(hook)

state_dict() 方法註冊一個 post-hook。

它應該具有以下簽名:

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

註冊的 hooks 可以就地修改 state_dict

register_state_dict_pre_hook(hook)

state_dict() 方法註冊一個 pre-hook。

它應該具有以下簽名:

hook(module, prefix, keep_vars) -> None

註冊的 hooks 可用於在呼叫 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]

傳回 reset 鍵的列表。

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

鍵會依照資料樹中的深度排序。

property reward_key

環境的獎勵鍵。

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

如果環境中有超過一個獎勵鍵,此函數將引發例外。

property reward_keys: List[NestedKey]

環境的獎勵鍵。

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

鍵會依照資料樹中的深度排序。

property reward_spec: TensorSpec

reward 規格。

reward_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)

如果獎勵規格作為複合規格提供,並且僅包含一個葉子,則此函數將僅傳回該葉子。

>>> 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([]))

要取得完整的 spec,請使用

>>> env.output_spec["full_reward_spec"]

這個 property 是可變的。

範例

>>> 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 之前達到 done 狀態,則實際步數可能會更少。

  • policy (callable, optional) – 可呼叫物件,用於計算所需的動作。如果沒有提供 policy,將會使用 env.rand_step() 呼叫動作。該 policy 可以是任何可呼叫物件,它讀取一個 tensordict 或者讀取排序過的觀察條目序列,該序列按照 env.observation_spec.keys() 排序。預設為 None。

  • callback (Callable[[TensorDict], Any], optional) – 在每次迭代中使用給定的 TensorDict 呼叫的函式。預設為 Nonecallback 的輸出將不會被收集,使用者有責任在 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) – 待辦事項

  • return_contiguous (bool) – 如果為 False,將返回 LazyStackedTensorDict。預設為 True。

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

  • 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

回傳:

包含結果 trajectory 的 TensorDict 物件。

傳回的資料將標記有 "time" 維度名稱,用於 tensordict 的最後一個維度(位於 env.ndim 索引)。

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 (常規的 ModuleTensorDictModule) 也很容易

範例

>>> 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 的 lazy stack

非連續 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_seed(seed: Optional[int] = None, static_seed: bool = False) Optional[int]

設定環境的 seed 並傳回要使用的下一個 seed(如果存在單個環境,則為輸入 seed)。

參數:
  • seed (int) – 要設定的 seed。seed 僅在環境中本機設定。若要處理全域 seed,請參閱 manual_seed()

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

回傳:

即,如果與此環境同時建立另一個環境,則應使用的 seed。

回傳類型:

表示 "下一個 seed" 的整數

set_submodule(target: str, module: Module) None

設定由 target 給定的 submodule(如果存在),否則拋出錯誤。

例如,假設你有一個 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 AA 有一個巢狀子模組 net_b,它本身有兩個子模組 net_clinearnet_c 然後有一個子模組 conv。)

若要使用新的 submodule Linear 覆寫 Conv2d,您可以呼叫 set_submodule("net_b.net_c.conv", nn.Linear(33, 16))

參數:
  • target – 要尋找的子模組的完整限定字串名稱。(有關如何指定完整限定字串,請參閱以上範例。)

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

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() 也接受位置引數,依序對應 destinationprefixkeep_vars。然而,這種做法已被棄用,未來版本將強制使用關鍵字引數。

警告

請避免使用引數 destination,因為它並非為終端使用者所設計。

參數:
  • destination (dict, optional) – 如果提供,模組的狀態將更新到此字典中,並傳回相同的物件。否則,將建立並傳回一個 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

狀態規格。

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

在 TorchRL 中,即使它們嚴格來說並非「狀態」,環境的所有輸入(非動作)都會儲存在 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 接受單一引數,tensordict,通常帶有一個 'action' 鍵,指示要採取的動作。Step 將呼叫一個原地外 (out-place) 的私有方法 _step,這是要由 EnvBase 子類別重新編寫的方法。

參數:

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

回傳:

輸入的 tensordict,就地修改,包含產生的觀測、done 狀態和獎勵(以及其他需要的)。

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)

其簽名檔與 torch.Tensor.to() 類似,但僅接受浮點數或複數 dtype。此外,此方法僅將浮點數或複數參數和緩衝區轉換為 dtype (如果已給定)。整數參數和緩衝區將被移動到 device,如果已給定,但 dtype 不會改變。當設定 non_blocking 時,它會嘗試相對於主機非同步地轉換/移動,例如,將具有釘選記憶體的 CPU 張量移動到 CUDA 裝置。

請參閱以下範例。

注意

這個方法會就地修改模組。

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

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

  • tensor (torch.Tensor) – 張量,其 dtype 和裝置是此模組中所有參數和緩衝區的目標 dtype 和裝置

  • 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

將模組設定為訓練模式。

這只對某些模組有任何影響。請參閱特定模組的文件,了解它們在訓練/評估模式下的行為詳情 (如果受到影響),例如 DropoutBatchNorm 等。

參數:

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

回傳:

self

回傳類型:

Module

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

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

注意

這個方法會就地修改模組。

參數:

dst_type (type or 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()

文件

存取 PyTorch 的完整開發人員文件

檢視文件

教學

取得為初學者和進階開發人員提供的深入教學

檢視教學

資源

尋找開發資源並獲得您的問題解答

檢視資源