指令微调配方详解：打造高质量监督微调数据集

{
  "messages": [
    {"role": "system", "content": "你是一个专业的编程助手。"},
    {"role": "user", "content": "解释什么是递归"},
    {"role": "assistant", "content": "递归是一种编程技术..."}
  ]
}

class InstructionDataBuilder:
    async def build_from_seeds(self, seed_instructions, expansion_rate=10):
        """从种子指令扩展"""
        expanded = []
        
        for seed in seed_instructions:
            # 1. 改写指令
            rewrites = await self.rewrite_instruction(seed, n=expansion_rate//2)
            
            # 2. 生成变体
            variants = await self.generate_variants(seed, n=expansion_rate//2)
            
            expanded.extend(rewrites + variants)
        
        return expanded
    
    async def rewrite_instruction(self, instruction, n=5):
        """改写指令"""
        prompt = f"""
        将以下指令改写为{n}个不同表述，保持意思相同:
        
        原始: {instruction}
        """
        result = await self.llm.call(prompt)
        return result["rewrites"]

class SelfInstruct:
    async def generate(self, seed_tasks, num_tasks=1000):
        """Self-Instruct生成"""
        tasks = list(seed_tasks)
        
        while len(tasks) < num_tasks:
            # 1. 随机选择种子任务作为示例
            examples = random.sample(tasks, min(3, len(tasks)))
            
            # 2. 生成新指令
            new_instruction = await self.llm.generate(
                f"基于以下示例生成一个新的指令:\n{examples}"
            )
            
            # 3. 过滤低质量
            if self.is_quality(new_instruction):
                # 4. 生成回答
                response = await self.llm.generate(new_instruction)
                
                tasks.append({
                    "instruction": new_instruction,
                    "response": response
                })
        
        return tasks

class EvolInstruct:
    """逐步进化指令复杂度"""
    
    async def evolve(self, instruction):
        """进化指令"""
        strategies = [
            "增加约束条件",
            "增加推理步骤",
            "增加领域深度",
            "增加多步骤要求",
            "增加边界条件处理"
        ]
        
        strategy = random.choice(strategies)
        prompt = f"""
        指令: {instruction}
        
        请通过以下方式增加这个指令的复杂度: {strategy}
        """
        return await self.llm.call(prompt)

class QualityFilter:
    def filter(self, dataset):
        filtered = []
        
        for sample in dataset:
            # 1. 长度检查
            if len(sample["response"]) < 10:
                continue
            
            # 2. 重复检查
            if self.is_duplicate(sample, filtered):
                continue
            
            # 3. 格式检查
            if not self.validate_format(sample):
                continue
            
            # 4. 内容质量
            if not self.check_content_quality(sample):
                continue
            
            filtered.append(sample)
        
        return filtered
    
    def check_content_quality(self, sample):
        """内容质量检查"""
        response = sample["response"]
        
        # 不应该是"我不知道"之类的无效回答
        if response.strip() in ["我不知道", "无法回答", "I don't know"]:
            return False
        
        # 不应该是重复内容
        if len(set(response.split())) / len(response.split()) < 0.3:
            return False
        
        return True

class Deduplicator:
    def deduplicate(self, dataset):
        """多级去重"""
        # 1. 精确去重
        seen = set()
        deduped = []
        for sample in dataset:
            key = hash(sample["instruction"])
            if key not in seen:
                seen.add(key)
                deduped.append(sample)
        
        # 2. 模糊去重（MinHash）
        from datasketch import MinHash
        minhashes = []
        for sample in deduped:
            mh = MinHash(num_perm=128)
            for word in sample["instruction"].split():
                mh.update(word.encode())
            minhashes.append(mh)
        
        # 移除相似度>0.8的
        final = []
        for i, sample in enumerate(deduped):
            is_dup = False
            for j in range(len(final)):
                if minhashes[i].jaccard(minhashes[final[j]["index"]]) > 0.8:
                    is_dup = True
                    break
            if not is_dup:
                final.append({"index": i, "sample": sample})
        
        return [f["sample"] for f in final]

class DataMixer:
    def create_mix(self, datasets):
        """创建数据混合"""
        # 2026年经验配比
        mix = {
            "general_qa": 0.30,        # 通用问答
            "coding": 0.20,            # 编程
            "reasoning": 0.15,         # 推理
            "math": 0.10,              # 数学
            "creative_writing": 0.10,  # 创意写作
            "safety": 0.05,            # 安全
            "multi_turn": 0.05,        # 多轮对话
            "tool_use": 0.05,          # 工具使用
        }
        
        total = sum(v for v in mix.values())
        assert abs(total - 1.0) < 0.01
        
        mixed = []
        for category, ratio in mix.items():
            n = int(total_samples * ratio)
            sampled = self.sample_from(datasets[category], n)
            mixed.extend(sampled)
        
        random.shuffle(mixed)
        return mixed

from trl import SFTTrainer, SFTConfig

config = SFTConfig(
    output_dir="./sft-output",
    num_train_epochs=3,
    per_device_train_batch_size=8,
    gradient_accumulation_steps=4,
    learning_rate=2e-5,
    warmup_ratio=0.03,
    lr_scheduler_type="cosine",
    max_seq_length=2048,
    bf16=True,
    gradient_checkpointing=True,
    save_strategy="epoch",
    evaluation_strategy="epoch",
)

trainer = SFTTrainer(
    model=base_model,
    args=config,
    train_dataset=train_data,
    eval_dataset=eval_data,
    tokenizer=tokenizer,
)

trainer.train()

async def evaluate_sft(model, eval_set):
    """评估SFT模型"""
    metrics = {}
    
    # 1. 自动评估
    metrics["loss"] = model.evaluate(eval_set)
    
    # 2. 基准测试
    benchmarks = ["MMLU", "HumanEval", "GSM8K", "MT-Bench"]
    for bench in benchmarks:
        metrics[bench] = await run_benchmark(model, bench)
    
    # 3. 人工评估
    samples = generate_samples(model, n=100)
    metrics["human_score"] = await human_eval(samples)
    
    return metrics