LoRA 微调实战 2026：从数据准备到部署的完整流程

数据准备 → 格式转换 → 训练配置 → LoRA训练 → 评估 → 合并 → 部署

# 推荐格式：ShareGPT / OpenAI Messages
{
  "messages": [
    {"role": "system", "content": "你是一个专业的技术顾问。"},
    {"role": "user", "content": "解释一下 RAG 的工作原理"},
    {"role": "assistant", "content": "RAG（检索增强生成）是一种..."}
  ]
}

import json
from pathlib import Path

class SFTDataBuilder:
    def __init__(self, output_dir: str):
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(parents=True, exist_ok=True)
    
    def build_from_qa_pairs(self, qa_pairs: list, system_prompt: str):
        """从问答对构建训练数据"""
        samples = []
        for qa in qa_pairs:
            sample = {
                "messages": [
                    {"role": "system", "content": system_prompt},
                    {"role": "user", "content": qa["question"]},
                    {"role": "assistant", "content": qa["answer"]}
                ]
            }
            samples.append(sample)
        
        # 划分训练/验证集
        split = int(len(samples) * 0.95)
        
        train_path = self.output_dir / "train.jsonl"
        val_path = self.output_dir / "val.jsonl"
        
        with open(train_path, 'w', encoding='utf-8') as f:
            for s in samples[:split]:
                f.write(json.dumps(s, ensure_ascii=False) + '\n')
        
        with open(val_path, 'w', encoding='utf-8') as f:
            for s in samples[split:]:
                f.write(json.dumps(s, ensure_ascii=False) + '\n')
        
        print(f"训练集: {split} 条 → {train_path}")
        print(f"验证集: {len(samples) - split} 条 → {val_path}")
    
    def build_from_conversations(self, conversations: list):
        """从多轮对话构建训练数据"""
        samples = []
        for conv in conversations:
            messages = []
            for turn in conv:
                messages.append({"role": turn["role"], "content": turn["content"]})
            samples.append({"messages": messages})
        
        return samples

class DataQualityChecker:
    def check(self, data_path: str):
        issues = []
        
        with open(data_path, 'r', encoding='utf-8') as f:
            lines = f.readlines()
        
        for i, line in enumerate(lines):
            sample = json.loads(line)
            
            # 1. 检查消息格式
            if "messages" not in sample:
                issues.append(f"Line {i}: 缺少 messages 字段")
                continue
            
            # 2. 检查角色顺序
            roles = [m["role"] for m in sample["messages"]]
            if roles[-1] != "assistant":
                issues.append(f"Line {i}: 最后一条消息不是 assistant")
            
            # 3. 检查内容长度
            for msg in sample["messages"]:
                if len(msg["content"]) < 5:
                    issues.append(f"Line {i}: 消息内容过短")
                if len(msg["content"]) > 8000:
                    issues.append(f"Line {i}: 消息内容过长 ({len(msg['content'])} chars)")
            
            # 4. 检查 assistant 回复质量
            assistant_msgs = [m for m in sample["messages"] if m["role"] == "assistant"]
            for msg in assistant_msgs:
                if msg["content"].startswith("我是一个AI"):
                    issues.append(f"Line {i}: assistant 回复包含模板化语言")
                if len(msg["content"]) < 20:
                    issues.append(f"Line {i}: assistant 回复过短")
        
        # 5. 统计
        stats = {
            "total_samples": len(lines),
            "avg_turns": np.mean([len(json.loads(l)["messages"]) for l in lines]),
            "avg_assistant_len": np.mean([
                len(m["content"]) 
                for l in lines 
                for m in json.loads(l)["messages"] 
                if m["role"] == "assistant"
            ]),
            "issues_found": len(issues),
        }
        
        return {"issues": issues[:20], "stats": stats}

# train_lora.py
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer, TrainingArguments
from peft import LoraConfig, get_peft_model, TaskType
from trl import SFTTrainer, SFTConfig

# 1. 加载模型和分词器
model_name = "Qwen/Qwen2.5-7B-Instruct"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    torch_dtype=torch.bfloat16,
    device_map="auto",
    attn_implementation="flash_attention_2"
)

# 2. LoRA 配置
lora_config = LoraConfig(
    task_type=TaskType.CAUSAL_LM,
    r=64,                    # 秩，常用 8/16/32/64
    lora_alpha=128,          # alpha = 2 * r 是常见默认值
    lora_dropout=0.05,
    target_modules=[
        "q_proj",
        "k_proj", 
        "v_proj",
        "o_proj",
        "gate_proj",
        "up_proj",
        "down_proj"
    ],
    bias="none"
)

# 3. 应用 LoRA
model = get_peft_model(model, lora_config)
model.print_trainable_parameters()
# 输出: trainable params: 39,976,960 || all params: 7,621,836,800 || trainable%: 0.5247%

# 4. 训练配置
training_args = SFTConfig(
    output_dir="./output/qwen2.5-7b-lora",
    num_train_epochs=3,
    per_device_train_batch_size=4,
    gradient_accumulation_steps=4,   # 有效 batch_size = 16
    learning_rate=2e-4,
    lr_scheduler_type="cosine",
    warmup_ratio=0.05,
    bf16=True,
    logging_steps=10,
    save_strategy="steps",
    save_steps=200,
    save_total_limit=3,
    eval_strategy="steps",
    eval_steps=200,
    load_best_model_at_end=True,
    metric_for_best_model="eval_loss",
    greater_is_better=False,
    gradient_checkpointing=True,
    max_seq_length=2048,
    dataset_text_field="messages",
)

# 5. 加载数据
from datasets import load_dataset
dataset = load_dataset("json", data_files={
    "train": "data/train.jsonl",
    "validation": "data/val.jsonl"
})

# 6. 启动训练
trainer = SFTTrainer(
    model=model,
    args=training_args,
    train_dataset=dataset["train"],
    eval_dataset=dataset["validation"],
    processing_class=tokenizer,
)

trainer.train()

from transformers import BitsAndBytesConfig

# 4bit 量化配置
bnb_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",
    bnb_4bit_compute_dtype=torch.bfloat16,
    bnb_4bit_use_double_quant=True
)

# 用量化方式加载
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    quantization_config=bnb_config,
    device_map="auto"
)

# 后续 LoRA 配置和训练步骤相同

class LoRAEvaluator:
    def __init__(self, model, tokenizer):
        self.model = model
        self.tokenizer = tokenizer
    
    def evaluate(self, eval_dataset):
        results = {
            "loss": [],
            "bleu": [],
            "rouge_l": [],
            "accuracy": []
        }
        
        for sample in eval_dataset:
            # 生成
            inputs = self.tokenizer.apply_chat_template(
                sample["messages"][:-1],
                return_tensors="pt",
                add_generation_prompt=True
            ).to(self.model.device)
            
            with torch.no_grad():
                output = self.model.generate(
                    inputs,
                    max_new_tokens=512,
                    temperature=0.1,
                    do_sample=False
                )
            
            pred = self.tokenizer.decode(output[0][inputs.shape[1]:], skip_special_tokens=True)
            gt = sample["messages"][-1]["content"]
            
            # 计算指标
            results["bleu"].append(bleu_score(pred, gt))
            results["rouge_l"].append(rouge_l_score(pred, gt))
            results["accuracy"].append(self._check_accuracy(pred, gt))
        
        return {k: np.mean(v) for k, v in results.items()}

# 合并 LoRA 权重到基础模型
def merge_and_save(model, output_path: str):
    merged_model = model.merge_and_unload()
    merged_model.save_pretrained(output_path)
    tokenizer.save_pretrained(output_path)
    print(f"合并模型已保存到 {output_path}")

# 导出为 GGUF 格式（用于 llama.cpp / Ollama 部署）
def export_gguf(model_path: str, output_path: str):
    """
    使用 llama.cpp 的 convert 脚本转为 GGUF 格式
    """
    import subprocess
    subprocess.run([
        "python", "llama.cpp/convert_hf_to_gguf.py",
        model_path,
        "--outfile", output_path,
        "--outtype", "f16"  # 或 q4_k_m 量化
    ])

# 部署到 vLLM
def deploy_vllm(model_path: str, port: int = 8000):
    """
    用 vLLM 启动推理服务
    """
    import subprocess
    subprocess.run([
        "python", "-m", "vllm.entrypoints.openai.api_server",
        "--model", model_path,
        "--port", str(port),
        "--tensor-parallel-size", "1",
        "--gpu-memory-utilization", "0.9"
    ])