98_数据增强：提升LLM微调效果的关键技术

LLM微调数据增强流程图
原始数据 → 质量评估 → 数据清洗 → 增强变换 → 质量控制 → 训练数据
                   ↑              ↓
                领域知识 ← 评估与迭代

import nltk
from nltk.corpus import wordnet
import random
import re

# 下载必要的NLTK资源
# nltk.download('wordnet')
# nltk.download('averaged_perceptron_tagger')
# nltk.download('punkt')

def get_synonyms(word):
    """获取词语的同义词列表"""
    synonyms = set()
    for syn in wordnet.synsets(word):
        for lemma in syn.lemmas():
            synonym = lemma.name().replace('_', ' ')
            if synonym != word:
                synonyms.add(synonym)
    return list(synonyms)

def synonym_replacement(sentence, n=1):
    """同义词替换增强
    
    Args:
        sentence: 原始句子
        n: 每句话替换的词语数量
    
    Returns:
        增强后的句子
    """
    words = nltk.word_tokenize(sentence)
    new_words = words.copy()
    random_word_list = list(set([word for word in words if word.isalnum()]))
    random.shuffle(random_word_list)
    
    num_replaced = 0
    for random_word in random_word_list:
        synonyms = get_synonyms(random_word)
        if len(synonyms) >= 1:
            synonym = random.choice(synonyms)
            new_words = [synonym if word == random_word else word for word in new_words]
            num_replaced += 1
        if num_replaced >= n:
            break
    
    sentence = ' '.join(new_words)
    return sentence

# 示例
original_text = "数据增强是提升LLM微调效果的重要技术。"
enhanced_text = synonym_replacement(original_text, n=2)
print(f"原始文本: {original_text}")
print(f"增强文本: {enhanced_text}")

def random_insertion(sentence, n=1):
    """随机插入增强
    
    Args:
        sentence: 原始句子
        n: 要插入的词语数量
    
    Returns:
        增强后的句子
    """
    words = nltk.word_tokenize(sentence)
    new_words = words.copy()
    
    for _ in range(n):
        add_word(new_words)
    
    sentence = ' '.join(new_words)
    return sentence

def add_word(new_words):
    """向词语列表中随机添加一个同义词"""
    synonyms = []
    counter = 0
    
    while len(synonyms) < 1 and counter < 10:
        random_word = new_words[random.randint(0, len(new_words)-1)]
        synonyms = get_synonyms(random_word)
        counter += 1
    
    if len(synonyms) >= 1:
        random_synonym = random.choice(synonyms)
        random_idx = random.randint(0, len(new_words)-1)
        new_words.insert(random_idx, random_synonym)

# 示例
original_text = "数据增强可以提高模型的泛化能力。"
enhanced_text = random_insertion(original_text, n=2)
print(f"原始文本: {original_text}")
print(f"增强文本: {enhanced_text}")

def random_swap(sentence, n=1):
    """随机交换增强
    
    Args:
        sentence: 原始句子
        n: 要交换的词语对数量
    
    Returns:
        增强后的句子
    """
    words = nltk.word_tokenize(sentence)
    new_words = words.copy()
    
    for _ in range(n):
        new_words = swap_word(new_words)
    
    sentence = ' '.join(new_words)
    return sentence

def swap_word(new_words):
    """随机交换两个词语的位置"""
    random_idx_1 = random.randint(0, len(new_words)-1)
    random_idx_2 = random_idx_1
    counter = 0
    
    while random_idx_2 == random_idx_1 and counter < 10:
        random_idx_2 = random.randint(0, len(new_words)-1)
        counter += 1
    
    new_words[random_idx_1], new_words[random_idx_2] = new_words[random_idx_2], new_words[random_idx_1]
    return new_words

# 示例
original_text = "LLM微调需要高质量的数据和适当的策略。"
enhanced_text = random_swap(original_text, n=2)
print(f"原始文本: {original_text}")
print(f"增强文本: {enhanced_text}")

def random_deletion(sentence, p=0.1):
    """随机删除增强
    
    Args:
        sentence: 原始句子
        p: 每个词语被删除的概率
    
    Returns:
        增强后的句子
    """
    words = nltk.word_tokenize(sentence)
    
    # 如果句子只有一个词语，则不进行删除
    if len(words) <= 1:
        return sentence
    
    new_words = []
    
    # 遍历每个词语，根据概率决定是否保留
    for word in words:
        r = random.uniform(0, 1)
        if r > p:
            new_words.append(word)
    
    # 如果所有词语都被删除，则随机保留一个
    if len(new_words) == 0:
        rand_int = random.randint(0, len(words)-1)
        return words[rand_int]
    
    sentence = ' '.join(new_words)
    return sentence

# 示例
original_text = "数据增强技术可以有效提高LLM微调的效果。"
enhanced_text = random_deletion(original_text, p=0.2)
print(f"原始文本: {original_text}")
print(f"增强文本: {enhanced_text}")

from transformers import AutoModelForSeq2SeqLM, AutoTokenizer
import torch

def back_translate(text, source_lang="zh", target_langs=["en", "fr", "de"]):
    """
    使用回译进行文本增强
    
    Args:
        text: 原始文本
        source_lang: 源语言
        target_langs: 目标语言列表
    
    Returns:
        回译增强后的文本列表
    """
    results = []
    
    for target_lang in target_langs:
        # 确定翻译方向
        translate_to = target_lang
        translate_back = source_lang
        
        # 加载翻译模型
        model_name = f"Helsinki-NLP/opus-mt-{source_lang}-{target_lang}" if source_lang != "zh" else f"Helsinki-NLP/opus-mt-zh-en"
        try:
            tokenizer = AutoTokenizer.from_pretrained(model_name)
            model = AutoModelForSeq2SeqLM.from_pretrained(model_name)
        except:
            print(f"模型 {model_name} 不可用，跳过...")
            continue
        
        # 翻译到目标语言
        inputs = tokenizer(text, return_tensors="pt", truncation=True, max_length=512)
        with torch.no_grad():
            translated_tokens = model.generate(**inputs, max_length=512)
        intermediate_text = tokenizer.decode(translated_tokens[0], skip_special_tokens=True)
        
        # 翻译回源语言
        reverse_model_name = f"Helsinki-NLP/opus-mt-{target_lang}-{source_lang}" if source_lang != "zh" else f"Helsinki-NLP/opus-mt-en-zh"
        try:
            reverse_tokenizer = AutoTokenizer.from_pretrained(reverse_model_name)
            reverse_model = AutoModelForSeq2SeqLM.from_pretrained(reverse_model_name)
        except:
            print(f"反向模型 {reverse_model_name} 不可用，跳过...")
            continue
        
        reverse_inputs = reverse_tokenizer(intermediate_text, return_tensors="pt", truncation=True, max_length=512)
        with torch.no_grad():
            back_translated_tokens = reverse_model.generate(**reverse_inputs, max_length=512)
        back_translated_text = reverse_tokenizer.decode(back_translated_tokens[0], skip_special_tokens=True)
        
        results.append(back_translated_text)
    
    return results

# 示例
original_text = "数据增强是提升LLM微调效果的关键技术之一。"
enhanced_texts = back_translate(original_text, source_lang="zh", target_langs=["en"])
print(f"原始文本: {original_text}")
for i, text in enumerate(enhanced_texts):
    print(f"增强文本 {i+1}: {text}")

from transformers import BertTokenizer, BertForMaskedLM
import torch
import random

def masked_language_model_augmentation(text, model_name="bert-base-chinese", mask_ratio=0.15, num_augmentations=3):
    """
    使用掩码语言模型进行文本增强
    
    Args:
        text: 原始文本
        model_name: 预训练模型名称
        mask_ratio: 掩码比例
        num_augmentations: 生成的增强样本数量
    
    Returns:
        增强后的文本列表
    """
    # 加载模型和分词器
    tokenizer = BertTokenizer.from_pretrained(model_name)
    model = BertForMaskedLM.from_pretrained(model_name)
    model.eval()
    
    augmented_texts = []
    
    for _ in range(num_augmentations):
        # 分词
        tokens = tokenizer.tokenize(text)
        original_tokens = tokens.copy()
        
        # 确定要掩码的token位置
        num_tokens = len(tokens)
        num_to_mask = max(1, int(num_tokens * mask_ratio))
        
        # 随机选择位置进行掩码，但避免[CLS]和[SEP]
        mask_positions = random.sample(range(num_tokens), num_to_mask)
        
        # 应用掩码
        masked_tokens = original_tokens.copy()
        for pos in mask_positions:
            masked_tokens[pos] = tokenizer.mask_token
        
        # 转换为模型输入
        input_ids = tokenizer.convert_tokens_to_ids(masked_tokens)
        attention_mask = [1] * len(input_ids)
        
        # 转换为tensor
        input_ids = torch.tensor([input_ids])
        attention_mask = torch.tensor([attention_mask])
        
        # 使用模型预测掩码位置
        with torch.no_grad():
            outputs = model(input_ids, attention_mask=attention_mask)
            predictions = outputs.logits
        
        # 对每个掩码位置选择最可能的token
        for pos in mask_positions:
            mask_token_logits = predictions[0, pos]
            # 选择概率最高的token
            predicted_token_id = torch.argmax(mask_token_logits).item()
            predicted_token = tokenizer.convert_ids_to_tokens([predicted_token_id])[0]
            
            # 如果预测结果是subtoken，保留原始token
            if predicted_token.startswith('##'):
                predicted_token = original_tokens[pos]
            
            masked_tokens[pos] = predicted_token
        
        # 转换回文本
        augmented_text = tokenizer.convert_tokens_to_string(masked_tokens)
        augmented_text = augmented_text.replace(' ', '')
        augmented_texts.append(augmented_text)
    
    return augmented_texts

# 示例
original_text = "数据增强技术可以显著提升LLM微调的效果和泛化能力。"
enhanced_texts = masked_language_model_augmentation(original_text, mask_ratio=0.2, num_augmentations=3)
print(f"原始文本: {original_text}")
for i, text in enumerate(enhanced_texts):
    print(f"增强文本 {i+1}: {text}")

from transformers import AutoTokenizer, AutoModelForCausalLM
import torch

def gpt_paraphrase(text, model_name="uer/gpt2-chinese-cluecorpussmall", num_augmentations=3, temperature=0.7, top_k=50, top_p=0.95):
    """
    使用GPT模型进行文本改写增强
    
    Args:
        text: 原始文本
        model_name: 预训练模型名称
        num_augmentations: 生成的增强样本数量
        temperature: 生成温度参数
        top_k: 采样时保留的最高概率词汇数量
        top_p: 累积概率阈值
    
    Returns:
        增强后的文本列表
    """
    # 加载模型和分词器
    tokenizer = AutoTokenizer.from_pretrained(model_name)
    model = AutoModelForCausalLM.from_pretrained(model_name)
    model.eval()
    
    # 构建提示
    prompt = f"将下面的句子改写为不同的表达方式，保持原意不变：\n{text}\n改写："
    
    augmented_texts = []
    
    for _ in range(num_augmentations):
        # 编码提示
        input_ids = tokenizer.encode(prompt, return_tensors="pt")
        
        # 生成改写文本
        with torch.no_grad():
            output = model.generate(
                input_ids,
                max_length=input_ids.shape[1] + 100,
                temperature=temperature,
                top_k=top_k,
                top_p=top_p,
                do_sample=True,
                pad_token_id=tokenizer.eos_token_id
            )
        
        # 解码输出
        generated_text = tokenizer.decode(output[0], skip_special_tokens=True)
        
        # 提取改写部分
        if "改写：" in generated_text:
            paraphrased = generated_text.split("改写：")[1].strip()
        else:
            paraphrased = generated_text[len(prompt):].strip()
        
        augmented_texts.append(paraphrased)
    
    return augmented_texts

# 示例
original_text = "数据增强是提升LLM微调效果的重要技术手段。"
enhanced_texts = gpt_paraphrase(original_text, num_augmentations=3)
print(f"原始文本: {original_text}")
for i, text in enumerate(enhanced_texts):
    print(f"增强文本 {i+1}: {text}")

def eda_augmentation(text, alpha_sr=0.1, alpha_ri=0.1, alpha_rs=0.1, p_rd=0.1, num_augmentations=9):
    """
    实现EDA数据增强
    
    Args:
        text: 原始文本
        alpha_sr: 同义词替换的词语比例
        alpha_ri: 随机插入的词语比例
        alpha_rs: 随机交换的词语对比例
        p_rd: 随机删除的概率
        num_augmentations: 生成的增强样本数量
    
    Returns:
        增强后的文本列表
    """
    words = nltk.word_tokenize(text)
    num_words = len(words)
    
    augmented_texts = []
    
    # 计算每种增强方法的参数
    n_sr = max(1, int(alpha_sr * num_words))
    n_ri = max(1, int(alpha_ri * num_words))
    n_rs = max(1, int(alpha_rs * num_words))
    
    # 生成同义词替换样本
    sr_count = min(num_augmentations, 3)
    for _ in range(sr_count):
        augmented_texts.append(synonym_replacement(text, n=n_sr))
    
    # 生成随机插入样本
    ri_count = min(num_augmentations - sr_count, 2)
    for _ in range(ri_count):
        augmented_texts.append(random_insertion(text, n=n_ri))
    
    # 生成随机交换样本
    rs_count = min(num_augmentations - sr_count - ri_count, 2)
    for _ in range(rs_count):
        augmented_texts.append(random_swap(text, n=n_rs))
    
    # 生成随机删除样本
    rd_count = min(num_augmentations - sr_count - ri_count - rs_count, 2)
    for _ in range(rd_count):
        augmented_texts.append(random_deletion(text, p=p_rd))
    
    # 如果还需要更多样本，随机选择一种方法继续生成
    remaining = num_augmentations - len(augmented_texts)
    for _ in range(remaining):
        method = random.choice([0, 1, 2, 3])  # 0: SR, 1: RI, 2: RS, 3: RD
        if method == 0:
            augmented_texts.append(synonym_replacement(text, n=n_sr))
        elif method == 1:
            augmented_texts.append(random_insertion(text, n=n_ri))
        elif method == 2:
            augmented_texts.append(random_swap(text, n=n_rs))
        else:
            augmented_texts.append(random_deletion(text, p=p_rd))
    
    return augmented_texts

# 示例
original_text = "数据增强是一种有效的LLM微调技术。"
enhanced_texts = eda_augmentation(original_text, num_augmentations=5)
print(f"原始文本: {original_text}")
for i, text in enumerate(enhanced_texts):
    print(f"增强文本 {i+1}: {text}")

def generate_contrastive_samples(text, other_texts, num_positive=3, num_negative=5):
    """
    生成对比学习所需的正负样本
    
    Args:
        text: 锚文本
        other_texts: 其他类别文本列表
        num_positive: 生成的正样本数量
        num_negative: 选择的负样本数量
    
    Returns:
        包含锚文本、正样本和负样本的字典
    """
    # 生成正样本
    positive_samples = []
    
    # 使用回译生成正样本
    back_translated = back_translate(text, target_langs=["en"])
    positive_samples.extend(back_translated[:min(num_positive, len(back_translated))])
    
    # 如果正样本不足，使用EDA补充
    if len(positive_samples) < num_positive:
        eda_samples = eda_augmentation(text, num_augmentations=num_positive - len(positive_samples))
        positive_samples.extend(eda_samples)
    
    # 选择负样本
    negative_samples = random.sample(other_texts, min(num_negative, len(other_texts)))
    
    return {
        "anchor": text,
        "positive": positive_samples[:num_positive],
        "negative": negative_samples[:num_negative]
    }

# 示例
original_text = "数据增强是提升LLM微调效果的关键技术。"
other_texts = [
    "模型压缩可以有效减少LLM的推理时间和资源消耗。",
    "知识蒸馏是一种将大模型知识迁移到小模型的技术。",
    "量化是降低LLM内存占用的常用方法。",
    "剪枝可以移除模型中不重要的神经元，提高效率。"
]

contrastive_samples = generate_contrastive_samples(original_text, other_texts)
print(f"锚文本: {contrastive_samples['anchor']}")
print("正样本:")
for i, pos in enumerate(contrastive_samples['positive']):
    print(f"  {i+1}: {pos}")
print("负样本:")
for i, neg in enumerate(contrastive_samples['negative']):
    print(f"  {i+1}: {neg}")

def instruction_rewriting(original_instruction, num_variations=3):
    """
    生成指令的多种变体
    
    Args:
        original_instruction: 原始指令
        num_variations: 生成的指令变体数量
    
    Returns:
        指令变体列表
    """
    # 可以使用GPT模型进行指令重写
    # 这里使用简单的模板作为示例
    templates = [
        "请你{instruction}",
        "要求：{instruction}",
        "任务：{instruction}",
        "需要你完成：{instruction}",
        "目标：{instruction}"
    ]
    
    variations = []
    for _ in range(num_variations):
        template = random.choice(templates)
        variation = template.format(instruction=original_instruction)
        variations.append(variation)
    
    return variations

# 示例
original_instruction = "总结以下文章的主要内容"
instruction_variations = instruction_rewriting(original_instruction)
print(f"原始指令: {original_instruction}")
for i, var in enumerate(instruction_variations):
    print(f"指令变体 {i+1}: {var}")

def generate_instruction_response_pairs(topic, num_pairs=5):
    """
    生成指令-响应对
    
    Args:
        topic: 主题
        num_pairs: 生成的指令-响应对数量
    
    Returns:
        指令-响应对列表
    """
    # 这里使用预定义的指令模板作为示例
    # 实际应用中可以使用GPT等模型生成
    instruction_templates = [
        f"解释{topic}的基本概念",
        f"列举{topic}的主要应用场景",
        f"分析{topic}的优缺点",
        f"比较{topic}与其他相关技术的区别",
        f"总结{topic}的最新研究进展"
    ]
    
    # 简单的响应生成逻辑
    def generate_response(instruction):
        return f"这是关于'{instruction}'的示例回答，实际应用中应使用LLM生成更详细的内容。"
    
    pairs = []
    for i in range(min(num_pairs, len(instruction_templates))):
        instruction = instruction_templates[i]
        response = generate_response(instruction)
        pairs.append({
            "instruction": instruction,
            "response": response
        })
    
    return pairs

# 示例
topic = "数据增强"
pairs = generate_instruction_response_pairs(topic)
for i, pair in enumerate(pairs):
    print(f"对 {i+1}:")
    print(f"  指令: {pair['instruction']}")
    print(f"  响应: {pair['response']}")

def dialogue_reordering(dialogue, keep_turns=0.8):
    """
    对话顺序重排增强
    
    Args:
        dialogue: 原始对话列表，每个元素是{"role": "user/assistant", "content": "..."}
        keep_turns: 保留的轮次比例
    
    Returns:
        重排后的对话
    """
    # 确保对话从user开始
    if not dialogue or dialogue[0]["role"] != "user":
        return dialogue
    
    # 计算保留的轮次数
    num_turns = len(dialogue)
    num_to_keep = max(2, int(num_turns * keep_turns))
    
    # 确保保留的是完整的用户-助手交互
    if num_to_keep % 2 != 0:
        num_to_keep -= 1
    
    # 选择起始点
    start_idx = random.randint(0, num_turns - num_to_keep)
    # 确保起始点是user
    if start_idx % 2 != 0:
        start_idx -= 1
        if start_idx < 0:
            start_idx = 0
    
    # 提取子对话
    reordered_dialogue = dialogue[start_idx:start_idx + num_to_keep]
    
    return reordered_dialogue

# 示例
original_dialogue = [
    {"role": "user", "content": "什么是数据增强？"},
    {"role": "assistant", "content": "数据增强是通过变换原始数据生成新样本的技术，用于扩充训练数据。"},
    {"role": "user", "content": "数据增强有哪些常用方法？"},
    {"role": "assistant", "content": "常用的数据增强方法包括同义词替换、回译、随机插入等。"},
    {"role": "user", "content": "在LLM微调中如何应用数据增强？"},
    {"role": "assistant", "content": "在LLM微调中，数据增强可以用于扩充训练数据，减少过拟合风险。"}
]

reordered = dialogue_reordering(original_dialogue)
print("原始对话:")
for turn in original_dialogue:
    print(f"{turn['role']}: {turn['content']}")
print("\n重排对话:")
for turn in reordered:
    print(f"{turn['role']}: {turn['content']}")

def inject_dialogue_noise(dialogue, noise_level=0.1):
    """
    在对话中注入噪声
    
    Args:
        dialogue: 原始对话列表
        noise_level: 噪声注入水平 (0-1)
    
    Returns:
        注入噪声后的对话
    """
    noisy_dialogue = []
    
    for turn in dialogue:
        # 只处理content部分
        content = turn["content"]
        
        # 随机决定是否注入噪声
        if random.random() < noise_level:
            # 拼写错误示例（简化版）
            words = content.split()
            if len(words) > 1:
                # 随机选择一个词语
                idx = random.randint(0, len(words)-1)
                word = words[idx]
                
                # 如果词语长度大于2，随机替换一个字符
                if len(word) > 2:
                    char_idx = random.randint(1, len(word)-2)  # 避免替换首末字符
                    new_char = random.choice('abcdefghijklmnopqrstuvwxyz')
                    new_word = word[:char_idx] + new_char + word[char_idx+1:]
                    words[idx] = new_word
            
            content = ' '.join(words)
        
        noisy_dialogue.append({"role": turn["role"], "content": content})
    
    return noisy_dialogue

# 示例
noisy = inject_dialogue_noise(original_dialogue, noise_level=0.3)
print("注入噪声后的对话:")
for turn in noisy:
    print(f"{turn['role']}: {turn['content']}")

from sentence_transformers import SentenceTransformer
from sklearn.metrics.pairwise import cosine_similarity

def evaluate_semantic_similarity(original_text, augmented_texts):
    """
    评估增强文本与原始文本的语义相似度
    
    Args:
        original_text: 原始文本
        augmented_texts: 增强文本列表
    
    Returns:
        相似度列表
    """
    # 加载预训练的语义相似度模型
    model = SentenceTransformer('distiluse-base-multilingual-cased-v1')  # 支持多语言
    
    # 编码原始文本和增强文本
    original_embedding = model.encode([original_text])
    augmented_embeddings = model.encode(augmented_texts)
    
    # 计算余弦相似度
    similarities = []
    for emb in augmented_embeddings:
        similarity = cosine_similarity([original_embedding[0]], [emb])[0][0]
        similarities.append(similarity)
    
    return similarities

# 示例
original_text = "数据增强可以有效提升LLM微调的效果。"
augmented_texts = [
    "数据增强能够显著提高LLM微调的性能。",
    "通过数据增强技术，可以让LLM微调的效果更好。",
    "数据增强对LLM微调有很大帮助。"
]

similarities = evaluate_semantic_similarity(original_text, augmented_texts)
for i, (text, sim) in enumerate(zip(augmented_texts, similarities)):
    print(f"增强文本 {i+1}: {text}")
    print(f"相似度: {sim:.4f}")

def filter_by_similarity(original_text, augmented_texts, threshold=0.8):
    """
    基于语义相似度过滤增强文本
    
    Args:
        original_text: 原始文本
        augmented_texts: 增强文本列表
        threshold: 相似度阈值
    
    Returns:
        过滤后的增强文本列表
    """
    similarities = evaluate_semantic_similarity(original_text, augmented_texts)
    filtered_texts = []
    
    for text, sim in zip(augmented_texts, similarities):
        if sim >= threshold:
            filtered_texts.append(text)
    
    return filtered_texts

# 示例
filtered = filter_by_similarity(original_text, augmented_texts, threshold=0.85)
print(f"过滤后保留 {len(filtered)} 个样本:")
for text in filtered:
    print(f"- {text}")

def filter_for_diversity(augmented_texts, min_diversity=0.7):
    """
    过滤增强文本以确保多样性
    
    Args:
        augmented_texts: 增强文本列表
        min_diversity: 最小多样性阈值（样本之间的最大相似度）
    
    Returns:
        具有足够多样性的文本列表
    """
    if len(augmented_texts) <= 1:
        return augmented_texts
    
    # 加载预训练模型
    model = SentenceTransformer('distiluse-base-multilingual-cased-v1')
    
    # 编码所有文本
    embeddings = model.encode(augmented_texts)
    
    # 选择多样性样本
    selected = [0]  # 先选择第一个样本
    remaining = list(range(1, len(augmented_texts)))
    
    while remaining and len(selected) < len(augmented_texts):
        best_idx = -1
        best_score = float('-inf')
        
        for idx in remaining:
            # 计算与已选样本的最小相似度
            min_sim = float('inf')
            for sel_idx in selected:
                sim = cosine_similarity([embeddings[idx]], [embeddings[sel_idx]])[0][0]
                if sim < min_sim:
                    min_sim = sim
            
            # 选择与已选样本最不相似的样本
            if min_sim > best_score:
                best_score = min_sim
                best_idx = idx
        
        if best_score >= min_diversity:
            break
        
        selected.append(best_idx)
        remaining.remove(best_idx)
    
    # 返回选择的样本
    return [augmented_texts[i] for i in selected]

# 示例
diverse_texts = filter_for_diversity(augmented_texts)
print(f"多样性过滤后保留 {len(diverse_texts)} 个样本:")
for text in diverse_texts:
    print(f"- {text}")

def calculate_augmentation_ratios(labels):
    """
    根据类别频率计算增强比例
    
    Args:
        labels: 样本标签列表
    
    Returns:
        每个类别的增强比例字典
    """
    from collections import Counter
    
    # 计算每个类别的样本数量
    label_counts = Counter(labels)
    total_samples = len(labels)
    num_classes = len(label_counts)
    
    # 计算理想的样本数量（所有类别样本数相等）
    ideal_count = total_samples / num_classes
    
    # 计算每个类别的增强比例
    augmentation_ratios = {}
    for label, count in label_counts.items():
        # 如果样本数少于理想数量，增加增强比例
        if count < ideal_count:
            ratio = (ideal_count - count) / count
            augmentation_ratios[label] = ratio
        else:
            augmentation_ratios[label] = 0.0
    
    return augmentation_ratios

# 示例
labels = [0, 0, 0, 0, 1, 1, 2]  # 不平衡的标签分布
ratios = calculate_augmentation_ratios(labels)
print("各类别的增强比例:")
for label, ratio in ratios.items():
    print(f"类别 {label}: {ratio:.2f}")

增强强度调优流程图
开始 → 确定初始强度 → 训练模型 → 评估性能 → 过拟合? → 降低强度 → 继续训练
                             ↓否                      
                          欠拟合? → 增加强度 → 继续训练
                             ↓否
                           结束

数据增强实施路径图
原始数据 → 数据清洗 → 增强策略选择 → 增强实施 → 质量评估 → 过滤 → 训练数据 → 模型微调 → 效果评估 → 策略优化