6장: 데이터 증강 기법

import random
 
 
# 한국어 유의어 사전 (예시)
SYNONYM_DICT = {
    "빠른": ["신속한", "재빠른", "민첩한"],
    "큰": ["거대한", "방대한", "대규모의"],
    "좋은": ["우수한", "양호한", "훌륭한"],
    "만들다": ["생성하다", "구축하다", "제작하다"],
    "사용하다": ["활용하다", "이용하다", "적용하다"],
    "중요한": ["핵심적인", "필수적인", "결정적인"],
}
 
 
def synonym_replacement(
    text: str,
    replacement_ratio: float = 0.1,
    synonym_dict: dict = SYNONYM_DICT,
) -> str:
    """텍스트의 일부 단어를 동의어로 치환합니다."""
    words = text.split()
    num_replacements = max(1, int(len(words) * replacement_ratio))
 
    replaceable = [
        (i, w) for i, w in enumerate(words)
        if w in synonym_dict
    ]
 
    if not replaceable:
        return text
 
    to_replace = random.sample(
        replaceable, min(num_replacements, len(replaceable))
    )
 
    for idx, word in to_replace:
        words[idx] = random.choice(synonym_dict[word])
 
    return " ".join(words)

async def back_translate(
    text: str,
    intermediate_lang: str = "en",
    model_fn=None,
) -> str:
    """역번역으로 텍스트 변형을 생성합니다."""
    # 한국어 -> 영어
    forward_prompt = (
        f"다음 한국어 텍스트를 영어로 자연스럽게 번역하세요.\n\n"
        f"한국어: {text}\n영어:"
    )
    english = await model_fn(forward_prompt)
 
    # 영어 -> 한국어
    backward_prompt = (
        f"다음 영어 텍스트를 한국어로 자연스럽게 번역하세요.\n\n"
        f"영어: {english}\n한국어:"
    )
    korean = await model_fn(backward_prompt)
 
    return korean

import random
 
 
def random_insertion(text: str, n: int = 1) -> str:
    """랜덤 위치에 의미 없는 부사/접속사를 삽입합니다."""
    fillers = ["또한", "그리고", "특히", "실제로", "결국"]
    words = text.split()
    for _ in range(n):
        pos = random.randint(0, len(words))
        words.insert(pos, random.choice(fillers))
    return " ".join(words)
 
 
def random_deletion(text: str, p: float = 0.1) -> str:
    """각 단어를 확률 p로 삭제합니다."""
    words = text.split()
    if len(words) <= 1:
        return text
    remaining = [w for w in words if random.random() > p]
    return " ".join(remaining) if remaining else words[0]
 
 
def random_swap(text: str, n: int = 1) -> str:
    """랜덤으로 두 단어의 위치를 교환합니다."""
    words = text.split()
    for _ in range(n):
        if len(words) < 2:
            break
        i, j = random.sample(range(len(words)), 2)
        words[i], words[j] = words[j], words[i]
    return " ".join(words)

PARAPHRASE_PROMPT = """다음 텍스트를 의미를 완전히 보존하면서 다른 표현으로 바꿔 작성하세요.
 
원본: {text}
 
규칙:
- 핵심 정보와 의미를 반드시 보존합니다.
- 문장 구조, 어순, 표현을 변경합니다.
- 원본과 최소 30% 이상 표현이 달라야 합니다.
- 자연스러운 한국어를 사용합니다.
 
패러프레이즈:"""
 
 
async def generate_paraphrases(
    text: str,
    num_variants: int = 3,
    model_fn=None,
) -> list[str]:
    """LLM으로 패러프레이즈를 생성합니다."""
    variants = []
    for i in range(num_variants):
        result = await model_fn(
            PARAPHRASE_PROMPT.format(text=text),
            temperature=0.6 + i * 0.2,  # 점진적으로 다양성 증가
        )
        variants.append(result)
    return variants

STYLE_PROMPTS = {
    "formal": "격식체 학술 논문 스타일로 다시 작성하세요.",
    "casual": "친근하고 구어체적인 스타일로 다시 작성하세요.",
    "technical": "기술 전문가가 동료에게 설명하는 스타일로 다시 작성하세요.",
    "beginner": "프로그래밍 입문자가 이해할 수 있는 수준으로 다시 작성하세요.",
    "concise": "핵심만 간결하게 3문장 이내로 요약하세요.",
    "detailed": "예시와 비유를 추가하여 더 상세하게 확장하세요.",
}
 
 
async def style_transfer(
    text: str,
    target_style: str,
    model_fn=None,
) -> str:
    """텍스트를 지정된 스타일로 변환합니다."""
    prompt = (
        f"다음 텍스트를 {STYLE_PROMPTS[target_style]}\n\n"
        f"원본: {text}\n\n변환된 텍스트:"
    )
    return await model_fn(prompt, temperature=0.7)

REWRITE_PROMPT = """다음 텍스트를 주어진 조건에 맞게 재작성하세요.
 
원본: {text}
조건: {condition}
 
규칙:
- 조건에 맞게 내용을 수정합니다.
- 전체적인 구조와 맥락은 유지합니다.
- 수정된 부분이 자연스러워야 합니다.
 
재작성:"""
 
CONDITIONS = [
    "Python 대신 JavaScript로 예시를 변경",
    "초보자 관점에서 더 기초적인 설명 추가",
    "보안 관련 주의사항을 강조",
    "성능 최적화 관점에서 재구성",
    "실패 사례와 트러블슈팅 추가",
]

ADVERSARIAL_PROMPT = """다음 분류 태스크에서 AI 모델이 틀리기 쉬운
어려운 예제를 생성하세요.
 
태스크: {task_description}
카테고리: {categories}
 
생성 전략:
1. 두 카테고리의 경계에 있는 모호한 예제
2. 일반적인 패턴에 반하는 예외 사례
3. 맥락에 따라 다른 카테고리로 분류될 수 있는 예제
4. 부정, 이중부정, 반어법 등이 포함된 예제
 
각 전략별로 3개씩 생성하세요.
 
JSON 형식:
[{{
  "text": "예제 텍스트",
  "label": "정답 카테고리",
  "strategy": "사용된 전략",
  "difficulty_reason": "어려운 이유"
}}]"""

from collections import Counter
 
 
def analyze_model_errors(
    predictions: list[str],
    ground_truth: list[str],
    texts: list[str],
) -> dict:
    """모델의 오류 패턴을 분석합니다."""
    errors = []
    for pred, gt, text in zip(predictions, ground_truth, texts):
        if pred != gt:
            errors.append({
                "text": text,
                "predicted": pred,
                "actual": gt,
            })
 
    # 혼동 패턴 분석
    confusion_patterns = Counter(
        (e["actual"], e["predicted"]) for e in errors
    )
 
    return {
        "total_errors": len(errors),
        "confusion_patterns": confusion_patterns.most_common(10),
        "error_samples": errors[:20],
    }
 
 
TARGETED_GEN_PROMPT = """모델이 '{actual}' 카테고리를 '{predicted}'로
자주 혼동합니다.
 
다음은 오분류된 예시입니다:
{error_samples}
 
이 혼동 패턴을 해소할 수 있는 학습 데이터를 생성하세요.
특히 두 카테고리의 차이를 명확히 보여주는 예제를 포함합니다.
 
각 카테고리별 5개, 총 10개를 생성하세요."""

EDGE_CASE_TAXONOMY = {
    "텍스트 형식": [
        "극단적으로 짧은 입력 (1-3 단어)",
        "극단적으로 긴 입력 (1000단어 이상)",
        "특수 문자/이모티콘만으로 구성된 입력",
        "코드와 자연어가 혼재된 입력",
        "여러 언어가 혼합된 입력",
    ],
    "내용 특수성": [
        "질문 형태이지만 답변이 불가능한 입력",
        "자기 모순적인 지시",
        "맥락 없이 대명사만 사용한 입력",
        "중의적 표현이 포함된 입력",
        "시간에 민감한 정보 질문",
    ],
    "도메인 경계": [
        "두 도메인에 걸쳐 있는 질문",
        "일반 상식과 전문 지식의 경계",
        "문화/지역에 따라 답이 달라지는 질문",
    ],
}
 
 
EDGE_CASE_PROMPT = """다음 유형의 엣지 케이스 데이터를 생성하세요.
 
엣지 케이스 유형: {edge_type}
도메인: {domain}
생성 수: {count}개
 
각 예제에 대해:
1. 입력 텍스트
2. 기대되는 모델 동작
3. 이 엣지 케이스가 중요한 이유
 
를 포함하세요."""

from sklearn.model_selection import train_test_split
import numpy as np
 
 
def find_optimal_ratio(
    original_data: list[dict],
    augment_fn,
    evaluate_fn,
    ratios: list[float] = [0.5, 1.0, 2.0, 3.0, 5.0, 10.0],
) -> dict:
    """최적의 증강 비율을 탐색합니다."""
    results = {}
 
    # 원본 데이터를 train/val로 분리
    train_data, val_data = train_test_split(
        original_data, test_size=0.2, random_state=42
    )
 
    # 기준: 증강 없이 학습
    baseline_score = evaluate_fn(train_data, val_data)
    results["baseline"] = baseline_score
 
    for ratio in ratios:
        # 증강 데이터 생성
        num_augmented = int(len(train_data) * ratio)
        augmented = augment_fn(train_data, num_augmented)
 
        # 원본 + 증강으로 학습
        combined = train_data + augmented
        score = evaluate_fn(combined, val_data)
        results[f"ratio_{ratio}"] = score
 
    # 최적 비율 결정
    best_ratio = max(
        results.items(),
        key=lambda x: x[1] if x[0] != "baseline" else -1
    )
 
    return {
        "results": results,
        "best_ratio": best_ratio[0],
        "best_score": best_ratio[1],
        "improvement_over_baseline": best_ratio[1] - baseline_score,
    }

AUGMENTATION_MIX = {
    "paraphrase": 0.3,      # 30% - LLM 패러프레이징
    "back_translate": 0.2,   # 20% - 역번역
    "style_transfer": 0.15,  # 15% - 스타일 변환
    "hard_examples": 0.15,   # 15% - 어려운 예제
    "edge_cases": 0.1,       # 10% - 엣지 케이스
    "traditional": 0.1,      # 10% - 전통적 증강
}
 
 
def mixed_augmentation(
    data: list[dict],
    total_augmented: int,
    mix: dict = AUGMENTATION_MIX,
) -> list[dict]:
    """혼합 증강 전략을 적용합니다."""
    augmented = []
 
    for method, ratio in mix.items():
        count = int(total_augmented * ratio)
        # 각 기법별 증강 수행
        method_fn = get_augmentation_fn(method)
        batch = method_fn(data, count)
        augmented.extend(batch)
 
    return augmented[:total_augmented]