Ray RLlib Episode 数据结构与 SingleAgentEpisode 实战指南：轨迹存储、Getter API 与 Lookback 缓冲

• 人工智能
• 分布式训练
• 强化学习
• 任务调度
• 模型推理服务

导读

本文是 Ray RLlib 新 API 栈中 Episode（回合）数据结构的完整技术指南，主体基于仓库文档 single-agent-episode.md 展开。你将掌握SingleAgentEpisode的构造与数据填充、五类 Getter API 的索引规则、numpy'ized 状态转换，以及cut()与 lookback buffer（回看缓冲）在分段采样中的关键作用——这些都是编写自定义 Connector 管道、从历史轨迹中提取特征（如 Transformer 所需的最近 n 帧观测、DQN 所需的(obs, next_obs)对）时的必备能力。

一、为什么 RLlib 选择用 Episode 而非 Tensor Batch 传递轨迹

在 RLlib 的新 API 栈中，所有轨迹数据都以Episode的形式存储和传输：

• 单智能体场景使用 {py:class}~ray.rllib.env.single_agent_episode.SingleAgentEpisode（源码位于 python/ray/rllib/env/single_agent_episode.py）；
• 多智能体场景使用 {py:class}~ray.rllib.env.multi_agent_episode.MultiAgentEpisode（源码位于 python/ray/rllib/env/multi_agent_episode.py）。

数据在组件之间流动（如EnvRunner→Learner、ReplayBuffer→Learner）时保持"整段轨迹"的形式，只有到神经网络前向传播之前，所谓的 Connector 管道 才会把多个 Episode 翻译成 Tensor Batch（并可能搬运到 GPU）。这与传统 RL 框架"边采样边拼 batch"的做法截然不同，其设计原则是：尽可能长时间地让轨迹数据保持回合形态。

这种设计带来两个核心优势：

1. 360° 可见性（完整历史可访问）。你可以从 Episode 中提取任意片段信息，供自定义组件进一步处理。例如 Transformer 模型需要的不是最近一次观测，而是最近 n 个观测的完整序列——借助get_observations()，你可以在自定义ConnectorV2管道内提取这段历史并注入神经网络 batch。

2. 更省内存。以 DQN 为例，训练 batch 中同时需要观测与下一观测来计算 TD 误差损失，这会让已经很大的观测张量翻倍。而使用 Episode 对象时，从 reset 到 terminal 的全部观测只存一条观测记录（observation track），内存占用显著降低。

二、创建并填充一个 SingleAgentEpisode

RLlib 正常运行时会在内部创建并搬运SingleAgentEpisode实例，例如从EnvRunner到Learner。但为了理解其数据模型，我们可以手动构造并填充一个初始为空的 Episode。以下是仓库示例 doc/source/rllib/doc_code/sa_episode.py 中的完整代码：

from ray.rllib.env.single_agent_episode import SingleAgentEpisode # Construct a new episode (without any data in it yet). episode = SingleAgentEpisode() assert len(episode) == 0 episode.add_env_reset(observation="obs_0", infos="info_0") # Even with the initial obs/infos, the episode is still considered len=0. assert len(episode) == 0 # Fill the episode with some fake data (5 timesteps). for i in range(5): episode.add_env_step( observation=f"obs_{i+1}", action=f"act_{i}", reward=f"rew_{i}", terminated=False, truncated=False, infos=f"info_{i+1}", ) assert len(episode) == 5

关键点解读：

• add_env_reset()与add_env_step()是对gym.Env的reset()/step()返回值的封装。从源码看，SingleAgentEpisode的 docstring 明确说明这两个方法是往进行中的 Episode 追加数据的主 API（见 single_agent_episode.py）。
• len(episode)统计的是 step 数，而不是观测数：调用add_env_reset()之后长度仍为 0；每调用一次add_env_step()长度加 1。这是因为 Episode 的观测序列天然比动作/奖励序列多一个（多出的是 reset 观测）。
• 构造函数支持丰富的关键字参数，包括id_（不传时自动生成uuid4().hex唯一标识）、observation_space/action_space（提供后可在 numpy'ized 状态下校验新追加的数据）、terminated/truncated、extra_model_outputs、t_started、len_lookback_buffer（默认"auto"，见 源码init）。

填好的 Episode 结构如下：以单个 reset 观测开头，之后每个时间步追加一个(observation, action, reward)三元组。由于 reset 观测的存在，每个时间步上 Episode 总是比动作/奖励多一个观测。除数据外，Episode 还有两个重要的布尔属性terminated与truncated，以及字符串类型的id_。

三、Getter API：五类字段的灵活索引

SingleAgentEpisode的五个字段——observations、actions、rewards、infos、extra_model_outputs——都有对应的 getter 方法：get_observations()、get_actions()、get_rewards()、get_infos()、get_extra_model_outputs()（定义位置见 single_agent_episode.py）。

索引规则（核心）：

• 传单个索引→ 返回单个元素（不是 size-1 的列表或 batch）；
• 传索引列表或slice 对象→ 返回元素列表（该规则适用于非 numpy'ized 的 Episode）。

示例代码（同样取自 sa_episode.py）：

from ray.rllib.utils.test_utils import check # Get the very first observation ("reset observation"). Note that a single observation # is returned here (not a list of size 1 or a batch of size 1). check(episode.get_observations(0), "obs_0") # ... which is the same as using the indexing operator on the Episode's # `observations` property: check(episode.observations[0], "obs_0") # You can also get several observations at once by providing a list of indices: check(episode.get_observations([1, 2]), ["obs_1", "obs_2"]) # .. or a slice of observations by providing a python slice object: check(episode.get_observations(slice(1, 3)), ["obs_1", "obs_2"]) # Similarly for getting rewards: Get the last reward. check(episode.get_rewards(-1), "rew_4") check(episode.rewards[-1], "rew_4") # Similarly for getting actions: Get the first action in the episode # (single item, not batched). This works regardless of the action space. check(episode.get_actions(0), "act_0") check(episode.actions[0], "act_0") # Finally, you can slice the entire episode using the []-operator with a slice notation: sliced_episode = episode[3:4] check(list(sliced_episode.observations), ["obs_3", "obs_4"]) check(list(sliced_episode.actions), ["act_3"]) check(list(sliced_episode.rewards), ["rew_3"])

几点补充：

• 负索引（如get_rewards(-1)）表示"从末尾往前数"，与 Python 列表语义一致。
• 对extra_model_outputs的 getter 稍复杂，因为这类数据带子键（如action_logp、state_outs），需要传入子键名，详见get_extra_model_outputs()的签名与实现。
• episode[3:4]这种切片操作返回一个新的 Episode 对象（子片段），其observations/actions/rewards等属性已被裁剪为对应区间，这在离线数据回放、细粒度诊断场景非常实用。

四、numpy'ized 与非 numpy'ized 状态

Episode 内部数据存在两种状态：

一个 numpy'ized 的 Episode 未必已 terminated 或 truncated——底层环境不一定宣告回合结束，也可能只是未达到最大时间步。这意味着"是否 numpy'ized"与"是否完成"是两个完全独立的维度。

示例代码：

# Episodes start in the non-numpy'ized state (in which data is stored # under the hood in lists). assert episode.is_numpy is False # Call `to_numpy()` to convert all stored data from lists of individual (possibly # complex) items to numpy arrays. Note that RLlib normally performs this method call, # so users don't need to call `to_numpy()` themselves. episode.to_numpy() assert episode.is_numpy is True

以复杂的 dict 观测为例对比两种状态：

• 非 numpy'ized：Episode 保存了三个独立的复杂 dict 观测，每个 dict 的结构匹配 gymnasium 环境的 observation space；
• numpy'ized：整个观测记录变成一个与 observation space 同构的复杂 dict，其叶子位置是NDArray，每个 NDArray 在 axis 0 上多出一个 batch 维，长度等于已存 Episode 的长度（此处为 3）。

使用要点：to_numpy()在 RLlib 内部通常会被自动调用（例如数据从 EnvRunner 送往 Learner 之前），普通用户无需手动调用。若你在构造函数中传入了observation_space或action_space，在 numpy'ized 状态下追加数据时会自动校验数据是否符合空间定义（见 源码init的说明）。

五、Episode.cut()：分段采样与 Lookback 缓冲

5.1 cut() 做了什么

在从 RL 环境采样时，EnvRunner有时必须停止向进行中的 Episode 追加数据，把已采集的部分返回（例如一次sample()调用已经收集够数据）。此时它会调用cut()，返回一个新的续接 chunk，下一轮采样从该 chunk 继续采集。示例：

# An ongoing episode (of length 5): assert len(episode) == 5 assert episode.is_done is False # During an `EnvRunner.sample()` rollout, when enough data has been collected into # one or more Episodes, the `EnvRunner` calls the `cut()` method, interrupting # the ongoing Episode and returning a new continuation chunk (with which the # `EnvRunner` can continue collecting data during the next call to `sample()`): continuation_episode = episode.cut() # The length is still 5, but the length of the continuation chunk is 0. assert len(episode) == 5 assert len(continuation_episode) == 0 # Thanks to the lookback buffer, we can still access the most recent observation # in the continuation chunk: check(continuation_episode.get_observations(-1), "obs_5")

从源码实现看，cut(len_lookback_buffer=0)会返回一个与self同 ID的续接 chunk（single_agent_episode.py）：

• 续接 chunk 以self最近的一条观测和 info 初始化（类似一次env.reset()），因此即使 lookback 为 0，也能通过get_observations(-1)拿到上一段最后的观测；
• 若传入len_lookback_buffer > 0，则从self右端（末尾）取对应数量的 observations/actions/rewards/extra_model_outputs 作为新 chunk 的 lookback 数据，新 chunk 的len仍为 0，t_started保持为self.t；
• 若self中的数据不足以满足 lookback 请求，该值会被自动调低；
• 自定义数据（custom_data）会被深拷贝到续接 chunk，保证自定义字段的延续性。

5.2 Lookback 机制与 episode_lookback_horizon

Lookback 机制让 Connector 能从续接 chunk 内部访问被 cut 掉的上一个 chunk 的最近H个时间步，H是可通过配置调整的参数。默认 lookback 视野H为 1，即 cut 之后你仍可以：

• get_actions(-1)拿到最近一次动作；
• get_rewards(-1)拿到最近一次奖励；
• get_observations([-2, -1])拿到最近两次观测。

如果需要访问更早的数据，通过AlgorithmConfig调整：

from ray.rllib.algorithms.algorithm_config import AlgorithmConfig config = AlgorithmConfig() # Change the lookback horizon setting, in case your connector (pipelines) need # to access data further in the past. config.env_runners(episode_lookback_horizon=10)

该配置项在 algorithm_config.py 中默认值为1，其 docstring 明确指出（L2001-L2007）：它表示生成新 chunk 继续采样时，从上一个 chunk 保留多少时间步的数据；取值越大，env-to-module connector 能回看的时间越长。例如你的自定义 connector 与 RLModule 需要最近 10 个奖励作为输入，就必须把episode_lookback_horizon至少设为 10。该配置同时被单/多智能体 EnvRunner（single_agent_env_runner.py、multi_agent_env_runner.py）以及离线预学习（offline_prelearner.py）读取。

5.3 Lookback 缓冲区与 getter 的深层用法

下面演示如何在 Connector 内利用 lookback 缓冲访问更早的数据。想象你要写一个 Connector，把最近 5 个奖励拼进模型计算动作的 tensor batch：

# Construct a new episode (with some data in its lookback buffer). episode = SingleAgentEpisode( observations=["o0", "o1", "o2", "o3"], actions=["a0", "a1", "a2"], rewards=[0.0, 1.0, 2.0], len_lookback_buffer=3, ) # Since our lookback buffer is 3, all data already specified in the constructor should # now be in the lookback buffer (and not be part of the `episode` chunk), meaning # the length of `episode` should still be 0. assert len(episode) == 0 # .. and trying to get the first reward will hence lead to an IndexError. try: episode.get_rewards(0) except IndexError: pass # Get the last 3 rewards (using the lookback buffer). check(episode.get_rewards(slice(-3, None)), [0.0, 1.0, 2.0]) # Instead, `SingleAgentEpisode` getters offer some useful options to solve this problem: last_5_rewards = episode.get_rewards(slice(-5, None), fill=0.0)

这里引入了fill参数：当请求的时间步超出了 lookback 缓冲的覆盖范围时，用固定值填充。手动补零不仅繁琐，在面对复杂（嵌套）观测/动作时几乎不可维护。

另一个实用的 getter 参数是neg_index_as_lookback：置为 True 时，负索引的含义从"从末尾往前数"切换为"往 lookback 缓冲里数"。借此可以一边循环全局时间步，一边从每个时间步回看固定步数：

# Construct a new episode (len=3 and lookback buffer=3). episode = SingleAgentEpisode( observations=["o-3", "o-2", "o-1", "o0", "o1", "o2", "o3"], actions=["a-3", "a-2", "a-1", "a0", "a1", "a2"], rewards=[-3.0, -2.0, -1.0, 0.0, 1.0, 2.0], len_lookback_buffer=3, ) assert len(episode) == 3 # In case you want to loop through global timesteps 0 to 2 (timesteps -3, -2, and -1 # being the lookback buffer) and at each such global timestep look 2 timesteps back, # you can do so easily using the `neg_index_as_lookback` arg like so: for global_ts in [0, 1, 2]: rewards = episode.get_rewards( slice(global_ts - 2, global_ts + 1), # Switch behavior of negative indices from "from-the-end" to # "into the lookback buffer": neg_index_as_lookback=True, ) print(rewards) # The expected output should be: # [-2.0, -1.0, 0.0] # global ts=0 (plus looking back 2 ts) # [-1.0, 0.0, 1.0] # global ts=1 (plus looking back 2 ts) # [0.0, 1.0, 2.0] # global ts=2 (plus looking back 2 ts)

注意len(episode)不包含 lookback 缓冲——本例中len(episode) == 3（只有 o0~o2 三个实际时间步），而 o-3、o-2、o-1 是缓冲数据。

fill与neg_index_as_lookback可组合使用，例如"从 ts=1 往前看 4 个奖励，超出范围用 0.0 填充"：

episode.get_rewards(slice(-4, 0), neg_index_as_lookback=True, fill=0.0) # 期望返回 [0.0, 0.0, 0.0, r0]，其中 r0 是 episode 收到的第一个奖励

5.4 复杂（嵌套离散）动作空间的进阶处理

当动作空间是嵌套的gym.spaces.Dict且包含Discrete/MultiDiscrete子空间时，getter 提供两个关键能力：

1. fill=...仍然生效，会以该值填充所有子空间（Box、Discrete 均适用）；
2. one_hot_discrete=True会把离散/多离散子组件自动转为 one-hot（或 multi-one-hot）张量，直接作为神经网络输入。

完整示例（出自 sa_episode.py）：

import gymnasium as gym import numpy as np episode = SingleAgentEpisode( action_space=gym.spaces.Dict( { "a": gym.spaces.Discrete(3), "b": gym.spaces.MultiDiscrete([2, 3]), "c": gym.spaces.Box(-1.0, 1.0, (2,)), } ) ) # ... fill episode with data ... episode.add_env_reset(observation=0) episode.add_env_step( observation=1, action={"a": 0, "b": np.array([1, 2]), "c": np.array([0.5, -0.5], np.float32)}, reward=1.0, ) # In your connector prev_4_a = [] # Note here that len(episode) does NOT include the lookback buffer. for ts in range(len(episode)): prev_4_a.append( episode.get_actions( indices=slice(ts - 4, ts), # Make sure negative indices are interpreted as "into lookback buffer" neg_index_as_lookback=True, # Zero-out everything even further before the lookback buffer. fill=0.0, # Take care of discrete components (get ready as NN input). one_hot_discrete=True, ) ) # Finally, convert from list of batch items to a struct (same as action space) # of batched (numpy) arrays, in which all leafs have B==len(prev_4_a). from ray.rllib.utils.spaces.space_utils import batch prev_4_actions_col = batch(prev_4_a)

这段代码展示了完整闭环：先解释手动补齐缺失时间步的痛点 → 用fill+neg_index_as_lookback+one_hot_discrete优雅解决 → 最后用ray.rllib.utils.spaces.space_utils.batch把逐条结果拼成与动作空间同构、叶子 batch 维为B的批量结构，直接喂给模型前向。

六、与 Connector 管道的衔接（最佳实践）

Episode 是 Connector 管道的输入载体。理解二者关系有助于把握"何时调用这些 API"：

• Env-to-Module 方向：数据以 Episode 列表形式进入ConnectorV2管道，管道内通常逐个 Episode 提取特征（正是本文 getter API 的用武之地），最终聚合成 tensor batch 供RLModule前向。这就是为什么官方建议在自定义 connector 中通过get_observations()等 getter 为 Transformer 等模型拼装历史序列。
• Module-to-Env 方向：模型输出（动作、状态等）又通过反向管道写回 Episode（如extra_model_outputs中的action_logp、state_outs），供损失计算与日志使用。
• episode_lookback_horizon与 cut 的配合：分段采样场景下，lookback 视野必须覆盖 connector 所需的历史长度，否则在 chunk 起始处会拿不到足够的历史数据。配置逻辑可参考 connector-v2.md 与 AlgorithmConfig 源码。

从源码结构看，SingleAgentEpisode的底层存储是 InfiniteLookbackBuffer（无界回看缓冲）：非 numpy'ized 阶段它就是普通 Python 列表的封装，numpy'ized 后其叶子变为 NumPy 数组；所有 getter 的索引、切片、lookback 语义都由该缓冲提供。文档还提示，多智能体场景的MultiAgentEpisode是类似的设计，Ray 团队正在撰写与之对应的详细说明（当前仅在本页发布了一个 note 占位）。

七、小结

对任何想深入 RLlib 新 API 栈、编写自定义 Connector 或离线数据管道的开发者而言，SingleAgentEpisode都是最核心的数据契约。掌握了上述 API 与 lookback 语义，你就能在 EnvRunner、ReplayBuffer、Learner 之间自由地搬运和加工整段轨迹数据，而不必退回到笨拙的逐 batch 拼接。

相关参考：

• 本文主文档：doc/source/rllib/single-agent-episode.md
• 可运行示例：doc/source/rllib/doc_code/sa_episode.py
• 核心实现：python/ray/rllib/env/single_agent_episode.py
• 配置项定义：python/ray/rllib/algorithms/algorithm_config.py
• 底层缓冲：python/ray/rllib/env/utils/infinite_lookback_buffer.py
• 多智能体版本：python/ray/rllib/env/multi_agent_episode.py

• 人工智能
• 分布式训练
• 强化学习
• 任务调度
• 模型推理服务