Scale AI와 데이터 라벨링의 세계: AI 훈련 데이터 산업의 모든 것과 커리어 가이드

Scale AI 사업 구조
├── Data Engine (핵심 사업)
│   ├── 이미지/비디오 라벨링 (자율주행, 로보틱스)
│   ├── 텍스트 라벨링 (NLP, LLM)
│   ├── 3D 포인트 클라우드 (LiDAR)
│   └── RLHF 데이터 (LLM 정렬)
├── Government (정부 사업)
│   ├── 미 국방부 계약
│   ├── 위성 이미지 분석
│   └── 정보 분석 지원
├── Generative AI Platform
│   ├── LLM 평가 (Model Evaluation)
│   ├── Fine-tuning 데이터
│   └── Safety 데이터 (유해성 분류)
└── Enterprise Solutions
    ├── 맞춤형 파이프라인
    ├── 품질 관리 도구
    └── 분석 대시보드

{
  "label": "car",
  "bbox": {
    "x_min": 120,
    "y_min": 80,
    "x_max": 350,
    "y_max": 240
  },
  "confidence": 0.95
}

# Panoptic Segmentation 라벨 예시
panoptic_label = {
    "segments": [
        {"id": 1, "category": "road", "is_thing": False},      # stuff (배경)
        {"id": 2, "category": "sky", "is_thing": False},        # stuff
        {"id": 3, "category": "car", "is_thing": True, "instance_id": 1},  # thing
        {"id": 4, "category": "car", "is_thing": True, "instance_id": 2},  # thing
        {"id": 5, "category": "person", "is_thing": True, "instance_id": 1}
    ]
}

"[Apple:ORG]의 CEO [팀 쿡:PERSON]이 [쿠퍼티노:LOC]에서
 신제품을 발표했다."

# 3D Bounding Box 라벨
lidar_annotation = {
    "label": "vehicle",
    "center": {"x": 15.2, "y": -3.4, "z": 0.8},
    "dimensions": {"length": 4.5, "width": 1.8, "height": 1.5},
    "rotation": {"yaw": 0.35, "pitch": 0.0, "roll": 0.0},
    "num_points": 342,
    "tracking_id": "veh_0042",
    "attributes": {
        "vehicle_type": "sedan",
        "occlusion": "partial",
        "truncation": 0.0
    }
}

프롬프트: "양자역학을 초등학생에게 설명해줘"

응답 A: "양자역학은 아주 작은 세계의 규칙이야..."
응답 B: "양자역학이란 원자보다 작은 입자들의..."

평가: A > B (이유: 더 쉬운 비유 사용, 연령에 적합한 어휘)

RLHF 파이프라인 5단계
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

Step 1: 프롬프트 수집
  └─ 다양성 확보 (주제, 난이도, 언어, 문화)
  └─ 안전성 테스트 프롬프트 포함
  └─ Red-teaming 프롬프트 포함

Step 2: AI 응답 생성
  └─ 동일 프롬프트에 복수 응답 생성 (보통 2-4개)
  └─ 다른 temperature/sampling 설정 사용
  └─ 다른 모델 버전 사용 가능

Step 3: 인간 평가
  └─ Comparison: A vs B 선택
  └─ Rating: 유용성, 정확성, 안전성 각각 점수
  └─ Correction: 직접 수정하여 "gold response" 생성

Step 4: Reward Model 학습
  └─ 인간 선호 데이터로 보상 함수 학습
  └─ 브래들리-테리 모델 기반

Step 5: 정책 최적화
  └─ PPO (Proximal Policy Optimization) 또는
  └─ DPO (Direct Preference Optimization) 사용

Scale AI의 차별점
━━━━━━━━━━━━━━━━━━━━━━━━

강점:
  ✓ 최대 규모의 숙련된 라벨러 네트워크
  ✓ 정부/국방 보안 인증 (FedRAMP)
  ✓ 3D 포인트 클라우드 업계 최고
  ✓ RLHF 데이터 파이프라인 검증됨
  ✓ 자동 품질 관리 시스템

약점:
  ✗ 높은 가격 (소규모 팀에 부담)
  ✗ 최소 계약 규모 존재
  ✗ 셀프서비스 옵션 제한적
  ✗ 커스터마이징에 시간 소요

# Labelbox Python SDK 예시
import labelbox as lb

client = lb.Client(api_key="YOUR_API_KEY")
project = client.create_project(name="Object Detection v2")

# 데이터셋 연결
dataset = client.create_dataset(name="street_images_2025")

# 온톨로지(라벨 스키마) 정의
ontology_builder = lb.OntologyBuilder(
    tools=[
        lb.Tool(tool=lb.Tool.Type.BBOX, name="Vehicle"),
        lb.Tool(tool=lb.Tool.Type.BBOX, name="Pedestrian"),
        lb.Tool(tool=lb.Tool.Type.POLYGON, name="Road"),
        lb.Tool(tool=lb.Tool.Type.SEGMENTATION, name="Sidewalk"),
    ],
    classifications=[
        lb.Classification(
            class_type=lb.Classification.Type.RADIO,
            name="Weather",
            options=[
                lb.Option(value="sunny"),
                lb.Option(value="rainy"),
                lb.Option(value="cloudy"),
            ]
        )
    ]
)

from snorkel.labeling import labeling_function, PandasLFApplier
from snorkel.labeling.model import LabelModel

# 라벨링 함수 정의
@labeling_function()
def lf_keyword_positive(record):
    """긍정 키워드가 포함되면 POSITIVE"""
    positive_words = ["great", "excellent", "amazing", "love"]
    if any(w in record.text.lower() for w in positive_words):
        return 1  # POSITIVE
    return -1  # ABSTAIN

@labeling_function()
def lf_keyword_negative(record):
    """부정 키워드가 포함되면 NEGATIVE"""
    negative_words = ["terrible", "awful", "hate", "worst"]
    if any(w in record.text.lower() for w in negative_words):
        return 0  # NEGATIVE
    return -1  # ABSTAIN

@labeling_function()
def lf_short_review(record):
    """짧은 리뷰는 부정적인 경향"""
    if len(record.text.split()) < 5:
        return 0  # NEGATIVE
    return -1  # ABSTAIN

# 라벨 모델로 노이즈 있는 라벨 통합
applier = PandasLFApplier(lfs=[
    lf_keyword_positive,
    lf_keyword_negative,
    lf_short_review
])
L_train = applier.apply(df_train)

label_model = LabelModel(cardinality=2, verbose=True)
label_model.fit(L_train, n_epochs=500, seed=42)
predictions = label_model.predict(L_train)

# 설치 및 실행
pip install label-studio
label-studio start

# Docker로 실행
docker run -it -p 8080:8080 \
  -v label-studio-data:/label-studio/data \
  heartexlabs/label-studio:latest

class QualityMonitor:
    """라벨링 품질 모니터링 시스템"""

    def __init__(self, golden_set_ratio=0.05):
        self.golden_set_ratio = golden_set_ratio
        self.annotator_scores = {}

    def inject_golden_items(self, task_batch, golden_items):
        """작업 배치에 골든 아이템을 무작위로 삽입"""
        import random
        n_golden = max(1, int(len(task_batch) * self.golden_set_ratio))
        selected_golden = random.sample(golden_items, min(n_golden, len(golden_items)))

        mixed_batch = task_batch.copy()
        for item in selected_golden:
            pos = random.randint(0, len(mixed_batch))
            mixed_batch.insert(pos, {**item, "_is_golden": True})
        return mixed_batch

    def evaluate_annotator(self, annotator_id, submissions):
        """라벨러의 골든 아이템 정확도 평가"""
        golden_results = [s for s in submissions if s.get("_is_golden")]
        if not golden_results:
            return None

        correct = sum(
            1 for s in golden_results
            if s["submitted_label"] == s["golden_label"]
        )
        accuracy = correct / len(golden_results)
        self.annotator_scores[annotator_id] = accuracy

        if accuracy < 0.80:
            self._flag_for_retraining(annotator_id)
        return accuracy

from sklearn.metrics import cohen_kappa_score
import numpy as np

def compute_cohens_kappa(annotator1_labels, annotator2_labels):
    """두 라벨러 간의 Cohen's Kappa 계산"""
    kappa = cohen_kappa_score(annotator1_labels, annotator2_labels)
    # 해석 기준:
    # < 0.20: 거의 일치하지 않음 (Poor)
    # 0.21-0.40: 약한 일치 (Fair)
    # 0.41-0.60: 보통 일치 (Moderate)
    # 0.61-0.80: 상당한 일치 (Substantial)
    # 0.81-1.00: 거의 완벽한 일치 (Almost Perfect)
    return kappa

def compute_fleiss_kappa(rating_matrix):
    """3명 이상의 라벨러에 대한 Fleiss' Kappa 계산"""
    n_items, n_categories = rating_matrix.shape
    n_raters = rating_matrix.sum(axis=1)[0]

    # 각 아이템별 일치도
    p_i = (np.sum(rating_matrix ** 2, axis=1) - n_raters) / (n_raters * (n_raters - 1))
    p_bar = np.mean(p_i)

    # 우연에 의한 일치도
    p_j = np.sum(rating_matrix, axis=0) / (n_items * n_raters)
    p_e = np.sum(p_j ** 2)

    # Fleiss' Kappa
    kappa = (p_bar - p_e) / (1 - p_e)
    return kappa

품질 관리 워크플로
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

1. 동일 데이터를 3명의 라벨러에게 배정
2. 3명 모두 일치 → 해당 라벨 채택
3. 2명 일치, 1명 불일치 → 다수결 채택 + 불일치자 검토
4. 3명 모두 불일치 → 전문가 중재자에게 에스컬레이션
5. 반복적 불일치 → 가이드라인 업데이트 필요 신호

class AnomalyDetector:
    """비정상적인 라벨링 패턴을 탐지하는 시스템"""

    def detect_speed_anomaly(self, annotator_id, task_times):
        """비정상적으로 빠른 라벨링 탐지 (랜덤 클릭 의심)"""
        median_time = np.median(task_times)
        threshold = median_time * 0.3  # 중앙값의 30% 미만은 의심

        suspicious_count = sum(1 for t in task_times if t < threshold)
        if suspicious_count / len(task_times) > 0.2:
            return {"status": "flagged", "reason": "speed_anomaly"}
        return {"status": "ok"}

    def detect_pattern_anomaly(self, annotator_id, labels):
        """동일 라벨 반복 패턴 탐지"""
        from collections import Counter
        counter = Counter(labels)
        most_common_ratio = counter.most_common(1)[0][1] / len(labels)

        if most_common_ratio > 0.85:  # 85% 이상 동일 라벨
            return {"status": "flagged", "reason": "pattern_anomaly"}
        return {"status": "ok"}

class PreLabelingPipeline:
    """사전 라벨링 파이프라인"""

    def __init__(self, model, confidence_threshold=0.85):
        self.model = model
        self.confidence_threshold = confidence_threshold

    def pre_label(self, data_batch):
        """모델로 1차 라벨링 후 신뢰도별 분류"""
        results = []
        for item in data_batch:
            prediction = self.model.predict(item)
            confidence = prediction["confidence"]

            if confidence >= self.confidence_threshold:
                # 높은 신뢰도: 자동 승인 후 샘플링 검수
                results.append({
                    "item": item,
                    "label": prediction["label"],
                    "route": "auto_approve",
                    "confidence": confidence
                })
            elif confidence >= 0.5:
                # 중간 신뢰도: 사람이 검수 (pre-label 참고)
                results.append({
                    "item": item,
                    "suggested_label": prediction["label"],
                    "route": "human_review",
                    "confidence": confidence
                })
            else:
                # 낮은 신뢰도: 처음부터 사람이 라벨링
                results.append({
                    "item": item,
                    "route": "human_label",
                    "confidence": confidence
                })
        return results

import numpy as np

class ActiveLearningSelector:
    """능동 학습 샘플 선택기"""

    def uncertainty_sampling(self, model, unlabeled_pool, n_select=100):
        """불확실성 기반 샘플링"""
        predictions = model.predict_proba(unlabeled_pool)
        # 엔트로피가 높은 샘플 = 가장 불확실한 샘플
        entropies = -np.sum(predictions * np.log(predictions + 1e-10), axis=1)
        top_indices = np.argsort(entropies)[-n_select:]
        return unlabeled_pool[top_indices]

    def diversity_sampling(self, embeddings, n_select=100):
        """다양성 기반 샘플링 (클러스터 중심에서 먼 샘플)"""
        from sklearn.cluster import KMeans
        kmeans = KMeans(n_clusters=n_select, random_state=42)
        kmeans.fit(embeddings)
        # 각 클러스터에서 중심에 가장 가까운 샘플 선택
        selected = []
        for i in range(n_select):
            cluster_mask = kmeans.labels_ == i
            cluster_points = embeddings[cluster_mask]
            distances = np.linalg.norm(
                cluster_points - kmeans.cluster_centers_[i], axis=1
            )
            selected.append(np.where(cluster_mask)[0][np.argmin(distances)])
        return selected

    def badge_sampling(self, model, unlabeled_pool, n_select=100):
        """BADGE: 불확실성 + 다양성 결합"""
        # Gradient embedding 계산 후 K-Means++로 다양한 샘플 선택
        gradients = self._compute_gradient_embeddings(model, unlabeled_pool)
        return self.diversity_sampling(gradients, n_select)

데이터 플라이휠 사이클
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

  더 많은 데이터 ──→ 더 좋은 모델
       ↑                    │
       │                    ↓
  더 많은 사용자 ←── 더 좋은 서비스

핵심: 이 사이클이 돌수록 진입 장벽이 높아진다 (경쟁 해자, Moat)

# ML Data Engineer의 일상적인 작업 예시
# Airflow DAG로 라벨링 파이프라인 자동화

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

default_args = {
    "owner": "ml-data-team",
    "retries": 2,
    "retry_delay": timedelta(minutes=5),
}

dag = DAG(
    "labeling_pipeline_v2",
    default_args=default_args,
    schedule_interval="@daily",
    start_date=datetime(2025, 1, 1),
    catchup=False,
)

def extract_raw_data(**kwargs):
    """S3에서 미라벨링 데이터 추출"""
    pass

def run_pre_labeling(**kwargs):
    """모델로 사전 라벨링 실행"""
    pass

def distribute_to_annotators(**kwargs):
    """라벨링 플랫폼에 작업 배분"""
    pass

def quality_check(**kwargs):
    """완료된 라벨링의 품질 검증"""
    pass

def export_training_data(**kwargs):
    """검증된 데이터를 학습 데이터셋으로 내보내기"""
    pass

extract = PythonOperator(task_id="extract", python_callable=extract_raw_data, dag=dag)
pre_label = PythonOperator(task_id="pre_label", python_callable=run_pre_labeling, dag=dag)
distribute = PythonOperator(task_id="distribute", python_callable=distribute_to_annotators, dag=dag)
qa = PythonOperator(task_id="quality_check", python_callable=quality_check, dag=dag)
export = PythonOperator(task_id="export", python_callable=export_training_data, dag=dag)

extract >> pre_label >> distribute >> qa >> export

Data Labeling 커리어 경로
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

[Entry Level]
  Annotation Specialist ($15-25/hr)
  │
  ├──→ Quality Reviewer ($25-40/hr)
  │     │
  │     └──→ Data Quality Manager ($70K-$120K)
  │           │
  │           └──→ Head of Data Operations ($150K+)
  │
  ├──→ RLHF Specialist ($80K-$150K)
  │     │
  │     └──→ AI Safety Researcher ($150K-$250K)
  │
  └──→ [기술 전환]
        │
        ├──→ ML Data Engineer ($120K-$180K)
        │     │
        │     └──→ Senior ML Engineer ($180K-$250K)
        │
        └──→ Annotation Platform Engineer ($130K-$200K)
              │
              └──→ Engineering Manager ($200K+)

AI 성능의 결정 요인 (2025년 이후)
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

알고리즘 차이:        ████░░░░░░ 점점 줄어드는 중
컴퓨팅 파워:         ██████░░░░ 돈으로 해결 가능
데이터 양:           ██████████ 거의 포화 (인터넷 데이터 한계)
데이터 품질:         ██████████ ← 최종 병목, 인간 노동 필요

결론: 고품질 데이터를 확보할 수 있는 기업이 AI 경쟁에서 승리한다