Reverb

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

目录

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

安装

请注意，Reverb 尚未针对生产环境进行充分优化，尽管我们尽力保持其正常运行，但仍可能出现故障或段错误。

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

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

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

$ pip install dm-reverb[tensorflow]

# 不包含 TensorFlow 安装及版本依赖检查。
$ pip install dm-reverb

每日构建

$ pip install dm-reverb-nightly[tensorflow]

# 不包含 TensorFlow 安装及版本依赖检查。
$ pip install dm-reverb-nightly

从源代码构建

本指南 详细介绍了如何从源代码构建 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 的版本。

版本	分支 / 标签	TensorFlow 版本
每日构建	master	tf-nightly
0.14.0	v0.14.0	2.14.0
0.13.0	v0.13.0	2.14.0
0.12.0	v0.12.0	2.13.0
0.11.0	v0.11.0	2.12.0
0.10.0	v0.10.0	2.11.0
0.9.0	v0.9.0	2.10.0
0.8.0	v0.8.0	2.9.0
0.7.x	v0.7.0	2.8.0

快速入门

启动一个 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())

向表中写入一些数据：

# 创建一个单条目，并包含数据元素 [0, 1]。
client.insert([0, 1], priorities={'my_table': 1.0})

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

# 追加三个数据元素，并插入一个单条目，该条目引用所有这些元素，形式为 {'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})

  # 创建一个引用所有数据的条目。
  writer.create_item(
      table='my_table',
      priority=1.0,
      trajectory={
          'a': writer.history['a'][:],
          'b': writer.history['b'][:],
      })

  # 等待条目被插入并由服务器确认。
  writer.flush()

我们添加到 Reverb 中的条目可以通过采样来读取：

# client.sample() 返回一个生成器。
print(list(client.sample('my_table', num_samples=2)))

欲了解更多信息，请参阅 Reverb 教程，其中包含交互式教程。

详细概述

经验回放已成为训练离策略强化学习策略的重要工具。它被用于诸如 深度 Q 网络 (DQN)、软演员-评论家 (SAC)、深度确定性策略梯度 (DDPG) 以及 事后经验回放 (HER) 等算法中。然而，构建一个高效、易用且可扩展的经验回放系统可能颇具挑战性。为了获得良好的性能，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 控制，即数据项选择策略。结合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, et al.。

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 及其变体，以及Distributed Distributional Deterministic Policy Gradients。

队列

最多可容纳 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),
)

# 或者使用辅助类方法 `.queue`。
reverb.Table.queue(name='my_queue', max_size=1000)

使用队列的算法示例有 IMPALA 以及异步实现的 Proximal Policy Optimization。

数据项选择策略

Reverb 定义了几种可用于数据项采样或移除的选择器：

• Uniform: 在所有数据项中均匀采样。
• Prioritized: 根据存储的优先级按比例采样。
• FIFO: 选择最早的数据。
• LIFO: 选择最新的数据。
• MinHeap: 选择优先级最低的数据。
• MaxHeap: 选择优先级最高的数据。

以上任何一种策略都可以用于从表中采样或移除数据项。这为用户提供了灵活性，可以根据自身需求创建定制化的表。

速率限制

速率限制器允许用户强制执行某些条件，以控制何时可以从表中插入和/或采样数据项。以下是 Reverb 中当前可用的速率限制器列表：

• MinSize: 设置表中必须至少有多少个数据项才能进行采样。
• SampleToInsertRatio: 通过阻塞插入和/或采样请求来设定平均插入与采样的比率。这对于控制每个数据项在被移除前被采样的次数非常有用。
• Queue: 数据项在被移除前恰好被采样一次。
• Stack: 数据项在被移除前恰好被采样一次。

分片

Reverb 服务器彼此独立，在将系统扩展到多服务器架构时，数据不会跨多个节点复制。这使得 Reverb 不适合作为传统数据库使用，但同时也带来了易于扩展的优势——在可以接受一定程度数据丢失的情况下，系统可以轻松地水平扩展。

分布式系统只需增加 Reverb 服务器的数量即可实现水平扩展。当与兼容 gRPC 的负载均衡器配合使用时，只需将负载均衡目标地址提供给 Reverb 客户端，操作就会自动分配到不同的节点上。具体行为细节请参阅相关方法和类的文档。

如果您的环境中没有负载均衡器，或者需要更精细的控制，仍然可以通过几乎相同的方式扩展系统。只需增加 Reverb 服务器的数量，并为每台服务器创建单独的客户端即可。

检查点

Reverb 支持检查点功能，可以将 Reverb 服务器的状态和内容保存到持久化存储中。在创建检查点的过程中，Server 会序列化所有必要的数据和元数据，以便后续重建。在此期间，Server 会阻止所有传入的插入、采样、更新和删除请求。

创建检查点的操作由 Reverb 客户端发起：

# client.checkpoint() 返回检查点写入的路径。
checkpoint_path = client.checkpoint()

从检查点恢复 reverb.Server：

# 检查点管理器接受检查点写入的根目录路径。如果我们传递上述写入的检查点的根目录，
# 那么新的服务器将加载旧服务器写入的最新检查点。
checkpointer = reverb.platform.checkpointers_lib.DefaultCheckpointer(
  path=checkpoint_path.rsplit('/', 1)[0])

# 传递给 `tables=` 的参数必须与写入检查点的 `Server` 所使用的参数相同。
server = reverb.Server(tables=[...], checkpointer=checkpointer)

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

使用 reverb_server 启动 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: 一种经验回放框架},
      author={Albin Cassirer 和 Gabriel Barth-Maron、Eugene Brevdo、Sabela Ramos、Toby Boyd、Thibault Sottiaux 和 Manuel Kroiss},
      year={2021},
      eprint={2102.04736},
      archivePrefix={arXiv},
      primaryClass={cs.LG}
}

Reverb 快速上手指南

Reverb 是一个高效且易用的数据存储与传输系统，专为机器学习研究设计。它主要用作分布式强化学习算法的经验回放（Experience Replay）系统，同时也支持 FIFO、LIFO 和优先队列等多种数据结构。

环境准备

在开始之前，请确保满足以下系统要求和依赖条件：

• 操作系统：仅支持 Linux 系统（Windows 和 macOS 暂不支持）。
• Python 版本：支持 Python 3.9 - 3.11（注意：0.13.0 及以上版本已放弃对 Python 3.8 的支持）。
• 核心依赖：
  • TensorFlow（推荐安装以确保兼容性，版本需与 Reverb 对应）。
  • 底层库依赖：protoc 和 absl。

注意：Reverb 目前尚未针对生产环境进行加固，开发过程中可能会遇到崩溃或段错误（segfault），建议仅在研究和实验环境中使用。

TensorFlow 版本对应表

若通过 pip install dm-reverb[tensorflow] 安装，系统将自动匹配正确的 TensorFlow 版本。常见版本对应关系如下：

Reverb 版本	TensorFlow 版本	备注
Nightly	tf-nightly	开发版
0.14.0 / 0.13.0	2.14.0
0.12.0	2.13.0
0.11.0	2.12.0	首个支持 Python 3.11 的版本
0.10.0	2.11.0	最后一个支持 Python 3.7 的版本

安装步骤

推荐使用 pip 进行安装。国内用户若遇网络问题，可配置国内镜像源（如清华源、阿里源）加速下载。

1. 标准安装（推荐）

安装包含 TensorFlow 依赖的完整版本，确保兼容性：

pip install dm-reverb[tensorflow]

如果已单独安装 TensorFlow 或不需要版本检查，可运行：

pip install dm-reverb

2. 安装夜间构建版 (Nightly Builds)

如需体验最新功能（可能不稳定）：

pip install dm-reverb-nightly[tensorflow]

3. 国内镜像加速示例

使用清华大学镜像源加速安装：

pip install dm-reverb[tensorflow] -i https://pypi.tuna.tsinghua.edu.cn/simple

4. 源码编译

如需从源码构建（例如需要特定定制或使用 Docker 环境），请参考官方指南： 如何构建 Reverb

基本使用

以下是启动服务器、创建客户端、写入数据及采样数据的最小化工作流程。

1. 启动 Reverb 服务器

定义一个表（Table），配置采样策略（Sampler）、移除策略（Remover）、最大容量及速率限制器。

import reverb

server = reverb.Server(tables=[
    reverb.Table(
        name='my_table',
        sampler=reverb.selectors.Uniform(),       # 均匀采样
        remover=reverb.selectors.Fifo(),          # 先进先出移除
        max_size=100,                             # 最大容量 100
        rate_limiter=reverb.rate_limiters.MinSize(1)), # 至少 1 条数据才允许采样
    ],
)

2. 创建客户端并连接

连接到本地运行的服务器。

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

3. 写入数据

方式 A：插入单个条目 直接插入包含数据元素 [0, 1] 的单个条目。

# 创建单个条目和数据元素 [0, 1]
client.insert([0, 1], priorities={'my_table': 1.0})

方式 B：写入轨迹（Trajectory） 适用于序列数据，一个条目可引用多个数据元素。

# 追加三个数据元素，并创建一个引用所有元素的条目 {'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})

  # 创建引用所有数据的条目
  writer.create_item(
      table='my_table',
      priority=1.0,
      trajectory={
          'a': writer.history['a'][:],
          'b': writer.history['b'][:],
      })

  # 阻塞直到条目被服务器确认插入
  writer.flush()

4. 采样数据

从表中采样数据进行训练。client.sample() 返回一个生成器。

# 从 'my_table' 采样 2 个样本
print(list(client.sample('my_table', num_samples=2)))

提示：更多高级用法（如优先经验回放、队列模式等）及交互式教程，请参阅 Reverb Tutorial。