从小数据走到大模型：AI数据增强笔记

# 图像增强示例
from tensorflow.keras.preprocessing.image import ImageDataGenerator

datagen = ImageDataGenerator(
    rotation_range=20,      # 随机旋转
    width_shift_range=0.2,  # 水平平移
    height_shift_range=0.2, # 垂直平移
    horizontal_flip=True    # 水平翻转
)

# 一张图片可以变出 10+ 张"不同"的图片

import nlpaug.augmenter.word as naw

# 同义词替换
aug_syn = naw.SynonymAug(aug_src='wordnet')
augmented = aug_syn.augment("患者出现发热症状")
# 可能变成："患者显现发烧症状"

# 随机删除
aug_del = naw.RandomWordAug(action='delete')
augmented = aug_del.augment("患者出现发热症状")
# 可能变成："患者发热症状"

# 随机交换
aug_swap = naw.RandomWordAug(action='swap')
augmented = aug_swap.augment("患者出现发热症状")
# 可能变成："患者症状出现发热"

from transformers import pipeline

# 使用 BERT 做掩码语言模型填充
fill_mask = pipeline("fill-mask", model="bert-base-chinese")

def contextual_augment(text, n=5):
    # 随机选择一个词进行 mask
    words = list(text)
    mask_idx = random.randint(0, len(words)-1)
    words[mask_idx] = '[MASK]'
    masked_text = ''.join(words)

    # 用 BERT 预测可能的词
    predictions = fill_mask(masked_text)
    return [pred['sequence'] for pred in predictions[:n]]

# 示例
results = contextual_augment("患者出现发热症状")
# 可能得到："患者表现出发热症状"、"患者出现发烧症状" 等

import openai

def generate_synthetic_samples(original_samples, n_per_sample=3):
    prompt = """
    你是一个医疗数据增强助手。请根据以下医疗文本，生成语义相似但表述不同的新文本。
    要求：
    1. 保持医疗语义准确
    2. 改变句式结构
    3. 替换同义词
    4. 保持专业性和准确性

    原始文本：{original_text}

    请生成 {n} 个不同的版本：
    """

    synthetic_samples = []
    for sample in original_samples:
        response = openai.ChatCompletion.create(
            model="gpt-4",
            messages=[{
                "role": "user",
                "content": prompt.format(
                    original_text=sample,
                    n=n_per_sample
                )
            }],
            temperature=0.7  # 控制多样性
        )
        synthetic_samples.extend(response.choices[0].message.content.split('\n'))

    return synthetic_samples

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

# 使用语义相似度评估增强质量
model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')

def evaluate_augmentation_quality(original, augmented):
    original_emb = model.encode([original])
    augmented_emb = model.encode(augmented)

    similarities = cosine_similarity(original_emb, augmented_emb)[0]
    return similarities

# 使用示例
original = "患者出现发热症状"
augmented = [
    "患者表现出发热症状",
    "患者有发烧的征象", 
    "发热是患者的症状表现"
]

similarities = evaluate_augmentation_quality(original, augmented)
print(similarities)
# 输出：[0.85, 0.82, 0.78] - 都在合理范围内

# 控制增强比例
def smart_augmentation(original_data, augment_ratio=0.3):
    """
    只对部分数据进行增强，控制总体数据量增长
    augment_ratio: 增强数据占总数据的比例
    """
    augment_count = int(len(original_data) * augment_ratio)
    to_augment = random.sample(original_data, augment_count)

    # 每条原始数据生成 2-3 条增强样本
    augmented_samples = []
    for sample in to_augment:
        augmented_samples.extend(augment_sample(sample, n=2))

    return original_data + augmented_samples

# 医疗领域保护词汇
PROTECTED_TERMS = {
    '心肌梗死', '心力衰竭', '高血压', '糖尿病',
    '冠状动脉', '心律失常', '脑卒中', '肺炎'
}

def safe_augmentation(text):
    # 先识别保护词汇
    tokens = tokenize(text)
    protected_positions = [i for i, token in enumerate(tokens) 
                          if token in PROTECTED_TERMS]

    # 只对非保护词汇做增强
    augmented_tokens = tokens.copy()
    for i in range(len(augmented_tokens)):
        if i not in protected_positions:
            # 应用增强
            augmented_tokens[i] = augment_word(augmented_tokens[i])

    return detokenize(augmented_tokens)

from collections import Counter

def balanced_augmentation(data, labels, target_samples_per_class=100):
    """
    确保每个类别的样本数量相对均衡
    """
    # 统计每个类别的现有样本数
    class_counts = Counter(labels)
    augmented_data = []
    augmented_labels = []

    for class_name, count in class_counts.items():
        class_data = [d for d, l in zip(data, labels) if l == class_name]

        if count < target_samples_per_class:
            # 需要增强
            needed = target_samples_per_class - count
            aug_per_sample = max(1, needed // count)

            for sample in class_data:
                augmented = augment_sample(sample, n=aug_per_sample)
                augmented_data.extend(augmented)
                augmented_labels.extend([class_name] * len(augmented))

    return data + augmented_data, labels + augmented_labels

# 实验配置
experiment_config = {
    'original_data_size': 200,
    'test_data_size': 100,
    'model': 'bert-base-chinese',
    'epochs': 10,
    'batch_size': 16
}

# 对比方案
schemes = {
    '原始数据': {
        'train_size': 200,
        'augmentation': None
    },
    '基础增强': {
        'train_size': 600,  # 每条生成3条
        'augmentation': 'nlpaug'
    },
    '上下文增强': {
        'train_size': 800,  # 每条生成4条
        'augmentation': 'bert'
    },
    '合成数据': {
        'train_size': 1000,  # 每条生成5条
        'augmentation': 'llm'
    }
}