TorchTitan 数据流水线实战指南基于 Grain 的源、数据集、打包与分布式加载【免费下载链接】torchtitanA PyTorch native platform for training generative AI models项目地址: https://gitcode.com/GitHub_Trending/to/torchtitan本文系统讲解 TorchTitan 中统一基于 Grain配套源码在 torchtitan/components/data/ 目录。统一数据流水线的心智模型TorchTitan 用一个基于 Grain 的流水线同时服务文本预训练、SFT 与图像训练共分五层。先建立整体直觉1. Define a source (e.g. jsonl): class: SourceConfig output: RandomAccessDataSource | IterDataset 2. Define a dataset (filter/process applied to the source): class: SingleDatasetConfig does: pre-filter - process - post-filter output: MapDataset | IterDataset 3. Compose datasets (optional): class: e.g. FirstFitPackingConfig(datasetDatasetMixConfig(...)) input: one or more child DatasetConfig values does: mix, concatenate, and/or pack output: MapDataset | IterDataset 4. Dataloader: runtime: GrainDataLoader config: GrainDataLoader.Config input: MapDataset | IterDataset does: convert to iterable if needed - batch - collate - prefetch output: TrainerBatch 5. Trainer: input: TrainerBatch does: model forward and backward这五层分别对应 torchtitan/components/data/ 下的sources.py源、dataset.py数据集与组合、packing.py打包、loader.py加载器和collators.py批处理公共类型定义在 types.py。值得注意的细节SourceConfig与DatasetConfig在源码中都是协议Protocol而非基类只要实现约定的build()方法即可参与组装而SingleDatasetConfig、DatasetMixConfig等则是冻结数据类frozen dataclass以配置值Config value的形式被声明由 GrainDataLoader 在初始化时统一构建为 Grain 数据集图。从源码结构看这套设计的核心是“声明式配置 延迟构建”你在配置中描述流水线形状真正的构建发生在 loader.py 的GrainDataLoader.__init__中它把 tokenizer、max_context_length、num_tokens_per_batch等运行时值打包进DatasetBuildContext把seed/shuffle/repeat/dp_rank/dp_world_size打包进DatasetIterationPolicy再自上而下构建整张数据集图。文本预训练从 JSONL 到打包好的 Token 批本地 JSONL 源JSONL 文件要求每个非空行恰好是一个 JSON 对象{title: First, body: The first document.} {title: Second, body: The second document.}最简的文本预训练配置from torchtitan.components.data import ( ConcatThenSplitPackingConfig, GrainDataLoader, IndexedJsonlSource, SingleDatasetConfig, ) from torchtitan.hf_datasets.text_datasets import TextProcessor def article_text(row): return row[title] \n\n row[body] books_ds SingleDatasetConfig( sourceIndexedJsonlSource.Config( patterns( /datasets/books/*.jsonl, ), ), processorTextProcessor.Config(text_fnarticle_text), post_filters(lambda sample: sample is not None,), ) books_packed_ds ConcatThenSplitPackingConfig(datasetbooks_ds) config.dataloader GrainDataLoader.Config( datasetbooks_packed_ds, )这里IndexedJsonlSource实现见 sources.py提供的是基于字节偏移的随机访问初始化时扫描每个匹配的 JSONL 文件记录每个非空行的(path_id, byte_offset)__getitem__(index)直接seek到偏移处读取该行并json.loads因此既不需要把整个语料加载进内存也能支持dataset[index]式随机访问。有几点实现约束值得注意每个 glob pattern 必须至少匹配一个文件否则抛出FileNotFoundError多个 pattern 解析出的路径若出现重复会抛出ValueError拒绝防止同一文件被索引两次sources.py源码中的 TODO 也说明了当前限制每个 rank 和 worker 启动时都会重新扫描全部 JSONL 文件未来计划构建一个可被所有进程 mmap 的共享偏移索引。TextProcessortext_datasets.py把text_fn返回的字符串编码为带 BOS/EOS 的 token 序列然后前移一位切分为 next-token 对齐的TextSequence(input_ids, labels)——注意这个移位发生在数据集处理器里而不是 trainer 中TextSequence的注释明确强调“trainer does not do it”。短于 2 个 token 的样本返回None由post_filters过滤掉。ConcatThenSplitPackingConfig实现于 packing.py先把文档拼接成连续 token 流再切分成固定长度num_tokens_per_batch的行并丢弃未填满的行_packing_output_is_full。当设置了max_num_documents每行最多包含的文档片段数时会切换到文档感知document-aware实现在行内保留文档边界、记录 remainder 状态以保证续训精确恢复见 packing.py。Hugging Face 源随机访问还是流式需要物化materialize数据集时用随机访问源from torchtitan.components.data import HuggingFaceRandomAccessSource source HuggingFaceRandomAccessSource.Config( pathopenai/gsm8k, namemain, splittrain, )语料不应物化时用流式源from torchtitan.components.data import HuggingFaceStreamingSource source HuggingFaceStreamingSource.Config( pathallenai/c4, nameen, splittrain, )两类源都接受path、split、name、revision以及透传的load_dataset_kwargs并在__post_init__中拒绝把split/name/revision/streaming重复放进 kwargssources.py。它们的底层实现差异很大HuggingFaceRandomAccessSource以streamingFalse加载一个datasets.Dataset非Dataset类型会直接报错随后只做len/__getitem__包装HuggingFaceStreamingSource本身继承grain.IterDataset以streamingTrue加载IterableDataset并用datasets.distributed.split_dataset_by_node在源级按 DP 坐标切分dp_rank/dp_world_size来自DatasetIterationPolicy。它还要求 HF 数据集实现state_dict()/load_state_dict()否则拒绝使用——这是精确续训的前提。内部的_HuggingFaceCursorIterator把流式游标暴露给 Grain 的检查点递归记录epoch与 HF 内部状态repeatTrue时在每个 epoch 递增epoch、必要时set_epoch重设 shuffle 后再从头迭代sources.py。添加自定义源预 token 化数据示例当需要把“每条索引对应一篇已 token 化的文档”暴露为随机访问源时可以自定义源并复用既有的处理与打包配置。完整的可运行示例见 torchtitan/components/data/README.md核心是把 memmap 的 token 数组与文档偏移数组包装成RandomAccessDataSourcefrom dataclasses import dataclass import numpy as np from torchtitan.components.data import ( ConcatThenSplitPackingConfig, DatasetBuildContext, DatasetIterationPolicy, RandomAccessDataSource, SampleProcessor, SingleDatasetConfig, TextSequence, ) from torchtitan.config import Configurable class PretokenizedMemmapSource(Configurable, RandomAccessDataSource): dataclass(kw_onlyTrue, slotsTrue) class Config(Configurable.Config): tokens_path: str document_offsets_path: str def __init__( self, config: Config, *, dataset_iteration_policy: DatasetIterationPolicy, ): del dataset_iteration_policy self.tokens np.memmap(config.tokens_path, dtypenp.uint32, moder) self.offsets np.load(config.document_offsets_path) def __len__(self): return len(self.offsets) - 1 def __getitem__(self, index): start, end self.offsets[index : index 2] return np.asarray(self.tokens[start:end], dtypenp.int64) class TokensToTextSequence(SampleProcessor): dataclass(kw_onlyTrue, slotsTrue) class Config(SampleProcessor.Config): pass def __init__(self, config: Config, *, context: DatasetBuildContext): del config, context def __call__(self, token_ids, rng): del rng if len(token_ids) 2: return None return TextSequence( input_idstoken_ids[:-1], labelstoken_ids[1:], ) token_documents_ds SingleDatasetConfig( sourcePretokenizedMemmapSource.Config( tokens_pathtokens.bin, document_offsets_pathdocument_offsets.npy, ), processorTokensToTextSequence.Config(), post_filters(lambda sample: sample is not None,), ) packed_tokens_ds ConcatThenSplitPackingConfig( datasettoken_documents_ds, )这个例子展示了三层复用自定义源只负责“按索引返回原始行”SampleProcessordataset.py负责把原始行转成TextSequence其中rng参数是 Grain 提供的确定性随机数生成器可安全用于数据增强打包与加载则完全复用通用组件。SingleDatasetConfig在构建时会执行pre_filters - processor(random_map) - post_filters - (shuffle) - DP shard - (repeat)的固定顺序dataset.py这解释了为什么post_filters里常见的lambda sample: sample is not None能过滤掉处理器返回的None。SFT换处理器与打包策略不换加载器SFT 与预训练的差异只体现在处理器和打包策略上加载器保持不变from torchtitan.components.data import ( FirstFitPackingConfig, GrainDataLoader, HuggingFaceRandomAccessSource, SingleDatasetConfig, ) from torchtitan.hf_datasets.text_datasets import ChatProcessor def gsm8k_messages(row): return [ {role: user, content: row[question]}, {role: assistant, content: row[answer]}, ] gsm8k_ds SingleDatasetConfig( sourceHuggingFaceRandomAccessSource.Config( pathopenai/gsm8k, namemain, splittrain, ), processorChatProcessor.Config(messages_fngsm8k_messages), post_filters(lambda sample: sample is not None,), ) gsm8k_packed_ds FirstFitPackingConfig(datasetgsm8k_ds) config.dataloader GrainDataLoader.Config( datasetgsm8k_packed_ds, )无 renderer 时的单轮对话处理不配置 renderer 时ChatProcessor走 tokenizer 的 chat template 处理单轮[user, assistant]对话先校验消息严格为两轮且 role 分别为user、assistant否则ValueError再用apply_chat_template渲染完整对话并追加 EOS生成 next-token 的 input/label 对最后把 prompt 部分的 label 置为IGNORE_INDEX即 -100见 torchtitan/components/loss.py 的IGNORE_INDEX只对 assistant 回答计算损失。prompt/response 边界的定位方式是单独用add_generation_promptTrue渲染 prompt并要求它的 token 序列恰好是完整渲染 token 序列的前缀否则抛出ValueErrortext_datasets.py。这里选择“报错”而非“丢弃样本”是有意的前缀不匹配意味着标签边界无法确定且这类问题通常是模板系统性的静默丢弃会让模型只在小部分数据上训练。超出max_context_length的样本则被整条丢弃因为超长是个别样本问题。ChatProcessor也会在首个样本时打印完整渲染文本便于人工核对。使用 renderer 处理多轮对话多轮对话需要显式选择模型对应的 rendererfrom renderers import Qwen3RendererConfig from torchtitan.components.renderer import RenderersLibraryConfig processor ChatProcessor.Config( messages_fnlambda row: row[messages], rendererRenderersLibraryConfig(renderers_configQwen3RendererConfig()), )RenderersLibraryConfigrenderer.py基于renderers库通过RendererTokenizerWrapper把 TorchTitan 已加载的 tokenizer 适配成 renderer 需要的接口字符偏移、token 到 id 的查找、raw 编码不会二次加载 tokenizer。它明确禁止auto与default类型的 renderer前者依赖name_or_path精确匹配且可能落到不受支持的 DefaultRenderer后者依赖 HF 的apply_chat_template而 TorchTitan 的模板渲染缺少其特殊 token 变量会产生静默不同的 token。使用 renderer 时一次渲染即返回 token 与 loss maskmask 只监督模型生成的 token含回合终止符排除 prompt token 与模板脚手架。约束包括对话必须以 assistant 消息结尾否则报错每个对话是一个样本打包时在对话之间重置位置而不是回合之间超过max_context_length的样本整条丢弃格式化与思维链reasoning保留策略由所选 renderer 决定例如 Qwen3 会在最后一个 user 提问之前省略 assistant 的 reasoning被省略的 token 不计损失若想对每个回合的 reasoning 都训练需要在源数据集里准备独立的对话前缀thinking_retention控制的是 renderer 的 rollout 桥接不作用于此处使用的完整渲染。混合与拼接数据集加权混合DatasetMixConfig 与 WeightedDataset把权重放在每个数据集旁边from torchtitan.components.data import DatasetMixConfig, WeightedDataset # books_ds and code_ds are SingleDatasetConfig values. pretraining_mix_ds DatasetMixConfig( datasets( WeightedDataset(datasetbooks_ds, weight0.75), WeightedDataset(datasetcode_ds, weight0.25), ), )示例中概率之和为 1.0但实现上接受任意正相对权重并在内部归一化——weight0.75配weight0.25表示第一个数据集被抽中的频率是第二个的 3 倍。DatasetMixConfigdataset.py在构建时校验所有权重有限且为正并给每个子数据集分配seed index的偏移插入或重排子数据集会重新播种其后的所有子数据集从而改变数据顺序——断点续训要求代码与配置完全不变的原因之一。混合的语义取决于子数据集的粒度这由 README 中的两个小节给出先混合再打包权重按“文档”计数packed_pretraining_ds ConcatThenSplitPackingConfig( datasetpretraining_mix_ds, )先打包再混合权重按“定长行”计数books_packed_ds ConcatThenSplitPackingConfig(datasetbooks_ds) code_packed_ds ConcatThenSplitPackingConfig(datasetcode_ds) token_ratio_mix_ds DatasetMixConfig( datasets( WeightedDataset(datasetbooks_packed_ds, weight0.67), WeightedDataset(datasetcode_packed_ds, weight0.33), ), )源码注释dataset.py补充了微妙差异全 map 路径下MapDataset.filter会把被拒索引留作None因此权重作用于尝试取样的索引而非被接受的样本而一旦混入 iterable 子数据集权重则作用于各子数据集实际发射的元素——混合TextSequence子项按文档计数混合打包好的定长子项则按物理 token 计数。repeatTrue时混合是无限的repeatFalse时混合在第一个耗尽的子数据集处停止较大的子数据集不会被完整覆盖。从源码结构看按观察到的文档数或监督 token 数自动调节权重的功能尚未实现源码中有 TODO 注释文档也明确说明自定义混合可维护各数据集的移动平均并自行再平衡权重。拼接把有限数据集当作一个语料拼接用于把多个有限数据集视为一个语料每行出现一次占比由各数据集大小决定需要显式权重或流式数据集时改用混合from torchtitan.components.data import DatasetConcatConfig pretraining_corpus_ds DatasetConcatConfig( datasets(books_ds, code_ds, math_ds), )若希望某个有限数据集的每一行在一个 epoch 内出现多次可以在拼接中重复该子项pretraining_corpus_ds DatasetConcatConfig( datasets(books_ds,) * 3 (code_ds, math_ds), )此时每个books_ds行在合并后的有限索引空间中出现 3 次当shuffleTrue时TorchTitan 会在 DP 分片之前对合并后的索引空间做全局 shuffle。实现上dataset.pyDatasetConcatConfig要求全部子项为 map 风格构建时先把每个子数据集以shuffleFalse, repeatFalse, dp_rank0, dp_world_size1的迭代策略构建再MapDataset.concatenate合并最后统一全局 shuffle、DP 分片、repeat。若需求是“相对采样频率”而非“精确的有限重复”应改用混合pretraining_mix_ds DatasetMixConfig( datasets( WeightedDataset(datasetbooks_ds, weight0.6), WeightedDataset(datasetcode_ds, weight0.2), WeightedDataset(datasetmath_ds, weight0.2), ), )最后注意GrainDataLoader.Config(repeatTrue)重复的是整个拼接/混合后的数据集并不会改变某个子数据集的相对贡献。图像与多模态数据图像训练复用同一套 source / dataset / loader / sharding / checkpoint 契约处理器保留模态相关的样本字典collator 生成模型专属的 batch。Qwen 多模态示例from torchtitan.components.data import ( GrainDataLoader, HuggingFaceStreamingSource, SingleDatasetConfig, ) from torchtitan.hf_datasets.multimodal.mm_collator import MultiModalCollator from torchtitan.hf_datasets.multimodal.mm_datasets import ( MMSamplePackingConfig, MultiModalProcessor, _process_cc12_wd_sample, ) mm_processor MultiModalProcessor.Config( sample_processor_process_cc12_wd_sample, ) mm_ds SingleDatasetConfig( sourceHuggingFaceStreamingSource.Config( pathpixparse/cc12m-wds, splittrain, ), processormm_processor, post_filters(lambda sample: sample is not None,), ) packed_mm_ds MMSamplePackingConfig( datasetmm_ds, num_packing_bins8, ) config.dataloader GrainDataLoader.Config( datasetpacked_mm_ds, collatorMultiModalCollator.Config( build_mrope_positionsTrue, patch_sizemm_processor.patch_size, temporal_patch_sizemm_processor.temporal_patch_size, spatial_merge_sizemm_processor.spatial_merge_size, ), streaming_shuffle_buffer_size128, )几个关键语义均有源码对应MultiModalProcessormm_datasets.py适配 Grain 的 map 契约内部调用sample_processor如_process_cc12_wd_sample、_process_obelics_sample。处理流程为解码图像/视频字节 - 缩放到patch_size * spatial_merge_size的倍数并归一化 - 在文本中插入|vision_start||image_pad|...|vision_end|占位 token - 编码文本vision 占位 token 在 labels 中被置为IGNORE_INDEXmm_datasets.py。默认配置包括patch_size16、temporal_patch_size2、spatial_merge_size2、min_pixels65_536、max_pixels16_777_216、max_patches4096等。处理后的样本若超过max_context_length会被整条跳过。MMSamplePackingConfig把整个多模态文档装进定长行先过滤超长样本再用FirstFitPackIterDataset打包input_ids/labels/positions为定长结构pixel_values作为 meta features 保留num_packing_bins是保持打开的候选打包行数量而不是输入样本缓冲区——更大的值可以减少 padding 但会滞留更多媒体数据mm_datasets.py。MultiModalCollatormm_collator.py负责两件事collate_images把图像/视频张量切块patch并 pad 到统一 patch 数产出pixel_values与grid_thw形状(num_images, 3)乘积给出每个条目的 patch 序列长度collate_text拼接整条样本并只 pad token 批的尾部。当build_mrope_positionsTrue时还会在 CPU 数据侧构建三维时间/高/宽的 MRoPE 位置 ID_build_mrope_positions返回(num_tokens, 3)此时patch_order必须为blockraster 顺序会使 MRoPE 与 patch 序列失同步。max_images_per_batch默认 128限制每个批的视觉条目数。自定义图像增强应放进SampleProcessor。Loader 策略一次配置全局生效GrainDataLoader 完整配置运行级行为在GrainDataLoader.Config中一次配好loader.pyimport grain.python as grain config.dataloader GrainDataLoader.Config( datasetpacked_pretraining_ds, seed42, shuffleTrue, repeatTrue, streaming_shuffle_buffer_size1_000, read_optionsgrain.ReadOptions( num_threads16, prefetch_buffer_size500, ), num_prefetch_batches2, )各字段的默认值与作用源码注释字段默认值说明dataset必填任意DatasetConfig叶子、混合、拼接或打包后的数据集collatorTextCollator.Config行到批的转换多模态用MultiModalCollator.Configseed42全局随机种子贯穿 shuffle、random_map、打包shuffleTrue是否全局/流式 shufflerepeatTrue是否无限重复见下文分布式小节中的限制streaming_shuffle_buffer_size1_000每个 rank 保留的流式原始行数用于近似 shuffleread_optionsgrain.ReadOptions()MapDataset转IterDataset时的并发读参数num_prefetch_batches2预留给 trainer 的完整 collate 批数量GrainDataLoader.__init__里还有一道防线当dp_world_size 1且repeatFalse时直接抛ValueError提示必须用repeatTrue配合 trainer 控制的步数原因见“分布式与检查点行为”一节。加载器构建完数据集图后执行batch(collator.num_rows_per_batch(), drop_remainderconfig.repeat, batch_fncollator)再用ThreadPrefetchIterDataset预取num_prefetch_batches个完整批。TextCollatorcollators.py把若干TextSequence行合并进**预分配、页锁定pin_memory**的定长张量padding 位置 label 为IGNORE_INDEX、padding_maskTrue并在 batch 字典中预先计算num_valid_tokens参与损失的 token 数避免 trainer 在关键路径上重复扫描。读取带索引的数据集MapDataset 与 IterDataset 的转换规则随机访问源构成MapDataset支持dataset[index]当后续某个阶段需要顺序消费样本时它会转成IterDataset。转换时机遵循以下规则README 原文all children are MapDataset: DatasetMixConfig remains a MapDataset any child is IterDataset: DatasetMixConfig converts each MapDataset child packing: converts its child if needed and returns IterDataset GrainDataLoader: converts a MapDataset if no earlier stage didread_options控制每一次MapDataset到IterDataset的转换grain.ReadOptions( num_threads16, # indexed samples read concurrently prefetch_buffer_size500, # samples waiting for the consumer )每次转换都有自己独立的线程和缓冲区全 map 的混合只转换一次混有流的混合会为每个 map 子数据集各自转换一次。loader.py 中的 TODO 也提到当前只使用多线程而非多进程做 CPU 密集处理未来计划在更早的边界引入共享 worker 池。流式 shuffle 与就绪批streaming_shuffle_buffer_size保留用于近似 shuffle的原始行数每个 rank。缓冲区越大混合越均匀但内存占用越高。流式源在SingleDatasetConfig._build_iter_dataset里通过grain.experimental.WindowShuffleIterDataset实现dataset.py窗口大小即此值。num_prefetch_batches允许等待 trainer 的完整 collate 批数量trainer computes batch 10 background thread prepares batches 11 and 12即 trainer 计算第 10 批时后台线程已在准备第 11、12 批。分布式与检查点行为有效数据并行effective DP决定数据归属只有有效 DP 坐标负责选取数据effective DP data_parallel_replicate_degree * data_parallel_shard_degree different effective-DP ranks - disjoint source rows TP/PP/CP peers - same rows for their effective-DP coordinate也就是说TP/PP/CP 的 peer rank 与它们的有效 DP 坐标共享相同数据行而不同的有效 DP rank 之间数据行互不相交。这是保证前向/反向集合通信语义一致的关键。数据所有权的判定时机数据归属在批处理之前决定不同源/组合的归属方式不同random-access source - global shuffle - contiguous balanced DP shard Hugging Face stream - source-level DP shard DatasetMixConfig - combines children already owned by this DP rank DatasetConcatConfig - concatenates, globally shuffles, then DP-shards packing - packs samples locally on each DP rank GrainDataLoader - batches and collates that ranks samples对随机访问训练而言每个 rank 拿到的是全局 shuffle 后索引空间的连续切片而不是原始语料的连续区域。SingleDatasetConfig与DatasetConcatConfig中divmod分片逻辑dataset.py保证了dp_world_size不整除时余数行被均衡分配。当len(dataset) dp_world_size时会直接报错避免分片后出现空集。有限数据与 hang 的陷阱当有效 DP 大于 1 时repeatFalse会被拒绝loader.py因为各 rank 可能在不同的 step耗尽数据导致训练集合通信挂起。正确做法是repeatTrue让 trainer 的步数控制训练何时停止。加载器还定义了DataloaderExhaustedErrorloader.py刻意继承Exception而非StopIteration以避免 PEP 479 把生成器内部的StopIteration包装成RuntimeError导致程序崩溃。state_dict 与断点续训GrainDataLoader.state_dict()loader.py记录以下内容源游标与 shuffle/repeat 进度含 HF 流式源的 epoch 与内部状态mix 子数据集与打包缓冲区的状态如 document-aware 打包的 remainder 与偏移batching 与 prefetch 状态有效 DP 维度dp_world_size。恢复load_state_dict的硬性要求代码、配置、源内容、tokenizer 与有效 DP 维度都必须与保存时一致。源码会显式校验state_dict[version] 1以及dp_world_size是否变化并确保 checkpoint 中包含当前dp_rank_{i}的条目缺失或版本不符都会抛错。这也解释了上文混合时“插入或重排子数据集会重新播种其后所有子数据集”为何影响续训一致性。配套的单元测试覆盖了这些行为例如test_indexed_jsonl_random_access、test_hf_shuffled_repeat_advances_epoch、test_hf_resume_mid_second_epoch、test_loader_requires_repeat_with_data_parallelism、test_weighted_map_mix_keeps_weight_with_dataset、test_concat_shards_after_one_global_index_space、test_document_aware_concat_packing_restores_exactly等见 tests/unit_tests/cpu/components/data/test_grain_data.py可作为理解各阶段精确语义的补充证据。小结TorchTitan 的数据流水线把“源、处理、组合、加载”四层职责分离全部以声明式 Config 表达最终由GrainDataLoader统一构建并接入 trainer。文本预训练、SFT 与多模态训练共用同一套骨架差异仅在SampleProcessor与打包/ collator 的选择上分布式场景下“有效 DP 决定数据归属 全局 shuffle 后分片 repeatTrue”的约定加上可递归恢复的 Grain 状态为大规模多机训练提供了可复现与可断点续训的坚实基础。动手实践时建议从本文的 JSONL 最小示例出发逐步替换源、处理器与组合方式并用仓库中的单元测试验证每一步的语义。【免费下载链接】torchtitanA PyTorch native platform for training generative AI models项目地址: https://gitcode.com/GitHub_Trending/to/torchtitan创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
TorchTitan 数据流水线实战指南：基于 Grain 的源、数据集、打包与分布式加载
TorchTitan 数据流水线实战指南基于 Grain 的源、数据集、打包与分布式加载【免费下载链接】torchtitanA PyTorch native platform for training generative AI models项目地址: https://gitcode.com/GitHub_Trending/to/torchtitan本文系统讲解 TorchTitan 中统一基于 Grain配套源码在 torchtitan/components/data/ 目录。统一数据流水线的心智模型TorchTitan 用一个基于 Grain 的流水线同时服务文本预训练、SFT 与图像训练共分五层。先建立整体直觉1. Define a source (e.g. jsonl): class: SourceConfig output: RandomAccessDataSource | IterDataset 2. Define a dataset (filter/process applied to the source): class: SingleDatasetConfig does: pre-filter - process - post-filter output: MapDataset | IterDataset 3. Compose datasets (optional): class: e.g. FirstFitPackingConfig(datasetDatasetMixConfig(...)) input: one or more child DatasetConfig values does: mix, concatenate, and/or pack output: MapDataset | IterDataset 4. Dataloader: runtime: GrainDataLoader config: GrainDataLoader.Config input: MapDataset | IterDataset does: convert to iterable if needed - batch - collate - prefetch output: TrainerBatch 5. Trainer: input: TrainerBatch does: model forward and backward这五层分别对应 torchtitan/components/data/ 下的sources.py源、dataset.py数据集与组合、packing.py打包、loader.py加载器和collators.py批处理公共类型定义在 types.py。值得注意的细节SourceConfig与DatasetConfig在源码中都是协议Protocol而非基类只要实现约定的build()方法即可参与组装而SingleDatasetConfig、DatasetMixConfig等则是冻结数据类frozen dataclass以配置值Config value的形式被声明由 GrainDataLoader 在初始化时统一构建为 Grain 数据集图。从源码结构看这套设计的核心是“声明式配置 延迟构建”你在配置中描述流水线形状真正的构建发生在 loader.py 的GrainDataLoader.__init__中它把 tokenizer、max_context_length、num_tokens_per_batch等运行时值打包进DatasetBuildContext把seed/shuffle/repeat/dp_rank/dp_world_size打包进DatasetIterationPolicy再自上而下构建整张数据集图。文本预训练从 JSONL 到打包好的 Token 批本地 JSONL 源JSONL 文件要求每个非空行恰好是一个 JSON 对象{title: First, body: The first document.} {title: Second, body: The second document.}最简的文本预训练配置from torchtitan.components.data import ( ConcatThenSplitPackingConfig, GrainDataLoader, IndexedJsonlSource, SingleDatasetConfig, ) from torchtitan.hf_datasets.text_datasets import TextProcessor def article_text(row): return row[title] \n\n row[body] books_ds SingleDatasetConfig( sourceIndexedJsonlSource.Config( patterns( /datasets/books/*.jsonl, ), ), processorTextProcessor.Config(text_fnarticle_text), post_filters(lambda sample: sample is not None,), ) books_packed_ds ConcatThenSplitPackingConfig(datasetbooks_ds) config.dataloader GrainDataLoader.Config( datasetbooks_packed_ds, )这里IndexedJsonlSource实现见 sources.py提供的是基于字节偏移的随机访问初始化时扫描每个匹配的 JSONL 文件记录每个非空行的(path_id, byte_offset)__getitem__(index)直接seek到偏移处读取该行并json.loads因此既不需要把整个语料加载进内存也能支持dataset[index]式随机访问。有几点实现约束值得注意每个 glob pattern 必须至少匹配一个文件否则抛出FileNotFoundError多个 pattern 解析出的路径若出现重复会抛出ValueError拒绝防止同一文件被索引两次sources.py源码中的 TODO 也说明了当前限制每个 rank 和 worker 启动时都会重新扫描全部 JSONL 文件未来计划构建一个可被所有进程 mmap 的共享偏移索引。TextProcessortext_datasets.py把text_fn返回的字符串编码为带 BOS/EOS 的 token 序列然后前移一位切分为 next-token 对齐的TextSequence(input_ids, labels)——注意这个移位发生在数据集处理器里而不是 trainer 中TextSequence的注释明确强调“trainer does not do it”。短于 2 个 token 的样本返回None由post_filters过滤掉。ConcatThenSplitPackingConfig实现于 packing.py先把文档拼接成连续 token 流再切分成固定长度num_tokens_per_batch的行并丢弃未填满的行_packing_output_is_full。当设置了max_num_documents每行最多包含的文档片段数时会切换到文档感知document-aware实现在行内保留文档边界、记录 remainder 状态以保证续训精确恢复见 packing.py。Hugging Face 源随机访问还是流式需要物化materialize数据集时用随机访问源from torchtitan.components.data import HuggingFaceRandomAccessSource source HuggingFaceRandomAccessSource.Config( pathopenai/gsm8k, namemain, splittrain, )语料不应物化时用流式源from torchtitan.components.data import HuggingFaceStreamingSource source HuggingFaceStreamingSource.Config( pathallenai/c4, nameen, splittrain, )两类源都接受path、split、name、revision以及透传的load_dataset_kwargs并在__post_init__中拒绝把split/name/revision/streaming重复放进 kwargssources.py。它们的底层实现差异很大HuggingFaceRandomAccessSource以streamingFalse加载一个datasets.Dataset非Dataset类型会直接报错随后只做len/__getitem__包装HuggingFaceStreamingSource本身继承grain.IterDataset以streamingTrue加载IterableDataset并用datasets.distributed.split_dataset_by_node在源级按 DP 坐标切分dp_rank/dp_world_size来自DatasetIterationPolicy。它还要求 HF 数据集实现state_dict()/load_state_dict()否则拒绝使用——这是精确续训的前提。内部的_HuggingFaceCursorIterator把流式游标暴露给 Grain 的检查点递归记录epoch与 HF 内部状态repeatTrue时在每个 epoch 递增epoch、必要时set_epoch重设 shuffle 后再从头迭代sources.py。添加自定义源预 token 化数据示例当需要把“每条索引对应一篇已 token 化的文档”暴露为随机访问源时可以自定义源并复用既有的处理与打包配置。完整的可运行示例见 torchtitan/components/data/README.md核心是把 memmap 的 token 数组与文档偏移数组包装成RandomAccessDataSourcefrom dataclasses import dataclass import numpy as np from torchtitan.components.data import ( ConcatThenSplitPackingConfig, DatasetBuildContext, DatasetIterationPolicy, RandomAccessDataSource, SampleProcessor, SingleDatasetConfig, TextSequence, ) from torchtitan.config import Configurable class PretokenizedMemmapSource(Configurable, RandomAccessDataSource): dataclass(kw_onlyTrue, slotsTrue) class Config(Configurable.Config): tokens_path: str document_offsets_path: str def __init__( self, config: Config, *, dataset_iteration_policy: DatasetIterationPolicy, ): del dataset_iteration_policy self.tokens np.memmap(config.tokens_path, dtypenp.uint32, moder) self.offsets np.load(config.document_offsets_path) def __len__(self): return len(self.offsets) - 1 def __getitem__(self, index): start, end self.offsets[index : index 2] return np.asarray(self.tokens[start:end], dtypenp.int64) class TokensToTextSequence(SampleProcessor): dataclass(kw_onlyTrue, slotsTrue) class Config(SampleProcessor.Config): pass def __init__(self, config: Config, *, context: DatasetBuildContext): del config, context def __call__(self, token_ids, rng): del rng if len(token_ids) 2: return None return TextSequence( input_idstoken_ids[:-1], labelstoken_ids[1:], ) token_documents_ds SingleDatasetConfig( sourcePretokenizedMemmapSource.Config( tokens_pathtokens.bin, document_offsets_pathdocument_offsets.npy, ), processorTokensToTextSequence.Config(), post_filters(lambda sample: sample is not None,), ) packed_tokens_ds ConcatThenSplitPackingConfig( datasettoken_documents_ds, )这个例子展示了三层复用自定义源只负责“按索引返回原始行”SampleProcessordataset.py负责把原始行转成TextSequence其中rng参数是 Grain 提供的确定性随机数生成器可安全用于数据增强打包与加载则完全复用通用组件。SingleDatasetConfig在构建时会执行pre_filters - processor(random_map) - post_filters - (shuffle) - DP shard - (repeat)的固定顺序dataset.py这解释了为什么post_filters里常见的lambda sample: sample is not None能过滤掉处理器返回的None。SFT换处理器与打包策略不换加载器SFT 与预训练的差异只体现在处理器和打包策略上加载器保持不变from torchtitan.components.data import ( FirstFitPackingConfig, GrainDataLoader, HuggingFaceRandomAccessSource, SingleDatasetConfig, ) from torchtitan.hf_datasets.text_datasets import ChatProcessor def gsm8k_messages(row): return [ {role: user, content: row[question]}, {role: assistant, content: row[answer]}, ] gsm8k_ds SingleDatasetConfig( sourceHuggingFaceRandomAccessSource.Config( pathopenai/gsm8k, namemain, splittrain, ), processorChatProcessor.Config(messages_fngsm8k_messages), post_filters(lambda sample: sample is not None,), ) gsm8k_packed_ds FirstFitPackingConfig(datasetgsm8k_ds) config.dataloader GrainDataLoader.Config( datasetgsm8k_packed_ds, )无 renderer 时的单轮对话处理不配置 renderer 时ChatProcessor走 tokenizer 的 chat template 处理单轮[user, assistant]对话先校验消息严格为两轮且 role 分别为user、assistant否则ValueError再用apply_chat_template渲染完整对话并追加 EOS生成 next-token 的 input/label 对最后把 prompt 部分的 label 置为IGNORE_INDEX即 -100见 torchtitan/components/loss.py 的IGNORE_INDEX只对 assistant 回答计算损失。prompt/response 边界的定位方式是单独用add_generation_promptTrue渲染 prompt并要求它的 token 序列恰好是完整渲染 token 序列的前缀否则抛出ValueErrortext_datasets.py。这里选择“报错”而非“丢弃样本”是有意的前缀不匹配意味着标签边界无法确定且这类问题通常是模板系统性的静默丢弃会让模型只在小部分数据上训练。超出max_context_length的样本则被整条丢弃因为超长是个别样本问题。ChatProcessor也会在首个样本时打印完整渲染文本便于人工核对。使用 renderer 处理多轮对话多轮对话需要显式选择模型对应的 rendererfrom renderers import Qwen3RendererConfig from torchtitan.components.renderer import RenderersLibraryConfig processor ChatProcessor.Config( messages_fnlambda row: row[messages], rendererRenderersLibraryConfig(renderers_configQwen3RendererConfig()), )RenderersLibraryConfigrenderer.py基于renderers库通过RendererTokenizerWrapper把 TorchTitan 已加载的 tokenizer 适配成 renderer 需要的接口字符偏移、token 到 id 的查找、raw 编码不会二次加载 tokenizer。它明确禁止auto与default类型的 renderer前者依赖name_or_path精确匹配且可能落到不受支持的 DefaultRenderer后者依赖 HF 的apply_chat_template而 TorchTitan 的模板渲染缺少其特殊 token 变量会产生静默不同的 token。使用 renderer 时一次渲染即返回 token 与 loss maskmask 只监督模型生成的 token含回合终止符排除 prompt token 与模板脚手架。约束包括对话必须以 assistant 消息结尾否则报错每个对话是一个样本打包时在对话之间重置位置而不是回合之间超过max_context_length的样本整条丢弃格式化与思维链reasoning保留策略由所选 renderer 决定例如 Qwen3 会在最后一个 user 提问之前省略 assistant 的 reasoning被省略的 token 不计损失若想对每个回合的 reasoning 都训练需要在源数据集里准备独立的对话前缀thinking_retention控制的是 renderer 的 rollout 桥接不作用于此处使用的完整渲染。混合与拼接数据集加权混合DatasetMixConfig 与 WeightedDataset把权重放在每个数据集旁边from torchtitan.components.data import DatasetMixConfig, WeightedDataset # books_ds and code_ds are SingleDatasetConfig values. pretraining_mix_ds DatasetMixConfig( datasets( WeightedDataset(datasetbooks_ds, weight0.75), WeightedDataset(datasetcode_ds, weight0.25), ), )示例中概率之和为 1.0但实现上接受任意正相对权重并在内部归一化——weight0.75配weight0.25表示第一个数据集被抽中的频率是第二个的 3 倍。DatasetMixConfigdataset.py在构建时校验所有权重有限且为正并给每个子数据集分配seed index的偏移插入或重排子数据集会重新播种其后的所有子数据集从而改变数据顺序——断点续训要求代码与配置完全不变的原因之一。混合的语义取决于子数据集的粒度这由 README 中的两个小节给出先混合再打包权重按“文档”计数packed_pretraining_ds ConcatThenSplitPackingConfig( datasetpretraining_mix_ds, )先打包再混合权重按“定长行”计数books_packed_ds ConcatThenSplitPackingConfig(datasetbooks_ds) code_packed_ds ConcatThenSplitPackingConfig(datasetcode_ds) token_ratio_mix_ds DatasetMixConfig( datasets( WeightedDataset(datasetbooks_packed_ds, weight0.67), WeightedDataset(datasetcode_packed_ds, weight0.33), ), )源码注释dataset.py补充了微妙差异全 map 路径下MapDataset.filter会把被拒索引留作None因此权重作用于尝试取样的索引而非被接受的样本而一旦混入 iterable 子数据集权重则作用于各子数据集实际发射的元素——混合TextSequence子项按文档计数混合打包好的定长子项则按物理 token 计数。repeatTrue时混合是无限的repeatFalse时混合在第一个耗尽的子数据集处停止较大的子数据集不会被完整覆盖。从源码结构看按观察到的文档数或监督 token 数自动调节权重的功能尚未实现源码中有 TODO 注释文档也明确说明自定义混合可维护各数据集的移动平均并自行再平衡权重。拼接把有限数据集当作一个语料拼接用于把多个有限数据集视为一个语料每行出现一次占比由各数据集大小决定需要显式权重或流式数据集时改用混合from torchtitan.components.data import DatasetConcatConfig pretraining_corpus_ds DatasetConcatConfig( datasets(books_ds, code_ds, math_ds), )若希望某个有限数据集的每一行在一个 epoch 内出现多次可以在拼接中重复该子项pretraining_corpus_ds DatasetConcatConfig( datasets(books_ds,) * 3 (code_ds, math_ds), )此时每个books_ds行在合并后的有限索引空间中出现 3 次当shuffleTrue时TorchTitan 会在 DP 分片之前对合并后的索引空间做全局 shuffle。实现上dataset.pyDatasetConcatConfig要求全部子项为 map 风格构建时先把每个子数据集以shuffleFalse, repeatFalse, dp_rank0, dp_world_size1的迭代策略构建再MapDataset.concatenate合并最后统一全局 shuffle、DP 分片、repeat。若需求是“相对采样频率”而非“精确的有限重复”应改用混合pretraining_mix_ds DatasetMixConfig( datasets( WeightedDataset(datasetbooks_ds, weight0.6), WeightedDataset(datasetcode_ds, weight0.2), WeightedDataset(datasetmath_ds, weight0.2), ), )最后注意GrainDataLoader.Config(repeatTrue)重复的是整个拼接/混合后的数据集并不会改变某个子数据集的相对贡献。图像与多模态数据图像训练复用同一套 source / dataset / loader / sharding / checkpoint 契约处理器保留模态相关的样本字典collator 生成模型专属的 batch。Qwen 多模态示例from torchtitan.components.data import ( GrainDataLoader, HuggingFaceStreamingSource, SingleDatasetConfig, ) from torchtitan.hf_datasets.multimodal.mm_collator import MultiModalCollator from torchtitan.hf_datasets.multimodal.mm_datasets import ( MMSamplePackingConfig, MultiModalProcessor, _process_cc12_wd_sample, ) mm_processor MultiModalProcessor.Config( sample_processor_process_cc12_wd_sample, ) mm_ds SingleDatasetConfig( sourceHuggingFaceStreamingSource.Config( pathpixparse/cc12m-wds, splittrain, ), processormm_processor, post_filters(lambda sample: sample is not None,), ) packed_mm_ds MMSamplePackingConfig( datasetmm_ds, num_packing_bins8, ) config.dataloader GrainDataLoader.Config( datasetpacked_mm_ds, collatorMultiModalCollator.Config( build_mrope_positionsTrue, patch_sizemm_processor.patch_size, temporal_patch_sizemm_processor.temporal_patch_size, spatial_merge_sizemm_processor.spatial_merge_size, ), streaming_shuffle_buffer_size128, )几个关键语义均有源码对应MultiModalProcessormm_datasets.py适配 Grain 的 map 契约内部调用sample_processor如_process_cc12_wd_sample、_process_obelics_sample。处理流程为解码图像/视频字节 - 缩放到patch_size * spatial_merge_size的倍数并归一化 - 在文本中插入|vision_start||image_pad|...|vision_end|占位 token - 编码文本vision 占位 token 在 labels 中被置为IGNORE_INDEXmm_datasets.py。默认配置包括patch_size16、temporal_patch_size2、spatial_merge_size2、min_pixels65_536、max_pixels16_777_216、max_patches4096等。处理后的样本若超过max_context_length会被整条跳过。MMSamplePackingConfig把整个多模态文档装进定长行先过滤超长样本再用FirstFitPackIterDataset打包input_ids/labels/positions为定长结构pixel_values作为 meta features 保留num_packing_bins是保持打开的候选打包行数量而不是输入样本缓冲区——更大的值可以减少 padding 但会滞留更多媒体数据mm_datasets.py。MultiModalCollatormm_collator.py负责两件事collate_images把图像/视频张量切块patch并 pad 到统一 patch 数产出pixel_values与grid_thw形状(num_images, 3)乘积给出每个条目的 patch 序列长度collate_text拼接整条样本并只 pad token 批的尾部。当build_mrope_positionsTrue时还会在 CPU 数据侧构建三维时间/高/宽的 MRoPE 位置 ID_build_mrope_positions返回(num_tokens, 3)此时patch_order必须为blockraster 顺序会使 MRoPE 与 patch 序列失同步。max_images_per_batch默认 128限制每个批的视觉条目数。自定义图像增强应放进SampleProcessor。Loader 策略一次配置全局生效GrainDataLoader 完整配置运行级行为在GrainDataLoader.Config中一次配好loader.pyimport grain.python as grain config.dataloader GrainDataLoader.Config( datasetpacked_pretraining_ds, seed42, shuffleTrue, repeatTrue, streaming_shuffle_buffer_size1_000, read_optionsgrain.ReadOptions( num_threads16, prefetch_buffer_size500, ), num_prefetch_batches2, )各字段的默认值与作用源码注释字段默认值说明dataset必填任意DatasetConfig叶子、混合、拼接或打包后的数据集collatorTextCollator.Config行到批的转换多模态用MultiModalCollator.Configseed42全局随机种子贯穿 shuffle、random_map、打包shuffleTrue是否全局/流式 shufflerepeatTrue是否无限重复见下文分布式小节中的限制streaming_shuffle_buffer_size1_000每个 rank 保留的流式原始行数用于近似 shuffleread_optionsgrain.ReadOptions()MapDataset转IterDataset时的并发读参数num_prefetch_batches2预留给 trainer 的完整 collate 批数量GrainDataLoader.__init__里还有一道防线当dp_world_size 1且repeatFalse时直接抛ValueError提示必须用repeatTrue配合 trainer 控制的步数原因见“分布式与检查点行为”一节。加载器构建完数据集图后执行batch(collator.num_rows_per_batch(), drop_remainderconfig.repeat, batch_fncollator)再用ThreadPrefetchIterDataset预取num_prefetch_batches个完整批。TextCollatorcollators.py把若干TextSequence行合并进**预分配、页锁定pin_memory**的定长张量padding 位置 label 为IGNORE_INDEX、padding_maskTrue并在 batch 字典中预先计算num_valid_tokens参与损失的 token 数避免 trainer 在关键路径上重复扫描。读取带索引的数据集MapDataset 与 IterDataset 的转换规则随机访问源构成MapDataset支持dataset[index]当后续某个阶段需要顺序消费样本时它会转成IterDataset。转换时机遵循以下规则README 原文all children are MapDataset: DatasetMixConfig remains a MapDataset any child is IterDataset: DatasetMixConfig converts each MapDataset child packing: converts its child if needed and returns IterDataset GrainDataLoader: converts a MapDataset if no earlier stage didread_options控制每一次MapDataset到IterDataset的转换grain.ReadOptions( num_threads16, # indexed samples read concurrently prefetch_buffer_size500, # samples waiting for the consumer )每次转换都有自己独立的线程和缓冲区全 map 的混合只转换一次混有流的混合会为每个 map 子数据集各自转换一次。loader.py 中的 TODO 也提到当前只使用多线程而非多进程做 CPU 密集处理未来计划在更早的边界引入共享 worker 池。流式 shuffle 与就绪批streaming_shuffle_buffer_size保留用于近似 shuffle的原始行数每个 rank。缓冲区越大混合越均匀但内存占用越高。流式源在SingleDatasetConfig._build_iter_dataset里通过grain.experimental.WindowShuffleIterDataset实现dataset.py窗口大小即此值。num_prefetch_batches允许等待 trainer 的完整 collate 批数量trainer computes batch 10 background thread prepares batches 11 and 12即 trainer 计算第 10 批时后台线程已在准备第 11、12 批。分布式与检查点行为有效数据并行effective DP决定数据归属只有有效 DP 坐标负责选取数据effective DP data_parallel_replicate_degree * data_parallel_shard_degree different effective-DP ranks - disjoint source rows TP/PP/CP peers - same rows for their effective-DP coordinate也就是说TP/PP/CP 的 peer rank 与它们的有效 DP 坐标共享相同数据行而不同的有效 DP rank 之间数据行互不相交。这是保证前向/反向集合通信语义一致的关键。数据所有权的判定时机数据归属在批处理之前决定不同源/组合的归属方式不同random-access source - global shuffle - contiguous balanced DP shard Hugging Face stream - source-level DP shard DatasetMixConfig - combines children already owned by this DP rank DatasetConcatConfig - concatenates, globally shuffles, then DP-shards packing - packs samples locally on each DP rank GrainDataLoader - batches and collates that ranks samples对随机访问训练而言每个 rank 拿到的是全局 shuffle 后索引空间的连续切片而不是原始语料的连续区域。SingleDatasetConfig与DatasetConcatConfig中divmod分片逻辑dataset.py保证了dp_world_size不整除时余数行被均衡分配。当len(dataset) dp_world_size时会直接报错避免分片后出现空集。有限数据与 hang 的陷阱当有效 DP 大于 1 时repeatFalse会被拒绝loader.py因为各 rank 可能在不同的 step耗尽数据导致训练集合通信挂起。正确做法是repeatTrue让 trainer 的步数控制训练何时停止。加载器还定义了DataloaderExhaustedErrorloader.py刻意继承Exception而非StopIteration以避免 PEP 479 把生成器内部的StopIteration包装成RuntimeError导致程序崩溃。state_dict 与断点续训GrainDataLoader.state_dict()loader.py记录以下内容源游标与 shuffle/repeat 进度含 HF 流式源的 epoch 与内部状态mix 子数据集与打包缓冲区的状态如 document-aware 打包的 remainder 与偏移batching 与 prefetch 状态有效 DP 维度dp_world_size。恢复load_state_dict的硬性要求代码、配置、源内容、tokenizer 与有效 DP 维度都必须与保存时一致。源码会显式校验state_dict[version] 1以及dp_world_size是否变化并确保 checkpoint 中包含当前dp_rank_{i}的条目缺失或版本不符都会抛错。这也解释了上文混合时“插入或重排子数据集会重新播种其后所有子数据集”为何影响续训一致性。配套的单元测试覆盖了这些行为例如test_indexed_jsonl_random_access、test_hf_shuffled_repeat_advances_epoch、test_hf_resume_mid_second_epoch、test_loader_requires_repeat_with_data_parallelism、test_weighted_map_mix_keeps_weight_with_dataset、test_concat_shards_after_one_global_index_space、test_document_aware_concat_packing_restores_exactly等见 tests/unit_tests/cpu/components/data/test_grain_data.py可作为理解各阶段精确语义的补充证据。小结TorchTitan 的数据流水线把“源、处理、组合、加载”四层职责分离全部以声明式 Config 表达最终由GrainDataLoader统一构建并接入 trainer。文本预训练、SFT 与多模态训练共用同一套骨架差异仅在SampleProcessor与打包/ collator 的选择上分布式场景下“有效 DP 决定数据归属 全局 shuffle 后分片 repeatTrue”的约定加上可递归恢复的 Grain 状态为大规模多机训练提供了可复现与可断点续训的坚实基础。动手实践时建议从本文的 JSONL 最小示例出发逐步替换源、处理器与组合方式并用仓库中的单元测试验证每一步的语义。【免费下载链接】torchtitanA PyTorch native platform for training generative AI models项目地址: https://gitcode.com/GitHub_Trending/to/torchtitan创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考