执行摘要

• 一句话：磁盘级delta权重同步，替代NCCL传输
• 推荐动作：值得精读，特别是 disk_delta.py 中的工具函数设计以及 UpdateWeightFromDiskDelta 如何通过继承复用 UpdateWeightFromDistributed。配置参数的变化也需关注以便兼容旧脚本。

功能与动机

PR描述：For non-colocated training/inference across clusters, ship only the changed bytes between weight syncs instead of a full checkpoint; replaces the NCCL delta transport from #1806. 使rollout侧与trainer解耦，可自由使用任何低精度或并行方案。

实现拆解

1. 基础工具层：新增 slime/utils/disk_delta.py，提供字节级delta编解码（overwrite_encode）、多种校验和（xxh3-128/blake3/adler32）、本地检查点初始化（init_local_checkpoint）和delta应用（apply_deltas），使用线程池并行处理张量。
2. 发布端核心：新增 slime/backends/megatron_utils/update_weight/update_weight_from_disk_delta.py，实现 UpdateWeightFromDiskDelta 类。首次调用捕获基线快照，后续调用对每个HF张量进行字节级diff，压缩后写入标准HF检查点目录。支持自定义pre-push钩子。
3. Rollout引擎端：修改 slime/backends/sglang_utils/sglang_engine.py，新增 sync_local_checkpoint() 方法，调用 init_local_checkpoint 确保本地副本存在，然后调用 apply_deltas 应用发布的最新delta。启动时在后台线程预拷贝基座检查点以重叠启动时间。
4. 配置与验证：修改 slime/utils/arguments.py，将 --update-weight-encoding 替换为 --update-weight-delta-encoding（choices: xor/overwrite），新增 --update-weight-delta-checksum，删除 --update-weight-delta-dir，调整帮助文本明确delta只支持disk transport。验证逻辑内联化。
5. 清理与依赖：删除旧文件 update_weight_from_distributed_delta.py（864行），删除旧的示例脚本 examples/delta_weight_sync/run-glm4.7-355B-A32B-delta.sh，更新sglang patch去除delta接收端代码，调整actor.py和rollout.py的导入。文档同步更新。

关键文件：

• slime/backends/megatron_utils/update_weight/update_weight_from_distributed_delta.py（模块 增量传输(旧)；类别 source；类型 deletion；符号 ParamDiff, EncodedChunk, empty, _checksum）: 被完全删除的旧delta同步实现，包含NCCL和磁盘双传输方式，是本次重构的直接替代目标。
• slime/backends/megatron_utils/update_weight/update_weight_from_disk_delta.py（模块 权重同步；类别 source；类型 dependency-wiring；符号 UpdateWeightFromDiskDelta, init, connect_rollout_engines, disconnect_rollout_engines）: 新增的磁盘delta同步实现，核心发布端类，继承UpdateWeightFromDistributed，只支持disk transport。
• slime/utils/disk_delta.py（模块 增量工具；类别 source；类型 dependency-wiring；符号 overwrite_encode, _Adler32, checksum, init_local_checkpoint）: 新增的磁盘delta工具模块，提供overwrite_encode、checksum、init_local_checkpoint等核心函数。
• slime/backends/sglang_utils/sglang_engine.py（模块 SGLang引擎；类别 source；类型 core-logic；符号 sync_local_checkpoint）: 修改SGLang引擎，新增sync_local_checkpoint方法，去掉旧delta代码。

关键符号：UpdateWeightFromDiskDelta.init, UpdateWeightFromDiskDelta.update_weights, UpdateWeightFromDiskDelta._capture_baseline, UpdateWeightFromDiskDelta._publish, UpdateWeightFromDiskDelta._write_delta_files, overwrite_encode, checksum, init_local_checkpoint, apply_deltas, sync_local_checkpoint

关键源码片段

slime/backends/megatron_utils/update_weight/update_weight_from_disk_delta.py

新增的磁盘delta同步实现，核心发布端类，继承UpdateWeightFromDistributed，只支持disk transport。

from slime.utils.disk_delta import NUM_WORKERS, checksum, make_tensor_reader, overwrite_encode
​
class UpdateWeightFromDiskDelta(UpdateWeightFromDistributed):
    """
    Delta weight sync over a shared filesystem.
    PP-src ranks diff each gathered HF tensor against a CPU snapshot of the previous sync
    and publish the changes as a canonical HF checkpoint dir;
    every rollout host applies the delta into its local checkpoint and reloads via the ordinary
    update_weights_from_disk path, so sglang needs no delta support.
    """
​
    def __init__(self, args, model, weights_getter, *, model_name, quantization_config):
        super().__init__(args, model, weights_getter, model_name=model_name, quantization_config=quantization_config)
        self.delta_dir = args.update_weight_disk_dir
        os.makedirs(self.delta_dir, exist_ok=True)
        self.delta_encoding = args.update_weight_delta_encoding # "xor" or "overwrite"
        self.checksum_algorithm = args.update_weight_delta_checksum # e.g. "xxh3-128"
        self._snapshot: dict[str, np.ndarray] = {}
        self._baseline_captured = False
        self._commit_hook: Callable | None = None
        if args.custom_delta_pre_push_path:
            from slime.utils.misc import load_function
            self._commit_hook = load_function(args.custom_delta_pre_push_path)
​
    @torch.no_grad()
    def update_weights(self) -> None:
        if not self._baseline_captured:
            self._capture_baseline()
            self._baseline_captured = True
            return
        self.weight_version += 1
        if dist.get_rank() == 0:
            ray.get([engine.pause_generation.remote() for engine in self.rollout_engines])
            ray.get([engine.flush_cache.remote() for engine in self.rollout_engines])
        dist.barrier(group=get_gloo_group())
        self._publish()
        self._reload_engines()
        self._record_metrics()
​
    def _capture_baseline(self) -> None:
        if dist.get_rank() == 0:
            shutil.rmtree(self.delta_dir, ignore_errors=True)
            os.makedirs(self.delta_dir, exist_ok=True)
            if self._commit_hook is not None:
                self._commit_hook(self.args, self.delta_dir, list(self.rollout_engines))
        dist.barrier(group=get_gloo_group())
        read_hf = make_tensor_reader(self.args.hf_checkpoint)
        for name, tensor in self._iter_hf_tensors():
            try:
                self._snapshot[name] = read_hf(name)
            except KeyError:
                self._snapshot[name] = tensor.cpu().numpy()

slime/backends/sglang_utils/sglang_engine.py

修改SGLang引擎，新增sync_local_checkpoint方法，去掉旧delta代码。

def sync_local_checkpoint(self, target_version: int):
    """Apply the published deltas into this host's local checkpoint up to target_version;
    the engine reloads it afterwards. Assumes this actor shares the checkpoint filesystem
    with the sglang it drives (true for slime-launched engines)."""
    from slime.utils.disk_delta import apply_deltas, init_local_checkpoint
​
    # idempotent: ensure local checkpoint exists
    init_local_checkpoint(self.args.update_weight_local_checkpoint_dir, self.args.hf_checkpoint)
    # non-POSIX filesystems may need a pre-read hook for consistency
    if self.args.custom_delta_pre_read_path:
        from slime.utils.misc import load_function
        load_function(self.args.custom_delta_pre_read_path)(
            self.args.update_weight_disk_dir, target_version)
    apply_deltas(
        self.args.update_weight_local_checkpoint_dir,
        self.args.update_weight_disk_dir,
        target_version,
    )

评论区精华

Review评论只有1条，无实质性内容；设计决定已在PR body和commit message中明确记录。主要讨论包括：将delta传输从NCCL迁移到磁盘，使得rollout引擎无需了解delta细节；配置参数简化，移除旧alias。

• 整体设计评估 (design): 设计被接受，无需修改。

风险与影响

• 风险：

  1. 文件系统一致性：非POSIX文件系统可能出现跨主机写后读不一致，通过custom_delta_pre_read_path钩子解决。
  2. 锁竞争：使用fcntl.flock进行主机级互斥，高并发时可能成为瓶颈。
  3. 配置兼容性：旧配置中使用 --update-weight-encoding 和 --update-weight-delta-dir 的脚本需要迁移到新参数。
  4. 性能风险：磁盘I/O和压缩/解压缩可能增加同步延迟，但利用线程池和zstd轻量压缩缓解。
     - 影响：用户/系统：需要更新启动参数；delta模式仅支持disk transport，不再支持NCCL delta；跨集群训练/推理支持更灵活部署，但依赖共享文件系统。
     团队：代码结构更清晰，decoupled design便于后续维护；新增 disk_delta.py 工具模块可供其他组件复用。

• 风险标记：核心路径变更, 配置兼容性, 跨文件系统依赖, 锁竞争风险

关联脉络

• PR #1806 PR #1806 (前序: NCCL delta weight sync): 本PR替代了#1806中的NCCL delta传输方案，采用纯磁盘方式。
• PR #2181 PR #2181 (后续: 增量权重同步优化): 作为系列PR的一部分，本PR实现磁盘delta基础，后续PR将在此基础上优化。