BurnInTransform¶
- class torchrl.envs.transforms.BurnInTransform(modules: Sequence[TensorDictModuleBase], burn_in: int, out_keys: Sequence[NestedKey] | None = None)[source]¶
用於部分預熱(burn-in)資料序列的 Transform。
當無法獲得最新的迴圈狀態時,此 transform 非常有用。它從取樣到的序列資料切片中沿時間維度預熱若干步,並返回剩餘的資料序列,其中預熱的資料位於其初始時間步。此 transform 旨在用作回放緩衝區的 transform,而非環境 transform。
- 引數:
modules (TensorDictModule 的序列) – 用於預熱資料序列的模組列表。
burn_in (int) – 要預熱的時間步數。
out_keys (NestedKey 的序列, 可選) – 目標鍵。預設為
` (所有指向下一時間步的模組輸出鍵 (例如,如果一個模組的輸出鍵包含 "hidden") –
("next" –
module). ("hidden"),則預設為 "hidden") –
注意
此 transform 期望輸入的 TensorDicts 的最後一個維度是時間維度。它還假設所有提供的模組都能處理序列資料。
示例
>>> import torch >>> from tensordict import TensorDict >>> from torchrl.envs.transforms import BurnInTransform >>> from torchrl.modules import GRUModule >>> gru_module = GRUModule( ... input_size=10, ... hidden_size=10, ... in_keys=["observation", "hidden"], ... out_keys=["intermediate", ("next", "hidden")], ... default_recurrent_mode=True, ... ) >>> burn_in_transform = BurnInTransform( ... modules=[gru_module], ... burn_in=5, ... ) >>> td = TensorDict({ ... "observation": torch.randn(2, 10, 10), ... "hidden": torch.randn(2, 10, gru_module.gru.num_layers, 10), ... "is_init": torch.zeros(2, 10, 1), ... }, batch_size=[2, 10]) >>> td = burn_in_transform(td) >>> td.shape torch.Size([2, 5]) >>> td.get("hidden").abs().sum() tensor(86.3008)
>>> from torchrl.data import LazyMemmapStorage, TensorDictReplayBuffer >>> buffer = TensorDictReplayBuffer( ... storage=LazyMemmapStorage(2), ... batch_size=1, ... ) >>> buffer.append_transform(burn_in_transform) >>> td = TensorDict({ ... "observation": torch.randn(2, 10, 10), ... "hidden": torch.randn(2, 10, gru_module.gru.num_layers, 10), ... "is_init": torch.zeros(2, 10, 1), ... }, batch_size=[2, 10]) >>> buffer.extend(td) >>> td = buffer.sample(1) >>> td.shape torch.Size([1, 5]) >>> td.get("hidden").abs().sum() tensor(37.0344)