实战 | Qwen3大模型微调入门实战（完整代码）

pip install swanlab modelscope==1.22.0 "transformers>=4.50.0" datasets==3.2.0 accelerate pandas addict

{“question”: “我父亲刚刚被诊断为活动性出血，医生说需要立即处理，我们该怎么做？”,“think”: “嗯，用户的问题是关于病人出现活动性出血时应采取哪些一般处理措施，...”,“answer”: “首先，您父亲需要卧床休息，活动性出血期间暂时不要进食。为了...”,}

<think>嗯，用户的问题是关于病人出现活动性出血时应采取哪些一般处理措施，...</think>
首先，您父亲需要卧床休息，活动性出血期间暂时不要进食。为了...

from modelscope.msdatasets import MsDatasetimport jsonimport random
random.seed(42)
ds = MsDataset.load('krisfu/delicate_medical_r1_data', subset_name='default', split='train')data_list = list(ds)random.shuffle(data_list)
split_idx = int(len(data_list) * 0.9)
train_data = data_list[:split_idx]val_data = data_list[split_idx:]
with open('train.jsonl', 'w', encoding='utf-8') as f:    for item in train_data:        json.dump(item, f, ensure_ascii=False)        f.write('\n')
with open('val.jsonl', 'w', encoding='utf-8') as f:    for item in val_data:        json.dump(item, f, ensure_ascii=False)        f.write('\n')
print(f"The dataset has been split successfully.")print(f"Train Set Size：{len(train_data)}")print(f"Val Set Size：{len(val_data)}")

from modelscope import snapshot_download, AutoTokenizerfrom transformers import AutoModelForCausalLM, TrainingArguments, Trainer, DataCollatorForSeq2Seq
# 在modelscope上下载Qwen模型到本地目录下model_dir = snapshot_download("Qwen/Qwen3-1.7B", cache_dir="./", revision="master")
# Transformers加载模型权重tokenizer = AutoTokenizer.from_pretrained("./Qwen/Qwen3-1.7B", use_fast=False, trust_remote_code=True)model = AutoModelForCausalLM.from_pretrained("./Qwen/Qwen3-1.7B", device_map="auto", torch_dtype=torch.bfloat16)

import jsonimport pandas as pdimport torchfrom datasets import Datasetfrom modelscope import snapshot_download, AutoTokenizerfrom transformers import AutoModelForCausalLM, TrainingArguments, Trainer, DataCollatorForSeq2Seqimport osimport swanlab
os.environ["SWANLAB_PROJECT"]="qwen3-sft-medical"PROMPT = "你是一个医学专家，你需要根据用户的问题，给出带有思考的回答。"MAX_LENGTH = 2048swanlab.config.update({    "model": "Qwen/Qwen3-1.7B",    "prompt": PROMPT,    "data_max_length": MAX_LENGTH,    })
def dataset_jsonl_transfer(origin_path, new_path):    """    将原始数据集转换为大模型微调所需数据格式的新数据集    """    messages = []    # 读取旧的JSONL文件    with open(origin_path, "r") as file:        for line in file:            # 解析每一行的json数据            data = json.loads(line)            input = data["question"]            output = f"<think>{data["think"]}</think> \n {data["answer"]}"            message = {                "instruction": PROMPT,                "input": f"{input}",                "output": output,            }            messages.append(message)    # 保存重构后的JSONL文件    with open(new_path, "w", encoding="utf-8") as file:        for message in messages:            file.write(json.dumps(message, ensure_ascii=False) + "\n")
def process_func(example):    """    将数据集进行预处理    """     input_ids, attention_mask, labels = [], [], []    instruction = tokenizer(        f"<|im_start|>system\n{PROMPT}<|im_end|>\n<|im_start|>user\n{example['input']}<|im_end|>\n<|im_start|>assistant\n",        add_special_tokens=False,    )    response = tokenizer(f"{example['output']}", add_special_tokens=False)    input_ids = instruction["input_ids"] + response["input_ids"] + [tokenizer.pad_token_id]    attention_mask = (        instruction["attention_mask"] + response["attention_mask"] + [1]    )    labels = [-100] * len(instruction["input_ids"]) + response["input_ids"] + [tokenizer.pad_token_id]    if len(input_ids) > MAX_LENGTH:  # 做一个截断        input_ids = input_ids[:MAX_LENGTH]        attention_mask = attention_mask[:MAX_LENGTH]        labels = labels[:MAX_LENGTH]    return {"input_ids": input_ids, "attention_mask": attention_mask, "labels": labels}   
def predict(messages, model, tokenizer):    device = "cuda"    text = tokenizer.apply_chat_template(        messages,        tokenize=False,        add_generation_prompt=True    )    model_inputs = tokenizer([text], return_tensors="pt").to(device)    generated_ids = model.generate(        model_inputs.input_ids,        max_new_tokens=MAX_LENGTH,    )    generated_ids = [        output_ids[len(input_ids):] for input_ids, output_ids in zip(model_inputs.input_ids, generated_ids)    ]    response = tokenizer.batch_decode(generated_ids, skip_special_tokens=True)[0]    return response
# 在modelscope上下载Qwen模型到本地目录下model_dir = snapshot_download("Qwen/Qwen3-1.7B", cache_dir="/root/autodl-tmp/", revision="master")
# Transformers加载模型权重tokenizer = AutoTokenizer.from_pretrained("/root/autodl-tmp/Qwen/Qwen3-1.7B", use_fast=False, trust_remote_code=True)model = AutoModelForCausalLM.from_pretrained("/root/autodl-tmp/Qwen/Qwen3-1.7B", device_map="auto", torch_dtype=torch.bfloat16)model.enable_input_require_grads()  # 开启梯度检查点时，要执行该方法
# 加载、处理数据集和测试集train_dataset_path = "train.jsonl"test_dataset_path = "val.jsonl"train_jsonl_new_path = "train_format.jsonl"test_jsonl_new_path = "val_format.jsonl"
if not os.path.exists(train_jsonl_new_path):    dataset_jsonl_transfer(train_dataset_path, train_jsonl_new_path)if not os.path.exists(test_jsonl_new_path):    dataset_jsonl_transfer(test_dataset_path, test_jsonl_new_path)
# 得到训练集train_df = pd.read_json(train_jsonl_new_path, lines=True)train_ds = Dataset.from_pandas(train_df)train_dataset = train_ds.map(process_func, remove_columns=train_ds.column_names)
# 得到验证集eval_df = pd.read_json(test_jsonl_new_path, lines=True)eval_ds = Dataset.from_pandas(eval_df)eval_dataset = eval_ds.map(process_func, remove_columns=eval_ds.column_names)
args = TrainingArguments(    output_dir="/root/autodl-tmp/output/Qwen3-1.7B",    per_device_train_batch_size=1,    per_device_eval_batch_size=1,    gradient_accumulation_steps=4,    eval_strategy="steps",    eval_steps=100,    logging_steps=10,    num_train_epochs=2,    save_steps=400,    learning_rate=1e-4,    save_on_each_node=True,    gradient_checkpointing=True,    report_to="swanlab",    run_name="qwen3-1.7B",)
trainer = Trainer(    model=model,    args=args,    train_dataset=train_dataset,    eval_dataset=eval_dataset,    data_collator=DataCollatorForSeq2Seq(tokenizer=tokenizer, padding=True),)
trainer.train()
# 用测试集的前3条，主观看模型test_df = pd.read_json(test_jsonl_new_path, lines=True)[:3]test_text_list = []for index, row in test_df.iterrows():    instruction = row['instruction']    input_value = row['input']    messages = [        {"role": "system", "content": f"{instruction}"},        {"role": "user", "content": f"{input_value}"}    ]    response = predict(messages, model, tokenizer)    response_text = f"""    Question: {input_value}    LLM:{response}    """
    test_text_list.append(swanlab.Text(response_text))    print(response_text)
swanlab.log({"Prediction": test_text_list})
swanlab.finish()

import torchfrom transformers import AutoModelForCausalLM, AutoTokenizerdef predict(messages, model, tokenizer):    device = "cuda"    text = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)    model_inputs = tokenizer([text], return_tensors="pt").to(device)    generated_ids = model.generate(model_inputs.input_ids, max_new_tokens=2048)    generated_ids = [output_ids[len(input_ids):] for input_ids, output_ids in zip(model_inputs.input_ids, generated_ids)]    response = tokenizer.batch_decode(generated_ids, skip_special_tokens=True)[0]    return response# 加载原下载路径的tokenizer和modeltokenizer = AutoTokenizer.from_pretrained("./output/Qwen3-1.7B/checkpoint-1000", use_fast=False, trust_remote_code=True)model = AutoModelForCausalLM.from_pretrained("./output/Qwen3-1.7B/checkpoint-1000", device_map="auto", torch_dtype=torch.bfloat16)test_texts = {    'instruction': "你是一个医学专家，你需要根据用户的问题，给出带有思考的回答。",    'input': "医生，我最近被诊断为糖尿病，听说碳水化合物的选择很重要，我应该选择什么样的碳水化合物呢？"}instruction = test_texts['instruction']input_value = test_texts['input']messages = [    {"role": "system", "content": f"{instruction}"},    {"role": "user", "content": f"{input_value}"}]response = predict(messages, model, tokenizer)print(response)