5장: 데이터 품질 검증과 필터링 파이프라인

import numpy as np
from scipy import stats
from scipy.spatial.distance import jensenshannon
 
 
def compute_fidelity_metrics(
    real: np.ndarray, synthetic: np.ndarray
) -> dict:
    """충실도 메트릭을 계산합니다."""
    metrics = {}
 
    # KS 검정: 두 분포의 최대 차이
    ks_stat, ks_pvalue = stats.ks_2samp(real, synthetic)
    metrics["ks_statistic"] = ks_stat
    metrics["ks_pvalue"] = ks_pvalue
 
    # JS 발산: 두 분포의 대칭적 거리 (0~1)
    # 히스토그램으로 이산화
    bins = np.linspace(
        min(real.min(), synthetic.min()),
        max(real.max(), synthetic.max()),
        50,
    )
    real_hist, _ = np.histogram(real, bins=bins, density=True)
    synth_hist, _ = np.histogram(synthetic, bins=bins, density=True)
 
    # 0 방지를 위한 스무딩
    real_hist = real_hist + 1e-10
    synth_hist = synth_hist + 1e-10
 
    js_div = jensenshannon(real_hist, synth_hist)
    metrics["js_divergence"] = js_div
 
    # 기본 통계량 비교
    metrics["mean_diff"] = abs(real.mean() - synthetic.mean())
    metrics["std_diff"] = abs(real.std() - synthetic.std())
    metrics["median_diff"] = abs(
        np.median(real) - np.median(synthetic)
    )
 
    return metrics

from sklearn.model_selection import cross_val_score
from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.linear_model import LogisticRegression
import numpy as np
 
 
def tstr_evaluation(
    synthetic_X: np.ndarray,
    synthetic_y: np.ndarray,
    real_X: np.ndarray,
    real_y: np.ndarray,
) -> dict:
    """TSTR 평가를 수행합니다."""
    classifiers = {
        "logistic_regression": LogisticRegression(max_iter=1000),
        "random_forest": RandomForestClassifier(n_estimators=100),
        "gradient_boosting": GradientBoostingClassifier(
            n_estimators=100
        ),
    }
 
    results = {}
 
    for name, clf in classifiers.items():
        # 합성 데이터로 학습
        clf.fit(synthetic_X, synthetic_y)
 
        # 실제 데이터로 평가
        score = clf.score(real_X, real_y)
        results[f"tstr_{name}"] = score
 
    # 비교: 실제 데이터로 학습했을 때의 성능 (TRTR)
    for name, clf in classifiers.items():
        scores = cross_val_score(clf, real_X, real_y, cv=5)
        results[f"trtr_{name}"] = scores.mean()
 
    # TSTR / TRTR 비율 계산
    for name in classifiers:
        tstr = results[f"tstr_{name}"]
        trtr = results[f"trtr_{name}"]
        results[f"ratio_{name}"] = tstr / trtr if trtr > 0 else 0
 
    return results

from sklearn.neighbors import NearestNeighbors
import numpy as np
 
 
def distance_to_closest_record(
    synthetic: np.ndarray, real: np.ndarray
) -> dict:
    """합성 데이터와 실제 데이터 간 최근접 거리를 계산합니다."""
    nn = NearestNeighbors(n_neighbors=1, metric="euclidean")
    nn.fit(real)
 
    distances, _ = nn.kneighbors(synthetic)
    distances = distances.flatten()
 
    return {
        "min_distance": distances.min(),
        "mean_distance": distances.mean(),
        "median_distance": np.median(distances),
        "pct_below_threshold": (distances < 0.01).mean(),
    }

JUDGE_PROMPT = """당신은 AI 학습 데이터의 품질을 평가하는 전문가입니다.
다음 지시-응답 쌍의 품질을 5개 차원에서 1~5점으로 평가하세요.
 
지시문: {instruction}
응답문: {response}
 
평가 차원:
1. 정확성(Accuracy): 정보가 사실에 부합하는가
2. 완전성(Completeness): 지시에 충분히 답변했는가
3. 명확성(Clarity): 설명이 이해하기 쉬운가
4. 관련성(Relevance): 지시와 응답이 일관되는가
5. 유용성(Helpfulness): 실제 사용자에게 도움이 되는가
 
각 차원에 대해 점수와 간단한 근거를 제공하세요.
 
JSON 형식으로 출력:
{{
  "accuracy": {{"score": N, "reason": "..."}},
  "completeness": {{"score": N, "reason": "..."}},
  "clarity": {{"score": N, "reason": "..."}},
  "relevance": {{"score": N, "reason": "..."}},
  "helpfulness": {{"score": N, "reason": "..."}},
  "overall": N,
  "recommendation": "accept/revise/reject"
}}"""
 
 
import json
from dataclasses import dataclass
 
 
@dataclass
class QualityScore:
    accuracy: float
    completeness: float
    clarity: float
    relevance: float
    helpfulness: float
    overall: float
    recommendation: str
 
 
async def evaluate_with_llm_judge(
    instruction: str,
    response: str,
    model_fn,
) -> QualityScore:
    """LLM-as-Judge로 데이터 품질을 평가합니다."""
    prompt = JUDGE_PROMPT.format(
        instruction=instruction,
        response=response,
    )
 
    result = await model_fn(prompt)
    parsed = json.loads(result)
 
    return QualityScore(
        accuracy=parsed["accuracy"]["score"],
        completeness=parsed["completeness"]["score"],
        clarity=parsed["clarity"]["score"],
        relevance=parsed["relevance"]["score"],
        helpfulness=parsed["helpfulness"]["score"],
        overall=parsed["overall"],
        recommendation=parsed["recommendation"],
    )

import hashlib
from collections import defaultdict
 
 
def exact_dedup(data: list[dict], key: str = "text") -> list[dict]:
    """정확한 중복을 해시 기반으로 제거합니다."""
    seen_hashes: set[str] = set()
    unique_data: list[dict] = []
 
    for item in data:
        text_hash = hashlib.md5(
            item[key].encode("utf-8")
        ).hexdigest()
        if text_hash not in seen_hashes:
            seen_hashes.add(text_hash)
            unique_data.append(item)
 
    return unique_data

from datasketch import MinHash, MinHashLSH
 
 
def near_dedup_minhash(
    data: list[dict],
    key: str = "text",
    threshold: float = 0.8,
    num_perm: int = 128,
) -> list[dict]:
    """MinHash LSH로 유사 중복을 제거합니다."""
    lsh = MinHashLSH(threshold=threshold, num_perm=num_perm)
    minhashes: dict[int, MinHash] = {}
 
    # MinHash 생성 및 LSH에 삽입
    for idx, item in enumerate(data):
        mh = MinHash(num_perm=num_perm)
        # 3-gram 기반 shingling
        text = item[key]
        for i in range(len(text) - 2):
            shingle = text[i:i + 3]
            mh.update(shingle.encode("utf-8"))
        minhashes[idx] = mh
 
        try:
            lsh.insert(str(idx), mh)
        except ValueError:
            pass  # 이미 유사한 항목이 존재
 
    # 중복 그룹에서 대표만 유지
    keep_indices: set[int] = set()
    visited: set[int] = set()
 
    for idx in range(len(data)):
        if idx in visited:
            continue
        keep_indices.add(idx)
        # 유사 항목 탐색
        result = lsh.query(minhashes[idx])
        for r in result:
            visited.add(int(r))
 
    return [data[i] for i in sorted(keep_indices)]

import numpy as np
from sklearn.cluster import KMeans
 
 
def diversity_sampling(
    embeddings: np.ndarray,
    data: list[dict],
    target_size: int,
    n_clusters: int = 50,
) -> list[dict]:
    """클러스터링 기반 다양성 샘플링을 수행합니다."""
    # K-Means 클러스터링
    kmeans = KMeans(n_clusters=n_clusters, random_state=42)
    labels = kmeans.fit_predict(embeddings)
 
    # 각 클러스터에서 비례 샘플링
    samples_per_cluster = max(1, target_size // n_clusters)
    selected_indices: list[int] = []
 
    for cluster_id in range(n_clusters):
        cluster_indices = np.where(labels == cluster_id)[0]
 
        if len(cluster_indices) <= samples_per_cluster:
            selected_indices.extend(cluster_indices.tolist())
        else:
            # 클러스터 중심에서 가장 먼 것부터 선택 (다양성 최대화)
            center = kmeans.cluster_centers_[cluster_id]
            distances = np.linalg.norm(
                embeddings[cluster_indices] - center, axis=1
            )
            sorted_idx = np.argsort(-distances)
            selected = cluster_indices[
                sorted_idx[:samples_per_cluster]
            ]
            selected_indices.extend(selected.tolist())
 
    return [data[i] for i in selected_indices[:target_size]]

from dataclasses import dataclass
from enum import Enum
 
 
class FilterResult(Enum):
    ACCEPT = "accept"
    REVISE = "revise"
    REJECT = "reject"
 
 
@dataclass
class PipelineConfig:
    min_length: int = 50
    max_length: int = 5000
    min_quality_score: float = 3.5
    dedup_threshold: float = 0.8
    diversity_target: int = 10000
    enable_human_review: bool = True
    human_sample_rate: float = 0.05
 
 
class QualityPipeline:
    """합성 데이터 품질 검증 통합 파이프라인"""
 
    def __init__(self, config: PipelineConfig):
        self.config = config
        self.stats = {
            "total": 0,
            "format_filtered": 0,
            "quality_filtered": 0,
            "dedup_filtered": 0,
            "final": 0,
        }
 
    def run(self, data: list[dict]) -> list[dict]:
        """전체 파이프라인을 실행합니다."""
        self.stats["total"] = len(data)
 
        # 1단계: 형식 필터링
        data = self._format_filter(data)
        self.stats["format_filtered"] = (
            self.stats["total"] - len(data)
        )
 
        # 2단계: 품질 평가 및 필터링
        data = self._quality_filter(data)
        self.stats["quality_filtered"] = (
            self.stats["total"]
            - self.stats["format_filtered"]
            - len(data)
        )
 
        # 3단계: 중복 제거
        data = self._dedup_filter(data)
 
        # 4단계: 다양성 샘플링
        if len(data) > self.config.diversity_target:
            data = self._diversity_sample(data)
 
        self.stats["final"] = len(data)
        return data
 
    def _format_filter(self, data: list[dict]) -> list[dict]:
        """형식 기반 필터링"""
        filtered = []
        for item in data:
            text = item.get("response", item.get("text", ""))
            if self.config.min_length <= len(text) <= self.config.max_length:
                filtered.append(item)
        return filtered
 
    def _quality_filter(self, data: list[dict]) -> list[dict]:
        """LLM-as-Judge 기반 품질 필터링"""
        # 실제 구현에서는 LLM-as-Judge 호출
        return [
            item for item in data
            if item.get("quality_score", 5) >= self.config.min_quality_score
        ]
 
    def _dedup_filter(self, data: list[dict]) -> list[dict]:
        """중복 제거"""
        data = exact_dedup(data, key="response")
        data = near_dedup_minhash(
            data,
            key="response",
            threshold=self.config.dedup_threshold,
        )
        return data
 
    def _diversity_sample(self, data: list[dict]) -> list[dict]:
        """다양성 기반 샘플링"""
        # 임베딩 생성 후 다양성 샘플링
        return data[:self.config.diversity_target]
 
    def report(self) -> str:
        """파이프라인 실행 결과를 보고합니다."""
        return (
            f"총 입력: {self.stats['total']}\n"
            f"형식 필터: -{self.stats['format_filtered']}\n"
            f"품질 필터: -{self.stats['quality_filtered']}\n"
            f"중복 제거: -{self.stats['dedup_filtered']}\n"
            f"최종 출력: {self.stats['final']}\n"
            f"통과율: "
            f"{self.stats['final'] / self.stats['total'] * 100:.1f}%"
        )