大模型蒸馏：从GPT-5到7B的能力迁移方案

Teacher模型 (GPT-5, 175B+)
    │
    │ 知识转移
    ▼
Student模型 (Qwen2.5-7B)
    │
    │ 微调 + 对齐
    ▼
蒸馏模型 (7B, 接近GPT-5效果)

import openai
import json
from tqdm import tqdm

client = openai.OpenAI(api_key="your-key")

def generate_training_data(prompts, teacher_model="gpt-5"):
    """用GPT-5生成高质量训练数据"""
    
    training_data = []
    
    for prompt in tqdm(prompts):
        # 系统提示：让Teacher模型输出详细、有条理的回答
        system_prompt = """你是一个专业的AI助手。请按照以下要求回答问题：
1. 回答要详细且有条理
2. 使用适当的格式（标题、列表、代码块）
3. 如果适用，提供示例
4. 确保回答准确且有帮助"""
        
        response = client.chat.completions.create(
            model=teacher_model,
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": prompt},
            ],
            temperature=0.7,  # 适度多样性
            max_tokens=2000,
        )
        
        training_data.append({
            "instruction": prompt,
            "output": response.choices[0].message.content,
            "metadata": {
                "teacher_model": teacher_model,
                "temperature": 0.7,
            }
        })
    
    return training_data

# 生成多样化提示词
def generate_diverse_prompts(domains, num_per_domain=500):
    """生成覆盖多个领域的多样化提示词"""
    prompts = []
    
    for domain in domains:
        # 用Teacher模型生成领域特定提示词
        meta_prompt = f"为'{domain}'领域生成{num_per_domain}个多样化的用户问题。"
        domain_prompts = client.chat.completions.create(
            model="gpt-5",
            messages=[{"role": "user", "content": meta_prompt}],
        )
        prompts.extend(domain_prompts.choices[0].message.content.split("\n"))
    
    return prompts

# 生成数据
domains = ["编程开发", "数据分析", "文案写作", "技术咨询", "数学推理", "创意设计"]
prompts = generate_diverse_prompts(domains, num_per_domain=500)
training_data = generate_training_data(prompts)

# 保存
with open("distill_training_data.json", "w", encoding="utf-8") as f:
    json.dump(training_data, f, ensure_ascii=False, indent=2)

def quality_filter(data):
    """过滤低质量训练数据"""
    filtered = []
    
    for item in data:
        output = item["output"]
        
        # 长度过滤
        if len(output) < 50 or len(output) > 5000:
            continue
        
        # 质量信号检查
        quality_signals = [
            len(output.split("\n")) > 3,  # 有结构
            any(kw in output for kw in ["1.", "首先", "因为", "因此"]),  # 有逻辑
            not output.startswith("I cannot") and not output.startswith("我无法"),  # 非拒答
        ]
        
        if sum(quality_signals) >= 2:
            filtered.append(item)
    
    return filtered

from transformers import AutoModelForCausalLM, AutoTokenizer, TrainingArguments
from trl import SFTTrainer
from peft import LoraConfig, get_peft_model

# 加载Student模型
model = AutoModelForCausalLM.from_pretrained(
    "Qwen/Qwen2.5-7B",
    torch_dtype=torch.bfloat16,
    device_map="auto",
)

# LoRA配置
lora_config = LoraConfig(
    r=64,
    lora_alpha=16,
    target_modules=["q_proj", "k_proj", "v_proj", "o_proj", 
                    "gate_proj", "up_proj", "down_proj"],
    lora_dropout=0.05,
    bias="none",
    task_type="CAUSAL_LM",
)

# 训练
training_args = TrainingArguments(
    output_dir="./distilled-7b",
    num_train_epochs=3,
    per_device_train_batch_size=4,
    gradient_accumulation_steps=4,
    learning_rate=2e-4,
    lr_scheduler_type="cosine",
    warmup_ratio=0.03,
    bf16=True,
    logging_steps=10,
    save_steps=500,
)

trainer = SFTTrainer(
    model=model,
    args=training_args,
    train_dataset=dataset,
    tokenizer=tokenizer,
    max_seq_length=2048,
    peft_config=lora_config,
)

trainer.train()

import torch
import torch.nn.functional as F

class DistillationTrainer:
    def __init__(self, teacher_model, student_model, temperature=2.0, alpha=0.5):
        self.teacher = teacher_model
        self.student = student_model
        self.temperature = temperature  # 温度参数，软化概率分布
        self.alpha = alpha  # 蒸馏loss与SFT loss的权重
        
        # 冻结Teacher
        for param in self.teacher.parameters():
            param.requires_grad = False
        self.teacher.eval()
    
    def distillation_loss(self, student_logits, teacher_logits, labels):
        """蒸馏损失函数"""
        
        # KL散度损失（软标签）
        soft_loss = F.kl_div(
            F.log_softmax(student_logits / self.temperature, dim=-1),
            F.softmax(teacher_logits / self.temperature, dim=-1),
            reduction="batchmean",
        ) * (self.temperature ** 2)
        
        # 交叉熵损失（硬标签）
        hard_loss = F.cross_entropy(
            student_logits.view(-1, student_logits.size(-1)),
            labels.view(-1),
            ignore_index=-100,
        )
        
        # 加权组合
        return self.alpha * soft_loss + (1 - self.alpha) * hard_loss
    
    def train_step(self, batch):
        """单步训练"""
        input_ids = batch["input_ids"]
        attention_mask = batch["attention_mask"]
        labels = batch["labels"]
        
        # Teacher前向传播（不需要梯度）
        with torch.no_grad():
            teacher_outputs = self.teacher(
                input_ids=input_ids,
                attention_mask=attention_mask,
            )
            teacher_logits = teacher_outputs.logits
        
        # Student前向传播
        student_outputs = self.student(
            input_ids=input_ids,
            attention_mask=attention_mask,
            labels=labels,
        )
        student_logits = student_outputs.logits
        
        # 计算蒸馏loss
        loss = self.distillation_loss(student_logits, teacher_logits, labels)
        
        return loss

GPT-5 (API)  ──生成数据──►  训练数据集
                                │
Qwen2.5-72B (开源) ──►  Teacher (白盒蒸馏)
                                │
                                ▼
                        Student (Qwen2.5-7B)

def hybrid_distillation_pipeline():
    """混合蒸馏pipeline"""
    
    # 阶段1: GPT-5生成高质量数据
    training_data = generate_with_gpt5(prompts)
    
    # 阶段2: SFT初步训练
    sft_model = train_sft(student_model, training_data)
    
    # 阶段3: 用开源大模型做白盒蒸馏精调
    teacher = load_model("Qwen/Qwen2.5-72B-Instruct")
    distilled_model = distill(teacher=teacher, student=sft_model, data=training_data)
    
    # 阶段4: DPO偏好对齐
    preference_data = generate_preference_data_with_gpt5(distilled_model, prompts)
    final_model = train_dpo(distilled_model, preference_data)
    
    return final_model

# 问题：Student模型只学到了Teacher的"说话方式"而非"推理能力"
# 症状：输出格式很像GPT-5，但内容质量没有提升

# 解决：数据多样化 + 能力导向的评估
def diverse_training_data():
    """确保训练数据覆盖不同风格和能力"""
    data = {
        "reasoning": [],    # 推理类（数学、逻辑）
        "knowledge": [],    # 知识类（事实问答）
        "creation": [],     # 创作类（写作、代码）
        "instruction": [],  # 指令跟随类
    }
    # 每个类别至少占20%，避免单一风格过拟合

def check_capability_degradation(original_model, distilled_model, test_cases):
    """检查蒸馏后模型是否退化了原有能力"""
    
    for case in test_cases:
        original_output = original_model.generate(case)
        distilled_output = distilled_model.generate(case)
        
        # 对比两个输出
        score = evaluate_pair(original_output, distilled_output)
        
        if score < 0.7:
            print(f"退化警告: {case}")
            print(f"原始: {original_output[:100]}")
            print(f"蒸馏: {distilled_output[:100]}")

# 白盒蒸馏的温度参数T的影响
# T越大，概率分布越"软"，传递的"暗知识"越多

for T in [1.0, 2.0, 4.0, 8.0]:
    model = distill(temperature=T)
    score = evaluate(model)
    print(f"T={T}: score={score}")

# 典型结果:
# T=1.0: 76.2 (接近硬标签)
# T=2.0: 78.5 (推荐)
# T=4.0: 78.1 (略降)
# T=8.0: 76.8 (过软，信息丢失)