MultiStepActorWrapper¶
- class torchrl.modules.tensordict_module.MultiStepActorWrapper(*args, **kwargs)[源代码]¶
围绕多动作 actor 的包装器。
此类允许在环境中执行宏。actor 动作条目必须具有额外的时序维度才能被使用。它必须放置在输入 tensordict 的最后一个维度旁边(即在
tensordict.ndim
处)。如果未使用简单的启发式方法(任何以
"action"
字符串结尾的嵌套键)提供动作条目键,则会从 actor 自动检索这些键。输入 tensordict 中还必须存在一个
"is_init"
条目,以跟踪何时以及何时应中断当前收集,因为遇到了“完成”状态。与action_keys
不同,此键必须是唯一的。- 参数:
actor (TensorDictModuleBase) – 一个 actor。
n_steps (int) – actor 同时输出的动作数量(前瞻窗口)。
- 关键字参数:
action_keys (嵌套键列表, 可选) – 环境中的动作键。可以从
env.action_keys
中检索。默认值为actor
的所有out_keys
,这些键以"action"
字符串结尾。init_key (嵌套键, 可选) – 指示环境何时经历重置的条目的键。默认值为
"is_init"
,它是InitTracker
转换的out_key
。
示例
>>> import torch.nn >>> from torchrl.modules.tensordict_module.actors import MultiStepActorWrapper, Actor >>> from torchrl.envs import CatFrames, GymEnv, TransformedEnv, SerialEnv, InitTracker, Compose >>> from tensordict.nn import TensorDictSequential as Seq, TensorDictModule as Mod >>> >>> time_steps = 6 >>> n_obs = 4 >>> n_action = 2 >>> batch = 5 >>> >>> # Transforms a CatFrames in a stack of frames >>> def reshape_cat(data: torch.Tensor): ... return data.unflatten(-1, (time_steps, n_obs)) >>> # an actor that reads `time_steps` frames and outputs one action per frame >>> # (actions are conditioned on the observation of `time_steps` in the past) >>> actor_base = Seq( ... Mod(reshape_cat, in_keys=["obs_cat"], out_keys=["obs_cat_reshape"]), ... Mod(torch.nn.Linear(n_obs, n_action), in_keys=["obs_cat_reshape"], out_keys=["action"]) ... ) >>> # Wrap the actor to dispatch the actions >>> actor = MultiStepActorWrapper(actor_base, n_steps=time_steps) >>> >>> env = TransformedEnv( ... SerialEnv(batch, lambda: GymEnv("CartPole-v1")), ... Compose( ... InitTracker(), ... CatFrames(N=time_steps, in_keys=["observation"], out_keys=["obs_cat"], dim=-1) ... ) ... ) >>> >>> print(env.rollout(100, policy=actor, break_when_any_done=False)) TensorDict( fields={ action: Tensor(shape=torch.Size([5, 100, 2]), device=cpu, dtype=torch.float32, is_shared=False), action_orig: Tensor(shape=torch.Size([5, 100, 6, 2]), device=cpu, dtype=torch.float32, is_shared=False), counter: Tensor(shape=torch.Size([5, 100, 1]), device=cpu, dtype=torch.int32, is_shared=False), done: Tensor(shape=torch.Size([5, 100, 1]), device=cpu, dtype=torch.bool, is_shared=False), is_init: Tensor(shape=torch.Size([5, 100, 1]), device=cpu, dtype=torch.bool, is_shared=False), next: TensorDict( fields={ done: Tensor(shape=torch.Size([5, 100, 1]), device=cpu, dtype=torch.bool, is_shared=False), is_init: Tensor(shape=torch.Size([5, 100, 1]), device=cpu, dtype=torch.bool, is_shared=False), obs_cat: Tensor(shape=torch.Size([5, 100, 24]), device=cpu, dtype=torch.float32, is_shared=False), observation: Tensor(shape=torch.Size([5, 100, 4]), device=cpu, dtype=torch.float32, is_shared=False), reward: Tensor(shape=torch.Size([5, 100, 1]), device=cpu, dtype=torch.float32, is_shared=False), terminated: Tensor(shape=torch.Size([5, 100, 1]), device=cpu, dtype=torch.bool, is_shared=False), truncated: Tensor(shape=torch.Size([5, 100, 1]), device=cpu, dtype=torch.bool, is_shared=False)}, batch_size=torch.Size([5, 100]), device=cpu, is_shared=False), obs_cat: Tensor(shape=torch.Size([5, 100, 24]), device=cpu, dtype=torch.float32, is_shared=False), observation: Tensor(shape=torch.Size([5, 100, 4]), device=cpu, dtype=torch.float32, is_shared=False), terminated: Tensor(shape=torch.Size([5, 100, 1]), device=cpu, dtype=torch.bool, is_shared=False), truncated: Tensor(shape=torch.Size([5, 100, 1]), device=cpu, dtype=torch.bool, is_shared=False)}, batch_size=torch.Size([5, 100]), device=cpu, is_shared=False)
- forward(tensordict: TensorDictBase) TensorDictBase [源代码]¶
定义每次调用时执行的计算。
应由所有子类覆盖。
注意
虽然需要在此函数中定义前向传递的配方,但应在之后调用
Module
实例,而不是此函数,因为前者负责运行已注册的挂钩,而后者则会静默地忽略它们。
- property init_key: NestedKey¶
批次中给定元素的初始步骤指示符。