Firefly 是一个开源的大模型训练项目,支持对主流的大模型进行预训练、指令微调和DPO,包括但不限于Yi-1.5、Llama3、Gemma、Qwen1.5、MiniCPM、Llama、InternLM、Baichuan、ChatGLM、Yi、Deepseek、Qwen、Orion、Ziya、Xverse、Mistral、Mixtral-8x7B、Zephyr、Vicuna、Bloom等。 本项目支持全量参数训练、LoRA、QLoRA高效训练,支持预训练、SFT、DPO。
在此基础上,我们对Firefly项目进行剖析,力求了解其中的每一个细节,那么废话不多数,我们开始吧!
在笔者这一年来对大模型训练微调的代码框架理解,大模型微调的训练步骤是规范化的,如下:
- 1 Model、Tokenizer载入
- 2 Dataset载入
- pretrained dataset
- fine-tuning dataset
- dpo dataset
- 3 Datacollator载入
- 4 Trainer载入
除了上诉4个大步骤之外,还有一些配置的细节也一一向大家展开。
1 函数main
俗话说看问题要得先从一个口子看起,以此口子从宏观到微观,最终窥得全貌。如此解析一个库最重要的是从入口看起,对于python来说,入口只能是main函数了,我们直接定位到Firefly/train.py的main函数
args, training_args = setup_everything()
# 加载各种组件
trainer = init_components(args, training_args)
# 开始训练
logger.info("*** starting training ***")
train_result = trainer.train()
# 保存最好的checkpoint
final_save_path = join(training_args.output_dir)
trainer.save_model(final_save_path) # Saves the tokenizer too
# 保存训练指标
metrics = train_result.metrics
trainer.log_metrics("train", metrics)
trainer.save_metrics("train", metrics)
trainer.save_state()
由上述函数体,我们可以将main.py做的事情分为以下几项(与我们上述总结的不冲突):
- 1 载入配置文件args和training_args
- 2 载入各种组件,model、tokenizer、dataset、datacllator、trainer
- 3 保存模型
- 4 保存训练指标
2 模型配置
从setup_everything函数中,我们将一步一步解析Firefly对于库参数和模型训练参数的载入。
def setup_everything():
parser = argparse.ArgumentParser()
# parser.add_argument("--train_args_file", type=str, default='train_args/pretrain/full/bloom-1b1-pretrain-full.json', help="")
parser.add_argument("--train_args_file", type=str, default='train_args/sft/qlora/qwen-7b-sft-qlora.json', help="")
parser.add_argument("--local_rank", type=int, help="")
args = parser.parse_args()
train_args_file = args.train_args_file
# 读取训练的参数配置
parser = HfArgumentParser((CustomizedArguments, TrainingArguments))
# 解析得到自定义参数,以及自带参数
args, training_args = parser.parse_json_file(json_file=train_args_file)
# 创建输出目录
if not os.path.exists(training_args.output_dir):
os.makedirs(training_args.output_dir)
logger.add(join(training_args.output_dir, 'train.log'))
logger.info("train_args:{}".format(training_args))
# 加载训练配置文件
with open(train_args_file, "r") as f:
train_args = json.load(f)
# 保存训练参数到输出目录
with open(join(training_args.output_dir, 'train_args.json'), "w") as f:
json.dump(train_args, f, indent=4)
# 设置随机种子
set_seed(training_args.seed)
# check some setting
assert args.task_type in ['pretrain', 'sft', 'dpo'], "task_type should be in ['pretrain', 'sft', 'dpo']"
assert args.train_mode in ['full', 'lora', 'qlora'], "task_type should be in ['full', 'lora', 'qlora']"
assert sum([training_args.fp16, training_args.bf16]) == 1, "only one of fp16 and bf16 can be True"
# assert not (args.task_type == 'dpo' and args.use_unsloth), 'We have not tested Unsloth during DPO yet. Please set use_unsloth=False when task_type=dpo'
return args, training_args
上诉配置代码中,需要我们注意的步骤如下:
- 1 载入args(库的一些配置)和training_args(Trainer参数)
- 2 保存训练参数
- 3 保存训练日志
args配置详细细节:
Firefly/component/argument.py中(实际另外设置json存储也行):
@dataclass
class CustomizedArguments:
"""
一些自定义参数
"""
max_seq_length: int = field(metadata={"help": "输入最大长度"})
train_file: str = field(metadata={"help": "训练集。如果task_type=pretrain,请指定文件夹,将扫描其下面的所有jsonl文件"})
model_name_or_path: str = field(metadata={"help": "预训练权重路径"})
template_name: str = field(default="", metadata={"help": "sft时的数据格式"})
eval_file: Optional[str] = field(default="", metadata={"help": "验证集"})
max_prompt_length: int = field(default=512, metadata={"help": "dpo时,prompt的最大长度"})
beta: float = field(default=0.1, metadata={"help": "The beta factor in DPO loss"})
tokenize_num_workers: int = field(default=10, metadata={"help": "预训练时tokenize的线程数量"})
task_type: str = field(default="sft", metadata={"help": "预训练任务:[pretrain, sft]"})
train_mode: str = field(default="qlora", metadata={"help": "训练方式:[full, qlora]"})
lora_rank: Optional[int] = field(default=64, metadata={"help": "lora rank"})
lora_alpha: Optional[int] = field(default=16, metadata={"help": "lora alpha"})
lora_dropout: Optional[float] = field(default=0.05, metadata={"help": "lora dropout"})
use_unsloth: Optional[bool] = field(default=False, metadata={"help": "use sloth or not"})
training_args配置详细细节:
Firefly/train_args/sft/lora/qwen1.5-0.5b-sft-lora.json
{
"output_dir": "output/firefly-qwen1.5-0.5b-sft-huanhuan-lora",
"model_name_or_path": "qwen_0.5b",
"train_file": "./data/huanhuan/merged_huanhuan.jsonl",
"template_name": "qwen",
"train_mode": "lora",
"num_train_epochs": 1,
"per_device_train_batch_size": 16,
"gradient_accumulation_steps": 1,
"learning_rate": 2e-4,
"max_seq_length": 1024,
"logging_steps": 100,
"save_steps": 100,
"save_total_limit": 1,
"lr_scheduler_type": "constant_with_warmup",
"warmup_steps": 100,
"lora_rank": 64,
"lora_alpha": 128,
"lora_dropout": 0.05,
"gradient_checkpointing": true,
"disable_tqdm": false,
"optim": "paged_adamw_32bit",
"seed": 42,
"fp16": true,
"report_to": "tensorboard",
"dataloader_num_workers": 0,
"save_strategy": "steps",
"weight_decay": 0,
"max_grad_norm": 0.3,
"remove_unused_columns": false
}
3 模型和分词器载入
上一节我们重点关注了args和training_args的载入,了解了配置文件才可能对训练过程进行更细节的调整,毕竟项目主体构建完毕后,最后进行训练调整基本上都是和配置文件打交道。
本节重点关注模型载入和分词器载入,以及过程中遇到的一些细节。
3.1 Tokenizer
照例,我们先放上载入分词器的代码,然后总结需要注意的点。
def load_tokenizer(args):
config = AutoConfig.from_pretrained(args.model_name_or_path, trust_remote_code=True)
# 加载tokenzier
tokenizer = AutoTokenizer.from_pretrained(
args.model_name_or_path,
trust_remote_code=True,
# llama不支持fast
use_fast=False if config.model_type == 'llama' or config.model_type == 'internlm2' else True
)
# 部分模型的base与chat版本的tokenizer存在差异
if 'internlm2' in args.model_name_or_path.lower():
tokenizer._added_tokens_encoder.update({'<|im_start|>': 92543})
tokenizer._added_tokens_encoder.update({'<|im_end|>': 92542})
tokenizer._added_tokens_decoder.update({92543: AddedToken('<|im_start|>')})
tokenizer._added_tokens_decoder.update({92542: AddedToken('<|im_end|>')})
tokenizer.add_special_tokens({'additional_special_tokens': ['<|im_start|>', '<|im_end|>']})
elif 'orion' in args.model_name_or_path.lower():
tokenizer.add_special_tokens({'bos_token': '', 'eos_token': ''})
elif 'gemma' in args.model_name_or_path.lower():
tokenizer.add_special_tokens({'additional_special_tokens': ['', '']})
if tokenizer.__class__.__name__ == 'QWenTokenizer':
tokenizer.pad_token_id = tokenizer.eod_id
tokenizer.bos_token_id = tokenizer.eod_id
tokenizer.eos_token_id = tokenizer.eod_id
if tokenizer.pad_token is None:
tokenizer.pad_token = tokenizer.eos_token
assert tokenizer.pad_token_id is not None, "pad_token_id should not be None"
assert tokenizer.eos_token_id is not None, "eos_token_id should not be None"
logger.info(f'vocab_size of tokenizer: {tokenizer.vocab_size}')
return tokenizer
上述代码中值得注意的点:
- 1 AutoTokenizer.from_pretrained中use_fast参数,llama和internlm2不支持fast tokenizer
- 2 添加特殊字符,包括encode_token, decode_token, special_token,用于增加一些预训练词表中不存在但是微调过程中又需要的token
- 3 tokenizer.pad_token_id = tokenizer_eos_token_id
tokenizer中一般不设置pad_token_id,一般设置为 tokenizer_eos_token_id。
除此之外一般tokenizer还会设置padding的顺序,一般为left_padding.
tokenizer.padding_size = “left”
3.2 Model
照例,我们先放上载入Model的代码,然后哦中国结需要注意的点。
def load_model(args, training_args):
"""
加载模型
"""
assert training_args.bf16 or training_args.fp16, 'bf16 or fp16 should be True'
logger.info(f'Loading model from base model: {args.model_name_or_path}')
logger.info(f'Train model with {args.train_mode}')
# init model kwargs
# todo add flash attention
# attn_implementation = None
torch_dtype = torch.float16 if training_args.fp16 else torch.bfloat16
if args.train_mode == 'qlora':
quantization_config = BitsAndBytesConfig(
load_in_4bit=True,
bnb_4bit_compute_dtype=torch.float16 if training_args.fp16 else torch.bfloat16,
bnb_4bit_use_double_quant=True,
bnb_4bit_quant_type="nf4",
llm_int8_threshold=6.0,
llm_int8_has_fp16_weight=False,
)
else:
quantization_config = None
model_kwargs = dict(
trust_remote_code=True,
# attn_implementation=attn_implementation,
torch_dtype=torch_dtype,
use_cache=False if training_args.gradient_checkpointing else True,
device_map=get_kbit_device_map() if quantization_config is not None else None,
quantization_config=quantization_config,
)
model = AutoModelForCausalLM.from_pretrained(args.model_name_or_path, **model_kwargs)
# moe模型,需要考虑负载均衡的loss
if 'output_router_logits' in model.config.to_dict():
logger.info('set output_router_logits as True')
model.config.output_router_logits = True
# QLoRA: casts all the non int8 modules to full precision (fp32) for stability
if args.train_mode == 'qlora' and args.task_type in ['pretrain', 'sft']:
model = prepare_model_for_kbit_training(model, use_gradient_checkpointing=training_args.gradient_checkpointing)
# LoRA: Enables the gradients for the input embeddings
if args.train_mode == 'lora' and args.task_type in ['pretrain', 'sft']:
# For backward compatibility
if hasattr(model, "enable_input_require_grads"):
model.enable_input_require_grads()
else:
def make_inputs_require_grad(module, input, output):
output.requires_grad_(True)
model.get_input_embeddings().register_forward_hook(make_inputs_require_grad)
# init peft_config
if args.train_mode == 'full':
peft_config = None
else:
# 找到所有需要插入adapter的全连接层
target_modules = find_all_linear_names(model, args.train_mode)
peft_config = LoraConfig(
r=args.lora_rank,
lora_alpha=args.lora_alpha,
target_modules=target_modules,
lora_dropout=args.lora_dropout,
bias="none",
task_type="CAUSAL_LM",
)
# init peft model
if args.train_mode in ['lora', 'qlora'] and args.task_type in ['pretrain', 'sft']:
model = get_peft_model(model, peft_config)
# logger.info(f'memory footprint of model: {model.get_memory_footprint() / (1024 * 1024 * 1024)} GB')
model.print_trainable_parameters()
# init ref_model
if args.task_type == 'dpo':
ref_model = AutoModelForCausalLM.from_pretrained(args.model_name_or_path, **model_kwargs) if args.train_mode == 'full' else None
# pretrain和sft,不需要ref_model
else:
ref_model = None
# 计算模型参数量
total = sum(p.numel() for p in model.parameters())
logger.info("Total model params: %.2fM" % (total / 1e6))
return {
'model': model,
'ref_model': ref_model,
'peft_config': peft_config
}
上述代码注意的点:
- 1 AutoModelForCausalLM.from_pretrained中的参数分为两项,其一为载入模型的参数,其二为量化参数。
- 2 prepare_model_for_kbit_training,载入PEFT Model的准备工作
- 3 get_peft_model,载入预训练模型和LoraConfig配置文件。
AutoModelForCausalLM.from_pretrained:
model_kwargs = dict(
trust_remote_code=True,
# attn_implementation=attn_implementation,
torch_dtype=torch_dtype,
use_cache=False if training_args.gradient_checkpointing else True,
device_map=get_kbit_device_map() if quantization_config is not None else None,
quantization_config=quantization_config,
)
quantization_config = BitsAndBytesConfig(
load_in_4bit=True,
bnb_4bit_compute_dtype=torch.float16 if training_args.fp16 else torch.bfloat16,
bnb_4bit_use_double_quant=True,
bnb_4bit_quant_type="nf4",
llm_int8_threshold=6.0,
llm_int8_has_fp16_weight=False,
)
- trust_remote_code=True,当权重文件不在本地时会从远程下载,如果该项设置为False则无法从远程下载模型。
- torch_dtype模型载入的精度,由fp16和bf16可选。
- use_cache=False,如果训练时打开梯度检查点则需要关闭use_cache,这两项冲突。
- device_map,模型载入时使用的GPU
- quantization_config,load_in_4bit而且包含bnb参数为QLora,load_in_4bit=True(不包含bnb参数)和load_in_8bit=True使用为LLM.int8()量化方法
Prepare_model_for_kbit_training:
prepare_model_for_kbit_training(model, use_gradient_checkpointing=training_args.gradient_checkpointing)
use_gradient_checkpointing=True,即打开梯度检查点,该技术的原理是模型在训练过程中前向传播的激活值不进行存储,在反向传播的过程中重新计算,这极大的降低了显存消耗,虽然牺牲了一定的性能。与model.config.use_cache冲突,两者只能择一打开,个人建议打开梯度检查点。
Get_peft_model:
def find_all_linear_names(model, train_mode):
"""
找出所有全连接层,为所有全连接添加adapter
"""
assert train_mode in ['lora', 'qlora']
cls = bnb.nn.Linear4bit if train_mode == 'qlora' else nn.Linear
lora_module_names = set()
for name, module in model.named_modules():
if isinstance(module, cls):
names = name.split('.')
lora_module_names.add(names[0] if len(names) == 1 else names[-1])
if 'lm_head' in lora_module_names: # needed for 16-bit
lora_module_names.remove('lm_head')
lora_module_names = list(lora_module_names)
logger.info(f'LoRA target module names: {lora_module_names}')
return lora_module_names
peft_config = LoraConfig(
r=args.lora_rank,
lora_alpha=args.lora_alpha,
target_modules=target_modules,
lora_dropout=args.lora_dropout,
bias="none",
task_type="CAUSAL_LM",
)
model = get_peft_model(model, peft_config)
- r:LoRA的秩,秩越大则可训练的参数越多,一般来说越困难的任务的秩越大。
- lora_alpha:缩放系数,一般为r的两倍。
- target_modules:为LoRA作用的层,原始的LoRA只作用与Self-attention中的Wq、Wk、Wv、Wo的权重矩阵。此处使用find_all_linear_names,对全部的线性层使用LoRA。
- lora_dropout:正则化dropout系数
- task_type:“CASUAL_LM”
4 Dataset和DataCollator载入
上一节真是惊心动魄的一章节,我们分别了解Tokenizer和Model的载入,同时了解到一些载入过程中的小细节,能看到这里已经超过大多数人了,但是接下来才是微调过程中最重要的细节。
在大模型技术发展到如火如荼的今天,微调技术已经不是门槛,最重要的反而是高质量的数据,以及对数据的处理。同时在某种程度上预训练、微调、对齐的区别只在于对数据集的处理和微调函数的修改。
在此处我绘制了一幅图,解释了预训练、微调和对齐的区别。
预训练是一种自回归的训练方式,大模型预测token的下一个token的概率。
微调虽然也是自回归,但是会将不需要的prompt使用mask遮盖,只预测想要的答案。
对齐则需要准备chosen和reject两类数据集,其中chosen为人类想要的对齐形态。
好了经过上述的介绍,我们对数据准备有了一个大概认识了,那我们正式进入新世界的大门吧。
4.1 Dataset
Dataset分为SftDataset和DpoDataset,我们先介绍SftDataset。
4.1.1 SftDataset
作为一般流程,Dataset需要将输入中文经过tokenizer编码为token_id,同时做一定的处理,例如max_length。padding,max_length,tensor形式可以交给DataCollator处理。
首先我们需要知道SFT的数据需要为什么样子,如下图
从上看出,数据的核心在于conversation:human和assistant的多段对话。
我们上代码看看Firefly如何处理他们,此处我们关注Firefly/component/dataset.py下的UnifiedSFTDataset类中的__getitem__函数,只要是集成自torch的Dataset就关注该函数即可,这决定了怎么取数据。
def __getitem__(self, index):
# 每条数据拼接格式为: {system_format}{user_format}{assistant_format}{user_format}{assistant_format}...
data = self.data_list[index]
data = json.loads(data)
input_ids, target_mask = [], []
# setting system information
if self.system_format is not None:
system = data['system'].strip() if 'system' in data.keys() else self.system
# system信息不为空
if system is not None:
system_text = self.system_format.format(content=system)
input_ids = self.tokenizer.encode(system_text, add_special_tokens=False)
target_mask = [0] * len(input_ids)
conversations = data['conversation']
# 拼接多轮对话
for i, conv in enumerate(conversations):
human = conv['human'].strip()
assistant = conv['assistant'].strip()
human = self.user_format.format(content=human, stop_token=self.tokenizer.eos_token)
assistant = self.assistant_format.format(content=assistant, stop_token=self.tokenizer.eos_token)
input_tokens = self.tokenizer.encode(human, add_special_tokens=False)
output_tokens = self.tokenizer.encode(assistant, add_special_tokens=False)
input_ids += input_tokens + output_tokens
target_mask += [0] * len(input_tokens) + [1] * len(output_tokens)
assert len(input_ids) == len(target_mask)
# 对长度进行截断
input_ids = input_ids[:self.max_seq_length]
target_mask = target_mask[:self.max_seq_length]
attention_mask = [1] * len(input_ids)
assert len(input_ids) == len(target_mask) == len(attention_mask)
inputs = {
'input_ids': input_ids,
'attention_mask': attention_mask,
'target_mask': target_mask
}
return inputs
从代码中我们看出,input_ids由input_tokens和output_tokens组成的一个列表:
- input_tokens为human的问题
- output_tokens为assistant的回复
从代码中我们看出,target_mask由[0]*len(input_tokens)和[1]*len(output_tokens)组成
- human的问题不需要预测
- assistant的回复需要预测
这也很符合上图中微调的定义,即将不需要预测的token使用mask进行遮盖,只保留需要预测的token来计算loss。
4.2 DataCollator
经过Dataset的编码后,我们获得了数据的tokens的列表和mask列表,Collator需要做的是为输入数据添加padding_id,最大长度截断和label处理这三大功能。
我们直接上代码,代码在Firefly/component/collator.py下的SFTDataCollator类中。
def __call__(self, batch: List[Dict[str, Any]]) -> Dict[str, Any]:
# 找出batch中的最大长度
lengths = [len(x['input_ids']) for x in batch if x['input_ids'] is not None]
# 取出batch中的最大长度,如果超过max_seq_length,则取max_seq_length
batch_max_len = min(max(lengths), self.max_seq_length)
# batch_max_len = self.max_seq_length
input_ids_batch, attention_mask_batch, target_mask_batch = [], [], []
# truncate and padding
for x in batch:
input_ids = x['input_ids']
attention_mask = x['attention_mask']
target_mask = x['target_mask']
if input_ids is None:
logger.info('some input_ids is None')
continue
padding_len = batch_max_len - len(input_ids)
# padding
input_ids = input_ids + [self.pad_token_id] * padding_len
attention_mask = attention_mask + [0] * padding_len
target_mask = target_mask + [0] * padding_len
# truncate
input_ids = input_ids[:self.max_seq_length]
attention_mask = attention_mask[:self.max_seq_length]
target_mask = target_mask[:self.max_seq_length]
input_ids_batch.append(input_ids)
attention_mask_batch.append(attention_mask)
target_mask_batch.append(target_mask)
# 将list转换为tensor,得到最终的的模型输入
input_ids_batch = torch.tensor(input_ids_batch, dtype=torch.long)
attention_mask_batch = torch.tensor(attention_mask_batch, dtype=torch.long)
target_mask_batch = torch.tensor(target_mask_batch, dtype=torch.long)
labels = torch.where(target_mask_batch == 1, input_ids_batch, -100)
inputs = {
'input_ids': input_ids_batch,
'attention_mask': attention_mask_batch,
'labels': labels
}
return inputs
padding、max_length_trucate、label、list_to_tensor四大功能。
5 Trainer
到此处的我猜全都是勇士,好了,其实到此处已经快结束了,剩下的就是调用transformer库的Trainer类,将上述导入的配置、tokenizer、model、dataset和datacollator都加载进去就完成啦。
trainer = Trainer(
model=model,
args=training_args,
train_dataset=train_dataset,
tokenizer=tokenizer,
data_collator=data_collator,
)
train_result = trainer.train()
final_save_path = join(training_args.output_dir)
trainer.save_model(final_save_path) # Saves the tokenizer too
到这里就结束了,怎么样,是不是非常的符合规范,有没有一种戛然而止的感觉。
既然了解了项目的基本细节,那么我们一定要来一个刺激的实操环节了,来释放一下无处燃烧的激动了,此处可以从笔者之前的博客中找到答案,祝好。
不错不错