企业官网建设流程全解析

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

导读

SingleAgentEpisode是 Ray RLlib 新 API 栈（New API Stack）中描述单个智能体一个回合（Episode）的核心数据结构，负责按时间顺序记录 observations、infos、actions、rewards 以及各类模型附加输出（如action_logp、RNN 内部状态），并在此基础上提供灵活的切片、拼接、lookback 缓冲等能力。本篇文章以 single_agent_episode.rst 为骨架，结合 single_agent_episode.py 的完整源码实现与 test_single_agent_episode.py 测试用例，深入讲解该类的构造、数据写入、读取、回合分块（cut/slice/concat）以及序列化全流程，帮助你理解并直接使用这一新 API 栈核心组件。

一、SingleAgentEpisode 在 RLlib 新 API 栈中的定位

根据 new_api_stack.rst 的说明，Ray 2.40 及以后版本的 RLlib 默认启用新 API 栈，算法、示例脚本和文档均已迁移到新代码库。在新 API 栈中，环境交互的数据载体不再是旧式SampleBatch的单条记录流，而是以"回合"为单位组织的Episode对象：

• 单个智能体的回合数据 →SingleAgentEpisode；
• 多智能体环境的一次完整交互 →MultiAgentEpisode（内部由多个SingleAgentEpisode组合而成，见 multi_agent_episode.py 中对SingleAgentEpisode的引用与封装）。

从源码可见，SingleAgentEpisode被标记为@PublicAPI(stability="alpha")（见 single_agent_episode.py），是面向用户公开的 API。它在整个数据链路中的角色大致如下：

gym.Env.reset() ──► add_env_reset(obs, infos) gym.Env.step(a) ──► add_env_step(obs, action, reward, ...) （循环多步） │ ▼ get_observations / get_actions / get_rewards / ...（读取） cut() / slice() / concat_episode() （切分与拼接） to_numpy() / get_sample_batch() / get_state() （批量化与序列化）

EnvRunner（如 single_agent_env_runner.py）会在采样过程中持续创建并填充SingleAgentEpisode，随后将其交给 Learner 等下游模块使用。

二、构造函数与底层数据结构

SingleAgentEpisode的构造函数签名（见 single_agent_episode.py）如下：

SingleAgentEpisode( id_=None, # 回合唯一 ID，缺省时自动生成 uuid4().hex *, observations=None, # 观测列表或 InfiniteLookbackBuffer observation_space=None, # gym.Space，用于数据合法性校验 infos=None, # info 字典列表或 InfiniteLookbackBuffer actions=None, # 动作列表或 InfiniteLookbackBuffer action_space=None, # gym.Space rewards=None, # 奖励列表或 InfiniteLookbackBuffer terminated=False, # 回合是否已终止 truncated=False, # 回合是否已被截断 extra_model_outputs=None, # {"action_logp": [...], ...} 等模型附加输出 t_started=None, # 回合（或分块）起始的全局时间步 len_lookback_buffer="auto", # lookback 缓冲长度，默认 "auto" agent_id=None, # 归属的智能体 ID（仅作引用） module_id=None, # 归属的 RLModule ID（仅作引用） multi_agent_episode_id=None, # 所属 MultiAgentEpisode 的 ID )

关键参数说明（源码 docstring 归纳）：

• len_lookback_buffer：指定保留在回合数据最左侧的"历史缓冲"大小。缓冲区的数据不属于当前回合分块本身，仅用于向后"回看"以派生新数据（例如做观测帧叠加时，切分后的新分块需要回看上一分块的末尾数据）。默认值"auto"会把构造时传入的全部数据都解释为 lookback 缓冲。
• t_started：回合（分块）起始时间步。若传入了数据，默认从最后一个观测开始计数；显式传入则以给定值为准。
• terminated/truncated：分别表示回合被环境终止或截断，两者均为 False 时回合仍在进行中。

内部存储：InfiniteLookbackBuffer

所有时序数据（observations、infos、actions、rewards、extra_model_outputs）都不是直接存为普通 list，而是存放在InfiniteLookbackBuffer中（见 infinite_lookback_buffer.py）。该缓冲的核心语义：

• 维护一个lookback长度，即数据左侧有多少个时间步属于"历史区"而不属于当前回合；
• 提供append、extend、pop、finalize（列表转 numpy 批）、get（按索引/切片/回看语义读取）、set（按索引覆盖写入）等方法；
• 支持嵌套结构（如 dict 观测空间），finalize()后叶子变为np.ndarray。

值得注意的是，奖励缓冲在构造时绑定了_REWARDS_BOX_SPACE = gym.spaces.Box(-inf, inf, (), np.float32)（见 single_agent_episode.py），保证奖励数据以 float32 标量语义存储。

时间步与长度的约定

源码中回合的时序关系非常严格（这也是 MDP 数据形态的直接体现）：

• observations / infos：从 t0（reset 观测）一直记录到 T；
• actions / rewards / extra_model_outputs：从 t1 记录到 T；
• 因此恒有len(observations) == len(infos) == len(actions) + 1 == len(rewards) + 1（见validate()，single_agent_episode.py）；
• len(episode)返回self.t - self.t_started，不包含 lookback 缓冲，且只有调用过add_env_step后才会从 0 变为 1。

测试 test_single_agent_episode.py 的test_init印证了这些约定：传入 100 条数据、len_lookback_buffer=10时，len(episode) == 90，而len(episode.rewards.data) == 100，说明 10 个时间步进入了 lookback 缓冲。

三、写入数据：add_env_reset 与 add_env_step

文档把数据写入划分为两个方法，正好对应gym.Env的两个核心接口。

add_env_reset：写入 reset 初始数据

episode.add_env_reset(observation=obs, infos=info)

对应env.reset()的返回（初始观测 + 初始 info）。源码（single_agent_episode.py）规定调用前必须满足：

• 尚未 reset 过（len(observations) == 0）；
• 回合尚未 done；
• t == t_started == 0（只能用于全新空回合）。

调用后 observations 与 infos 各追加一项，但t保持为 0（智能体还没有真正"走一步"）。is_reset属性此时返回 True。

add_env_step：写入一步环境交互

episode.add_env_step( observation=obs, # env.step 返回的下一个观测 action=action, # 本次采取的动作 reward=reward, # 本次获得的奖励 terminated=term, # 是否终止 truncated=trunc, # 是否截断 infos=info, # 环境返回的 info extra_model_outputs={ # 可选：模型附加输出 "action_logp": logp, "state_out": state, }, )

源码行为（single_agent_episode.py）：

• 向 observations / actions / rewards / infos 各追加一项，self.t += 1；
• extra_model_outputs中每个 key 若不存在则新建InfiniteLookbackBuffer([v])，存在则append(v)；
• 更新is_terminated/is_truncated标志；
• 已 done 的回合不允许再追加数据（会触发断言）；
• 若 episode 已被to_numpy()且设置了 space，每隔 100 步校验一次观测/动作是否落在observation_space/action_space内。

# 来自类 docstring 的最小可用示例（CartPole-v1） import gymnasium as gym from ray.rllib.env.single_agent_episode import SingleAgentEpisode episode = SingleAgentEpisode() env = gym.make("CartPole-v1") obs, infos = env.reset() episode.add_env_reset(obs, infos) for _ in range(5): action = env.action_space.sample() obs, reward, term, trunc, infos = env.step(action) episode.add_env_step( observation=obs, action=action, reward=reward, terminated=term, truncated=trunc, infos=infos, ) assert len(episode) == 5 # reset 数据不计入长度

四、读取环境数据：五种 getter 与索引语义

文档将get_observations、get_infos、get_actions、get_rewards、get_extra_model_outputs归为"读取环境数据"一组。五个方法共享同一套索引语义，掌握其中一个即可触类旁通。

indices 参数：int / list / slice

• 单个 int：返回该索引处的单条数据（无 0 轴批量维度），例如episode.get_actions(-1)取最近一次动作；
• int 列表：按多个索引批量收集，返回带 0 轴（batch）的结果，例如episode.get_rewards([-1, 0])；
• slice 对象：返回一段区间，例如episode.get_rewards(slice(-4, None))等价于episode.rewards[-4:]；
• None：返回全部数据（ts=0 到末尾）。

三个高级参数：neg_index_as_lookback / fill / one_hot_discrete

这三个参数是SingleAgentEpisode最独特、也最实用的能力（源码实现见 single_agent_episode.py 的get_observationsdocstring 及 infinite_lookback_buffer.py）：

1. neg_index_as_lookback=True：默认情况下负索引表示"从末尾倒数"；开启后，负索引被解释为"ts=0 左侧"，即回溯进入 lookback 缓冲。例如观测为[4, 5, 6, 7, 8, 9]（其中[4, 5, 6]是 lookback，ts=0 对应 7），则get_observations(-1, neg_index_as_lookback=True)返回 6，get_observations(slice(-2, 1), neg_index_as_lookback=True)返回[5, 6, 7]。

2. fill=...：当请求的索引区间超出 episode 边界（含 lookback 左侧）时，用该值填充越界部分，天然实现零填充。例如观测[10, 11, 12, 13, 14]（lookback=2）时，get_observations(slice(-7, -2), fill=0.0)返回[0.0, 0.0, 10, 11, 12]。对于嵌套的 dict 观测空间，fill会作用于所有叶子。

3. one_hot_discrete=True：将观测/动作空间中Discrete/MultiDiscrete子分量自动转为 one-hot（或 multi-one-hot）向量，便于直接作为神经网络输入。配合fill=0使用时，越界填充出来的 one-hot 向量是全零的"zero-hot"（注意与[1, 0, 0, 0]的差异）。

# 类 docstring 中的典型用法：在 connector 里为每个时间步构造前 4 步动作 prev_4_a = [] for ts in range(len(episode)): prev_4_a.append( episode.get_actions( indices=slice(ts - 4, ts), neg_index_as_lookback=True, # 负索引视为进入 lookback fill=0.0, # 更早的越界部分补零 one_hot_discrete=True, # 离散子分量转 one-hot ) ) from ray.rllib.utils.spaces.space_utils import batch prev_4_actions_col = batch(prev_4_a) # 汇总为叶子为 numpy 的批量结构

get_infos 与 get_extra_model_outputs 的差异

• get_infos(indices=...)：返回 info 字典，签名不含one_hot_discrete（info 本身无空间结构）。
• get_extra_model_outputs(key, indices=...)：第一个参数是key，从extra_model_outputs字典中按 key 取对应缓冲，例如episode.get_extra_model_outputs("action_logp", slice(None, -1))。

五、基础信息查询：get_return / get_duration_s / is_done / is_numpy / env_steps

文档将以下方法归为"获取基本信息的 API"：

• get_return()：返回回合累计回报，实现为sum(self.get_rewards())（single_agent_episode.py）。注意两点：① 忽略折扣因子、纯求和；② 若当前实例是cut()产生的延续分块，前一分块的奖励不计入，lookback 缓冲中的奖励同样排除。
• get_duration_s()：返回该回合（分块）的耗时秒数，即_last_step_time - _start_time；尚无 step 时返回 0.0。
• is_done：is_terminated or is_truncated，done 后不可再追加数据、不可被右侧拼接或生成后继分块。
• is_numpy：底层奖励缓冲是否已finalize()（列表转 numpy），即是否已调用过to_numpy()。
• env_steps()：返回环境步数，即len(self)（不含 lookback）；对单智能体而言agent_steps()与env_steps()数值相同（多智能体场景下二者才会不同）。

六、回合分块与生命周期：cut / slice / concat_episode / to_numpy

这是文档归为"创建与处理回合分块"的一组方法，也是新 API 栈支持"非完整回合采样"（partial episode sampling）的关键。

cut：从当前回合切出后继分块

successor = episode.cut(len_lookback_buffer=2)

cut()（single_agent_episode.py）返回一个同 ID、长度为 0的后继分块，用于在不打断真实 gym 回合的前提下把当前分块交出去、稍后继续构建：

• len_lookback_buffer=0时，后继只携带self最后一个观测（类似一次 reset），长度为 0；
• len_lookback_buffer=2时，self.actions[-2:]会作为 lookback 放进后继（注意t_started与t仍保持等于切分点，而不是回退 2 步）；
• 如果self数据量不足以满足请求的 lookback 长度，该值会被自动调低；
• 前提是self尚未 done。

测试 test_single_agent_episode.py 的test_cut覆盖了这一行为。

slice：按区间切出子回合

slice(slice_)（single_agent_episode.py）返回一个新的SingleAgentEpisode，包含指定区间的数据，并自动把原 episode 的 lookback 前置到结果中。例如self含 o0~o4、a1~a4（len=4）时，self.slice(slice(1, 3))得到观测 o1,o2,o3 与动作 a2,a3——因为观测总是比动作多一个（reset 观测）。切片到末尾时，is_terminated/is_truncated状态会保留；未到末尾则重置为 False。同时支持 Python 切片语法糖：

slice_1 = episode[:1] # 等价 episode.slice(slice(None, 1)) slice_2 = episode[-2:] # 取最后两个时间步，lookback 自动前置

concat_episode：拼接续接的分块

concat_episode(other)（single_agent_episode.py）把other拼接到self右侧，拼接合法性检查包括：

• 两者id_必须一致；
• self.t == other.t_started（时间步必须无缝衔接）；
• other.observations[0]与self.observations[-1]逐叶子np.array_equal（边界观测必须相同）；
• self未 done。

拼接时会把self末尾的观测与 info 弹出（因为它与other开头的观测重复），再扩展各自的数据，并合并custom_data（other优先，因为作为后继分块它拥有更完整的版本）。测试中的test_concat_episode与test_concat_episode_with_complex_obs（test_single_agent_episode.py、test_single_agent_episode.py）分别验证了普通与嵌套观测空间的拼接。

to_numpy：冻结数据并转 numpy

to_numpy()（single_agent_episode.py）把所有列表形式的数据（含嵌套结构）转换为叶子为 numpy 数组的批量结构，0 轴大小为请求批长度。调用后：

• is_numpy变为 True；
• 不能再调用add_env_step追加数据；
• infos永远不会被 numpy 化，始终保留原始 dict 列表（因为 env 返回的 info 高度异构）。

# 源码 docstring 示例 episode = SingleAgentEpisode( observations=[0, 1, 2, 3], actions=[1, 2, 3], rewards=[1, 2, 3], len_lookback_buffer=0, ) episode.to_numpy() assert episode.is_numpy assert isinstance(episode.actions[0:2], np.ndarray)

七、数据校验：validate

validate()（single_agent_episode.py）保证回合数据自洽：

• len(observations) == len(infos)；
• 空回合时，actions / rewards / extra_model_outputs 长度也必须为 0；
• 非空回合必须满足观测比动作/奖励恰好多 1（MDP 的 reset/末观测逻辑），且每个extra_model_outputs缓冲长度等于观测数减 1。

该校验在构造函数、add_env_reset、add_env_step、concat_episode、to_numpy、from_state等关键路径上都会被调用，保证任何阶段的数据都是合法的 MDP 轨迹。

八、更多实用 API：setter、SampleBatch 转换与状态序列化

虽然 single_agent_episode.rst 的 autosummary 只列出上述方法，源码中还提供了与之配套的实用接口，理解它们有助于掌握完整的数据流：

• 覆盖写 setter 族：set_observations/set_actions/set_rewards/set_extra_model_outputs（single_agent_episode.py），用于后处理阶段整体或按索引改写数据（例如修改奖励、修正观测），同样支持at_indices与neg_index_as_lookback。
• get_data_dict()/get_sample_batch()：把回合转换为以Columns（EPS_ID、T、OBS、ACTIONS、REWARDS、TERMINATEDS、TRUNCATEDS等）为 key 的数据字典或SampleBatch，是回合数据流向训练管线的主要出口（single_agent_episode.py）。
• get_state()/from_state()：回合整体可序列化为可 pickle 的 dict（gym space 会经gym_space_to_dict/gym_space_from_dict转换），用于分布式传输与回放；测试test_get_and_from_state（test_single_agent_episode.py）验证了往返一致性。
• custom_data：一个自由字典，可从回调等位置写入自定义指标（如渲染图片）；旧的add_temporary_timestep_data/get_temporary_timestep_data已标记@Deprecated，应改用custom_data。

九、测试与验证

仓库在 test_single_agent_episode.py 中提供了完整的单元测试，覆盖：初始化（空/指定t_started/带 lookback 与预置数据）、add_env_reset、add_env_step、各类 getter、cut、slice、concat_episode（含复杂观测）、get_state/from_state往返、setter 及其错误路径。阅读这些测试是快速掌握 API 语义的最佳方式，例如test_init印证了 lookback 对len(episode)的影响，test_getters则逐一验证了索引、切片、fill与neg_index_as_lookback的预期输出。

总结

SingleAgentEpisode是 RLlib 新 API 栈中"以回合为单位"组织单智能体轨迹的基础设施：通过add_env_reset/add_env_step增量写入，通过五组 getter 配合neg_index_as_lookback、fill、one_hot_discrete完成任意时刻/区间的灵活读取，通过cut/slice/concat_episode支撑非完整回合采样与分块续接，通过to_numpy与get_sample_batch衔接训练管线，并以get_state/from_state实现序列化。无论是自定义 EnvRunner、编写 connector 做特征工程，还是深度理解 RLlib 新 API 栈的数据流，掌握本文所述的核心方法族都能直接受益；进一步细节可深入阅读 single_agent_episode.py、infinite_lookback_buffer.py 以及对应的 测试文件。

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