RPCDataCollector¶
- 类 torchrl.collectors.distributed.RPCDataCollector(create_env_fn, policy, *, frames_per_batch: int, total_frames: int = -1, device: torch.device | List[torch.device] = None, storing_device: torch.device | List[torch.device] = None, env_device: torch.device | List[torch.device] = None, policy_device: torch.device | List[torch.device] = None, max_frames_per_traj: int = -1, init_random_frames: int = -1, reset_at_each_iter: bool = False, postproc: Callable | None = None, split_trajs: bool = False, exploration_type: ExporationType = InteractionType.RANDOM, collector_class=<class 'torchrl.collectors.collectors.SyncDataCollector'>, collector_kwargs=None, num_workers_per_collector=1, sync=False, slurm_kwargs=None, update_after_each_batch=False, max_weight_update_interval=-1, launcher='submitit', tcp_port=None, visible_devices=None, tensorpipe_options=None)[源代码]¶
一个基于 RPC 的分布式数据收集器。
支持同步和异步数据收集。
- 参数:
create_env_fn (可调用对象 或 可调用对象列表) – 可调用对象列表,每个对象返回一个
EnvBase
实例。policy (可调用对象) –
要在环境中执行的策略。必须接受
tensordict.tensordict.TensorDictBase
对象作为输入。如果提供None
,则使用的策略将是带有环境action_spec
的RandomPolicy
实例。接受的策略通常是TensorDictModuleBase
的子类。这是收集器的推荐用法。也接受其他可调用对象:如果策略不是TensorDictModuleBase
(例如,常规Module
实例),它将首先被包装在一个 nn.Module 中。然后,收集器将尝试评估这些模块是否需要包装在TensorDictModule
中。如果策略的 forward 方法签名与
forward(self, tensordict)
、forward(self, td)
或forward(self, <任意名称>: TensorDictBase)
中的任何一个匹配(或任何带有单个参数且类型为TensorDictBase
子类的类型提示),则策略将不会被包装在TensorDictModule
中。在所有其他情况下,将尝试按如下方式包装它:
TensorDictModule(policy, in_keys=env_obs_key, out_keys=env.action_keys)
。
- 关键字参数:
frames_per_batch (int) – 一个仅限关键字的参数,表示批次中的元素总数。
total_frames (int) –
一个仅限关键字的参数,表示收集器在其生命周期内返回的总帧数。如果
total_frames
不能被frames_per_batch
整除,则会引发异常。通过传入
total_frames=-1
可以创建无尽的收集器。默认为-1
(无尽的收集器)。device (int, str 或 torch.device, 可选) – 收集器的通用设备。
device
参数填充任何未指定的设备:如果device
不为None
且storing_device
、policy_device
或env_device
中的任何一个未指定,则其值将设置为device
。默认为None
(无默认设备)。支持设备列表。storing_device (int, str 或 torch.device, 可选) – 用于存储输出
TensorDict
的远程设备。如果传入了device
且storing_device
为None
,它将默认为device
指示的值。对于长轨迹,可能需要将数据存储在与执行策略和环境的设备不同的设备上。默认为None
(输出 tensordict 不在特定设备上,叶子张量位于创建它们的设备上)。支持设备列表。env_device (int, str 或 torch.device, 可选) – 环境应被转换到(或如果支持该功能则执行)的远程设备。如果未指定且环境有一个非
None
设备,env_device
将默认为该值。如果传入了device
且env_device=None
,它将默认为device
。如果如此指定的env_device
值与policy_device
不同且其中一个不为None
,数据在传递给环境之前将转换为env_device
(即,支持为策略和环境传递不同的设备)。默认为None
。支持设备列表。policy_device (int, str 或 torch.device, 可选) – 策略应被转换到的远程设备。如果传入了
device
且policy_device=None
,它将默认为device
。如果如此指定的policy_device
值与env_device
不同且其中一个不为None
,数据在传递给策略之前将转换为policy_device
(即,支持为策略和环境传递不同的设备)。默认为None
。支持设备列表。max_frames_per_traj (int, 可选) – 每个轨迹的最大步数。请注意,一个轨迹可以跨越多个批次(除非
reset_at_each_iter
设置为True
,见下文)。一旦轨迹达到n_steps
,环境将被重置。如果环境包装了多个环境,则会独立跟踪每个环境的步数。允许负值,此时将忽略此参数。默认为None
(即,无最大步数限制)。init_random_frames (int, 可选) – 在调用策略之前忽略策略的帧数。此功能主要用于离线/基于模型的设置,其中一批随机轨迹可用于初始化训练。如果提供,它将被向上取整到
frames_per_batch
的最近倍数。默认为None
(即,无随机帧)。reset_at_each_iter (bool, 可选) – 是否应在批次收集开始时重置环境。默认为
False
。postproc (可调用对象, 可选) – 一个后处理 transform,例如
Transform
或MultiStep
实例。默认为None
。split_trajs (bool, 可选) – 布尔值,指示是否应根据轨迹分割结果 TensorDict。有关更多信息,请参阅
split_trajectories()
。默认为False
。exploration_type (ExplorationType, 可选) – 收集数据时使用的交互模式。必须是
torchrl.envs.utils.ExplorationType.DETERMINISTIC
、torchrl.envs.utils.ExplorationType.RANDOM
、torchrl.envs.utils.ExplorationType.MODE
或torchrl.envs.utils.ExplorationType.MEAN
之一。默认为torchrl.envs.utils.ExplorationType.RANDOM
。collector_class (类型 或 str, 可选) –
远程节点的收集器类。可以是
SyncDataCollector
、MultiSyncDataCollector
、MultiaSyncDataCollector
或它们的派生类。字符串“single”、“sync”和“async”对应各自的类。默认为SyncDataCollector
。注意
对
MultiSyncDataCollector
和MultiaSyncDataCollector
的支持是实验性的,应始终优先使用SyncDataCollector
。如果需要在单个节点上执行多个同时环境,请考虑使用ParallelEnv
实例。collector_kwargs (dict 或 list, 可选) – 要传递给远程数据收集器的参数字典。如果提供一个列表,则每个元素将对应于指定收集器的一组独立的关键字参数。
num_workers_per_collector (int, 可选) – 在远程节点上使用的环境构造器副本数量。默认为 1(每个收集器一个环境)。在单个工作节点上,所有子工作器将执行相同的环境。如果需要执行不同的环境,应将它们分派到不同的工作节点,而不是子节点。
sync (bool, 可选) – 如果为
True
,则结果 tensordict 是在每个节点上收集的所有 tensordict 的堆叠。如果为False
(默认),每个 tensordict 来自一个单独的节点,采用“先就绪,先服务”的方式。slurm_kwargs (dict) – 要传递给 submitit 执行器的参数字典。
update_after_each_batch (bool, 可选) – 如果为
True
,则在每次收集后更新权重。对于sync=True
,这意味着所有工作器都将看到其权重被更新。对于sync=False
,只有从中收集数据的那个工作器会被更新。默认为False
,即必须通过update_policy_weights_()
手动执行更新。max_weight_update_interval (int, 可选) – 在一个工作器的策略权重更新之前,可以收集的最大批次数量。对于同步收集,此参数会被
update_after_each_batch
覆盖。对于异步收集,即使update_after_each_batch
开启,也可能存在某个工作器的参数长时间未更新的情况。默认为 -1(无强制更新)。launcher (str, 可选) – 如何启动作业。可以是“submitit”或用于多进程的“mp”之一。前者可以在多个节点上启动作业,而后者只能在单机上启动作业。“submitit”需要安装同名的库。要了解更多关于 submitit 的信息,请访问 https://github.com/facebookincubator/submitit 默认为“submitit”。
tcp_port (int, 可选) – 要使用的 TCP 端口。默认为 10003。
visible_devices (Union[int, torch.device, str] 列表, 可选) – 一个长度与节点数量相同的列表,包含用于将数据传递到主进程的设备。
tensorpipe_options (dict, 可选) – 要传递给
torch.distributed.rpc.TensorPipeRpcBackendOption
的关键字参数字典。