Reverb

Reverb是一个高效且易于使用的数据存储和传输系统，专为机器学习研究设计。Reverb主要用于作为分布式强化学习算法的经验回放系统，但该系统还支持多种数据结构表示，如FIFO、LIFO和优先队列。

目录

• 安装
• 快速开始
• 详细概述
  • 表格
  • 项目选择策略
  • 速率限制
  • 分片
  • 检查点
• 引用

安装

请注意，Reverb未经生产使用加固，虽然我们尽力保持系统正常运行，但某些东西可能会出错或段错误。

:warning: Reverb目前仅支持基于Linux的操作系统。

安装Reverb的推荐方法是使用pip。我们还提供了使用与我们用于发布的相同Docker镜像从源代码构建的说明。

TensorFlow可以单独安装或作为pip安装的一部分。将TensorFlow作为安装的一部分安装可确保兼容性。

$ pip install dm-reverb[tensorflow]

# Without Tensorflow install and version dependency check.
$ pip install dm-reverb

夜间构建

$ pip install dm-reverb-nightly[tensorflow]

# Without Tensorflow install and version dependency check.
$ pip install dm-reverb-nightly

调试构建

从版本0.6.0开始，Reverb的调试构建被上传到Google Cloud Storage。可以通过以下模式直接通过pip下载或安装构建。可以使用gsutils遍历目录结构以确保文件存在，例如gsutil ls gs://rl-infra-builds/dm_reverb/builds/dbg。要构建自己的调试二进制文件，请参阅构建说明。

对于python 3.8和3.9，请遵循以下模式

$ export reverb_version=0.8.0
# Python 3.9
$ export python_version=39
$ pip install https://storage.googleapis.com/rl-infra-builds/dm_reverb/builds/dbg/$reverb_version/dm_reverb-$reverb_version-cp$python_version-cp$python_version-manylinux2010_x86_64.whl

从源代码构建

本指南详细说明了如何从源代码构建Reverb。

Reverb版本发布

由于一些底层库，如protoc和absl，Reverb必须与TensorFlow的特定版本配合使用。如果以pip install dm-reverb[tensorflow]安装Reverb，则会安装正确的TensorFlow版本。下表列出了Reverb每个版本的关联TensorFlow版本以及一些感兴趣的版本

• 0.13.0已停止支持Python 3.8。
• 0.11.0是第一个支持Python 3.11的版本。
• 0.10.0是最后一个支持Python 3.7的版本。

快速开始

启动Reverb服务器就像

import reverb

server = reverb.Server(tables=[
    reverb.Table(
        name='my_table',
        sampler=reverb.selectors.Uniform(),
        remover=reverb.selectors.Fifo(),
        max_size=100,
        rate_limiter=reverb.rate_limiters.MinSize(1)),
    ],
)

创建一个客户端与服务器通信

client = reverb.Client(f'localhost:{server.port}')
print(client.server_info())

将一些数据写入表

# Creates a single item and data element [0, 1].
client.insert([0, 1], priorities={'my_table': 1.0})

一个项目也可以引用多个数据元素

# Appends three data elements and inserts a single item which references all
# of them as {'a': [2, 3, 4], 'b': [12, 13, 14]}.
with client.trajectory_writer(num_keep_alive_refs=3) as writer:
  writer.append({'a': 2, 'b': 12})
  writer.append({'a': 3, 'b': 13})
  writer.append({'a': 4, 'b': 14})

  # Create an item referencing all the data.
  writer.create_item(
      table='my_table',
      priority=1.0,
      trajectory={
          'a': writer.history['a'][:],
          'b': writer.history['b'][:],
      })

  # Block until the item has been inserted and confirmed by the server.
  writer.flush()

我们可以添加到Reverb的项目可以通过采样它们来读取

# client.sample() returns a generator.
print(list(client.sample('my_table', num_samples=2)))

继续使用Reverb教程进行交互式教程。

详细概述

经验回放已成为训练离线策略强化学习策略的重要工具。它被诸如深度Q网络（DQN）、软行为评论家（SAC）、深度确定性策略梯度（DDPG）和事后经验回放等算法所使用。然而，构建一个高效、易于使用且可扩展的回放系统可能具有挑战性。为了获得良好的性能，Reverb是用C++实现的，并且为了实现分布式使用，它提供了一种gRPC服务，用于添加、采样和更新表的内容。Python客户端以易于使用的方式公开了服务的全部功能。此外，还提供了本机TensorFlow操作，以便与TensorFlow和tf.data进行高性能集成。

尽管最初是为离线策略强化学习设计的，但Reverb的灵活性使其在在线策略强化学习或甚至（非）监督学习中也同样有用。一些有创意的用户甚至使用Reverb来存储和分发频繁更新的数据（如模型权重），充当内存中轻量级的分布式文件系统替代品，其中每个表代表一个文件。

表格

Reverb的Server由一个或多个表组成。表包含项目，每个项目引用一个或多个数据元素。表还定义了样本和移除选择策略、最大项目容量和速率限制器。

多个项目可以引用相同的数据元素，即使这些项目存在于不同的表中。这是因为项目只包含对数据元素的引用（而不是数据的副本）。这也意味着，只有当不存在包含对其引用的项目时，数据元素才会被删除。

例如，可以将一个表配置为优先经验回放（PER）用于转换（长度为2的序列），另一个表配置为长度为3的序列的（FIFO）队列。在这种情况下，PER数据可以用来训练DQN，而FIFO数据可以用来训练环境的转换模型。

当满足以下两种条件之一时，项目将从表中自动删除

1. 插入新项目会使表中的项目数量超过其最大容量。表的使用策略用于确定要删除的项目。

2. 一个项目的采样次数超过了表速率限制器允许的最大次数。此类项目将被删除。

不再由任何项目引用的数据元素也将被删除。

用户可以完全控制从Reverb表中采样和删除数据的方式。行为主要受提供给Table的作为sampler和remover的item selection strategies控制。结合rate_limiter和max_times_sampled，可以实现广泛的行为。一些常用的配置包括

均匀经验回放

维护一组N=1000最近插入的项目。通过设置sampler=reverb.selectors.Uniform()，选择项目的概率对所有项目相同。由于reverb.rate_limiters.MinSize(100)，采样请求将在100个项目插入之前阻塞。通过设置remover=reverb.selectors.Fifo()，当需要删除项目时，首先删除最旧的项目。

reverb.Table(
     name='my_uniform_experience_replay_buffer',
     sampler=reverb.selectors.Uniform(),
     remover=reverb.selectors.Fifo(),
     max_size=1000,
     rate_limiter=reverb.rate_limiters.MinSize(100),
)

使用均匀经验回放算法的例子包括SAC和DDPG。

优先经验回放

一组N=1000最近插入的项目。通过设置sampler=reverb.selectors.Prioritized(priority_exponent=0.8)，选择项目的概率与项目的优先级成正比。

注意：有关在此实现中使用的优先经验回放算法，请参阅Schaul, Tom, 等人。

reverb.Table(
     name='my_prioritized_experience_replay_buffer',
     sampler=reverb.selectors.Prioritized(0.8),
     remover=reverb.selectors.Fifo(),
     max_size=1000,
     rate_limiter=reverb.rate_limiters.MinSize(100),
)

使用优先经验回放的算法示例包括DQN及其变体，以及分布式分布式确定性策略梯度。

队列

最多包含N=1000个项的集合，其中最老的项在相同的操作中选中并移除。如果集合包含1000个项，则插入调用被阻塞，直到集合不再满，如果集合为空，则采样调用被阻塞，直到至少有一个项。

reverb.Table(
    name='my_queue',
    sampler=reverb.selectors.Fifo(),
    remover=reverb.selectors.Fifo(),
    max_size=1000,
    max_times_sampled=1,
    rate_limiter=reverb.rate_limiters.Queue(size=1000),
)

# Or use the helper classmethod `.queue`.
reverb.Table.queue(name='my_queue', max_size=1000)

使用队列的算法示例包括IMPALA和近端策略优化的异步实现。

项目选择策略

Reverb定义了多个选择器，可用于项目采样或删除

• 均匀：在所有项中均匀采样。
• 优先级：按存储的优先级比例采样。
• FIFO：选择最老的数据。
• LIFO：选择最新的数据。
• 最小堆：选择优先级最低的数据。
• 最大堆：选择优先级最高的数据。

这些策略中的任何一种都可以用于从表中采样或删除项目。这使用户能够创建最符合其需求的定制表。

速率限制

速率限制器允许用户在何时可以将项目插入到表中以及/或从表中采样时强制执行条件。以下是Reverb当前可用的速率限制器列表

• MinSize：设置在可以采样之前必须在表中存在的最小项数。
• SampleToInsertRatio：设置通过阻塞插入和/或采样请求的平均插入到样本的比率。这对于控制在删除之前每个项目被采样的次数很有用。
• Queue：在移除之前，项目仅被采样一次。
• Stack：在移除之前，项目仅被采样一次。

分片

Reverb服务器彼此之间互不了解，当将系统扩展到多服务器设置时，数据不会在超过一个节点上进行复制。这使得Reverb不适合作为传统数据库，但它在可接受一定程度数据丢失的系统扩展方面具有优势。

可以通过简单地增加Reverb服务器的数量来水平扩展分布式系统。当与兼容gRPC的负载均衡器一起使用时，可以将负载均衡的目标地址简单地提供给Reverb客户端，操作将自动分布在不同的节点上。您可以在相关方法和类的文档中找到特定行为的详细信息。

如果您的设置中没有负载均衡器或需要更多的控制，则系统仍然可以用几乎相同的方式扩展。只需增加Reverb服务器的数量，并为每个服务器创建单独的客户端。

检查点

Reverb支持检查点；Reverb服务器状态和内容可以存储到永久存储。在检查点过程中，Server将所有需要重建的数据和元数据序列化。在此过程中，Server将阻止所有传入的插入、采样、更新和删除请求。

检查点是通过Reverb Client的调用完成的

# client.checkpoint() returns the path the checkpoint was written to.
checkpoint_path = client.checkpoint()

从检查点恢复reverb.Server

# The checkpointer accepts the path of the root directory in which checkpoints
# are written. If we pass the root directory of the checkpoints written above
# then the new server will load the most recent checkpoint written from the old
# server.
checkpointer = reverb.platform.checkpointers_lib.DefaultCheckpointer(
  path=checkpoint_path.rsplit('/', 1)[0])

# The arguments passed to `tables=` must be the same as those used by the
# `Server` that wrote the checkpoint.
server = reverb.Server(tables=[...], checkpointer=checkpointer)

有关Reverb中检查点实现的详细信息，请参阅tfrecord_checkpointer.h。

使用reverb_server（beta）启动Reverb

使用pip安装dm-reverb将安装一个reverb_server脚本，该脚本接受其配置作为textproto。例如

$ reverb_server --config="
port: 8000
tables: {
  table_name: \"my_table\"
  sampler: {
    fifo: true
  }
  remover: {
    fifo: true
  }
  max_size: 200 max_times_sampled: 5
  rate_limiter: {
    min_size_to_sample: 1
    samples_per_insert: 1
    min_diff: $(python3 -c "import sys; print(-sys.float_info.max)")
    max_diff: $(python3 -c "import sys; print(sys.float_info.max)")
  }
}"

配置项 rate_limiter 等同于 Python 表达式 MinSize(1)，请参阅 rate_limiters.py。

引用

如果您使用此代码，请引用以下 Reverb 论文：

@misc{cassirer2021reverb,
      title={Reverb: A Framework For Experience Replay},
      author={Albin Cassirer and Gabriel Barth-Maron and Eugene Brevdo and Sabela Ramos and Toby Boyd and Thibault Sottiaux and Manuel Kroiss},
      year={2021},
      eprint={2102.04736},
      archivePrefix={arXiv},
      primaryClass={cs.LG}
}