AI大模型微调，从原理剖析到企业级落地实战

┌─────────────────────────────────────────────────────────┐
│                    三层训练架构                         │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  Layer 1: 预训练（Pre-training）                       │
│  数据：TB 级互联网公开文本                             │
│  目标：学习语言规律 + 世界知识                         │
│  成本：数千万美元，不可复制                           │
│  ↓                                                     │
│  Layer 2: 有监督微调（SFT - Supervised Fine-Tuning）  │
│  数据：数万条高质量对话数据                           │
│  目标：学习"对话格式"和"指令遵循"                     │
│  成本：数万美元，开源模型可做                         │
│  ↓                                                     │
│  Layer 3: 强化学习对齐（RLHF / DPO）                  │
│  数据：偏好排序数据                                   │
│  目标：符合人类价值观和偏好                           │
│  成本：数万美元，进阶选择                             │
│                                                         │
└─────────────────────────────────────────────────────────┘

预训练模型: P(y|x)   →  给定输入，生成输出的概率分布
微调后模型: P(y|x; θ_finetuned) 
           ↑
      调整参数 θ，使模型在"你的数据"上损失最小

原始权重 W (冻结，不训练)          低秩分解
    ████████          =     ████    ×    ████
    ████████               ████         ████
    ████████          +    ████    ×    ████  ← 只训练这两个小矩阵
    ████████               ████         ████
    (d × d)                 (d × r)   (r × d)
                               ↑
                          r << d (比如 r=8)

[
  {
    "conversations": [
      {"from": "human", "value": "什么是 LoRA？"},
      {"from": "gpt", "value": "LoRA 是一种参数高效微调方法..."},
      {"from": "human", "value": "它有什么优点？"},
      {"from": "gpt", "value": "主要有三个优点：1. 显存占用低..."}
    ]
  },
  {
    "conversations": [
      {"from": "human", "value": "帮我写一个 Python 函数，计算斐波那契数列"},
      {"from": "gpt", "value": "```python\ndef fibonacci(n):\n    ...\n```"}
    ]
  }
]

# 标注模板示例
标注指南：
1. 角色：你是一个专业的[领域]专家
2. 风格：[正式/亲切/幽默]
3. 约束：回答必须包含[具体要求]
4. 格式：如涉及代码，使用```包裹

# 标注工具推荐
- LabelStudio（开源免费）
- Doccano（轻量级）
- 飞桨数据标注平台

import openai

def generate_training_data(seed_questions):
    """用大模型生成高质量的训练数据"""
    system_prompt = """
    你是一个数据生成专家。请根据给定的问题，生成3个不同角度、不同问法的变体，
    并给出高质量的回答。回答要详细、准确、格式规范。
    """
    
    dataset = []
    for q in seed_questions:
        response = openai.ChatCompletion.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": f"原始问题：{q}"}
            ]
        )
        # 解析并添加到数据集
        dataset.append(parse_response(response))
    return dataset

def clean_chat_logs(raw_logs):
    """从真实对话日志中提取高质量训练数据"""
    cleaned = []
    for log in raw_logs:
        # 1. 过滤：去除单轮、过短、纯表情对话
        if len(log['messages']) < 2:
            continue
        # 2. 去重：相似度 > 0.9 的只保留一条
        # 3. 格式化：转为 ShareGPT 格式
        # 4. 脱敏：移除用户隐私信息
        cleaned.append(format_conversation(log))
    return cleaned

# 推荐配置
# GPU: A100 (40GB) 或 RTX 4090 (24GB) 或以上
# Python 3.10+

# 1. 安装核心依赖
pip install transformers datasets accelerate peft bitsandbytes
pip install flash-attn --no-build-isolation  # 可选，加速训练
pip install wandb  # 可选，训练监控

# 2. 创建项目结构
mkdir llm-finetune
cd llm-finetune
mkdir data models scripts outputs
touch train.py config.yaml

# train.py
import os
import torch
from transformers import (
    AutoModelForCausalLM,
    AutoTokenizer,
    TrainingArguments,
    Trainer,
    DataCollatorForSeq2Seq
)
from peft import LoraConfig, get_peft_model, prepare_model_for_kbit_training
from datasets import load_dataset
import bitsandbytes as bnb

# ==================== 配置 ====================
MODEL_NAME = "Qwen/Qwen2.5-7B"  # 或 "meta-llama/Meta-Llama-3-8B"
DATA_PATH = "./data/train.jsonl"
OUTPUT_DIR = "./outputs"
LORA_R = 8
LORA_ALPHA = 16
LORA_DROPOUT = 0.05
BATCH_SIZE = 4
GRADIENT_ACCUMULATION = 4
LEARNING_RATE = 2e-4
NUM_EPOCHS = 3
MAX_LENGTH = 2048

# ==================== 加载模型和分词器 ====================
def load_model_and_tokenizer():
    # 加载 4-bit 量化模型（显存节省 70%）
    model = AutoModelForCausalLM.from_pretrained(
        MODEL_NAME,
        torch_dtype=torch.bfloat16,
        device_map="auto",
        load_in_4bit=True,
        bnb_4bit_compute_dtype=torch.bfloat16,
        bnb_4bit_use_double_quant=True,
    )
    
    tokenizer = AutoTokenizer.from_pretrained(
        MODEL_NAME,
        trust_remote_code=True,
        padding_side="right"
    )
    if tokenizer.pad_token is None:
        tokenizer.pad_token = tokenizer.eos_token
    
    return model, tokenizer

# ==================== 配置 LoRA ====================
def setup_lora(model):
    # 准备模型用于 k-bit 训练
    model = prepare_model_for_kbit_training(model)
    
    lora_config = LoraConfig(
        r=LORA_R,
        lora_alpha=LORA_ALPHA,
        target_modules=["q_proj", "k_proj", "v_proj", "o_proj", "gate_proj", "up_proj", "down_proj"],
        lora_dropout=LORA_DROPOUT,
        bias="none",
        task_type="CAUSAL_LM"
    )
    
    model = get_peft_model(model, lora_config)
    model.print_trainable_parameters()  # 打印可训练参数量
    return model

# ==================== 数据加载 ====================
def load_and_prepare_data(tokenizer):
    dataset = load_dataset("json", data_files=DATA_PATH, split="train")
    
    def format_sharegpt(example):
        """将 ShareGPT 格式转为训练格式"""
        conversations = example["conversations"]
        # 构造输入：Human: xxx\nAssistant: xxx\nHuman: xxx\nAssistant: xxx
        text = ""
        for msg in conversations:
            if msg["from"] == "human":
                text += f"Human: {msg['value']}\n"
            else:
                text += f"Assistant: {msg['value']}\n"
        return {"text": text}
    
    dataset = dataset.map(format_sharegpt)
    
    def tokenize_function(examples):
        return tokenizer(
            examples["text"],
            truncation=True,
            max_length=MAX_LENGTH,
            padding=False,
        )
    
    tokenized_dataset = dataset.map(
        tokenize_function,
        batched=True,
        remove_columns=dataset.column_names
    )
    
    return tokenized_dataset

# ==================== 训练 ====================
def train():
    # 1. 加载
    model, tokenizer = load_model_and_tokenizer()
    model = setup_lora(model)
    
    # 2. 数据
    dataset = load_and_prepare_data(tokenizer)
    
    # 3. 训练参数
    training_args = TrainingArguments(
        output_dir=OUTPUT_DIR,
        num_train_epochs=NUM_EPOCHS,
        per_device_train_batch_size=BATCH_SIZE,
        gradient_accumulation_steps=GRADIENT_ACCUMULATION,
        learning_rate=LEARNING_RATE,
        warmup_ratio=0.03,
        lr_scheduler_type="cosine",
        logging_steps=10,
        save_steps=200,
        eval_steps=200,
        eval_strategy="steps",  # 如果有验证集
        save_total_limit=3,
        bf16=True,
        report_to="wandb",  # 或 "tensorboard"
        gradient_checkpointing=True,
        optim="adamw_8bit",
    )
    
    # 4. 数据整理器
    data_collator = DataCollatorForSeq2Seq(
        tokenizer=tokenizer,
        padding=True,
        return_tensors="pt"
    )
    
    # 5. 训练器
    trainer = Trainer(
        model=model,
        args=training_args,
        train_dataset=dataset,
        data_collator=data_collator,
    )
    
    # 6. 开始训练
    trainer.train()
    
    # 7. 保存
    model.save_pretrained(os.path.join(OUTPUT_DIR, "lora_weights"))
    tokenizer.save_pretrained(os.path.join(OUTPUT_DIR, "lora_weights"))
    
    print("🎉 训练完成！")

if __name__ == "__main__":
    train()

# 单卡训练
python train.py

# 多卡训练
torchrun --nproc_per_node=4 train.py

# 后台运行（推荐）
nohup python train.py > train.log 2>&1 &

# eval.py
import json
from transformers import pipeline

def evaluate_model(model_path, test_data_path):
    """在测试集上评估模型效果"""
    pipe = pipeline("text-generation", model=model_path)
    
    with open(test_data_path, 'r') as f:
        test_data = json.load(f)
    
    results = []
    for item in test_data[:100]:  # 取前100条
        input_text = item["conversations"][0]["value"]
        expected = item["conversations"][1]["value"]
        
        generated = pipe(input_text, max_new_tokens=256, temperature=0.1)[0]['generated_text']
        
        # 计算指标
        # BLEU, ROUGE, BERTScore 等
        results.append({
            "input": input_text,
            "expected": expected,
            "generated": generated
        })
    
    return results

测试组设计：
- 对照组 A：原始模型（无微调）+ 精心设计的 System Prompt
- 对照组 B：原始模型 + RAG（知识库检索）
- 实验组 C：微调后模型（不加任何 Prompt 优化）
- 实验组 D：微调后模型 + 简单 Prompt

盲测：让测试者不知道哪个是哪个，对 50 个问题打分
胜出标准：实验组 C/D 在相关性和准确性上明显优于对照组 A/B

# 1. 安装 vLLM
pip install vllm

# 2. 启动服务
python -m vllm.entrypoints.openai.api_server \
    --model /path/to/merged_model \  # 原始模型 + LoRA 合并后的路径
    --tensor-parallel-size 2 \        # 多卡并行
    --max-num-seqs 32 \
    --port 8000

# 3. 调用 API（与 OpenAI 兼容）
curl http://localhost:8000/v1/chat/completions \
  -H "Content-Type: application/json" \
  -d '{
    "model": "/path/to/merged_model",
    "messages": [{"role": "user", "content": "你好"}]
  }'

# 支持在推理时动态加载不同的 LoRA 适配器
from peft import PeftModel
from transformers import AutoModelForCausalLM

base_model = AutoModelForCausalLM.from_pretrained("Qwen/Qwen2.5-7B")

# 场景1：客服场景
customer_service_model = PeftModel.from_pretrained(
    base_model, 
    "./lora_weights/customer_service"
)

# 场景2：技术问答场景
tech_support_model = PeftModel.from_pretrained(
    base_model,
    "./lora_weights/tech_support"
)

# 根据请求路由到不同的适配器

# DPO 训练核心代码
from trl import DPOTrainer

dpo_trainer = DPOTrainer(
    model=model,
    ref_model=ref_model,  # 参考模型（冻结）
    args=training_args,
    train_dataset=preference_dataset,  # 格式: {prompt, chosen, rejected}
    tokenizer=tokenizer,
)

dpo_trainer.train()

# 增量微调：在已有微调模型基础上继续训练
def incremental_finetune(
    base_model_path,    # 上次微调后的模型
    new_data_path,      # 新数据
    output_path,
    learning_rate=1e-5  # 使用更小的学习率
):
    model = AutoModelForCausalLM.from_pretrained(base_model_path)
    # ... 同标准训练流程，但 learning_rate 降低 10 倍

微调模型 + 业务系统集成 + 持续数据飞轮 = 企业 AI 核心竞争力