回放缓冲区#

强化学习中回放缓冲区的快速入门#

当我们谈论强化学习中的回放缓冲区时，我们通常指的是一个缓冲区，它存储并回放从我们的智能体与环境交互中收集到的经验。在 Python 中，一个简单的缓冲区可以通过一个列表来实现，向其中添加元素，然后稍后从中采样。这类缓冲区主要用于离策略学习算法。这在直观上是合理的，因为这些算法可以从存储在缓冲区中的经验中学习，而这些经验是由先前策略版本（甚至完全不同的“行为策略”）产生的。

采样策略#

从回放缓冲区采样时，我们选择用哪些经验来训练我们的智能体。一种简单且已被证明对许多算法有效的策略是随机均匀地选择样本。一种更高级的策略（在许多情况下被证明效果更好）是优先经验回放（PER）。在 PER 中，缓冲区中的单个项目被分配一个（标量）优先级值，该值表示其重要性，或者用更简单的术语来说，我们期望从这些项目中学习多少。优先级较高的经验更有可能被采样。

驱逐策略#

缓冲区在存储经验方面自然是有限的。在算法运行过程中，缓冲区最终会达到其容量，为了给新经验腾出空间，我们需要删除（驱逐）旧的经验。这通常是按照先入先出的顺序进行的。对于你的算法而言，这意味着具有高容量的缓冲区提供了从旧样本中学习的机会，而较小的缓冲区则使学习过程更接近于在线策略。在实现水库采样的缓冲区中，会对此策略进行例外处理。

RLlib 中的回放缓冲区#

RLlib 内置了一套可扩展的回放缓冲区。所有这些缓冲区都支持两个基本方法：add() 和 sample()。我们提供了一个基础的 ReplayBuffer 类，你可以从中构建自己的缓冲区。在大多数算法中，我们需要 MultiAgentReplayBuffer。这是因为我们希望它们能够泛化到多智能体场景。因此，这些缓冲区的 add() 和 sample() 方法需要一个 policy_id 来处理每个策略的经验。请查看 MultiAgentReplayBuffer，以了解它是如何扩展我们的基础类的。你可以在 RLlib 的默认参数中找到缓冲区类型和修改其行为的参数。它们是 replay_buffer_config 的一部分。

基本用法#

在运行实验时，你很少需要定义自己的回放缓冲区子类，而是配置现有的缓冲区。以下内容来自 RLlib 的示例部分：它运行 R2D2 算法并使用 PER（默认情况下它不使用 PER）。突出显示的行专注于 PER 配置。

提示

由于其普遍性，大多数 Q-learning 算法都支持 PER。所需的优先级更新步骤已嵌入其训练迭代函数中。

警告

如果你的自定义缓冲区需要额外的交互，你也需要更改训练迭代函数！

指定缓冲区类型的方式与指定探索类型的方式相同。这里有三种指定类型的方法

除了 type 之外，你还可以指定 capacity 和其他参数。这些参数大多是缓冲区的构造函数参数。存在以下类别：

定义算法如何与回放缓冲区交互的参数。
例如，worker_side_prioritization 用来决定在哪里计算优先级
实例化回放缓冲区的构造函数参数。
例如，capacity 用来限制缓冲区的大小
底层回放缓冲区方法的调用参数。
例如，prioritized_replay_beta 由 MultiAgentPrioritizedReplayBuffer 用于调用每个底层 PrioritizedReplayBuffer 的 sample() 方法。

提示

大多数情况下，只有 1. 和 2. 是感兴趣的。3. 是一项高级功能，支持 MultiAgentReplayBuffer 实例化需要构造函数或默认调用参数的底层缓冲区的用例。

ReplayBuffer 基类#

基础的 ReplayBuffer 类仅支持在不同的 StorageUnit 中存储和回放经验。你可以使用 add() 方法向缓冲区的存储中添加数据，并使用 sample() 方法回放它。高级缓冲区类型在保留兼容性的同时添加功能。以下是与 ReplayBuffer 进行交互的最基本方案的示例。

# We choose fragments because it does not impose restrictions on our batch to be added
buffer = ReplayBuffer(capacity=2, storage_unit=StorageUnit.FRAGMENTS)
dummy_batch = SampleBatch({"a": [1], "b": [2]})
buffer.add(dummy_batch)
buffer.sample(2)
# Because elements can be sampled multiple times, we receive a concatenated version
# of dummy_batch `{a: [1, 1], b: [2, 2,]}`.

构建你自己的 ReplayBuffer#

以下是如何实现你自己的玩具 ReplayBuffer 类并让 SimpleQ 使用它的示例

class LessSampledReplayBuffer(ReplayBuffer):
    @override(ReplayBuffer)
    def sample(
        self, num_items: int, evict_sampled_more_then: int = 30, **kwargs
    ) -> Optional[SampleBatchType]:
        """Evicts experiences that have been sampled > evict_sampled_more_then times."""
        idxes = [random.randint(0, len(self) - 1) for _ in range(num_items)]
        often_sampled_idxes = list(
            filter(lambda x: self._hit_count[x] >= evict_sampled_more_then, set(idxes))
        )

        sample = self._encode_sample(idxes)
        self._num_timesteps_sampled += sample.count

        for idx in often_sampled_idxes:
            del self._storage[idx]
            self._hit_count = np.append(
                self._hit_count[:idx], self._hit_count[idx + 1 :]
            )

        return sample


config = (
    DQNConfig()
    .api_stack(
        enable_env_runner_and_connector_v2=False, enable_rl_module_and_learner=False
    )
    .environment(env="CartPole-v1")
    .training(replay_buffer_config={"type": LessSampledReplayBuffer})
)

tune.Tuner(
    "DQN",
    param_space=config,
    run_config=tune.RunConfig(
        stop={"training_iteration": 1},
    ),
).fit()

对于完整的实现，你应该考虑其他方法，如 get_state() 和 set_state()。一个更广泛的例子是我们对水库采样的实现，即 ReservoirReplayBuffer。

高级用法#

在 RLlib 中，所有回放缓冲区都实现了 ReplayBuffer 接口。因此，在可能的情况下，它们支持不同的 StorageUnit。回放缓冲区的 `storage_unit` 构造函数参数定义了经验的存储方式，因此也定义了它们被采样的单位。在调用 sample() 方法时，`num_items` 将与该 `storage_unit` 相关。

这是一个修改 `storage_unit` 和与自定义缓冲区交互的完整示例

# This line will make our buffer store only complete episodes found in a batch
config.training(replay_buffer_config={"storage_unit": StorageUnit.EPISODES})

less_sampled_buffer = LessSampledReplayBuffer(**config.replay_buffer_config)

# Gather some random experiences
env = RandomEnv()
terminated = truncated = False
batch = SampleBatch({})
t = 0
while not terminated and not truncated:
    obs, reward, terminated, truncated, info = env.step([0, 0])
    # Note that in order for RLlib to find out about start and end of an episode,
    # "t" and "terminateds" have to properly mark an episode's trajectory
    one_step_batch = SampleBatch(
        {
            "obs": [obs],
            "t": [t],
            "reward": [reward],
            "terminateds": [terminated],
            "truncateds": [truncated],
        }
    )
    batch = concat_samples([batch, one_step_batch])
    t += 1

less_sampled_buffer.add(batch)
for i in range(10):
    assert len(less_sampled_buffer._storage) == 1
    less_sampled_buffer.sample(num_items=1, evict_sampled_more_then=9)

assert len(less_sampled_buffer._storage) == 0

如上所述，RLlib 的 MultiAgentReplayBuffer 支持修改底层回放缓冲区。在底层，MultiAgentReplayBuffer 会为每个策略分别存储经验在独立的底层回放缓冲区中。你可以通过指定一个与父配置工作方式相同的底层 replay_buffer_config 来修改它们的行为。

下面是如何创建一个 MultiAgentReplayBuffer，其中包含一个替代的底层 ReplayBuffer。 MultiAgentReplayBuffer 可以保持不变。我们只需要指定我们自己的缓冲区以及一个默认的调用参数

config = (
    DQNConfig()
    .api_stack(
        enable_env_runner_and_connector_v2=False, enable_rl_module_and_learner=False
    )
    .training(
        replay_buffer_config={
            "type": "MultiAgentReplayBuffer",
            "underlying_replay_buffer_config": {
                "type": LessSampledReplayBuffer,
                # We can specify the default call argument
                # for the sample method of the underlying buffer method here.
                "evict_sampled_more_then": 20,
            },
        }
    )
    .environment(env="CartPole-v1")
)

tune.Tuner(
    "DQN",
    param_space=config.to_dict(),
    run_config=tune.RunConfig(
        stop={"env_runners/episode_return_mean": 40, "training_iteration": 7},
    ),
).fit()