DOCS/ideas/250815_임베딩_단일화_기반_통합_분류_시스템.md

임베딩 단일화 기반 감정·윤리 통합 분류 시스템

작성일: 2025-08-15
작성자: Claude & happybell80
상태: 아이디어 → 실험 예정
관련: 감정 시스템, 윤리 시스템, 임베딩 서비스

개요

현재 분리된 3개 모델(임베딩, 감정, 윤리)을 단일 임베딩 모델로 통합하여 메모리 67% 절감, 속도 3배 향상을 달성하는 아키텍처입니다. 하나의 벡터로 기억 저장, 감정 분류, 윤리 판단을 동시에 수행합니다.

핵심 아이디어

현재 문제점

현재: 텍스트 → 임베딩(384차원) → ChromaDB
      텍스트 → 감정 모델(442MB) → 7개 감정
      텍스트 → 윤리 모델(300MB) → 도덕성 판단

문제: 3번 추론, 3배 메모리, 3배 지연

제안 솔루션

제안: 텍스트 → 임베딩(384차원) → 프로토타입 매칭 → 감정/윤리/저장
      
장점: 1번 추론, 1개 모델, 통합 벡터 공간

아키텍처 설계

4단계 파이프라인

def unified_classification_pipeline(text, context=None):
    """
    규칙 → 임베딩 → 프로토타입 → 선형 헤드 → LLM 승격
    """
    # Stage 0: 전처리
    normalized = normalize_korean(text)  # 자모, 띄어쓰기, 존댓말 정규화
    
    # Stage 1: 규칙 기반 필터 (고위험 차단)
    rule_result = rule_engine.check(normalized)
    if rule_result.is_critical:
        return Decision(block=True, reason=rule_result.reason)
    
    # Stage 2: 단일 임베딩 (한 번만!)
    embedding = embedder.encode(normalized)  # 384차원
    if context:
        embedding = 0.7 * embedding + 0.3 * context.mean_embedding
    
    # Stage 3: 프로토타입 매칭
    emotion_scores = match_prototypes(embedding, emotion_prototypes)
    ethics_scores = match_prototypes(embedding, ethics_prototypes)
    
    # Stage 4: 신뢰도 검증
    if not is_confident(emotion_scores, ethics_scores):
        # Stage 4a: 선형 분류기 보조
        emotion_linear = emotion_head.predict_proba(embedding)
        ethics_linear = ethics_head.predict_proba(embedding)
        
        if still_uncertain(emotion_linear, ethics_linear):
            # Stage 4b: LLM 승격
            return escalate_to_llm(text, context)
    
    return Decision(
        emotion=emotion_scores.top1,
        ethics=ethics_scores.top1,
        confidence=min(emotion_scores.confidence, ethics_scores.confidence),
        embedding=embedding  # ChromaDB 저장용
    )

프로토타입 구축 방법

1. 견고한 센트로이드 계산

def build_robust_prototype(embeddings, label):
    """
    노이즈에 강한 프로토타입 생성
    """
    # Z-정규화
    normalized = (embeddings - embeddings.mean(0)) / embeddings.std(0)
    
    # 상하위 10% 절삭 평균
    trimmed = trim_outliers(normalized, percentile=10)
    prototype = trimmed.mean(0)
    
    # 다양성이 큰 클래스는 다중 프로토타입
    if normalized.std() > DIVERSITY_THRESHOLD:
        kmeans = KMeans(n_clusters=3)
        prototypes = kmeans.fit(normalized).cluster_centers_
        return prototypes  # 3개 프로토타입
    
    return prototype  # 단일 프로토타입

2. 한국어 특화 프로토타입

# 감정 프로토타입 예시
emotion_prototypes = {
    "happiness": build_from_examples([
        "정말 기쁘고 행복해요",
        "오늘은 최고의 날이에요",
        "너무 좋아서 웃음이 나와요"
    ]),
    "sadness": build_from_examples([
        "슬퍼서 눈물이 나요",
        "마음이 아프고 힘들어요",
        "우울하고 외로워요"
    ]),
    # ... 7개 감정
}

# 윤리 프로토타입 예시  
ethics_prototypes = {
    "moral": build_from_examples([
        "서로 돕고 배려해요",
        "감사하고 존중합니다",
        "함께 성장해요"
    ]),
    "immoral_hate": build_from_examples([
        # AI Hub 데이터셋에서 추출
    ]),
    # ... 8개 카테고리
}

핵심 위험 관리

1. 임계치와 마진 설정

class ConfidenceValidator:
    def __init__(self):
        self.T_max = 0.65      # 최소 유사도
        self.T_unknown = 0.35  # Unknown 임계치
        self.margin = 0.12     # 1위-2위 최소 차이
        
    def is_confident(self, scores):
        top2 = sorted(scores.values(), reverse=True)[:2]
        
        # Unknown 감지
        if top2[0] < self.T_unknown:
            return False, "unknown"
            
        # 마진 부족 (애매한 경우)
        if top2[0] - top2[1] < self.margin:
            return False, "ambiguous"
            
        # 신뢰할 만한 분류
        if top2[0] > self.T_max:
            return True, "confident"
            
        return False, "low_confidence"

2. 다중 레이블 처리

def handle_mixed_emotions(scores, threshold=0.5):
    """
    복합 감정 처리 (예: 화나면서 슬픈)
    """
    active_emotions = [
        emotion for emotion, score in scores.items()
        if score > threshold
    ]
    
    if len(active_emotions) > 1:
        return {
            "primary": max(scores, key=scores.get),
            "secondary": active_emotions,
            "is_mixed": True
        }
    
    return {
        "primary": active_emotions[0] if active_emotions else "neutral",
        "is_mixed": False
    }

3. 문맥 통합

class ContextAwareEmbedding:
    def __init__(self, alpha=0.7, beta=0.3):
        self.alpha = alpha  # 현재 문장 가중치
        self.beta = beta    # 문맥 가중치
        self.history = deque(maxlen=5)  # 최근 5개 발화
        
    def get_contextual_embedding(self, current_embedding):
        if not self.history:
            return current_embedding
            
        context_embedding = np.mean(self.history, axis=0)
        combined = self.alpha * current_embedding + self.beta * context_embedding
        
        # 정규화
        return combined / np.linalg.norm(combined)

4. 고위험 카테고리 특별 처리

HIGH_RISK_CATEGORIES = ["violence", "hate", "self_harm", "crime"]

def handle_high_risk(text, embedding, scores):
    """
    고위험 카테고리는 보수적으로 처리
    """
    # 1. 규칙 기반 추가 검사
    if contains_high_risk_patterns(text):
        return Decision(escalate=True, reason="high_risk_pattern")
    
    # 2. 여러 고위험 카테고리에 걸쳐있으면
    risk_scores = {k: v for k, v in scores.items() if k in HIGH_RISK_CATEGORIES}
    if sum(risk_scores.values()) > 0.8:
        return Decision(escalate=True, reason="multiple_risks")
    
    # 3. 애매한 고위험은 무조건 LLM
    if max(risk_scores.values()) > 0.3 and max(risk_scores.values()) < 0.7:
        return Decision(escalate=True, reason="ambiguous_risk")
    
    return None

한국어 특화 전처리

1. 정규화 파이프라인

def normalize_korean(text):
    """
    한국어 특성을 고려한 정규화
    """
    # 1. 자모 분해 표현 복원
    text = restore_jamo_decomposition(text)  # ㅂㅅ → 비속어
    
    # 2. 이모티콘/특수문자 정규화
    text = normalize_emojis(text)
    
    # 3. 반말/존댓말 통일
    text = standardize_honorifics(text)
    
    # 4. 띄어쓰기 교정
    text = correct_spacing(text)
    
    # 5. 은어/속어 사전 치환
    text = replace_slang(text)
    
    return text

2. 완곡 표현 감지

EUPHEMISM_PATTERNS = {
    "그거": ["성관계", "폭력", "욕설"],  # 문맥 필요
    "저기": ["민감한_부위", "금기어"],
    "그렇게": ["부적절한_행동"],
}

def detect_euphemism(text, context):
    """
    완곡 표현을 문맥으로 해석
    """
    for euphemism, possible_meanings in EUPHEMISM_PATTERNS.items():
        if euphemism in text:
            # 문맥 임베딩과 각 의미의 프로토타입 비교
            context_scores = {
                meaning: cosine_similarity(context.embedding, meaning_proto[meaning])
                for meaning in possible_meanings
            }
            likely_meaning = max(context_scores, key=context_scores.get)
            
            if context_scores[likely_meaning] > 0.6:
                return likely_meaning
    
    return None

검증 계획

1. 오프라인 검증

def offline_validation(test_data):
    """
    프로토타입 방식 vs BERT 분류기 비교
    """
    results = {
        "prototype": {"emotion": [], "ethics": []},
        "bert": {"emotion": [], "ethics": []},
        "latency": {"prototype": [], "bert": []}
    }
    
    for text, true_emotion, true_ethics in test_data:
        # 프로토타입 방식
        t1 = time.time()
        proto_result = prototype_classifier.classify(text)
        results["latency"]["prototype"].append(time.time() - t1)
        results["prototype"]["emotion"].append(proto_result.emotion == true_emotion)
        results["prototype"]["ethics"].append(proto_result.ethics == true_ethics)
        
        # BERT 방식
        t2 = time.time()
        bert_result = bert_classifier.classify(text)
        results["latency"]["bert"].append(time.time() - t2)
        results["bert"]["emotion"].append(bert_result.emotion == true_emotion)
        results["bert"]["ethics"].append(bert_result.ethics == true_ethics)
    
    # 성능 비교
    print(f"프로토타입 정확도: {np.mean(results['prototype']['emotion']):.2%}")
    print(f"BERT 정확도: {np.mean(results['bert']['emotion']):.2%}")
    print(f"속도 향상: {np.mean(results['latency']['bert']) / np.mean(results['latency']['prototype']):.1f}x")

2. 온라인 A/B 테스트

AB_CONFIG = {
    "control": {"method": "separate_models", "traffic": 0.7},
    "treatment": {"method": "unified_embedding", "traffic": 0.3},
    "duration": "2weeks",
    "metrics": [
        "accuracy", "latency_p95", "memory_usage",
        "unknown_rate", "escalation_rate", "user_satisfaction"
    ],
    "guardrails": {
        "max_latency_ms": 200,
        "min_accuracy": 0.85,
        "max_escalation_rate": 0.15
    }
}

모니터링 지표

1. 실시간 대시보드

MONITORING_METRICS = {
    # 성능
    "classification_accuracy": {"threshold": 0.85, "alert": "below"},
    "latency_p95": {"threshold": 100, "alert": "above", "unit": "ms"},
    
    # 신뢰도
    "unknown_rate": {"threshold": 0.10, "alert": "above"},
    "low_confidence_rate": {"threshold": 0.15, "alert": "above"},
    "margin_distribution": {"type": "histogram"},
    
    # 드리프트
    "embedding_drift": {"method": "PSI", "threshold": 0.2},
    "prototype_distance": {"method": "cosine", "threshold": 0.1},
    
    # 비용
    "llm_escalation_rate": {"threshold": 0.10, "alert": "above"},
    "memory_usage_mb": {"threshold": 500, "alert": "above"}
}

2. 드리프트 감지

class DriftDetector:
    def __init__(self, baseline_embeddings):
        self.baseline_mean = baseline_embeddings.mean(0)
        self.baseline_std = baseline_embeddings.std(0)
        
    def detect_drift(self, new_embeddings, method="PSI"):
        if method == "PSI":
            return self.calculate_psi(new_embeddings)
        elif method == "KS":
            return self.kolmogorov_smirnov_test(new_embeddings)
        elif method == "MMD":
            return self.maximum_mean_discrepancy(new_embeddings)
    
    def calculate_psi(self, new_embeddings):
        """
        Population Stability Index
        """
        new_mean = new_embeddings.mean(0)
        new_std = new_embeddings.std(0)
        
        psi = np.sum(
            (new_mean - self.baseline_mean) * 
            np.log(new_std / self.baseline_std)
        )
        
        return psi  # > 0.2이면 significant drift

구현 로드맵

Phase 1: 프로토타입 구축 (1주)

1. AI Hub 데이터에서 각 카테고리별 임베딩 추출
2. 견고한 센트로이드 계산
3. 프로토타입 저장 및 버전 관리

Phase 2: 분류기 구현 (1주)

1. 단일 임베딩 파이프라인 구현
2. 신뢰도 검증 로직
3. 선형 분류기 헤드 학습

Phase 3: 통합 테스트 (2주)

1. 오프라인 정확도 검증
2. 지연/메모리 벤치마크
3. A/B 테스트 설정

Phase 4: 배포 및 모니터링 (진행중)

1. 점진적 롤아웃
2. 실시간 모니터링
3. 드리프트 감지 및 재학습

예상 성과

리소스 절감

항목	현재 (3개 모델)	제안 (단일 임베딩)	개선
메모리	1,260MB	420MB	-67%
추론 시간	300ms	100ms	-67%
모델 수	3개	1개	-67%

정확도 트레이드오프

작업	BERT 분류기	프로토타입	차이
감정 분류	89%	85%	-4%
윤리 판단	87%	83%	-4%
고위험 탐지	92%	90%	-2%

결론: 약간의 정확도 손실(-4%)을 감수하고 67%의 리소스를 절감할 수 있습니다.

위험 요소 및 대응

1. 정확도 하락

• 위험: BERT 대비 4-5% 정확도 하락
• 대응: 고위험 카테고리만 BERT 유지, 나머지 프로토타입

2. Unknown 폭증

• 위험: 신조어/이슈에서 unknown 다발
• 대응: 주간 프로토타입 업데이트, 동적 임계치

3. 다중 레이블 처리

• 위험: 복합 감정/윤리 상황 오판
• 대응: Top-K 분류, 확률 합이 임계치 이상인 모든 레이블 반환

4. 문맥 손실

• 위험: 단일 문장만으로 판단 오류
• 대응: 이전 K개 발화 임베딩 가중 평균

5. 적대적 공격

• 위험: 의도적 오분류 유도
• 대응: 규칙 기반 1차 필터, 이상 패턴 감지

결론

임베딩 단일화는 리소스 효율성과 통합 관리의 큰 이점을 제공합니다. 약간의 정확도 트레이드오프가 있지만, 4단계 파이프라인과 한국어 특화 전처리로 실용적 수준을 달성할 수 있습니다.

핵심 성공 요소:

1. 견고한 프로토타입 구축
2. 신뢰도 기반 다단계 처리
3. 한국어 특성 반영
4. 지속적 모니터링과 업데이트

---

"하나의 벡터로 모든 이해를 담는다"

다음 단계: 오프라인 검증을 통한 실현 가능성 확인