고품질 학습 데이터셋 준비

데이터는 미세 조정에서 가장 중요한 요소입니다

지시 따르기 형식

import json

# OpenAI fine-tuning format (JSONL)
# Each line is one training example
openai_example = {
    'messages': [
        {'role': 'system', 'content': 'You are a JSON extraction agent.'},
        {'role': 'user', 'content': 'Extract: "John Smith, age 32, from Seattle, joined 2023-01-15"'},
        {'role': 'assistant', 'content': '{"name": "John Smith", "age": 32, "city": "Seattle", "join_date": "2023-01-15"}'}
    ]
}

# Alpaca format
alpaca_example = {
    'instruction': 'Extract structured data from the following text.',
    'input': 'John Smith, age 32, from Seattle, joined 2023-01-15',
    'output': '{"name": "John Smith", "age": 32, "city": "Seattle", "join_date": "2023-01-15"}'
}

# Write as JSONL
with open('train.jsonl', 'w') as f:
    f.write(json.dumps(openai_example) + '\n')
    # Add more examples here...

학습 데이터 수집: 세 가지 전략

from openai import OpenAI

client = OpenAI()

def generate_training_example_with_gpt4o(task_description: str, example_input: str) -> dict:
    '''Use GPT-4o to generate a training example for a smaller model.'''
    prompt = f'''You are creating a training example for fine-tuning a smaller model.
Task: {task_description}

Given this input:
{example_input}

Write the ideal assistant response that demonstrates the correct behavior for this task.
Be specific, accurate, and follow the expected format precisely.'''
    
    response = client.chat.completions.create(
        model='gpt-4o',  # teacher model
        messages=[{'role': 'user', 'content': prompt}]
    )
    
    return {
        'messages': [
            {'role': 'system', 'content': task_description},
            {'role': 'user', 'content': example_input},
            {'role': 'assistant', 'content': response.choices[0].message.content}
        ]
    }

품질 필터링 및 검증

import json

def validate_training_example(example: dict, schema_validator=None) -> dict:
    issues = []
    
    # Format check
    if 'messages' not in example:
        issues.append('Missing messages field')
        return {'valid': False, 'issues': issues}
    
    messages = example['messages']
    if not any(m['role'] == 'assistant' for m in messages):
        issues.append('No assistant message found')
    
    # Check assistant message quality
    assistant_content = next((m['content'] for m in messages if m['role'] == 'assistant'), '')
    
    if len(assistant_content) < 5:
        issues.append('Assistant response too short')
    
    # Schema validation for JSON output tasks
    if schema_validator:
        try:
            parsed = json.loads(assistant_content)
            schema_validator(parsed)  # raises if invalid
        except json.JSONDecodeError:
            issues.append('Assistant response is not valid JSON')
        except Exception as e:
            issues.append(f'Schema validation failed: {str(e)}')
    
    return {'valid': len(issues) == 0, 'issues': issues}

# Run validation on all examples before training
examples = load_training_examples('raw_dataset.jsonl')
valid_examples = [e for e in examples if validate_training_example(e)['valid']]
print(f'Valid examples: {len(valid_examples)}/{len(examples)}')

데이터 중복 제거

import hashlib
from datasketch import MinHash, MinHashLSH

def exact_deduplicate(examples: list[dict]) -> list[dict]:
    seen_hashes = set()
    unique = []
    
    for ex in examples:
        # Hash the user and assistant messages
        content = str(ex['messages'])
        h = hashlib.md5(content.encode()).hexdigest()
        if h not in seen_hashes:
            seen_hashes.add(h)
            unique.append(ex)
    
    print(f'Exact dedup: {len(examples)} -> {len(unique)} ({len(examples)-len(unique)} removed)')
    return unique

def near_deduplicate_by_input(examples: list[dict], similarity_threshold=0.85) -> list[dict]:
    # Build index of input texts
    inputs = [next((m['content'] for m in ex['messages'] if m['role'] == 'user'), '') for ex in examples]
    
    lsh = MinHashLSH(threshold=similarity_threshold, num_perm=128)
    unique_indices = set()
    
    for i, text in enumerate(inputs):
        m = MinHash(num_perm=128)
        for word in text.lower().split():
            m.update(word.encode('utf-8'))
        if not lsh.query(m):  # no similar items found
            lsh.insert(str(i), m)
            unique_indices.add(i)
    
    return [examples[i] for i in sorted(unique_indices)]

학습/검증 분할

import random
import json

def split_dataset(examples: list[dict], val_fraction=0.1, seed=42) -> tuple[list, list]:
    random.seed(seed)  # reproducible split
    shuffled = examples.copy()
    random.shuffle(shuffled)
    
    n_val = max(1, int(len(shuffled) * val_fraction))
    val_set = shuffled[:n_val]
    train_set = shuffled[n_val:]
    
    print(f'Train: {len(train_set)} examples, Validation: {len(val_set)} examples')
    return train_set, val_set

def save_jsonl(examples: list[dict], path: str):
    with open(path, 'w') as f:
        for ex in examples:
            f.write(json.dumps(ex) + '\n')

# Split and save
examples = load_training_examples('clean_dataset.jsonl')
train, val = split_dataset(examples, val_fraction=0.1)
save_jsonl(train, 'train.jsonl')
save_jsonl(val, 'validation.jsonl')
print(f'Saved train.jsonl ({len(train)}) and validation.jsonl ({len(val)})')

데이터셋 균형 맞추기

from collections import Counter
import random

def analyze_distribution(examples: list[dict], category_extractor) -> dict:
    categories = [category_extractor(ex) for ex in examples]
    counts = Counter(categories)
    print('Category distribution:')
    for cat, count in counts.most_common():
        print(f'  {cat}: {count} ({100*count/len(examples):.1f}%)')
    return counts

def oversample_minority(examples: list[dict], category_extractor, target_count: int) -> list[dict]:
    by_category = {}
    for ex in examples:
        cat = category_extractor(ex)
        by_category.setdefault(cat, []).append(ex)
    
    balanced = []
    for cat, cat_examples in by_category.items():
        if len(cat_examples) < target_count:
            # Oversample with replacement
            oversampled = random.choices(cat_examples, k=target_count)
            balanced.extend(oversampled)
        else:
            # Undersample to target_count
            balanced.extend(random.sample(cat_examples, target_count))
    
    random.shuffle(balanced)
    return balanced

데이터 정리 및 정규화

import json
import re

def normalize_json_output_example(example: dict) -> dict:
    '''Normalize JSON output in assistant messages for consistency.'''
    messages = example.get('messages', [])
    normalized = []
    
    for msg in messages:
        if msg['role'] == 'assistant':
            content = msg['content'].strip()
            
            # Try to parse and re-serialize JSON for consistent formatting
            try:
                parsed = json.loads(content)
                # Normalize: sort keys, consistent spacing
                content = json.dumps(parsed, ensure_ascii=False, sort_keys=True)
            except json.JSONDecodeError:
                pass  # Not JSON output - leave as is
            
            normalized.append({'role': 'assistant', 'content': content})
        else:
            normalized.append(msg)
    
    return {'messages': normalized}

def normalize_dataset(examples: list[dict]) -> list[dict]:
    return [normalize_json_output_example(ex) for ex in examples]

데이터셋 품질 지표 측정

import tiktoken

def analyze_dataset_quality(examples: list[dict], model='gpt-4o-mini') -> dict:
    encoder = tiktoken.encoding_for_model(model)
    
    token_counts = []
    for ex in examples:
        total_tokens = sum(
            len(encoder.encode(m['content']))
            for m in ex['messages']
        )
        token_counts.append(total_tokens)
    
    report = {
        'total_examples': len(examples),
        'avg_tokens_per_example': sum(token_counts) / len(token_counts),
        'max_tokens': max(token_counts),
        'min_tokens': min(token_counts),
        'examples_over_4k_tokens': sum(1 for t in token_counts if t > 4096),
        'estimated_training_tokens': sum(token_counts),
        'estimated_cost': sum(token_counts) / 1_000_000 * 8.0  # ~$8/1M tokens for gpt-4o-mini
    }
    
    for key, value in report.items():
        print(f'{key}: {value}')
    return report

반복적인 데이터셋 개선

예시 전체에서 시스템 프롬프트의 일관성

빠른 확인

레슨 요약