用LoRA在单卡A100上微调Qwen2.5-7B：从数据构造到loss曲线与推理对比的完整记录

{
  "instruction": "液压站蓄能器压力从12MPa掉到8MPa，可能是什么原因？",
  "input": "",
  "output": "【现象判断】蓄能器保压能力下降...\n【可能原因】1. 皮囊破损...\n【处置建议】..."
}

import json
import hashlib

def build_dataset(raw_path, out_path, min_len=20, max_len=1800):
    seen = set()
    kept = []
    with open(raw_path, encoding="utf-8") as f:
        for line in f:
            item = json.loads(line)
            text = item["instruction"] + item["output"]
            # 用 md5 去重，工单数据重复率很高
            h = hashlib.md5(text.encode()).hexdigest()
            if h in seen:
                continue
            seen.add(h)
            if not (min_len <= len(item["output"]) <= max_len):
                continue
            # 强制校验三段式格式，格式不对的直接丢弃而不是让模型学坏
            if not all(k in item["output"] for k in ["【现象判断】", "【可能原因】", "【处置建议】"]):
                continue
            kept.append(item)
    with open(out_path, "w", encoding="utf-8") as f:
        for item in kept:
            f.write(json.dumps(item, ensure_ascii=False) + "\n")
    print(f"raw={len(seen)}, kept={len(kept)}")

build_dataset("raw.jsonl", "train_clean.jsonl")

import torch
from datasets import load_dataset
from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig
from peft import LoraConfig, get_peft_model, prepare_model_for_kbit_training
from trl import SFTTrainer, SFTConfig

MODEL_PATH = "Qwen/Qwen2.5-7B-Instruct"

# QLoRA 用 4bit，纯 LoRA 把这行换成 None 即可
bnb_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",
    bnb_4bit_compute_dtype=torch.bfloat16,
    bnb_4bit_use_double_quant=True,
)

tokenizer = AutoTokenizer.from_pretrained(MODEL_PATH, trust_remote_code=True)
tokenizer.pad_token = tokenizer.eos_token

model = AutoModelForCausalLM.from_pretrained(
    MODEL_PATH,
    quantization_config=bnb_config,
    device_map="auto",
    torch_dtype=torch.bfloat16,
    trust_remote_code=True,
)
model = prepare_model_for_kbit_training(model)

lora_config = LoraConfig(
    r=16,
    lora_alpha=32,
    lora_dropout=0.05,
    target_modules=["q_proj", "k_proj", "v_proj", "o_proj",
                    "gate_proj", "up_proj", "down_proj"],
    bias="none",
    task_type="CAUSAL_LM",
)
model = get_peft_model(model, lora_config)
model.print_trainable_parameters()
# 输出: trainable params: 40,370,176 || all params: 7,655,986,688 || trainable%: 0.527

def format_sample(example):
    messages = [
        {"role": "system", "content": "你是工业设备运维专家，回答必须按【现象判断】【可能原因】【处置建议】三段式输出。"},
        {"role": "user", "content": example["instruction"]},
        {"role": "assistant", "content": example["output"]},
    ]
    return {"text": tokenizer.apply_chat_template(messages, tokenize=False)}

train_ds = load_dataset("json", data_files="train_clean.jsonl", split="train")
train_ds = train_ds.map(format_sample, remove_columns=train_ds.column_names)

args = SFTConfig(
    output_dir="./qwen2.5-7b-lora",
    per_device_train_batch_size=4,
    gradient_accumulation_steps=8,
    learning_rate=2e-4,
    num_train_epochs=3,
    lr_scheduler_type="cosine",
    warmup_ratio=0.03,
    logging_steps=10,
    save_strategy="epoch",
    bf16=True,
    optim="paged_adamw_8bit",
    max_seq_length=2048,
    dataset_text_field="text",
    report_to="none",
)

trainer = SFTTrainer(
    model=model,
    args=args,
    train_dataset=train_ds,
    tokenizer=tokenizer,
)
trainer.train()
trainer.save_model("./qwen2.5-7b-lora/final")

from peft import PeftModel

base = AutoModelForCausalLM.from_pretrained(
    MODEL_PATH, torch_dtype=torch.bfloat16, device_map="auto"
)
model = PeftModel.from_pretrained(base, "./qwen2.5-7b-lora/final")
model = model.merge_and_unload()  # 合并后推理速度提升约15%
model.eval()

def generate(prompt, max_new_tokens=512):
    messages = [
        {"role": "system", "content": "你是工业设备运维专家，回答必须按【现象判断】【可能原因】【处置建议】三段式输出。"},
        {"role": "user", "content": prompt},
    ]
    text = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
    inputs = tokenizer(text, return_tensors="pt").to(model.device)
    with torch.no_grad():
        out = model.generate(**inputs, max_new_tokens=max_new_tokens,
                             do_sample=False, temperature=1.0,
                             repetition_penalty=1.05)
    return tokenizer.decode(out[0][inputs["input_ids"].shape[1]:], skip_special_tokens=True)