Qwen3.5 × Twinkle 低コストモデルトレーニングのベストプラクティス

原创 通义实验室 2026-03-12 17:20 浙江

image

如何把模型训练过程变得简单高效？

大模型训练正在成为越来越多开发者和企业的刚需。但随着模型规模扩大，分布式训练的复杂度也呈指数级上升：显存不够用、通信开销大、配置太复杂......这些问题几乎每个做大模型训练的人都会遇到。

最新开源的 Twinkle 框架，正是为了解决这些问题而生。

今天，我们就以 Qwen3.5 的训练实践为主线，结合 Twinkle 分享一套高效、易用的大模型训练方案，帮助大家把原本复杂的训练过程变得简单高效。

Twinkle 是一个面向生产的大模型训练框架。它的核心设计非常容易理解：训练逻辑用 Python 代码表达，并基于Client-Server解耦的架构设计，支持从单卡到多机集群的多种训练场景。在给与最高灵活性和最简逻辑的同时，能支撑单租和多租服务化训练。同时Twinkle的客户端和服务端实现，均完整开源。

这意味着：

实验室里写的训练脚本，改一行代码就能多方式运行；

全开放的算法定制能力；

不需要维护多套代码来支持 torchrun、Ray、HTTP 等不同模式；

算法工程师专注写训练逻辑，框架自动处理分布式通信。

Twinkle 同时支持 Transformers和Megatron后端，以及多租户 LoRA 训练——多个用户共享一个基座模型，各自训练自己的适配器。

Twinkle GitHub⬇️

https://github.com/modelscope/twinkle

这也是最常见的训练场景：在本地一台机器上使用 1~8 张 GPU/NPU 进行模型训练。Twinkle 基于 PyTorch 原生接口，完整支持 FSDP2、DDP 等并行策略，让分布式训练的门槛降到最低。

基础篇：LoRA微调实战

下面是一个完整的 Qwen3.5-4B LoRA 微调代码：

from peft import LoraConfig

from tqdm import tqdm

import twinkle

from twinkle import DeviceMesh, get_device_placement, get_logger

from twinkle.dataloader import DataLoader

from twinkle.dataset import Dataset, DatasetMeta

from twinkle.model import TransformersModel

from twinkle.preprocessor import SelfCognitionProcessor

构造 device_mesh：fsdp=4, dp=2，共使用 8 张卡

device_mesh = DeviceMesh.from_sizes(fsdp_size=4, dp_size=2)

使用 torchrun 模式

twinkle.initialize(mode='local', global_device_mesh=device_mesh)

logger = get_logger()

def eval(model):

    # 验证集：100 条样本

    dataset = Dataset(dataset_meta=DatasetMeta('ms://swift/self-cognition', data_slice=range(100)))

    dataset.set_template('Template', model_id='ms://Qwen/Qwen3.5-4B')

    dataset.map(SelfCognitionProcessor('twinkle大模型', 'ModelScope社区'))

    dataset.encode()

    dataloader = DataLoader(dataset=dataset, batch_size=8)

    for step, batch in tqdm(enumerate(dataloader)):

        model.forward_only(inputs=batch)

        model.calculate_loss()

    metrics = model.calculate_metric(is_training=False)

    return metrics

def train():

    # 训练集：1000 条样本

    dataset = Dataset(dataset_meta=DatasetMeta('ms://swift/self-cognition', data_slice=range(1000)))

    # 设置模板，准备编码

    dataset.set_template('Template', model_id='ms://Qwen/Qwen3.5-4B')

    # 数据预处理：替换自我认知数据中的占位符

    dataset.map(SelfCognitionProcessor('twinkle大模型', 'ModelScope社区'))

    # 编码数据集

    dataset.encode()

    # 全局 batch size = 8，8 张卡每张处理 1 条

    dataloader = DataLoader(dataset=dataset, batch_size=8)

    # 加载模型

    model = TransformersModel(model_id='ms://Qwen/Qwen3.5-4B')

    model.model._no_split_modules = {'Qwen3_5DecoderLayer'}

    lora_config = LoraConfig(r=8, lora_alpha=32, target_modules='all-linear')

    # 添加 LoRA 适配器，命名为 'default'

    # 注释掉这行即可切换到全参数训练

    model.add_adapter_to_model('default', lora_config, gradient_accumulation_steps=2)

    # 为 LoRA 配置优化器

    model.set_optimizer(optimizer_cls='AdamW', lr=1e-4)

    # 配置学习率调度器

    model.set_lr_scheduler(

        scheduler_cls='CosineWarmupScheduler', num_warmup_steps=5, num_training_steps=len(dataloader))

    logger.info(get_device_placement())

    # 打印训练配置

    logger.info(model.get_train_configs())

    logger.info(f'Total steps: {len(dataloader)}')

    loss_metric = 99.0

    # LoRA 训练：约 8G * 8 显存占用

    # 全参数训练：约 18G * 8 显存占用

    for step, batch in enumerate(dataloader):

        # 前向 + 反向传播

        model.forward_backward(inputs=batch)

        # 梯度裁剪 + 优化器步进

        model.clip_grad_and_step()

        if step % 20 == 0:

            # 打印训练指标

            metric = model.calculate_metric(is_training=True)

            logger.info(f'Current is step {step} of {len(dataloader)}, metric: {metric}')

        if step > 0 and step % 40 == 0:

            # 定期验证

            metrics = eval(model)

            logger.info(f'Eval metric: {metrics}')

            metrics['step'] = step

            # 保存最优检查点

            if loss_metric > float(metrics['loss']):

                model.save(f'checkpoint-{step}')

                loss_metric = float(metrics['loss'])

    model.save(f'last-checkpoint')

if __name__ == '__main__':

    train()

启动命令

CUDA_VISIBLE_DEVICES=0,1,2,3,4,5,6,7 torchrun --nproc_per_node=8 fsdp2.py

关键设计说明

 DeviceMesh 并行策略：如何分配显卡？

device_mesh = DeviceMesh.from_sizes(fsdp_size=4, dp_size=2)

🔹 做法：让 8 张 GPU 协同工作，同时降低单卡显存占用，我们将 8 张卡分成「4 路 FSDP 分片 + 2 路数据并行」的混合并行。

🔹 效果：Qwen3.5-4B 在 bf16 精度下全参训练单卡约占用 18GB，LoRA 模式下单卡占用大约 8GB，使用游戏卡也可以流畅跑。

梯度累积策略：显存不够怎么办？

model.add_adapter_to_model('default', lora_config, gradient_accumulation_steps=2)

🔹 做法：在显存受限但希望进行较大 Batch Size 训练时，为了提升收敛稳定性，我们设置了 gradient_accumulation_steps=2，即每累积 2 个 Micro-Batch 再更新一次参数。

🔹 效果：等效于 Batch Size 翻倍，在不增加硬件成本的情况下获得更好的训练效果。

算法过程外露：谁掌握控制权？

🔹 做法：前向、反向、梯度裁剪、检查点保存——所有训练关键步骤都直接写在主循环里。

🔹 效果：开发者对训练过程有完整的控制权。底层的分布式通信由 Twinkle infra 负责，无论切换 Ray 还是 torchrun 模式，主循环代码都不需要改动。对于复杂的算法研究而言，这一点尤为关键。

进阶篇：Ray模式下的强化学习训练

当训练任务涉及强化学习（如GRPO、PPO等）时，往往需要模型训练和采样推理两个过程协同进行。 Twinkle 的 Ray 模式特别适合这类需要模型与采样器分离部署的 RL 算法。

以GRPO为例我们将为你展示如何在 Ray 模式下完成 RL 训练。

为什么选择GRPO？

与 PPO 不同，GRPO 不需要单独训练一个价值模型，而是通过组内采样结果的相对奖励来估计优势函数，这样做的好处是：

简化训练流程，减少一个需要训练的模块；

降低显存开销，让RL训练更加轻量；

训练稳定性更好，更容易收敛。

在下面的例子中，我们用 4 张卡跑模型训练，另外 4 张卡跑 vLLM 采样，两者通过 Ray 集群协调：

from typing import List, Dict, Any

from peft import LoraConfig

import twinkle

from twinkle import DeviceMesh, DeviceGroup, get_device_placement, get_logger

from twinkle.advantage import GRPOAdvantage

from twinkle.checkpoint_engine import CheckpointEngineManager

from twinkle.data_format import SamplingParams

from twinkle.dataloader import DataLoader

from twinkle.dataset import Dataset, DatasetMeta

from twinkle.model import TransformersModel

from twinkle.processor import InputProcessor

from twinkle.reward import GSM8KAccuracyReward, GSM8KFormatReward

from twinkle.sampler import vLLMSampler

from twinkle.templateimport Template

from twinkle.metric import CompletionRewardMetric

from twinkle.preprocessor.llm import GSM8KProcessor

logger = get_logger()

MODEL_ID = 'ms://Qwen/Qwen3.5-4B'

MODEL_GPUS = 4      # 模型训练用 4 张卡

SAMPLER_GPUS = 4    # vLLM 采样用 4 张卡

NUM_GPUS = MODEL_GPUS + SAMPLER_GPUS

NUM_GENERATIONS = 8     # 每组采样 8 个结果

MAX_NEW_TOKENS = 4096

LEARNING_RATE = 1e-5

MAX_STEPS = 200

BATCH_SIZE = 16

MINI_BATCH_SIZE = 16

MICRO_BATCH_SIZE = 2

ADAPTER_NAME = 'default'

def create_gsm8k_dataset():

    dataset = Dataset(DatasetMeta('ms://modelscope/gsm8k', subset_name='main', split='train'))

    dataset.set_template('Template', model_id=MODEL_ID, max_length=2048)

    dataset.map(GSM8KProcessor())

    dataset.encode(add_generation_prompt=True)

    return dataset

def compute_rewards(trajectories: List[Dict[str, Any]]):

    accuracy_reward_fn = GSM8KAccuracyReward()

    format_reward_fn = GSM8KFormatReward()

    accuracy_rewards = accuracy_reward_fn(trajectories)

    format_rewards = format_reward_fn(trajectories)

    total_rewards = [a + f for a, f in zip(accuracy_rewards, format_rewards)]

    return total_rewards, format_rewards, accuracy_rewards

def main():

    # 模型和采样器分到不同的 GPU 组

    device_groups = [

        DeviceGroup(name='model', ranks=list(range(MODEL_GPUS)), device_type='GPU'),

        DeviceGroup(name='sampler', ranks=list(range(MODEL_GPUS, NUM_GPUS)), device_type='GPU'),

    ]

    model_mesh = DeviceMesh.from_sizes(world_size=MODEL_GPUS, dp_size=MODEL_GPUS)

    sampler_mesh = DeviceMesh.from_sizes(world_size=SAMPLER_GPUS, dp_size=SAMPLER_GPUS)

    # Ray 模式初始化

    twinkle.initialize(mode='ray', nproc_per_node=NUM_GPUS, groups=device_groups, lazy_collect=False)

    lora_config = LoraConfig(target_modules='all-linear', r=32, lora_alpha=64, lora_dropout=0.05)

    # 模型部署在 'model' 组

    model = TransformersModel(model_id=MODEL_ID, device_mesh=model_mesh, remote_group='model')

    model.add_adapter_to_model(ADAPTER_NAME, lora_config, gradient_accumulation_steps=1)

    model.set_optimizer('AdamW', lr=LEARNING_RATE)

    model.set_lr_scheduler('CosineAnnealingLR', T_max=MAX_STEPS, eta_min=0)

    model.set_loss('GRPOLoss', epsilon=0.2)

    model.set_processor(InputProcessor)

    model.set_template('Template', model_id=MODEL_ID)

    # 采样器部署在 'sampler' 组

    sampler = vLLMSampler(

        model_id=MODEL_ID,

        engine_args={

            'gpu_memory_utilization': 0.8,

            'max_model_len': 4096,

            'max_lora_rank': 32,

            'enable_lora': False,

        },

        device_mesh=sampler_mesh,

        remote_group='sampler',

    )

    sampler.set_template(Template, model_id=MODEL_ID)

    ckpt_manager = CheckpointEngineManager(model=model, sampler=sampler)

    dataloader = DataLoader(

        dataset=create_gsm8k_dataset,

        batch_size=BATCH_SIZE,

        min_batch_size=BATCH_SIZE,

        device_mesh=model_mesh,

        remote_group='model',

    )

    advantage_fn = GRPOAdvantage()

    metrics = CompletionRewardMetric()

    sampling_params = SamplingParams(max_tokens=MAX_NEW_TOKENS)

    optim_step = 0

    logger.info(get_device_placement())

    for batch in dataloader:

        if optim_step >= MAX_STEPS:

            break

        metrics.reset()

        global_prompts = batch if isinstance(batch, list) else [batch]

        # 同步权重到采样器

        ckpt_manager.sync_weights(merge_and_sync=True)

        sampler.reset_prefix_cache()

        # 组采样：每个 prompt 采样 NUM_GENERATIONS 个结果

        sample_response = sampler.sample(

            global_prompts * NUM_GENERATIONS,

            sampling_params,

            num_samples=1,

        )

        all_input_data = []

        all_old_logps = []

        all_completion_lengths = []

        for sequence in sample_response.sequences:

            all_input_data.append(sequence.new_input_feature)

            all_old_logps.append(sequence.logprobs)

            all_completion_lengths.append(len(sequence.tokens))

        # 计算奖励

        total_rewards, format_rewards, accuracy_rewards = compute_rewards(all_input_data)

        metrics.accumulate(

            completion_lengths=all_completion_lengths,

            rewards={

                'total': total_rewards,

                'format': format_rewards,

                'accuracy': accuracy_rewards,

            },

        )

        # GRPO 优势估计：组内归一化

        advantages = advantage_fn(total_rewards, num_generations=NUM_GENERATIONS, scale='group').tolist()

        # Mini-batch 训练

        total_completions = len(all_input_data)

        for mb_start in range(0, total_completions, MINI_BATCH_SIZE):

            mb_end = min(mb_start + MINI_BATCH_SIZE, total_completions)

            mb_inputs = all_input_data[mb_start:mb_end]

            mb_old_logps = all_old_logps[mb_start:mb_end]

            mb_advantages = advantages[mb_start:mb_end]

            model.forward_backward(

                inputs=mb_inputs,

                old_logps=mb_old_logps,

                advantages=mb_advantages,

                micro_batch_size=MICRO_BATCH_SIZE,

            )

            model.clip_grad_and_step()

            optim_step += 1

            if optim_step >= MAX_STEPS:

                break

            log_dict = metrics.calculate()

            log_dict.update(model.calculate_metric(is_training=True))

            metrics.reset()

            logger.info(f'[Step {optim_step}/{MAX_STEPS}] {log_dict}')

    logger.info(f'Training completed. optim_steps={optim_step}')

    model.save('grpo-gsm8k-checkpoint')

if __name__ == '__main__':

    main()

启动命令（因为是 Ray 集群运行，所以启动命令非常简单）

python train.py

GRPO 训练的关键设计

1、模型与采样器分离：DeviceGroup 将 8 张卡分成两组，训练和采样互不干扰，采样流程可充分利用 vLLM 的高吞吐；

2、组采样策略：global_prompts * NUM_GENERATIONS 让每个问题采样多个结果，通过组内相对奖励估计优势——不需要单独训练价值模型；

3、权重同步：ckpt_manager.sync_weights() 在每次采样前将训练模型权重同步到 vLLM，确保采样始终使用最新策略；

4、算法组件外露：GRPOAdvantage 和 GRPOLoss 直接注册到模型，可替换为其他 RL 算法组件而不需修改其他任何代码。

这种写法的核心价值在于：整个 RL 训练流程——采样、奖励计算、优势估计、梯度更新，都展开在可见的 Python 主循环里，没有隐藏的魔法。不同 RL 算法的差异，往往只在于替换几个组件。

在实际生产环境中，算力资源和服务消费方往往是分离的。比如企业内部训推平台、云服务商的 Serverless 训练服务。这时候，就需要把训练能力以 API 的形式暴露出来。

Twinkle 支持两种 Client 接入方式：

Twinkle Client：和本地训练 API完 全一致，适合需要精细控制的场景

Tinker Client：兼容 Tinker 生态，调用方式更简洁

无论哪种方式，服务端只维护一份基座模型，多个客户端可以并行训练各自的 LoRA 适配器，实现真正的多租户隔离。

Twinkle Client：细粒度控制

Twinkle Client 提供与本地训练几乎完全一致的 API，适合需要精细控制训练流程的场景。

import dotenv

dotenv.load_dotenv('.env')

from peft import LoraConfig

from twinkle import get_logger

from twinkle.dataset import DatasetMeta

from twinkle_client import init_twinkle_client

from twinkle_client.dataloader import DataLoader

from twinkle_client.dataset import Dataset

from twinkle_client.model import MultiLoraTransformersModel

logger = get_logger()

初始化 Twinkle 客户端

client = init_twinkle_client(base_url='http://127.0.0.1:8000', api_key='EMPTY_TOKEN')

查询已有训练运行和检查点

runs = client.list_training_runs()

resume_path = None

for run in runs:

    logger.info(run.model_dump_json(indent=2))

    checkpoints = client.list_checkpoints(run.training_run_id)

    for checkpoint in checkpoints:

        logger.info(checkpoint.model_dump_json(indent=2))

        # 如需恢复训练，取消下面注释

        # resume_path = checkpoint.twinkle_path

def train():

    # 准备数据集

    dataset = Dataset(dataset_meta=DatasetMeta('ms://swift/self-cognition', data_slice=range(500)))

    dataset.set_template('Template', model_id='ms://Qwen/Qwen3.5-4B', max_length=512)

    dataset.map('SelfCognitionProcessor', init_args={'model_name': 'twinkle模型', 'model_author': 'ModelScope社区'})

    dataset.encode(batched=True)

    dataloader = DataLoader(dataset=dataset, batch_size=4)

    # 配置模型

    model = MultiLoraTransformersModel(model_id='ms://Qwen/Qwen3.5-4B')

    lora_config = LoraConfig(target_modules='all-linear')

    model.add_adapter_to_model('default', lora_config, gradient_accumulation_steps=2)

    model.set_template('Template')

    model.set_processor('InputProcessor', padding_side='right')

    model.set_loss('CrossEntropyLoss')

    model.set_optimizer('AdamW', lr=1e-4)

    model.set_lr_scheduler('LinearLR')

    # 恢复训练（如有检查点）

    if resume_path:

        logger.info(f'Resuming training from {resume_path}')

        model.load(resume_path, load_optimizer=True)

    logger.info(model.get_train_configs())

    for epoch in range(3):

        logger.info(f'Starting epoch {epoch}')

        for step, batch in enumerate(dataloader):

            # 前向 + 反向

            output = model.forward_backward(inputs=batch)

            if step % 2 == 0:

                logger.info(f'Current is step {step // 2}, loss: {output}')

            model.clip_grad_norm(1.0)

            model.step()

            model.zero_grad()

            model.lr_step()

        # 保存检查点

        twinkle_path = model.save(name=f'twinkle-epoch-{epoch}', save_optimizer=True)

        logger.info(f'Saved checkpoint: {twinkle_path}')

if __name__ == '__main__':

    train()

Twinkle Client 的特点：

API 与本地训练完全一致，无额外学习成本；

支持断点续训、检查点管理；

可动态切换 LoRA 适配器、损失函数、优化器等组件。

Tinker Client：简洁即用

Tinker 是一个轻量级训练 API。Twinkle 对 Tinker 客户端提供完整支持，几行代码就能拉起训练。已有 Tinker 代码的项目可以直接迎移到 Twinkle 服务端.

import os

from tinker import types

from tqdm import tqdm

from twinkle import init_tinker_client

from twinkle.dataloader import DataLoader

from twinkle.dataset import Dataset, DatasetMeta

from twinkle.preprocessor import SelfCognitionProcessor

from twinkle.server.tinker.common import input_feature_to_datum

初始化 Tinker 客户端（必须在导入 ServiceClient 之前）

init_tinker_client()

from tinker import ServiceClient

基座模型

base_model = 'Qwen/Qwen3.5-4B'

base_url = 'http://www.modelscope.cn/twinkle'

def train():

    # 准备数据集

    dataset = Dataset(dataset_meta=DatasetMeta('ms://swift/self-cognition', data_slice=range(500)))

    dataset.set_template('Template', model_id=f'ms://{base_model}', max_length=256)

    dataset.map(SelfCognitionProcessor('Twinkle模型', 'ModelScope团队'), load_from_cache_file=False)

    dataset.encode(batched=True, load_from_cache_file=False)

    dataloader = DataLoader(dataset=dataset, batch_size=8)

    # 初始化训练客户端

    service_client = ServiceClient(

        base_url=base_url,

        api_key=os.environ.get('MODELSCOPE_TOKEN')

    )

    training_client = service_client.create_lora_training_client(base_model=base_model, rank=16)

    # 训练循环

    for epoch in range(3):

        print(f'Epoch {epoch}')

        for step, batch in tqdm(enumerate(dataloader)):

            # 转换输入格式

            input_datum = [input_feature_to_datum(input_feature) for input_feature in batch]

            # 远端前向 + 反向

            fwdbwd_future = training_client.forward_backward(input_datum, 'cross_entropy')

            # 远端优化器步进

            optim_future = training_client.optim_step(types.AdamParams(learning_rate=1e-4))

            # 等待结果

            fwdbwd_result = fwdbwd_future.result()

            optim_result = optim_future.result()

            print(f'Training Metrics: {optim_result}')

        # 保存检查点

        save_future = training_client.save_state(f'twinkle-lora-{epoch}')

        save_result = save_future.result()

        print(f'Saved checkpoint to {save_result.path}')

if __name__ == '__main__':

    train()

Tinker Client 的特点：

调用方式极简，适合快速上手；

完全兼容 Tinker 生态，已有代码可无缝迁移；

支持魔搭官方训练环境（见下文）；

魔搭官方训练环境：零成本体验Twinkle

Twinkle 框架开源的同时，魔搭社区依托自身算力基础设施，提供了托管的模型训练服务（Training as a Service）。开发者无需准备 GPU 资源，通过 API 调用即可免费体验 Twinkle 的训练能力。

使用方式：

1、注册魔搭账号并申请加入 Twinkle-Explorers 组织；

2、在 Token 管理页面 获取 API Key；

3、使用上面的 Tinker Client 代码，修改 endpoint：

base_url = 'https://www.modelscope.cn/twinkle'

base_model = 'Qwen/Qwen3-30B-A3B-Instruct-2507'  # 官方环境当前部署的模型

选型建议：

如果你是算法研究员，需要频繁调整训练流程，从 torchrun 模式开始，验证完成后再考虑是否服务化；

如果你是平台开发者，需要为企业内部提供训练服务，部署 Twinkle Server，根据用户习惯提供 Twinkle Client 或 Tinker Client 两种接入方式；

如果你只是想快速体验 Twinkle 的能力，直接用魔搭官方环境，5 分钟跑通第一个训练任务。

我们希望为大模型开发者提供一个简单、高效、透明的训练工具。无论你是刚接触大模型训练的新手，还是深耕算法多年的专家，都能在 Twinkle 中找到适合自己的用法。

欢迎动手试试，体验“代码即配置”的畅快感。如果在使用中有任何问题或建议，也欢迎随时通过公众号后台反馈。

推荐阅读

CoPaw 开源！欢迎爆改你的专属智能搭档

image一句话即可自由生成！语音双模型上线

阅读原文

跳转微信打开