利用 transformers LoRA 微调模型

什么是 transformers

🤗 Transformers 提供了便于快速下载和使用的 API，让你可以把预训练模型用在给定文本、在你的数据集上微调然后通过 model hub 与社区共享。同时，每个定义的 Python 模块都是完全独立的，便于修改和快速进行研究实验。

https://github.com/huggingface/transformers

简单说就是利用这个库可以快速简单帮我们微调模型，很多东西都封装好了可以直接拿来用。

准备

租一个服务器，我尝试的是 4090
选择一个支持 transformers 容器镜像环境
如果没有，也可以选择自己安装 https://github.com/huggingface/transformers/blob/main/i18n/README_zh-hans.md#%E5%AE%89%E8%A3%85

步骤

下载基模，使用 hf-mirror 会快，或者镜像环境如果支持魔法可以直接魔法下

$ cd /tmp
$ git lfs install
$ git clone https://hf-mirror.com/Qwen/Qwen2.5-7B-Instruct
$ cd Qwen2.5-7B-Instruct

微调代码

数据集处理 data_preprocess.py

这个部分需要根据你实际的具体数据格式进行处理，其中的关键就是读取已有的数据格式，将已有的数据格式转换为训练所需的数据集格式而已。

import json
from torch.utils.data import Dataset

# 定义一个名为 MyDataset 的新类，它继承自 Dataset 类
class MyDataset(Dataset):
    # 定义类的初始化方法
    def __init__(self, data, tokenizer, args):
        super(MyDataset, self).__init__()
        self.data = data
        self.tokenizer = tokenizer
        self.prompt_column = args.prompt_column
        self.response_column = args.response_column
        self.max_source_length = args.max_source_length
        self.max_target_length = args.max_target_length

    # 返回数据集的长度
    def __len__(self):
        return len(self.data)

    # 通过索引访问数据集中的样本
    def __getitem__(self, i):
        item = self.data[i]

        # 获取消息列表
        messages = item["messages"]

        # 分离系统消息、用户消息和助手消息
        system_message = messages[0]  # 系统消息
        user_message = messages[1]    # 用户消息
        assistant_message = messages[2]  # 助手消息

        # 构建提示文本（系统消息 + 用户消息）
        prompt = build_prompt(system_message, user_message)

        # 构建响应文本（助手消息）
        response = build_response(assistant_message)

        # 对提示文本进行编码
        context = self.tokenizer(
            prompt,
            max_length=self.max_source_length,
            add_special_tokens=False)

        # 对响应文本进行编码
        response_encoding = self.tokenizer(
            response,
            max_length=self.max_target_length,
            add_special_tokens=False)

        # 将上下文和响应的 input_ids 连接起来
        input_ids = context["input_ids"] + response_encoding["input_ids"]

        # 将上下文和响应的注意力掩码连接起来
        attention_mask = context["attention_mask"] + response_encoding["attention_mask"]

        # 创建标签数组，标记上下文部分为 -100，响应部分使用真实的 input_ids
        labels = [-100] * len(context["input_ids"]) + response_encoding["input_ids"]

        # 确保输入 ID 和标签的长度一致
        assert len(input_ids) == len(labels), f"length mismatch: {len(input_ids)} vs {len(labels)}"

        # 返回编码后的数据
        return {
            "input_ids": input_ids,
            "attention_mask": attention_mask,
            "labels": labels
        }

# 构建提示文本
def build_prompt(system_message, user_message):
    # 获取系统消息内容和工具定义
    system_content = system_message["content"]
    tools = system_message.get("tools", [])

    # 获取用户消息内容
    user_content = user_message["content"]

    # 构建提示文本
    prompt = f"<|im_start|>system\n{system_content}"

    # 添加工具定义（如果有）
    if tools:
        tool_str = json.dumps(tools, ensure_ascii=False)
        prompt += f"\n\nTools: {tool_str}"

    # 添加用户消息
    prompt += f"\n<|im_end|>\n<|im_start|>user\n{user_content}\n<|im_end|>\n<|im_start|>assistant\n"

    return prompt

# 构建响应文本
def build_response(assistant_message):
    # 获取助手消息内容
    assistant_content = assistant_message["content"]

    # 构建响应文本
    response = f"{assistant_content}\n<|im_end|>"

    return response

看代码容易迷糊，直接看数据案例，首先是提示文本大概的样子，我们需要将原始的数据集转换为对话模型训练的格式，所以原始格式不重要，只要你最终能整成这样都可以。

<|im_start|>system
You are a helpful assistant.
<|im_end|>
<|im_start|>user
What is the capital of France?
<|im_end|>
<|im_start|>assistant
The capital of France is Paris.
<|im_end|>

然后是最终 tokenizer 后的样子

{
  'input_ids': tensor([ 101, 7429, 2105, 2019, 17341,  999, 2005, 3185, 4201, 14926, 2000, 7976, 22124,  102......]),
  'attention_mask': tensor([1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, ....]),
  'labels': tensor(1)
}

微调 finetune.py

可以参考

网上还有很多，就不一一举例，直接让 GPT 写一个就好了，因为 transformers 都封装好了，其中最关键的还是在参数的设置上

# 导入所需模块和库
import json
import torch
from transformers import (
    AutoTokenizer,
    AutoModelForCausalLM,
    BitsAndBytesConfig,
    DataCollatorForSeq2Seq,
    HfArgumentParser,
    TrainingArguments,
    Trainer
)
from peft import LoraConfig, TaskType, get_peft_model
from arguments import ModelArguments, DataTrainingArguments, PeftArguments
from data_preprocess import MyDataset

def main():
    # 解析命令行参数
    parser = HfArgumentParser((ModelArguments, DataTrainingArguments, PeftArguments, TrainingArguments))
    model_args, data_args, peft_args, training_args = parser.parse_args_into_dataclasses()

    # 加载预训练模型和分词器
    model = AutoModelForCausalLM.from_pretrained(model_args.model_name_or_path, torch_dtype=torch.bfloat16)
    tokenizer = AutoTokenizer.from_pretrained(model_args.model_name_or_path)

    # 设置LoRA配置
    lora_config = LoraConfig(
        inference_mode=False,
        task_type=TaskType.CAUSAL_LM,
        target_modules=["q_proj", "k_proj", "v_proj"],
        r=peft_args.lora_rank,
        lora_alpha=peft_args.lora_alpha,
        lora_dropout=peft_args.lora_dropout
    )
    # 应用LoRA配置到模型
    model = get_peft_model(model, lora_config).to("cuda")
    # 输出可训练参数数量
    model.print_trainable_parameters()

    # 设置数据规整器
    data_collator = DataCollatorForSeq2Seq(
        tokenizer=tokenizer,
        padding=True
    )

    # 读取训练数据
    if training_args.do_train:
        with open(data_args.train_file, "r", encoding="utf-8") as f:
            train_data = [json.loads(line) for line in f]
        # Dataset 处理数据
        train_dataset = Dataset(train_data, tokenizer, data_args)

    # 实例化Trainer
    trainer = Trainer(
        model=model,
        tokenizer=tokenizer,
        data_collator=data_collator,
        args=training_args,
        train_dataset=train_dataset if training_args.do_train else None,
        eval_dataset=eval_dataset if training_args.do_eval else None,
    )

    # 训练模型
    if training_args.do_train:
        model.gradient_checkpointing_enable()
        model.enable_input_require_grads()
        trainer.train()

if __name__ == "__main__":
    main()

启动脚本 train.sh

#! /usr/bin/env bash

set -ex

LR=2e-4

DATESTR=`date +%Y%m%d-%H%M%S`
RUN_NAME=ft_linkinstar_qwen2
OUTPUT_DIR=output/${RUN_NAME}-${DATESTR}
mkdir -p $OUTPUT_DIR

# 修改你的模型路径
MODEL_PATH="/tmp/Qwen2.5-7B-Instruct"

CUDA_VISIBLE_DEVICES=0 python finetune.py \
    --do_train \
    --train_file ../data/train.jsonl \
    --model_name_or_path "${MODEL_PATH}" \
    --output_dir $OUTPUT_DIR \
    --max_source_length 2048 \
    --max_target_length 1024 \
    --per_device_train_batch_size 1 \
    --per_device_eval_batch_size 1 \
    --gradient_accumulation_steps 4 \
    --evaluation_strategy steps \
    --eval_steps 300 \
    --num_train_epochs 3 \
    --logging_steps 30 \
    --logging_dir $OUTPUT_DIR/logs \
    --save_steps 200 \
    --learning_rate $LR \
    --lora_rank 8 \
    --lora_alpha 32 \
    --lora_dropout 0.1 2>&1 | tee ${OUTPUT_DIR}/train.log

其中如果你租的 GPU 多可以调整 per_device_train_batch_size 和 per_device_eval_batch_size 更大 2、4…
然后其中 LR 是学习率
response_column 和 prompt_column 是用来指定字段用的，不用看，因为没用到，我们直接处理数据内容为对应格式就好
gradient_accumulation_steps 如果 GPU 内存够 48+ 可以减少

使用

import argparse
from peft import PeftModel
from transformers import AutoTokenizer, AutoModelForCausalLM

def load_model(model_path, checkpoint_path):
    tokenizer = AutoTokenizer.from_pretrained(model_path)
    model = AutoModelForCausalLM.from_pretrained(model_path, torch_dtype=torch.bfloat16)
    model = PeftModel.from_pretrained(model, model_id=checkpoint_path).to("cuda").eval()
    return tokenizer, model

# 初始化全局变量
parser = argparse.ArgumentParser()
parser.add_argument("--model", type=str, default=None, required=True, help="main model weights")
parser.add_argument("--ckpt", type=str, default=None, required=True, help="The checkpoint path")
args = parser.parse_args()

# 加载模型
tokenizer, model = load_model(args.model, args.ckpt)

parser = argparse.ArgumentParser()
parser.add_argument("--model", type=str, default=None, required=True, help="main model weights")
parser.add_argument("--ckpt", type=str, default=None, required=True, help="The checkpoint path")
args = parser.parse_args()

tokenizer, model = load_model(args.model, args.ckpt)

# 获取模型生成的回复
def get_completion(prompt):
    inputs = tokenizer([prompt], return_tensors="pt").to("cuda")
    with torch.no_grad():
        outputs = model.generate(**inputs, max_new_tokens=1024)
        response = tokenizer.decode(outputs[:,inputs['input_ids'].shape[1]:][0], skip_special_tokens=True)
    return response