企业官网建设流程全解析

• 人工智能
• 分布式训练
• 强化学习
• 任务调度
• 模型推理服务

本篇文章围绕 Ray 官方仓库中 accelerate_example.rst 所展示的完整可运行示例，系统讲解如何用 Hugging Face Accelerate、Ray Train 与 Ray Data 三件套完成分布式数据并行训练，并在 GLUE MRPC 数据集上对 BERT 模型进行微调。读完本文，你将掌握TorchTrainer的用法、Accelerator.prepare()的分布式编排原理、Ray Data 与原生 PyTorchDataLoader两种数据接入方式的取舍，以及如何用 Accelerate 插件配置 DeepSpeed 与 FSDP，并了解仓库中的对应测试如何验证这些能力。

示例概览：一次集成 Accelerate、Ray Train 与 Ray Data 的分布式训练

该示例的定位在文档中写得很明确：使用 Hugging Face Accelerate、Ray Train 和 Ray Data 做分布式数据并行（Distributed Data Parallel, DDP）训练，微调一个 BERT 模型，代码改编自 Hugging Face Accelerate 官方仓库的examples/nlp_example.py。

三个组件各司其职：

• Hugging Face Accelerate：负责模型、优化器、学习率调度器与数据加载器的分布式编排（Accelerator.prepare()），并对梯度归约（all-reduce）、混合精度、梯度累积等底层细节做统一封装；
• Ray Train：通过TorchTrainer在多个 Ray worker 上拉起训练进程，配置 worker 数量、GPU 资源与持久化存储；
• Ray Data：以 Ray Dataset 的形式承载训练与验证数据，并在多个 worker 间自动分片，配合iter_torch_batches()高效产出 torch batch。

仓库中的完整实现位于 python/ray/train/examples/accelerate/accelerate_torch_trainer.py，另有不使用 Ray Data、改用原生 PyTorchDataLoader的姊妹版本 accelerate_torch_trainer_no_raydata.py。本文接下来会先给出文档 literalinclude 引用的完整代码，再逐段剖析，最后结合 huggingface-accelerate.rst 教程与 test_torch_accelerate.py 测试做深度扩展。

运行前置条件与依赖

示例依赖accelerate、evaluate、datasets、transformers与torch。仓库在 python/requirements/ml/train-requirements.txt 中约束了accelerate>=0.20.1，实际使用建议安装较新的 Accelerate 版本以获取完整的Accelerator.prepare()与插件能力。

pip install "accelerate>=0.20.1" datasets evaluate transformers torch

示例默认以num_workers=4、use_gpu=True运行，因此建议在具备至少 4 张 GPU 的环境（本地多卡机器或 Ray 集群）中执行；若仅想快速验证流程，可将use_gpu改为False（参见仓库测试 test_torch_accelerate.py 中对cpu与gpu两种模式的参数化测试）。

完整示例代码：带 Ray Data 的版本

以下为文档通过literalinclude引入的完整源码（即 accelerate_torch_trainer.py 的核心部分），它也是本文讲解的主体：

""" Minimal Ray Train and Accelerate example adapted from https://github.com/huggingface/accelerate/blob/main/examples/nlp_example.py Fine-tune a BERT model with Hugging Face Accelerate and Ray Train and Ray Data """ from tempfile import TemporaryDirectory import evaluate import torch from accelerate import Accelerator from datasets import load_dataset from torch.optim import AdamW from transformers import ( AutoModelForSequenceClassification, AutoTokenizer, get_linear_schedule_with_warmup, set_seed, ) import ray import ray.train from ray.train import Checkpoint, DataConfig, ScalingConfig from ray.train.torch import TorchTrainer def train_func(config): """Your training function that launches on each worker.""" # Unpack training configs lr = config["lr"] seed = config["seed"] num_epochs = config["num_epochs"] train_batch_size = config["train_batch_size"] eval_batch_size = config["eval_batch_size"] train_ds_size = config["train_dataset_size"] set_seed(seed) # Initialize accelerator accelerator = Accelerator() # Load datasets and metrics metric = evaluate.load("glue", "mrpc") # Prepare Ray Data loaders # ==================================================== train_ds = ray.train.get_dataset_shard("train") eval_ds = ray.train.get_dataset_shard("validation") tokenizer = AutoTokenizer.from_pretrained("bert-base-cased") def collate_fn(batch): outputs = tokenizer( list(batch["sentence1"]), list(batch["sentence2"]), truncation=True, padding="longest", return_tensors="pt", ) outputs["labels"] = torch.LongTensor(batch["label"]) outputs = {k: v.to(accelerator.device) for k, v in outputs.items()} return outputs train_dataloader = train_ds.iter_torch_batches( batch_size=train_batch_size, collate_fn=collate_fn ) eval_dataloader = eval_ds.iter_torch_batches( batch_size=eval_batch_size, collate_fn=collate_fn ) # ==================================================== # Instantiate the model, optimizer, lr_scheduler model = AutoModelForSequenceClassification.from_pretrained( "bert-base-cased", return_dict=True ) optimizer = AdamW(params=model.parameters(), lr=lr) steps_per_epoch = train_ds_size // (accelerator.num_processes * train_batch_size) lr_scheduler = get_linear_schedule_with_warmup( optimizer=optimizer, num_warmup_steps=100, num_training_steps=(steps_per_epoch * num_epochs), ) # Prepare everything with accelerator model, optimizer, lr_scheduler = accelerator.prepare(model, optimizer, lr_scheduler) for epoch in range(num_epochs): # Training model.train() for batch in train_dataloader: outputs = model(**batch) loss = outputs.loss accelerator.backward(loss) optimizer.step() lr_scheduler.step() optimizer.zero_grad() # Evaluation model.eval() for batch in eval_dataloader: with torch.no_grad(): outputs = model(**batch) predictions = outputs.logits.argmax(dim=-1) predictions, references = accelerator.gather_for_metrics( (predictions, batch["labels"]) ) metric.add_batch( predictions=predictions, references=references, ) eval_metric = metric.compute() accelerator.print(f"epoch {epoch}:", eval_metric) # Report checkpoint and metrics to Ray Train # ========================================== with TemporaryDirectory() as tmpdir: if accelerator.is_main_process: unwrapped_model = accelerator.unwrap_model(model) accelerator.save(unwrapped_model, f"{tmpdir}/ckpt_{epoch}.bin") checkpoint = Checkpoint.from_directory(tmpdir) else: checkpoint = None ray.train.report(metrics=eval_metric, checkpoint=checkpoint) if __name__ == "__main__": config = { "lr": 2e-5, "num_epochs": 3, "seed": 42, "train_batch_size": 16, "eval_batch_size": 32, } # Prepare Ray Datasets hf_datasets = load_dataset("nyu-mll/glue", "mrpc") ray_datasets = { "train": ray.data.from_huggingface(hf_datasets["train"]), "validation": ray.data.from_huggingface(hf_datasets["validation"]), } config["train_dataset_size"] = ray_datasets["train"].count() trainer = TorchTrainer( train_func, train_loop_config=config, datasets=ray_datasets, dataset_config=DataConfig(datasets_to_split=["train", "validation"]), scaling_config=ScalingConfig(num_workers=4, use_gpu=True), # If running in a multi-node cluster, this is where you # should configure the run's persistent storage that is accessible # across all worker nodes. # run_config=ray.train.RunConfig(storage_path="s3://..."), ) result = trainer.fit()

代码结构拆解

整份代码可分成三个层次：

1. train_func(config)：每个 worker 上执行的训练函数。Ray Train 会把同一个训练函数复制到每个 worker 进程中运行，因此函数内部的一切（Accelerator 初始化、数据加载、模型构建）都是"每进程一份"的视角；
2. collate_fn与iter_torch_batches：Ray Data → torch batch 的桥接。iter_torch_batches每次从分片后的 Ray Dataset 中取一个 batch，经collate_fn完成分词、padding 与迁移到accelerator.device，输出可直接喂给模型；
3. __main__部分：数据准备与TorchTrainer的组装。将 Hugging Face Datasets 转为 Ray Dataset，构造ScalingConfig与DataConfig，最后trainer.fit()启动分布式训练。

关键细节：为什么不需要prepare数据加载器

注意示例中accelerator.prepare(model, optimizer, lr_scheduler)只准备了三个对象，没有包含数据加载器。这与原生 PyTorchDataLoader版本不同（见下文"不使用 Ray Data 的版本"）。原因是：Ray Data 的iter_torch_batches()返回的是可迭代的 batch 生成器，本身已经按 worker 分片，DataConfig(datasets_to_split=...)会为每个 worker 分配不同的数据子集，因此在 Accelerate 侧无需再对数据做分布式采样器（DistributedSampler）处理。

steps_per_epoch的计算同样体现了分布式语义：train_ds_size // (accelerator.num_processes * train_batch_size)，其中accelerator.num_processes即参与训练的进程总数，这保证学习率调度器与真实的全局 batch 数一致。

关键细节：checkpoint 与指标上报

每个 epoch 结束后，通过ray.train.report(metrics=eval_metric, checkpoint=checkpoint)将验证指标与 checkpoint 上报给 Ray Train。这里有一个典型的多进程陷阱处理：

• 只有accelerator.is_main_process为真时才保存模型并构建Checkpoint.from_directory(tmpdir)，其他进程传checkpoint=None；
• 保存前用accelerator.unwrap_model(model)剥离 Accelerate/DDP 包装层，得到原始模型结构，便于后续加载。

上报的 checkpoint 会写入 Ray Train 的运行结果目录；若运行在多节点集群，应通过RunConfig(storage_path="s3://...")指定所有节点可访问的持久化存储（源码注释中已给出提示）。

不使用 Ray Data 的版本：原生 PyTorch DataLoader

如果不想引入 Ray Data，仓库还提供了 accelerate_torch_trainer_no_raydata.py。该版本在训练函数内部直接load_dataset("nyu-mll/glue", "mrpc")构建 PyTorchDataLoader，并与模型、优化器、调度器一起交给accelerator.prepare()：

train_dataloader = DataLoader( hf_datasets["train"], shuffle=True, collate_fn=collate_fn, batch_size=train_batch_size, drop_last=True, ) eval_dataloader = DataLoader( hf_datasets["validation"], shuffle=False, collate_fn=collate_fn, batch_size=eval_batch_size, drop_last=True, ) ... ( model, optimizer, train_dataloader, eval_dataloader, lr_scheduler, ) = accelerator.prepare( model, optimizer, train_dataloader, eval_dataloader, lr_scheduler )

这一版本更贴近 Hugging Face 官方nlp_example.py的写法：Accelerate 会自动为DataLoader注入DistributedSampler并完成进程间 shuffle 同步。两个版本的主要差异可总结如下：

选择哪种取决于你是否希望把数据预处理、分片纳入 Ray Data 的统一数据管线；两者都是仓库中验证过的可运行方案。

TorchTrainer 的组装与资源调度

__main__中的TorchTrainer是分布式训练的"启动器"，四个关键参数含义如下：

• train_func：要分发的训练函数，每个 worker 一份；
• train_loop_config：透传给train_func(config)的配置字典，示例中包括lr=2e-5、num_epochs=3、seed=42、train_batch_size=16、eval_batch_size=32，并在运行时追加train_dataset_size；
• datasets与dataset_config：datasets是以字符串为 key 的 Ray Dataset 字典；DataConfig(datasets_to_split=["train", "validation"])指定需要跨 worker 分片的数据集；
• scaling_config：ScalingConfig(num_workers=4, use_gpu=True)声明使用 4 个 worker 且每个 worker 使用 GPU。

result = trainer.fit()返回Result对象，其中包含每个 epoch 上报的指标与 checkpoint 的存储位置。

关于多节点集群，源码注释明确提示：此时应取消注释run_config=ray.train.RunConfig(storage_path="s3://...")，为所有 worker 节点配置共享的持久化存储。这一点在 huggingface-accelerate.rst 的示例中同样反复出现。

为什么不再需要accelerate config与accelerate launch

Accelerate 官方提供两条 CLI 路径：用accelerate config生成配置文件，再用accelerate launch启动训练。但 Ray Train 的场景中这两者都不必要，原因是：

1. TorchTrainer已经完成了 torch.distributed 环境的初始化：它会在每个 worker 上设置正确的RANK、LOCAL_RANK、WORLD_SIZE等环境变量并拉起训练进程；
2. 训练函数由 Ray Train 直接分发执行：Accelerator()在构造时会读取上述分布式环境变量，因此Accelerator.prepare()能正常工作，无需 CLI 参与。

换言之，Accelerate 的"分布式环境搭建"职责已被 Ray Train 接管，Accelerate 只需保留其"设备与组件编排"职责。仓库测试 test_torch_accelerate.py 对此做了直接验证：

assert accelerator.device == train.torch.get_device() assert accelerator.process_index == train.get_context().get_world_rank() assert accelerator.local_process_index == train.get_context().get_local_rank()

即 Accelerate 视角下的device、process_index、local_process_index与 Ray Train 视角下的设备、全局 rank、局部 rank 完全一致，证明两者在分布式环境上是天然对齐的。

深度配置：通过 Accelerator 插件启用 DeepSpeed 与 FSDP

huggingface-accelerate.rst 教程指出：在 Ray Train 中，所有 Accelerate 配置都在训练函数内的accelerate.Accelerator对象上完成，无需任何额外文件。

DeepSpeed：通过DeepSpeedPlugin传入配置字典

from accelerate import Accelerator, DeepSpeedPlugin DEEPSPEED_CONFIG = { "fp16": { "enabled": True }, "zero_optimization": { "stage": 3, "offload_optimizer": { "device": "cpu", "pin_memory": False }, "overlap_comm": True, "contiguous_gradients": True, "reduce_bucket_size": "auto", "stage3_prefetch_bucket_size": "auto", "stage3_param_persistence_threshold": "auto", "gather_16bit_weights_on_model_save": True, "round_robin_gradients": True }, "gradient_accumulation_steps": "auto", "gradient_clipping": "auto", "steps_per_print": 10, "train_batch_size": "auto", "train_micro_batch_size_per_gpu": "auto", "wall_clock_breakdown": False } def train_func(): ds_plugin = DeepSpeedPlugin(hf_ds_config=DEEPSPEED_CONFIG) accelerator = Accelerator( ..., deepspeed_plugin=ds_plugin, ) ...

要点说明：

• 配置以字典形式直接传给DeepSpeedPlugin(hf_ds_config=...)，避免维护额外的 JSON/配置文件；
• 多数字段可设为"auto"，由 Accelerate 根据实际分布式规模自动推导（例如train_batch_size、train_micro_batch_size_per_gpu、gradient_accumulation_steps）；
• zero_optimization.stage: 3启用 ZeRO-3，offload_optimizer.device: "cpu"将优化器状态卸载到 CPU，适合超大模型；
• 仓库测试 test_torch_accelerate.py 中另有一份带optimizer段与bf16段的 DeepSpeed 配置可作为补充参考。

FSDP：通过FullyShardedDataParallelPlugin配置

from torch.distributed.fsdp.fully_sharded_data_parallel import FullOptimStateDictConfig, FullStateDictConfig from accelerate import Accelerator, FullyShardedDataParallelPlugin def train_func(): fsdp_plugin = FullyShardedDataParallelPlugin( state_dict_config=FullStateDictConfig( offload_to_cpu=False, rank0_only=False ), optim_state_dict_config=FullOptimStateDictConfig( offload_to_cpu=False, rank0_only=False ) ) accelerator = Accelerator( ..., fsdp_plugin=fsdp_plugin, ) ...

这里通过state_dict_config与optim_state_dict_config控制模型与优化器状态字典的保存行为（是否卸载到 CPU、是否仅在 rank 0 保留），将其作为插件传入Accelerator即可，训练函数其余部分无需改动。

底层原理：Accelerate 与 Ray Train 的协作关系

从代码结构可以推断出两者的协作模型：

1. Ray Train 负责"进程与资源"：TorchTrainer按ScalingConfig启动 N 个 worker，每个 worker 是一个独立的 Python 进程，运行同一份train_func；
2. Accelerate 负责"组件编排"：每个 worker 内的Accelerator()读取 Ray Train 注入的分布式环境变量，识别自己是哪个 rank；accelerator.prepare()将模型包装为 DDP/FSDP/DeepSpeed 版本、将优化器与调度器适配为分布式步进语义；
3. 反向传播走 Accelerate 的accelerator.backward(loss)：它会自动处理梯度归约与（可选的）梯度累积，这是与原生loss.backward()的关键区别；
4. 指标聚合用accelerator.gather_for_metrics：在验证阶段收集所有进程的预测与标签，避免各 rank 各自计算导致指标偏差。

由此可以理解 huggingface-accelerate.rst 中特别强调的一条原则：使用 Accelerate 时，不要再调用 Ray Train 的prepare_model或prepare_data_loader等原生 PyTorch 辅助工具，分布式准备完全交给Accelerator对象及其prepare()方法完成，避免双重包装。

测试与验证：仓库如何保障该能力

仓库用 python/ray/train/tests/test_torch_accelerate.py 覆盖了 Accelerate 与 Ray Train 集成的三类场景：

• test_accelerate_base：在 2 节点 2 GPU（或 CPU）环境验证Accelerator与 Ray Train 的设备/rank 对齐，并跑通 3 个 epoch 的线性回归训练，断言 loss 单调下降；
• test_accelerate_deepspeed：用DeepSpeedPlugin+ 上述DEEPSPEED_CONFIG验证 ZeRO-2 配置下训练可正常收敛，覆盖了"通过插件注入 DeepSpeed"这条路径；
• test_accelerate_e2e：在纯 CPU 环境下验证accelerator.prepare(model)与 checkpoint 上报的全链路。

这些测试一方面印证了文档描述的能力（分布式环境由 Ray Train 提供、Accelerate 只管编排），另一方面也为读者提供了最小可复现的验证脚本——在本地启动ray.init(num_cpus=4)后，即可用 CPU 模式跑通 Accelerate + Ray Train 组合。

延伸阅读

• Get Started with Distributed Training using Hugging Face Accelerate：Ray Train 与 HF Accelerate 的入门教程，含 AccelerateTrainer 迁移指南（Ray 2.8 起推荐直接用TorchTrainer运行 Accelerate 代码，旧AccelerateTrainerAPI 已进入弃用计划）；
• Ray Train Examples：Examples 画廊的生成配置，本示例在画廊中名为 "Train a text classifier with Hugging Face Accelerate"（beginner 级别，框架标签为 accelerate / pytorch / hugging face），更多用例（DeepSpeed、Lightning、Transformer 等）均可从该入口找到；
• DeepSpeed 示例：Ray Train 与 DeepSpeed 的专项文档；
• 若需在本地修改并复现，可直接运行 accelerate_torch_trainer.py 或 accelerate_torch_trainer_no_raydata.py；注意模型与数据集首次运行时会从 Hugging Face Hub 下载bert-base-cased与 GLUE MRPC 数据，需要网络可达。

• 人工智能
• 分布式训练
• 强化学习
• 任务调度
• 模型推理服务