ConnectorV2 和 ConnectorV2 管道#
RLlib 以 SingleAgentEpisode 或 MultiAgentEpisode 对象的形式存储和传输所有轨迹数据。Connector 管道 是在模型前向传播之前，将这些 episode 数据转换为神经网络模型可读的张量批次的组件。
通用 ConnectorV2 管道：所有管道都由一个或多个 ConnectorV2 片段组成。调用管道时，您将传入一个 episode 列表、一个 RLModule 实例以及一个批次（最初可能是一个空字典）。管道中的每个 ConnectorV2 片段接收其前一个片段的输出（从左侧的批次开始），对 episodes、批次或两者都执行一些转换，然后将所有内容传递给下一个片段。因此，所有 ConnectorV2 片段都可以读取和写入提供的 episodes，将这些 episodes 的任何数据添加到批次中，或者更改批次中已有的数据。然后，管道返回最后一个片段的输出批次。#
注意
请注意，管道的批次输出仅在其后的 RLModule 前向传播或 Env.step() 调用期间有效。RLlib 之后会丢弃这些数据。然而，episode 列表可能会保留更长时间。例如，如果一个从环境到模块的管道读取一个 episode 的观测值，修改该观测值，然后将其写回 episode，那么后续的从模块到环境的管道就能看到修改后的观测值。此外，Learner 管道操作的是已经通过了环境到模块和模块到环境管道的同一个 episodes，因此可能已经发生了变化。
三种 ConnectorV2 管道类型#
RLlib 中有三种不同类型的 connector 管道
从环境到模块管道 (Env-to-module pipeline)，它为动作计算前向传播创建张量批次。
从模块到环境管道 (Module-to-env pipeline) (文档待定)，它将模型的输出转换为 RL 环境的动作。
Learner 连接器管道 (Learner connector pipeline)，它为模型更新创建训练批次。
ConnectorV2
ConnectorV2 管道：Connector 管道将 episodes 转换为批处理数据，您的模型可以处理这些数据（从环境到模块和 Learner），或者将模型的输出转换为您的（可能已向量化的）RL 环境在执行 step 时所需的动作批次（从模块到环境）。位于 EnvRunner 上的从环境到模块管道接收一个 episodes 列表作为输入，并为 RLModule 的前向传播输出一个批次，该前向传播计算下一个动作。位于同一 EnvRunner 上的从模块到环境管道接收该 RLModule 的输出，并将其转换为您 RL 环境下一次调用 step() 方法的动作。最后，位于 Learner 工作节点上的 Learner 连接器管道将 episodes 列表转换为下一个 RLModule 更新的训练批次。#
后续页面将更详细地讨论这三种管道类型，然而，它们都有一个共同点：
所有 connector 管道都是一个或多个
ConnectorV2片段的序列。您也可以嵌套这些管道，这意味着一些片段本身可以是 connector 管道。所有 connector 片段和管道都是 Python 可调用对象，覆盖了
__call__()方法。调用签名在不同管道类型之间是统一的。主要的、强制性的参数是 episodes 列表、待构建的批次以及
RLModule实例。有关更多详细信息，请参阅__call__()方法。所有 connector 管道都可以读写提供的 episodes 列表以及批次，从而根据需要执行数据转换。
批次构建阶段和格式#
当您将一系列输入 episodes 通过 connector 管道时，管道会从给定的数据中构建一个批次。这个批次始终以一个空的 Python 字典开始，并在通过管道的不同片段时经历不同的格式和阶段。
以下适用于所有 从环境到模块 和 Learner connector 管道（文档正在编写中）。
批次构建阶段和格式：在标准单智能体情况下，只有一个 ModuleID（DEFAULT_MODULE_ID）存在，批次开始时是一个空字典（左侧），然后经历一个“收集数据”阶段，在此阶段，connector 片段通过将单个项目存储在 a) 列名（例如 obs 或 rewards）和 b) 它们从中提取项目的 episode ID 下来添加它们。在大多数情况下，您的自定义 connector 片段在此阶段运行。一旦所有自定义片段执行了数据插入和转换，默认的 AgentToModuleMapping 片段会执行一个“按 ModuleID 重组”操作（中间），在此期间，批次的字典层级结构会变为以 ModuleID（DEFAULT_MODULE_ID）为顶层，下面是列名。在批次的最低层级，数据项仍然保留在 Python 列表中。最后，BatchIndividualItems 默认片段将 Python 列表转换为 NumPy 数组，从而将所有数据打包（右侧）。#
对于多智能体设置，其中有多个 ModuleID，默认的 AgentToModuleMapping connector 片段确保构建的输出批次将 module ID 映射到各自模块的前向传播批次。
多智能体批次构建：在多智能体设置中，默认的 AgentToModuleMapping connector 片段会按 ModuleID 然后按列名重组批次，以便 MultiRLModule 可以遍历其子模块，并为每个子模块提供一个用于前向传播的批次。#
RLlib 的 MultiRLModule 可以使用各个 ModuleIDs 下的单个批次，将前向传播分割为各个子模块的前向传播。请参阅 此处了解如何编写自己的多模块或多智能体前向传播逻辑，并覆盖 MultiRLModule 的此默认行为。
最后，如果您有一个有状态的 RLModule，例如 LSTM，RLlib 会向管道中添加两个额外的默认 connector 片段：AddTimeDimToBatchAndZeroPad 和 AddStatesFromEpisodesToBatch。
有状态模型的批次构建：对于有状态的 RLModule 实例，RLlib 会自动向管道中添加两个额外的默认 connector 片段。AddTimeDimToBatchAndZeroPad 片段将最低批次级别上的所有单个数据项列表转换为固定长度（max_seq_len，有关如何设置此值，请参见下面的注释）的序列，并在遇到 episode 结束时自动对其进行零填充。AddStatesFromEpisodesToBatch 片段将您的 RLModule 之前生成的 state_out 值作为 state_in 列名添加到批次中。请注意，RLlib 仅为每个序列的第一个时间步添加 state_in 值，因此也不会为 state_in 列中的数据添加时间维度。#
注意
要更改 AddTimeDimToBatchAndZeroPad 连接器的零填充序列长度，请在自定义模型的配置中设置
config.rl_module(model_config={"max_seq_len": ...})
以及 RLlib 的默认模型
from ray.rllib.core.rl_module.default_model_config import DefaultModelConfig
config.rl_module(model_config=DefaultModelConfig(max_seq_len=...))