Ray 在 OpenRLHF 里做什么

OpenRLHF 的核心差异化技术不是 PPO 算法本身（DeepSpeed-Chat 也有），而是用 Ray 把异构进程编排起来。 这一章把 Ray 在 OpenRLHF 里扮演的角色拆开，看看它解决了什么问题、用了哪些 API、 以及它和 verl / TRL 选择的不同路径。

5.1为什么需要 Ray

RLHF 训练有几个特征导致单进程 PyTorch 无法承载：

1. 同时运行 4 种异构进程（actor / ref / critic / reward）+ 1 种采样进程（vLLM）；
2. 它们对显存的需求不同，需要灵活的资源分配；
3. 需要在不同 GPU 子集之间动态调度，把同一组 GPU 在不同阶段分给不同角色（hybrid engine）；
4. 多机训练时，跨节点通信不能用普通的 NCCL（角色间的接口不是张量级 all-reduce，而是 RPC）。

Ray 正好提供这套能力：

5.2Ray 三大抽象在 OpenRLHF 里

5.3BaseDistributedActor：所有训练进程的基类

openrlhf/trainer/ray/launcher.py 里定义了 BaseDistributedActor，它就是把 PyTorch distributed worker 包成 Ray Actor：

class BaseDistributedActor:
    def __init__(self, world_size, rank, master_addr, master_port):
        # 把 Ray Actor 当成一个 torch.distributed 进程
        os.environ["MASTER_ADDR"] = master_addr
        os.environ["MASTER_PORT"] = str(master_port)
        os.environ["WORLD_SIZE"] = str(world_size)
        os.environ["RANK"] = str(rank)
        os.environ["LOCAL_RANK"] = str(ray.get_gpu_ids()[0])

关键点：每个 Ray Actor 既是 Ray 调度的单位，也是 PyTorch DDP/DeepSpeed 的一个 rank。 两套并行抽象在这一层 fuse 在一起。

5.4RayActorGroup：N 个 Actor 组成的"角色集群"

每个 RLHF 角色（actor / ref / critic / reward）被一个 RayActorGroup 包起来。它本质上是：

class RayActorGroup:
    def __init__(self, num_nodes, num_gpus_per_node, ray_actor_type):
        self.world_size = num_nodes * num_gpus_per_node
        # 创建 placement group: world_size 个 GPU 打包成一组
        self.pg = placement_group(
            [{"GPU": 1, "CPU": 1}] * self.world_size,
            strategy="STRICT_PACK",        # 尽量装在一起
        )
        # 在 placement group 里启 world_size 个 actor
        self.actors = [
            ray_actor_type.options(
                num_gpus=1,
                placement_group=self.pg,
                placement_group_bundle_index=i,
            ).remote(world_size=self.world_size, rank=i, ...)
            for i in range(self.world_size)
        ]

    def async_run_method(self, method, *args, **kwargs):
        # 调所有 actor 的同名方法
        return [a.__ray_call__.remote(method, *args, **kwargs) for a in self.actors]

controller 通过 async_run_method 一次性调用所有 worker，类似 verl 的 WorkerGroup。

5.5典型放置：8 卡机怎么分

看 train_ppo_ray_hybrid_engine.sh 那一长串 --xxx.num_nodes / num_gpus_per_node：

--ref.num_nodes 1 --ref.num_gpus_per_node 8
--reward.num_nodes 1 --reward.num_gpus_per_node 8
--critic.num_nodes 1 --critic.num_gpus_per_node 8
--actor.num_nodes 1 --actor.num_gpus_per_node 8
--vllm.num_engines 4 --vllm.tensor_parallel_size 2
--train.colocate_all                              # ★ 关键

看上去要 8×4 = 32 张卡（actor 8 + ref 8 + critic 8 + reward 8）+ vLLM 4×2 = 8 张，总 40 张。但其实只有 8 张物理卡。colocate_all 让所有角色共享同一组 8 张卡，靠 sleep mode 分时复用。

第 8 章会专门讲三种"colocate 程度"对应的放置方式。

5.6controller 跑在哪

OpenRLHF 的controller（算法主循环）跑在哪里？

从源码看，openrlhf.cli.train_ppo_ray 的 main 函数本身就是 controller：

# 简化版
def main(args):
    ray.init()
    # 起 4 个 RayActorGroup
    actor_group = RayActorGroup(...)
    ref_group = RayActorGroup(...)
    critic_group = RayActorGroup(...)
    reward_group = RayActorGroup(...)
    vllm_engines = [LLMRayActor.remote(...) for _ in range(num_engines)]

    # 起 RemoteExperienceMaker：负责跑主循环
    experience_maker = RemoteExperienceMaker(
        actor_model_group, critic_model_group, reward_model_group,
        initial_model_group, ...
    )

    # 主循环
    for episode in range(max_episodes):
        for step in range(steps_per_episode):
            experiences = experience_maker.make_experience(prompts)
            actor_group.training_step(experiences)
            critic_group.training_step(experiences)

controller 是一个普通 Python 进程，不是 Ray Actor，跑在你提交 train_ppo_ray 的那台机器上。 所有 worker 都是 Ray Actor，通过 controller 调度。

5.7placement_group 放置策略

Ray 的 placement_group 有 4 种放置策略，OpenRLHF 默认用 STRICT_PACK：

这样能保证一个角色（如 actor）的 N 个 worker 都在同一节点，TP 等 intra-node 通信走 NVLink。 跨节点（DP）走 IB，不在同一 placement group。

5.8跨 group 怎么通信

四个 RayActorGroup 之间不是用 NCCL 而是 Ray RPC（actor 方法调用）。例如 controller 让 actor group 同步权重到 vLLM：

def broadcast_weights_to_vllm(self):
    # 在 actor 端
    for name, param in self.model.named_parameters():
        # nccl group：actor 和 vllm 之间提前建好的 nccl 通信组
        torch.distributed.broadcast(param, src=0, group=actor_vllm_nccl_group)

注意：

• 角色内部（同一 group N 个 worker）的训练并行用 NCCL；
• 角色之间（不同 group）的 RPC 用 Ray，但权重广播会临时建一个 NCCL group；
• 这个 NCCL group 的建立用 stateless_init_process_group，是 OpenRLHF 自己的工具。

5.9故障 / 监控

Ray Dashboard

启动 Ray 后，浏览器打开 http://<head_ip>:8265 能看到：

• 所有 Actor 的状态（alive / dead）；
• 每个 Actor 的 stderr（多机时尤其有用）；
• 资源占用：CPU / GPU / mem；
• placement group 视图。

故障调试

5.10OpenRLHF 用 Ray 的方式 vs verl / TRL

OpenRLHF 和 verl 在这个层面非常像（都是"controller 单进程 + worker Ray actor"），主要差别在具体的角色组织和训练后端选择。

5.11RayActorGroup 关键设计：duplicate_actors

openrlhf/trainer/ray/launcher.py:202 的 RayActorGroup 不是简单的 "N 个 actor 包一组"。它内部维护 duplicate_actors = ring_attn_size × tensor_parallel_size 的概念：

为什么需要 duplicate_actors？因为 ring attention（sequence parallel）和 tensor parallel 内部需要同一份数据 —— 一个 8 卡机如果开 ring_attn_size=2, tp_size=2，4 个 effective actor 各拿 1/4 batch，但每个 actor 内部又复制成 2×2 = 4 份去做 SP+TP 计算。

5.12batched RPC：round-robin 数据分片

RayActorGroup.async_run_method_batch(method_name, **kwargs)（launcher.py:319）是 OpenRLHF 把 Ray 用得跟"分布式数据加载器"差不多的关键 API：

1. 把传入的 batch（如 prompts）切成 effective_actors 份；
2. 用 ray.put() 一次性把切片放进 Ray object store（零拷贝传递）；
3. 每个 actor 调 method_name(chunk_i, **kwargs)，相同 ring/TP 副本拿同一份 chunk；
4. 返回 list[ray.ObjectRef]，调用方决定何时 ray.get()。

对照 async_run_method（不带 batch）：所有 actor 拿完整 args，不切分，常用在 init / save 这类全局动作。区分这两个 method 是看懂 controller 代码的关键。

5.13Ray Actor 三层基类

看 launcher.py 顶上能找到三层基类，每层多挂一点东西：

"我想加一个新角色 worker" → 继承 BaseModelActor 加 @ray.remote(num_gpus=1)，封装到 RayActorGroup(MyActor, ...)。

5.14这章带走的

• Ray 在 OpenRLHF 里负责异构进程编排 + GPU 资源调度 + 跨节点 RPC；
• 每个角色 = 一个 RayActorGroup，包 N 个 BaseDistributedActor；
• 同一 group 内训练通信走 NCCL；不同 group 通信走 Ray RPC + 临时 NCCL；
• controller 是普通 Python 进程，遗忘了它就理解不了"算法主循环为什么看着像单机"；
• 遇到"训练 hang"先查 Ray Dashboard / ray list actors。