最近在看 verl 源码时，看到其主要使用了 Ray 实现资源调度和分布式训练的能力，然后看了下其他的强化学习框架，也有很多框架使用了 Ray，感觉 Ray 已经成为了现代强化学习领域实现分布式训练和资源调度的主流选择，所以还是有必要了解下它的一些核心概念和特性，这样看到相关代码时才能大概知道其作用。
Ray 是什么？
“An open source framework to build and scale your ML and Python applications easily.”
Ray 是由加州大学伯克利分校RISELabi开发的，RISELab是大数据和AI领域顶尖的实验室，Spark、vLLM 等也是出自这个实验室，并且这个实验室的成员还创建了Databricks(Spark创始人)、Anyscale(Ray初创成员) 这样的知名公司。
Ray 是一个开源的AI 计算引擎和统一框架，用于轻松构建和扩展机器学习与通用 Python 应用程序。它提供了一套统一的、高性能的分布式计算原语，支持从原型开发到生产部署的全生命周期。
Ray由一个核心分布式运行时（Ray Core）和一组用于简化机器学习计算的人工智能库组成：
核心组件
Ray 采用模块化设计，包含多个高层库，覆盖 ML 全流程：
| 组件 | 功能 | 关键能力 |
|---|---|---|
| Ray Core | 分布式 Python 基础 | 任务（Tasks）、参与者（Actors）、对象存储（Objects）——构建任意分布式应用 |
| Ray Data | 数据加载与预处理 | 高效、可扩展的数据摄取、转换、批处理（支持文本、图像、LLM 等） |
| Ray Train | 分布式模型训练 | 支持 PyTorch、TensorFlow、XGBoost、Hugging Face 等主流框架 |
| Ray Tune | 任意规模的超参数调优 | 自动调参、实验管理、支持多种搜索算法（Optuna, HyperOpt 等） |
| Ray Serve | 模型在线服务 | 低延迟、高并发的模型部署，支持 LLM、多模型组合、动态批处理 |
| Ray RLlib | 强化学习 | 工业级 RL 算法库，支持单智能体/多智能体、离线/在线训练 |
✅ 所有组件均基于 Ray Core 构建，共享同一套调度、容错、资源管理机制。
Ray Core
Ray Core 是 Ray 分布式计算框架的核心模块，它提供了一组极简但强大的原语（primitives），用于构建和扩展通用的 Python 分布式应用，能让你只需几行代码，就能将普通的 Python 或 Java 函数及类转换为分布式的无状态任务和有状态参与者。主要包含三个核心概念：
- Tasks（任务）：远程执行的无状态函数。
- Actors（参与者）：远程执行的有状态类实例。
- Objects（对象）：存储在分布式对象存储中的数据。
使用 Task
Task 是在 Ray 集群上并行化你的 Python 函数的最简单方式。要创建和运行一个任务：
- 使用
@ray.remote装饰函数，使其成为可远程执行的任务。 - 使用
.remote()调用该函数，而不是进行常规的函数调用 - 使用用
ray.get()从返回的future对象获取结果。
# Define the square task.@ray.remotedefsquare(x):returnx*x# Launch four parallel square tasks.futures=[square.remote(i)foriinrange(4)]# Retrieve results.print(ray.get(futures))# -> [0, 1, 4, 9]✅ 优势：自动并行化、跨节点调度、容错。
使用 Actors
Task 是无状态的，而 Actor 则是有状态的，Actor 在方法调用之间会保持其内部状态。当你实例化一个Ray Actor 时：
- Ray 会在集群中的某个位置启动一个专用的工作进程
- Actor的方法在那个特定的工作进程上运行，并且能够访问和修改其状态
- Actor 按照接收方法调用的顺序串行执行这些调用，以保持一致性
要创建和实例化 Actor：
- 用
@ray.remote装饰类，创建远程 Actor 实例。 - 每个 Actor 在独立进程中运行，维护内部状态。
- 方法调用也使用
.remote()，按提交顺序串行执行。
# Define the Counter actor.@ray.remoteclassCounter:def__init__(self):self.i=0defget(self):returnself.idefincr(self,value):self.i+=value# Create a Counter actor.c=Counter.remote()# Submit calls to the actor. These calls run asynchronously but in# submission order on the remote actor process.for_inrange(10):c.incr.remote(1)# Retrieve final actor state.print(ray.get(c.get.remote()))# -> 10✅ 适用场景：状态管理、模型服务、参数服务器等。
Objects
Ray的分布式对象存储能高效地跨集群管理数据。在Ray中处理对象主要有三种方式：
- 隐式创建：当 Task 和 Actor 返回值时，这些值会自动存储在 Ray 的分布式对象存储中，并返回可在之后被检索的对象引用。
- 显式创建：使用 ray.put() 将对象直接放入存储中。
- 传递引用：可以将对象引用传递给其他 Task 和 Actor，这样可以避免不必要的数据复制，并支持延迟执行。
importnumpyasnp# Define a task that sums the values in a matrix.@ray.remotedefsum_matrix(matrix):returnnp.sum(matrix)# Call the task with a literal argument value.print(ray.get(sum_matrix.remote(np.ones((100,100)))))# -> 10000.0# Put a large array into the object store.matrix_ref=ray.put(np.ones((1000,1000)))# Call the task with the object reference as an argument.print(ray.get(sum_matrix.remote(matrix_ref)))# -> 1000000.0总结
| 概念 | 作用 | 关键 API |
|---|---|---|
| Task | 无状态并行计算 | @ray.remote,.remote(),ray.get() |
| Actor | 有状态远程对象 | @ray.remote class, 方法.remote() |
| Object | 分布式共享数据 | ray.put() |
Ray Core 的设计哲学：保持 API 极简，但组合能力极强，可构建从简单并行脚本到复杂分布式系统的各类应用。
