4장: 구조화된 데이터와 멀티모달 합성

import pandas as pd
import numpy as np
from sdv.single_table import CTGANSynthesizer
from sdv.metadata import SingleTableMetadata
 
 
def synthesize_tabular_data(
    real_data: pd.DataFrame,
    num_samples: int,
    epochs: int = 300,
) -> pd.DataFrame:
    """CTGAN을 사용하여 테이블 데이터를 합성합니다."""
    # 메타데이터 자동 감지
    metadata = SingleTableMetadata()
    metadata.detect_from_dataframe(real_data)
 
    # CTGAN 모델 학습
    synthesizer = CTGANSynthesizer(
        metadata,
        epochs=epochs,
        batch_size=500,
        verbose=True,
    )
    synthesizer.fit(real_data)
 
    # 합성 데이터 생성
    synthetic_data = synthesizer.sample(num_rows=num_samples)
 
    return synthetic_data
 
 
def validate_statistical_properties(
    real: pd.DataFrame,
    synthetic: pd.DataFrame,
) -> dict:
    """합성 데이터의 통계적 특성을 검증합니다."""
    results = {}
 
    for col in real.columns:
        if real[col].dtype in [np.float64, np.int64]:
            # 수치형: KS 검정
            from scipy import stats
            ks_stat, p_value = stats.ks_2samp(real[col], synthetic[col])
            results[col] = {
                "ks_statistic": ks_stat,
                "p_value": p_value,
                "mean_diff": abs(
                    real[col].mean() - synthetic[col].mean()
                ),
                "std_diff": abs(
                    real[col].std() - synthetic[col].std()
                ),
            }
        else:
            # 범주형: 분포 비교
            real_dist = real[col].value_counts(normalize=True)
            synth_dist = synthetic[col].value_counts(normalize=True)
            results[col] = {
                "distribution_match": real_dist.to_dict(),
                "synthetic_distribution": synth_dist.to_dict(),
            }
 
    # 상관관계 보존 검증
    numeric_cols = real.select_dtypes(include=[np.number]).columns
    if len(numeric_cols) > 1:
        real_corr = real[numeric_cols].corr()
        synth_corr = synthetic[numeric_cols].corr()
        corr_diff = (real_corr - synth_corr).abs().mean().mean()
        results["correlation_preservation"] = {
            "mean_absolute_diff": corr_diff
        }
 
    return results

import json
from jsonschema import validate, ValidationError
 
USER_SCHEMA = {
    "type": "object",
    "properties": {
        "id": {"type": "integer", "minimum": 1},
        "name": {"type": "string", "minLength": 2, "maxLength": 50},
        "email": {"type": "string", "format": "email"},
        "age": {"type": "integer", "minimum": 18, "maximum": 100},
        "role": {"type": "string", "enum": ["admin", "user", "editor"]},
        "preferences": {
            "type": "object",
            "properties": {
                "language": {"type": "string"},
                "theme": {
                    "type": "string",
                    "enum": ["light", "dark"],
                },
                "notifications": {"type": "boolean"},
            },
        },
    },
    "required": ["id", "name", "email", "age", "role"],
}
 
JSON_GEN_PROMPT = """다음 JSON Schema에 맞는 현실적인 데이터 {count}개를 생성하세요.
 
Schema:
{schema}
 
규칙:
- 모든 필드가 스키마 제약 조건을 만족해야 합니다.
- 이메일, 이름 등은 현실적이지만 실제 존재하지 않는 값을 사용합니다.
- 다양한 값의 조합을 포함합니다.
- JSON 배열 형식으로 출력합니다."""
 
 
def generate_json_data(
    schema: dict, count: int, model_fn
) -> list[dict]:
    """JSON Schema에 맞는 합성 데이터를 생성합니다."""
    prompt = JSON_GEN_PROMPT.format(
        schema=json.dumps(schema, indent=2, ensure_ascii=False),
        count=count,
    )
 
    response = model_fn(prompt)
    data = json.loads(response)
 
    # 스키마 검증
    validated = []
    for item in data:
        try:
            validate(instance=item, schema=schema)
            validated.append(item)
        except ValidationError:
            pass  # 스키마 위반 데이터 폐기
 
    return validated

SQL_GEN_PROMPT = """다음 데이터베이스 스키마에 대해 자연어 질문과
SQL 쿼리 쌍을 생성하세요.
 
데이터베이스 스키마:
{db_schema}
 
난이도: {difficulty}
생성 수: {count}개
 
난이도 기준:
- 초급: 단일 테이블, 기본 SELECT/WHERE
- 중급: JOIN, GROUP BY, HAVING, 서브쿼리
- 고급: 윈도우 함수, CTE, 복합 서브쿼리, PIVOT
 
JSON 형식:
[
  {{
    "question": "자연어 질문",
    "sql": "SELECT ...",
    "difficulty": "난이도",
    "tables_used": ["테이블명"]
  }}
]"""
 
 
DB_SCHEMA_EXAMPLE = """
CREATE TABLE employees (
    id INT PRIMARY KEY,
    name VARCHAR(100),
    department_id INT,
    salary DECIMAL(10, 2),
    hire_date DATE,
    FOREIGN KEY (department_id) REFERENCES departments(id)
);
 
CREATE TABLE departments (
    id INT PRIMARY KEY,
    name VARCHAR(50),
    location VARCHAR(100)
);
 
CREATE TABLE projects (
    id INT PRIMARY KEY,
    name VARCHAR(100),
    department_id INT,
    budget DECIMAL(12, 2),
    start_date DATE,
    end_date DATE,
    FOREIGN KEY (department_id) REFERENCES departments(id)
);
"""

import sqlite3
 
 
def validate_sql_query(
    sql: str, db_schema: str, sample_data: str
) -> tuple[bool, str]:
    """SQL 쿼리의 유효성을 실행으로 검증합니다."""
    conn = sqlite3.connect(":memory:")
    cursor = conn.cursor()
 
    try:
        # 스키마 생성
        cursor.executescript(db_schema)
        # 샘플 데이터 삽입
        cursor.executescript(sample_data)
        # 쿼리 실행
        cursor.execute(sql)
        results = cursor.fetchall()
        return True, f"결과 {len(results)}행"
    except Exception as e:
        return False, str(e)
    finally:
        conn.close()

CAPTION_GEN_PROMPT = """다음 카테고리에 해당하는 이미지 캡션을 생성하세요.
 
카테고리: {category}
세부 주제: {subtopic}
스타일: {caption_style}
 
규칙:
- 시각적 요소를 구체적으로 묘사합니다.
- 색상, 구도, 개체 위치 등을 포함합니다.
- 이미지 생성 모델이 정확히 재현할 수 있을 정도로 상세합니다.
- 캡션 길이는 50~150 단어입니다.
 
생성할 캡션 수: {count}개"""
 
 
CAPTION_STYLES = {
    "descriptive": "객관적이고 상세한 묘사",
    "narrative": "이야기를 담은 서술",
    "technical": "기술적 특성 중심 설명",
    "alt_text": "접근성을 위한 대체 텍스트",
}
 
 
def generate_image_caption_pairs(
    categories: list[dict],
    caption_style: str = "descriptive",
    model_fn=None,
    image_gen_fn=None,
) -> list[dict]:
    """이미지-캡션 쌍을 합성 생성합니다."""
    pairs = []
 
    for cat in categories:
        # 1단계: 캡션 생성
        captions = model_fn(
            CAPTION_GEN_PROMPT.format(
                category=cat["category"],
                subtopic=cat["subtopic"],
                caption_style=CAPTION_STYLES[caption_style],
                count=cat["count"],
            )
        )
 
        # 2단계: 각 캡션에 대해 이미지 생성
        for caption in captions:
            image = image_gen_fn(caption)
            pairs.append({
                "caption": caption,
                "image_path": image,
                "category": cat["category"],
                "style": caption_style,
            })
 
    return pairs