LLM数据增强技术：用AI训练更好的AI

class TextAugmentor:
    def synonym_replace(self, text, replace_rate=0.1):
        """同义词替换"""
        words = text.split()
        n_replace = int(len(words) * replace_rate)
        
        for _ in range(n_replace):
            idx = random.randint(0, len(words)-1)
            synonyms = self.get_synonyms(words[idx])
            if synonyms:
                words[idx] = random.choice(synonyms)
        
        return " ".join(words)
    
    def back_translation(self, text):
        """回译增强"""
        # 中文→英文→中文
        en = await self.translate(text, "zh", "en")
        zh = await self.translate(en, "en", "zh")
        return zh
    
    def random_deletion(self, text, delete_rate=0.1):
        """随机删除"""
        words = text.split()
        kept = [w for w in words if random.random() > delete_rate]
        return " ".join(kept) if kept else words[0]

class AIGenerationAugmentor:
    async def generate_variations(self, instruction, n=5):
        """生成指令变体"""
        prompt = f"""
        为以下指令生成{n}个不同的表述方式:
        {instruction}
        
        要求:
        1. 保持语义相同
        2. 改变表述方式（正式/口语/简洁/详细）
        3. 适合不同教育水平的用户
        """
        return await self.llm.call(prompt)
    
    async def generate_edge_cases(self, instruction):
        """生成边界情况"""
        prompt = f"""
        对于以下指令，生成边界情况的变体:
        {instruction}
        
        考虑:
        1. 极端简短的输入
        2. 包含错误的输入
        3. 多语言混合输入
        4. 模糊/有歧义的输入
        """
        return await self.llm.call(prompt)

class SelfInstructGenerator:
    async def generate_dataset(self, seed_tasks, target_size=10000):
        """Self-Instruct生成大规模数据"""
        dataset = list(seed_tasks)
        
        while len(dataset) < target_size:
            # 1. 采样种子
            seeds = random.sample(dataset, min(3, len(dataset)))
            
            # 2. 生成新指令
            new_instruction = await self.llm.generate(
                f"基于以下示例生成一个新的不同的指令:\n{seeds}"
            )
            
            # 3. 质量过滤
            if self.passes_quality_check(new_instruction):
                # 4. 生成回答
                response = await self.llm.generate(new_instruction)
                
                # 5. 质量验证
                if self.verify_quality(new_instruction, response):
                    dataset.append({
                        "instruction": new_instruction,
                        "response": response
                    })
        
        return dataset

class SyntheticDataFilter:
    def filter(self, dataset):
        filtered = []
        
        for sample in dataset:
            # 1. 多样性检查
            if self.too_similar(sample, filtered):
                continue
            
            # 2. 复杂度检查
            if self.too_simple(sample):
                continue
            
            # 3. 事实准确性
            if not self.factually_correct(sample):
                continue
            
            # 4. 格式规范
            if not self.well_formatted(sample):
                continue
            
            filtered.append(sample)
        
        return filtered

class SyntheticDataDebiaser:
    async def debias(self, dataset):
        """去除合成数据中的偏见"""
        # 1. 分析分布
        distributions = self.analyze_distributions(dataset)
        
        # 2. 识别偏差
        biases = self.identify_biases(distributions)
        
        # 3. 补充不足
        for bias in biases:
            additional = await self.generate_compensating_data(bias)
            dataset.extend(additional)
        
        # 4. 重新平衡
        dataset = self.rebalance(dataset)
        
        return dataset

class CodeAugmentor:
    async def augment_code(self, code_snippet):
        """代码数据增强"""
        augmentations = []
        
        # 1. 变量重命名
        augmentations.append(await self.rename_variables(code_snippet))
        
        # 2. 注释添加/修改
        augmentations.append(await self.add_comments(code_snippet))
        
        # 3. 等价重构
        augmentations.append(await self.refactor(code_snippet))
        
        # 4. 语言转换
        augmentations.append(await self.translate_language(code_snippet, "Python", "JavaScript"))
        
        return augmentations

class ReasoningAugmentor:
    async def generate_reasoning_chains(self, question, answer):
        """生成多种推理路径"""
        prompt = f"""
        问题: {question}
        答案: {answer}
        
        请生成3种不同的推理路径来到达这个答案:
        1. 直接推理路径
        2. 反证法路径
        3. 类比推理路径
        """
        return await self.llm.call(prompt)

class AugmentedDataMixer:
    def mix(self, real_data, synthetic_data, ratio=0.3):
        """混合真实和合成数据"""
        # 合成数据占比不应过高
        n_synthetic = int(len(real_data) * ratio / (1 - ratio))
        synthetic_sample = random.sample(synthetic_data, min(n_synthetic, len(synthetic_data)))
        
        mixed = real_data + synthetic_sample
        random.shuffle(mixed)
        return mixed

class SyntheticDataEvaluator:
    async def evaluate(self, real_data, synthetic_data):
        metrics = {
            "diversity": self.compute_diversity(synthetic_data),
            "fidelity": await self.compute_fidelity(real_data, synthetic_data),
            "novelty": self.compute_novelty(real_data, synthetic_data),
            "utility": await self.compute_utility(real_data, synthetic_data)
        }
        
        # 效用测试：用合成数据训练，在真实数据上测试
        model = train(synthetic_data)
        metrics["downstream_performance"] = evaluate(model, real_data)
        
        return metrics