AstraFlow 架构解析：面向数据流的大规模强化学习系统

【Agentic RL / 强化学习框架】AstraFlow 架构解析：面向数据流的大规模强化学习系统

• 【Agentic RL / 强化学习框架】AstraFlow 架构解析：面向数据流的大规模强化学习系统
  • 0x00 概要
  • 0x01 现有问题
    • 1.1 现有 RL 框架的问题
    • 1.2 解决思路：dataflow-oriented coordination
  • 0x02 AstraFlow 是什么？
    • 2.1 用搭积木来理解 AstraFlow
    • 2.2 半专业人士——用系统架构视角来理解 AstraFlow
      • 四个核心组件
      • 这种设计解决了什么问题？
    • 2.3 专业人士——用系统设计视角来理解 AstraFlow
      • 现有 LLM RL 系统的根本局限
      • AstraFlow 的核心设计原则：Dataflow-Oriented Coordination
      • 数据流全景（一次训练迭代）
  • 0x03. 组件依赖关系与部署全景
    • 3.1 组件依赖图
    • 3.2 启动时序
    • 3.3 容错设计
    • 3.4 物理部署模式
    • 3.5. 技术栈总结
  • 0xFF 参考
• 附录：组件深度分析
  • 附录A. Dataflow 层 — 数据编排中枢
    • A.1 核心类及职责
    • A.2 DataAcquisition 双重线程设计
    • A.3 DataServing 缓冲区策略
    • A.4 RaaSPool 动态实例管理
    • A.5 版本屏障（Version Barrier）
    • A.6 支持的数据算法（可编程数据策略）
  • 附录B. RaaS — Remote Agentic Serving（远程推理服务）
    • B.1 核心类及职责
    • B.2 RaaS 的 HTTP API 设计
    • B.3 RaaS 作为 HTTP 客户端
    • B.4 权重同步流程（RaaS 侧）
    • B.5 推理引擎管理
  • 附录C. Train Worker — 分布式训练引擎
    • C.1 核心类及职责
    • C.2 Trainer 的 HTTP API 接口
    • C.3 AstraFlowPPOTrainer 训练循环
    • C.4 训练后端对比
    • C.5 配置系统
    • C.6 启动器
  • 附录D. WeightManager — 权重传输子系统
    • D.1 核心类及职责
    • D.2 权重传输全流程
    • D.3 双缓冲设计
    • D.4 GPU→CPU 复制策略对比
    • D.5 增量权重传输（Delta Transfer）
  • 附录E. Workflow — 可插拔的 Rollout 工作流
    • E.1 架构设计
    • E.2 如何添加一个新工作流
    • E.3 支持的工作流一览
    • E.4 支持的数据集
    • E.5 Reward Function API

0x00 概要

现有 LLM RL 系统每加一个新能力（多策略、弹性、跨地域）就要重写系统代码，根因是 Trainer 占据控制中心。AstraFlow 将其拆成三个独立组件：Dataflow Layer 负责数据协调，RaaS（Rollout-as-a-Service）定义 rollout 契约，Trainer + Weight Manager 只做优化与异步稀疏权重发布。三者以 streaming 交互，互不知晓对方拓扑，从而获得系统灵活性。

论文：AstraFlow: Dataflow-Oriented Reinforcement Learning for Agentic LLMs

代码：https://github.com/Infini-AI-Lab/astraflow

0x01 现有问题

1.1 现有 RL 框架的问题

1. 两个流水阶段的系统实现困境

LLM RL 训练分两个阶段：Rollout（vLLM/SGLang 生成 trajectory）与 Training（trainer 消费并优化）。主流的两种实现都有缺陷：

• Colocated synchronous（如 verl）：双方共享同一批 GPU、SPMD 跑，rollout 时 trainer 闲置、训练时 rollout 闲置，GPU 利用率差。
• Disaggregated async（如 AReaL、SLIME）：只是物理隔开、异步 overlap，计算分离 ≠ 控制分离。

2. 核心痛点：每加一个新能力就要重写系统

Agentic RL 需要支持 multi-policy 协同训练、弹性/异构/跨地域算力。现有系统如果增加任何能力，都要做专门的系统工程，组合起来复杂度爆炸。

3. 根因诊断：trainer 占据了控制中心

论文强调 "Compute decoupling is not enough." ——即便 AReaL 这类系统已把 GPU 物理分离，trainer 进程仍在充当全局 controller：决定何时启动 rollout、何时同步权重、用哪个 batch。其代价是 rollout 执行、dataflow 管理、训练、权重传输四件事之间缺少清晰的抽象边界。

1.2 解决思路：dataflow-oriented coordination

1. 翻转架构：从"老板下指令"到"流水线 + 货架"

• 传统模式：trainer 像老板，对 rollout workers 下指令，一切协调从中心发出。
• AstraFlow：Prompt 和 trajectory 先进 Dataflow Layer（共享货架），按规则缓存、过滤、采样；RaaS 自主取任务、投回结果；Trainer 自主取 batch、把权重 delta 投给 Weight Manager；任意 RaaS 按自己节奏拉取新权重。

2. 本质：streaming 取代同步 RPC 耦合

把 Spark/Flink/Ray Data 的 dataflow 范式迁移到 RL 训练：各方通过 stream 交互，只依赖接口契约，不感知对方的拓扑、数量、速度。这正是多策略协同、弹性伸缩、跨地域执行等所有系统灵活性的统一来源。

0x02 AstraFlow 是什么？

2.1 用搭积木来理解 AstraFlow

想象你在搭一座超级复杂的乐高城堡。通常我们搭乐高是这样的：

1. 先搭好城墙（造数据）
2. 然后找老师检查城墙好不好（推理/评估）
3. 根据老师意见修改城墙（训练模型）
4. 回到第 1 步

但问题来了：如果老师检查得很慢，搭城墙的工人就只能干等着。而且如果只有一个人搭城堡，有人请假整个项目就停了。

AstraFlow 就像是一个聪明的城堡建造队长，它想出了一个更好的办法：

• 分开干活，互不干扰：专门派人去建城墙（这叫"RaaS"），专门派人去根据反馈修改城墙（这叫"Trainer"），还有一个仓库管理员（这叫"Dataflow"）负责把建好的城墙收起来、把修改完的版本发给大家。大家不用等来等去。
• 可以很多人同时做不同的部分：不是只有一个人在搭城堡。而是有人搭城门、有人搭塔楼、有人搭护城河，大家各干各的，最后拼在一起。
• 随时加人，不用停工：如果发现搭城墙太慢了，直接叫几个新人来帮忙，新人一到就可以开始干活，不需要重新启动整个工程。
• 仓库在全世界都有人：AstraFlow 允许你在北京搭一部分城墙，在美国搭另一部分，然后通过网络把两部分拼起来——哪怕网络很慢也能正常工作。

简单说，AstraFlow 是一个能让 AI 训练变得更快、更灵活、更省钱的系统。它就像给 AI 训练装上了"超强项目管理器"。

2.2 半专业人士——用系统架构视角来理解 AstraFlow

传统 RL 训练循环：模型生成回答（rollout）→ 打分（reward）→ 根据分数更新模型（train）→ 用新模型继续生成。过这类系统的痛点如下：

1. 训得慢：生成回答和训练是串行的，生成时 GPU 空闲，训练时生成引擎空闲。
2. 改不动：想加一个"多模型协作训练"（比如一个模型出题、一个模型解题），需要对整个系统大改。
3. 不灵活：想用便宜的 GPU 做生成、贵的 GPU 做训练，系统不支持。想动态加机器，要停机重启。

AstraFlow 的核心理念是：把传统以 Trainer（训练器）为中心的控制架构，改造成以 Dataflow（数据流）为中心的协作架构。

四个核心组件

1. Dataflow（数据流层）：中央协调器。它不参与计算，只做数据管理——从 RaaS 收集 rollout 结果，缓冲、过滤、排序，然后喂给 Trainer。它也是"交通警察"，决定什么时候让 RaaS 更新权重。

2. RaaS（Remote Agentic Serving）：推理服务层。它只负责一件事——接收 Prompt，运行为每个任务定制的工作流（workflow），生成回答，返回结果。它不知道 Trainer 是怎么训练的，也不需要知道。

3. Trainer（训练器）：分布式训练引擎。它只从 Dataflow 拉取训练批次，做前向/反向传播，更新模型权重。它不知道 rollout 是怎么生成的。

4. WeightManager（权重管理）：独立的权重传输组件。它负责把 Trainer 更新后的模型权重通过 TCP 高效地传给 RaaS，让 RaaS 用新权重继续生成。

这种设计解决了什么问题？

• 完全异步：Trainer 训练一批数据的同时，RaaS 已经在生成下一批了。双方互不阻塞。
• 多策略协作：每个模型有自己的 Trainer 和权重流。Dataflow 通过一个"版本屏障"确保所有模型的版本一致后，才通知 RaaS 加载——无需修改系统代码。
• 弹性伸缩：想加推理能力？启动新 RaaS 实例，注册到 Dataflow 即可。想减？直接关掉。Dataflow 的 RaaSPool 自动处理心跳检测和负载均衡。
• 异构跨地域：RaaS 可以在不同规格的 GPU 上运行，甚至跨数据中心。WeightManager 的增量权重传输（只发送变化的部分）让跨地域训练成为可能。
• 可替换组件：Trainer 和 RaaS 都是"可插拔"的。你可以用自己写的训练器替换 AstraFlow 内置的 PPO Trainer，只要它和 Dataflow 通过 6 个 HTTP API 通信即可。

2.3 专业人士——用系统设计视角来理解 AstraFlow

我们来看更深层的问题。

现有 LLM RL 系统的根本局限

现有系统（如 verl、AReaL、SLIME）虽然实现了 rollout 和 training 的计算分离（disaggregation），但控制逻辑仍然集中在 Trainer 中。Trainer 不仅负责训练，还负责：

• 决定什么时候做 rollout、做多少
• 管理数据选择、回放（replay）、新鲜度过滤
• 协调权重同步

这种"Trainer-centered control"的设计，导致每增加一个新特性（弹性 rollout、多策略协作、跨地域训练）都需要对 Trainer 做专门的系统级修改。下表来自论文清晰地展示了这个差距：

AstraFlow 的核心设计原则：Dataflow-Oriented Coordination

AstraFlow 的核心洞察是：解耦不应该只发生在计算层面，更应该发生在控制层面。

传统架构中控制流：

Trainer → 控制 rollout 调度 → rollout 引擎
Trainer → 控制数据选择 → 训练数据
Trainer → 控制权重同步 → rollout 引擎

AstraFlow 改成了：

Dataflow ←→ 数据生产/消费 ←→ RaaS (自治控制循环)
Dataflow ←→ 数据生产/消费 ←→ Trainer (自治控制循环)
RaaS ←→ 权重拉取 ←→ Trainer (通过 WeightManager)

每个组件运行自己的自治控制循环，仅通过最小化的数据和权重接口交互。这使得：

• 多策略协作成为工作流层面的变化，而非系统层面的修改
• 弹性伸缩成为 RaaS 的生命周期管理，不涉及 Trainer
• 数据算法（动态采样、GRESO、回放）成为 Dataflow 的可编程策略，不侵入 Trainer

数据流全景（一次训练迭代）

1. RaaS 推理 (持续, 与 Data Acquisition 相互独立地并发运行):
   ┌────────────────────────────────────────────────────────┐
   │ RaaS 实例 (每个 GPU × 4)                                │
   │  接收 /submit → 执行 workflow (arun_episode)             │
   │    ├─ 调用推理引擎 (SGLang/vLLM) 生成 token              │
   │    ├─ 与环境交互 (代码执行/网页搜索/工具调用等多轮)          │
   │    ├─ 调用奖励函数计算得分                                │
   │    └─ 返回结构化 rollout (prompt, response, reward, ...) │
   └────────────────────────────────────────────────────────┘

2. Data Acquisition (持续后台运行, 与 RaaS 推理重叠):
   ┌─────────────────────────────────────────────┐
   │ DataAcquisition                              │
   │  ├─ Submit线程: 检查 RaaS 容量 → submit_auto │
   │  └─ Collect线程: pull_completed → 新鲜度过滤   │
   │              → RolloutBuffer                 │
   └─────────────────────────────────────────────┘
                ↓
   rollout 数据在 Dataflow 中持续缓冲、过滤、归一化

3. Trainer 获取批次:
   Trainer ──GET /batch──► Dataflow
   Dataflow ──返回 padded batch (fresh + replay 混合)

4. Trainer 训练步骤:
   Trainer ──forward/backward/optimizer──► PPO/GRPO/M2PO
   Trainer ──wm.offload()──► GPU→CPU (shard copy, ~0.5s for 1.7B)
   Trainer ──save checkpoint──► disk

5. 权重同步 (与步骤 3-4 异步重叠):
   Trainer ──POST /notify_version──► Dataflow
   Dataflow 版本屏障 (多模型: 等待所有 model_id 达到同一版本)
   Dataflow ──POST /notify_version──► 每个 RaaS (每个模型)
   RaaS ──GET /get_buffer_info──► Trainer SenderAgent (首次)
   RaaS ──POST /register_sglang_instance──► Trainer (首次)
   RaaS ──POST /request_transfer──► Trainer SenderAgent
   RaaS ←── 6路并行 TCP ── Trainer (sendfile zero-copy)
   RaaS: save safetensors → pause engine → load → resume

6. 回到步骤 1/2 (持续异步, 三轮流水线同时进行)

关键时序观察：

• Trainer GPU 只在 offload() 期间阻塞 (~0.5s)，其余时间在计算
• TCP 传输和 Trainer 训练完全重叠，因为双缓冲设计
• get_batch 是真正的同步点：Dataflow 确保 service_version 追上 trainer_version 后才提供下一批数据
• 增量权重传输（delta transfer）在 Qwen3-14B 上达到 98.9% 以上的稀疏度，每次迭代只需传输 ~1.5GB 而非 28GB

0x03. 组件依赖关系与部署全景

3.1 组件依赖图

3.2 启动时序

1. RaaS 启动:
   RaaS ──POST /register_raas──► Dataflow  (加入 RaaS 池)

2. Dataflow 注册工作流到 RaaS:
   Dataflow ──POST /register_workflow──► 每个 RaaS 实例

3. Trainer 启动:
   Trainer ──POST /ready──► Dataflow  (报告 batch_size 和 sender endpoint)
   Trainer 发送 recovered_version（恢复时）

4. Dataflow 开始数据采集:
   DataAcquisition 开始 submit + collect 循环
   (仅在 RaaS 和 Trainer 都就绪后开始)

5. 训练循环反复迭代...

3.3 容错设计

3.4 物理部署模式

所有组件作为独立进程运行，可以通过不同的启动器部署在不同的机器上：

3.5. 技术栈总结

0xFF 参考

下一代全异步RL架构？AstraFlow 论文解读

附录：组件深度分析

附录A. Dataflow 层 — 数据编排中枢

源码路径：astraflow/dataflow/

Dataflow 是 AstraFlow 的核心协调组件，它位于 RaaS（上游）和 Trainer（下游）之间，负责所有数据的管理和流转。它不参与计算，只做数据的管理和路由。

A.1 核心类及职责

A.2 DataAcquisition 双重线程设计

DataAcquisition 运行两个后台线程，实现rollout 生成和训练数据准备的完全解耦：

Submit 线程：

1. 调用 RaaSPool.get_raas_availability() 获取所有 RaaS 实例的容量信息
2. 通过 submit_auto() 将 prompt 路由到可用槽位最多的 RaaS 实例（容量负载均衡）
3. 支持 prompt curator（选择性 rollout）、动态采样等数据算法

Collect 线程：

1. 并行调用所有 RaaS 实例的 pull_completed()
2. 对返回的 rollout 应用过滤（新鲜度、奖励分布）
3. 计算奖励归一化和分组统计
4. 写入 RolloutBuffer

A.3 DataServing 缓冲区策略

AstraDataServing 管理每个模型的缓冲队列，核心设计：

• Fresh queue + Replay queue：按 replay_ratio 混合新鲜数据和回放数据
• 新鲜度过滤：每个 rollout 带有生成时的模型版本号。current_version - max_staleness 之前的样本自动丢弃
• 奖励归一化：按模型/策略分组归一化，避免奖励尺度差异
• EDF 排序：最早截止时间优先，优先发送最新的 rollout
• Padded batch：返回 [batch, seq_len, ...] 格式的填充张量

A.4 RaaSPool 动态实例管理

RaaSPool 实现了弹性 RaaS 实例池：

• 注册/注销：RaaS 实例通过 HTTP 动态加入或离开。新加入的实例会被"追赶"到当前权重版本
• 心跳检测：后台线程每 ~30s 轮询 /status，连续 2 次失败则标记为不可用
• Suspect-and-confirm 机制：数据路径上的失败先标记 suspect（跳过路由），心跳线程通过独立 /status 调用确认后才会注销。这避免了瞬时网络抖动导致的误判
• 容量路由：submit_auto 根据每个 RaaS 的 available_slots 做加权路由
• 并行收集：pull_completed 对所有存活实例做 fan-out，并行收集结果

这意味这你可以随时增加 RaaS 实例来扩展推理吞吐量——无需重启任何组件。

A.5 版本屏障（Version Barrier）

在多模型训练中，版本屏障是 Dataflow 的核心协调机制：

• 每个 Trainer 独立完成训练步骤后，调用 POST /notify_version 通知 Dataflow
• Dataflow 等待所有已注册的 model_id 达到同一版本
• 达到屏障后：
  • 非 eval 步骤：异步 fire-and-forget，每个模型的权重加载独立进行，新鲜度过滤器会处理短暂的不一致
  • eval 步骤：同步，屏障 leader 按顺序加载所有模型，保证原子性

这解决了多策略协作训练中的核心挑战：如何在不引入集中式调度器的情况下协调多个并行的 Trainer。

A.6 支持的数据算法（可编程数据策略）

Dataflow 暴露了可编程数据算法接口，这些都可以作为 Dataflow 策略实现，不需要修改 Trainer 或 RaaS：

附录B. RaaS — Remote Agentic Serving（远程推理服务）

源码路径：astraflow/raas/

RaaS 是推理服务层，将 rollout 生成抽象为"纯 agent 推理服务"。它的核心思想是：rollout 引擎不需要知道它是怎么被采样的、是不是要回放、数据要送到哪个 Trainer——它只需要接收 prompt、执行 workflow、返回结果。

B.1 核心类及职责

B.2 RaaS 的 HTTP API 设计

RaaS 的 API 设计体现了"最小化接口"原则——只暴露必要的方法，内部实现完全封装。

健康与状态（JSON）：

• GET /status — 就绪检查（ready/idle/starting/error），也作为心跳
• GET /availability — 容量信息（available/inflight slots）

Rollout（pickle 序列化）：

• POST /register_workflow — 注册工作流类 + 奖励函数
• POST /submit — 提交一个 prompt → 返回 task_id
• POST /pull — 批量拉取完成的 rollout 结果

权重同步：

• POST /notify_version — 权重更新通知，携带 {model_id, version, sender_endpoint}

评估：

• POST /eval_start / eval_submit / eval_pull / eval_end

生命周期：

• POST /shutdown — 优雅关闭

API 设计的关键点：所有数据通过 pickle/cloudpickle 序列化，Content-Type 为 application/octet-stream。这是因为 rollouts 中包含复杂 Python 对象（tool calls、环境交互记录等），JSON 无法表达。GET /status 和 GET /availability 是例外，返回 JSON。

B.3 RaaS 作为 HTTP 客户端

RaaS 不只是一个服务器，它还作为客户端与两个外部服务交互：

1. 启动时注册到 Dataflow：POST /register_raas 加入全局 RaaS 池
2. 从 Trainer 拉取权重（权重同步时）：
   • GET /get_buffer_info — 查询张量布局（首次）
   • POST /register_sglang_instance — 注册为权重接收方（首次）
   • POST /request_transfer — 请求 TCP 权重传输（每次）

B.4 权重同步流程（RaaS 侧）

Dataflow ──POST /notify_version──► RaaS
  {model_id, version, sender_endpoint}
                                   │
                                   │ acquire per-model asyncio.Lock
                                   │
                                   ├─ GET  /get_buffer_info ──► Trainer SenderAgent (首次)
                                   ├─ POST /register_sglang_instance ──► Trainer (首次)
                                   ├─ POST /request_transfer ──► Trainer
                                   ├─ ◄── TCP 6路并行传输 + ZMQ 信号
                                   │
                                   ├─ 保存 safetensors 到 /dev/shm
                                   ├─ engine.pause_generation()
                                   ├─ engine.load_weights_from_path(...)
                                   └─ engine.continue_generation()

B.5 推理引擎管理

引擎通过子进程管理（subprocess.Popen），RaaS 负责其完整生命周期——启动、健康检查、暂停（权重加载时）、恢复。这种设计意味着引擎运行在独立的进程空间中，权重加载失败不会影响 RaaS 本身的健康状态。

附录C. Train Worker — 分布式训练引擎

源码路径：astraflow/train_worker/

Train Worker 是 AstraFlow 的分布式训练引擎，支持 FSDP2 和 Megatron-Core 两种训练后端。它的设计原则是可替换性——与 Dataflow 仅通过 6 个 HTTP API 交互，没有任何共享库或类继承耦合。任何训练框架（PyTorch、JAX、自定义）只要实现这 6 个 API 就可以接入 AstraFlow。

C.1 核心类及职责

C.2 Trainer 的 HTTP API 接口

Trainer 与外部交互的全部接口：

Trainer → Dataflow（3 个出站调用）：

RaaS → Trainer（3 个入站端点，由 WeightManager 的 SenderAgent 提供）：

C.3 AstraFlowPPOTrainer 训练循环

while True:
    # 1. 从 Dataflow 获取训练批次
    batch = astraflow_client.get_batch()

    # 2. 分发到各 GPU
    distributed_batch = distribute(batch)

    # 3. 训练步骤（前向/反向/优化）
    loss = train_step(distributed_batch)

    # 4. 权重卸载到共享内存
    wm.offload(model.named_parameters(), version, rank, world_size)

    # 5. 保存检查点
    save_checkpoint(model, optimizer, version)

    # 6. 通知 Dataflow 新版本
    astraflow_client.notify_version_async(version)

C.4 训练后端对比

C.5 配置系统

astraflow/train_worker/api/cli_args.py（~1733 行）是集群配置的单一真理源，使用 Hydra/OmegaConf 验证。

主要配置 dataclass：

• PPOConfig / GRPOConfig — RL 算法参数（学习率、clip ratio、GAE lambda 等）
• TrainEngineConfig — 训练引擎配置
• TrainDatasetConfig / ValidDatasetConfig — 数据集路径、tokenizer、max_length 等
• ModelConfig — 模型架构配置

C.6 启动器

附录D. WeightManager — 权重传输子系统

源码路径：astraflow/core/weight_manager/

WeightManager 是 AstraFlow 中唯一被 Trainer 和 RaaS 共同引用的组件，但两者之间没有直接依赖：

train_worker/ ──import──► weight_manager/
raas/         ──import──► weight_manager/transfer/
train_worker/ ◄──► raas/   (没有交叉依赖)

D.1 核心类及职责

D.2 权重传输全流程

Trainer GPU
    │
    │ wm.offload(model.named_parameters(), version, rank, world_size)
    ▼
WeightManager (GPU → CPU 复制)
    │
    │ 策略选择:
    │   ├─ Shard copy (fast path): 每个 rank 只复制自己的 shard，无 NCCL 通信
    │   └─ All-gather (fallback):  跨节点非标准 sharding，rank 0 复制全量
    │
    │ 写入 /dev/shm 双缓冲区的空闲半区
    │ 原子切换缓冲区索引
    ▼
SenderAgent (CPU 子进程)
    │
    │ 暴露 HTTP 端点供 RaaS 调用:
    │   GET  /get_buffer_info        (查询张量布局)
    │   POST /register_sglang_instance (注册 RaaS)
    │   POST /request_transfer       (TCP 传输)
    │
    │ 可选: 异步计算增量 (delta):
    │   - 比较两个缓冲区半区找出变化的参数
    │   - 只发送变化的 bf16 元素
    │   - 1.7B 模型: ~95MB delta vs 3.9GB full → 40x 压缩
    ▼
RaaS (TCP 拉取)
    │
    │ 6 路并行 TCP 流 (sendfile zero-copy)
    │ ZMQ 信号确认传输完成
    │ 保存 safetensors 到 /dev/shm
    │ engine.pause_generation()
    │ engine.load_weights_from_path()
    │ engine.continue_generation()

D.3 双缓冲设计

Step N:     Trainer 写 Half 0  ←→  TCP 读 Half 1 (RaaS 拉取)
Step N+1:   Trainer 写 Half 1  ←→  TCP 读 Half 0 (RaaS 拉取)

• 缓冲区在 /dev/shm 中分配，大小为 2× 模型大小
• 缓冲区索引切换是一个 Python int 赋值（GIL 下原子操作）
• 无需锁竞争——Trainer 和 SenderAgent 永远不会争用同一个半区
• 优化：mlock + 透明大页 + madvise(MADV_SEQUENTIAL | MADV_WILLNEED) + CUDA 主机注册

D.4 GPU→CPU 复制策略对比

Shard copy 是标准 FSDP2 下的快路径——每个 rank 只复制自己的本地 shard 到缓冲区中的正确偏移位置，无需任何跨 rank 通信，所有 rank 并行写入。这是默认情况下的路径。

D.5 增量权重传输（Delta Transfer）

这是使跨地域异构训练可行的关键技术：

• 原理：只发送两个迭代之间发生变化的权重参数（bf16 元素）
• 压缩率：30-40×（1.7B 模型：~95MB delta vs 3.9GB full）
• 稀疏度：98.4%-99.2%（Qwen3-14B 在数学任务上达到 98.9%+）
• 计算方式：SenderAgent 子进程中异步比较两个缓冲区半区，和 Trainer 的检查点保存/日志记录重叠执行
• 应用方式：RaaS 侧使用 mmap 原地打补丁（~1.1s for 1.7B）
• 定期全量同步：每 20 次迭代做一次全量同步，防止增量误差累积

附录E. Workflow — 可插拔的 Rollout 工作流

源码路径：astraflow/core/workflow/

Workflow 是 AstraFlow 的主要扩展点。所有 rollout 工作流通过 @register_workflow 装饰器注册，所有奖励函数通过 @register_reward 注册。__init__.py 中导入所有 impl/ 和 reward/ 模块，使得注册装饰器在 import 时自动执行。

E.1 架构设计

astraflow/core/workflow/
├── api/                              # 基础接口
│   ├── workflow_api.py               # RolloutWorkflow ABC (arun_episode)
│   ├── reward_api.py                 # AsyncRewardWrapper (ProcessPoolExecutor)
│   ├── engine_api.py                 # InferenceEngine Protocol, EngineGroup
│   └── io_struct.py                  # ModelRequest/Response 数据类
│
├── registry.py                       # WORKFLOW_REGISTRY, REWARD_REGISTRY
│
├── impl/                             # 具体工作流实现
│   ├── rlvr.py                       # 标准 RLVR 数学推理
│   ├── multi_turn.py                 # 多轮对话 agent
│   ├── solve_and_verify.py           # Solver + Verifier 协作
│   ├── actor_and_verify.py           # Actor + Verify 多模型
│   ├── plan_and_solve.py             # 先规划后求解
│   ├── sm_lg_router.py               # 小模型/大模型路由
│   ├── spawn.py                      # 递归 agent 生成
│   ├── vision_rlvr.py                # 视觉-语言 RLVR
│   ├── code_actor_and_verify*.py     # 代码生成 + 验证
│   ├── code_solve_and_select.py      # 代码生成 + 选择器
│   ├── agentbench/                   # ALFWorld, WebShop 具身 agent
│   └── asearcher/                    # 搜索增强 agent
│
└── reward/                           # 奖励函数
    ├── math_verify.py                # @register_reward("math_verify")
    ├── livecodebench_reward.py       # @register_reward("livecodebench")
    ├── human_eval_reward.py          # @register_reward("human_eval")
    ├── geometry3k.py                 # vision math (legacy path)
    ├── clevr_count_70k.py            # CLEVR counting (legacy path)
    ├── textcraft_success.py          # TextCraft 任务成功
    └── deepdive_success.py           # DeepDive 任务成功

E.2 如何添加一个新工作流

只需四步：

1. 在 astraflow/core/workflow/impl/<name>.py 中创建新模块
2. 继承 RolloutWorkflow（来自 astraflow/core/workflow/api/workflow_api.py），实现 async arun_episode()
3. 用 @register_workflow("<name>") 装饰
4. 在 astraflow/core/workflow/__init__.py 中添加 import astraflow.core.workflow.impl.<name> 行

在 recipe 的 YAML 中通过 workflow_spec.workflow_cls: "<name>" 引用即可。

E.3 支持的工作流一览

E.4 支持的数据集

所有数据集加载器位于 astraflow/dataflow/dataset/，通过 YAML 中的 module:function 路径引用（如 dataset_fn: "astraflow.dataflow.dataset.deepscaler:get_deepscaler_rl_dataset"），没有中央注册表。

E.5 Reward Function API

奖励函数的标准签名：(prompt, completions, prompt_ids, completion_ids, **data)

其中 **data 携带数据集特定字段（如 answer），返回 float。

关键设计：阻塞式奖励代码应通过 AsyncRewardWrapper（astraflow/core/workflow/api/reward_api.py）包装，使用 ProcessPoolExecutor 将奖励计算推到独立进程，避免阻塞异步 workflow 的事件循环。