"""Phase T-2: Stage 1A/1B 검증 시스템. Stage 2 이후 검증(content_verifier.py)과 분리된 독립 모듈. AI(Kei)의 콘텐츠 분석 결과를 원본과 대조하여 검증. 검증 4계층: 1. 형식 검증 (Pydantic) — 값 범위, 유효 enum, null 체크 2. 내용 검증 (코드+대조) — 결과가 원본에 대해 적절한가 3. 모순 탐지 (결정 테이블) — purpose × relation_type 논리 모순 4. 피드백 생성 — Self-Refine 패턴: localization + evidence + instruction 도구: - kiwipiepy: 한국어 명사/키워드 추출 (T-2 조사: Windows 즉시 동작, Java 불필요) - regex: 관계 표현 패턴 (T-2 조사: 7개 relation_type별 15개+ 패턴) """ from __future__ import annotations import re import logging from typing import Any logger = logging.getLogger(__name__) # kiwipiepy lazy loading (첫 import 시 ~50MB 모델 다운로드) _kiwi = None def _get_kiwi(): global _kiwi if _kiwi is None: from kiwipiepy import Kiwi _kiwi = Kiwi() return _kiwi # ══════════════════════════════════════ # 한국어 키워드 추출 (kiwipiepy) # ══════════════════════════════════════ def extract_keywords_kiwi(text: str) -> set[str]: """kiwipiepy로 명사 + 영문 약어 추출. 기존 content_verifier.py의 regex extract_keywords()보다 정확: - "정립되지" → "정립" 추출 가능 (어미 분리) - "혼용되어" → "혼용" 추출 가능 - 복합 조사 "에서는" 등 정확 분리 """ kiwi = _get_kiwi() tokens = kiwi.tokenize(text) keywords = set() for t in tokens: # NNG: 일반명사, NNP: 고유명사, SL: 외국어(영문약어) if t.tag in ("NNG", "NNP", "SL") and len(t.form) >= 2: keywords.add(t.form) return keywords # ══════════════════════════════════════ # 관계 표현 패턴 (T-2 조사 결과) # ══════════════════════════════════════ RELATION_PATTERNS: dict[str, list[str]] = { "comparison": [ r"[Vv][Ss]\.?", r"에\s*비해", r"반면", r"차이점?", r"비교", r"대비", r"와\s*달리", r"과\s*달리", r"한편", r"에\s*반하여", r"그에\s*비해", r"상이", r"구분", r"차별화", ], "sequence": [ r"→", r"이후", r"다음", r"먼저", r"그\s*후", r"단계", r"순서", r"[0-9]+차", r"최종적", r"한\s*뒤", r"우선", r"이어서", r"점진적", r"과정", r"를\s*거쳐", ], "hierarchy": [ r"상위", r"하위", r"속하", r"의\s*일부", r"범주", r"구성요소", r"체계", r"분류", r"계층", r"로\s*나뉜?다", r"종속", r"수직적", r"상하\s*관계", r"아우르", r"광의|협의", ], "inclusion": [ r"포함", r"융합", r"통합", r"안에", r"속에", r"결합", r"합쳐", r"아우르", r"망라", r"수렴", r"내포", r"포괄", r"연계", r"접목", r"겹치|중복", ], "cause_effect": [ r"때문에", r"따라서", r"결과", r"원인", r"로\s*인해", r"하여", r"해서", r"초래", r"야기", r"기인", r"영향", r"유발", r"한\s*결과", r"므로", r"그래서", r"그러므로", r"에\s*의해", ], "definition": [ r"이란", r"정의", r"의미", r"개념", r"을\s*말한다", r"을\s*뜻한다", r"로\s*정의", r"가리킨다", r"라\s*함은", r"라고\s*한다", r"줄임말|약어|약자", r"에\s*해당", r"일컫", r"용어", ], } def detect_relation_evidence(text: str) -> dict[str, int]: """원본 텍스트에서 각 relation_type의 증거 수를 카운트.""" evidence = {} for rel_type, patterns in RELATION_PATTERNS.items(): count = sum(1 for p in patterns if re.search(p, text)) evidence[rel_type] = count return evidence # ══════════════════════════════════════ # 모순 결정 테이블 (데이터로 정의) # ══════════════════════════════════════ # purpose × relation_type 하드 모순 (이 조합은 논리적으로 불가능) CONTRADICTIONS: dict[str, list[str]] = { "결론강조": ["comparison", "sequence"], # 결론은 비교나 순서가 아님 "문제제기": ["sequence", "definition"], # 문제제기는 순서 나열이나 정의가 아님 "용어정의": ["hierarchy", "cause_effect"], # 정의 나열은 상하위나 인과가 아님 "구조시각화": ["none"], # 시각화할 관계가 없으면 구조시각화가 아님 } # 소프트 경고 (의심 수준) SOFT_WARNINGS: dict[str, list[str]] = { "핵심전달": ["definition"], # 핵심전달에 definition은 약간 의심 } # ══════════════════════════════════════ # Stage 1A 검증 # ══════════════════════════════════════ VALID_PURPOSES = {"문제제기", "근거사례", "핵심전달", "용어정의", "결론강조", "구조시각화"} VALID_ROLES = {"flow", "reference"} VALID_LAYERS = {"intro", "core", "supporting", "conclusion"} def validate_stage_1a( analysis: dict[str, Any], clean_text: str, ) -> list[dict]: """Stage 1A(Kei 꼭지 추출) 결과 검증. Args: analysis: Kei API 반환 dict clean_text: Stage 0에서 정규화된 텍스트 Returns: 에러 리스트 (빈 리스트 = 통과) """ errors = [] topics = analysis.get("topics", []) page_struct = analysis.get("page_structure", {}) # ── 형식 검증 ── if not topics: errors.append({ "severity": "FATAL", "field": "topics", "localization": "topics가 비어있음", "instruction": "콘텐츠에서 최소 1개 꼭지를 추출하라", }) return errors # Phase Y: page_structure 검증은 validate_page_structure()에서 별도 수행. # Stage 1A에서는 Kei 응답의 page_structure를 검증하지 않음. # 필수 필드 검증 for t in topics: tid = t.get("id", "?") if not t.get("title"): errors.append({ "severity": "RETRYABLE", "field": f"topics[{tid}].title", "localization": f"topic {tid}에 title 없음", "instruction": "각 topic에 title을 부여하라", }) if t.get("purpose") and t["purpose"] not in VALID_PURPOSES: errors.append({ "severity": "RETRYABLE", "field": f"topics[{tid}].purpose", "localization": f"topic {tid} purpose '{t['purpose']}' 유효하지 않음", "current_value": t["purpose"], "instruction": f"유효한 purpose: {VALID_PURPOSES}", }) # page_structure의 topic_ids가 실제 topics에 존재하는지 all_topic_ids = {t.get("id") for t in topics} for role, info in page_struct.items(): if not isinstance(info, dict): continue for tid in info.get("topic_ids", []): if tid not in all_topic_ids: errors.append({ "severity": "RETRYABLE", "field": f"page_structure.{role}.topic_ids", "localization": f"{role}에 존재하지 않는 topic_id {tid}", "instruction": f"topic_ids는 topics[].id에 존재하는 값만 사용하라. 현재 topics: {sorted(all_topic_ids)}", }) # ── 내용 검증 (원본 대조) ── if clean_text: # 원본 ## 섹션 수 vs topic 수 비교 original_sections = re.findall(r"^## .+$", clean_text, re.MULTILINE) # 유형 B에서는 하나의 섹션을 여러 꼭지로 나눌 수 있으므로 허용 폭 확대 _layout = analysis.get("layout_template", "A") max_diff = 4 if _layout == "B" else 2 if len(original_sections) > 0 and abs(len(topics) - len(original_sections)) > max_diff: errors.append({ "severity": "RETRYABLE", "field": "topics", "localization": f"원본 ## 섹션 {len(original_sections)}개, topic {len(topics)}개 (차이 {abs(len(topics) - len(original_sections))})", "evidence": f"원본 섹션: {[s[3:].strip()[:30] for s in original_sections]}", "instruction": "원본의 주요 섹션이 topic에 매핑되었는지 확인하라", }) # topic summary 키워드가 원본에 존재하는지 (kiwipiepy) try: orig_keywords = extract_keywords_kiwi(clean_text) for t in topics: summary = t.get("summary", "") if not summary: continue summary_kw = extract_keywords_kiwi(summary) if not summary_kw: continue overlap = summary_kw & orig_keywords rate = len(overlap) / len(summary_kw) if summary_kw else 1.0 if rate < 0.5: missing = summary_kw - orig_keywords errors.append({ "severity": "RETRYABLE", "field": f"topics[{t.get('id', '?')}].summary", "localization": f"summary 키워드 보존율 {rate:.0%}", "evidence": f"원본에 없는 키워드: {missing}", "instruction": f"summary에 원본에 없는 표현을 추가하지 마라. 원본 키워드로 수정하라.", }) except Exception as e: logger.warning(f"[T-2] kiwipiepy 키워드 검증 실패: {e}") return errors # ══════════════════════════════════════ # Stage 1B 검증 # ══════════════════════════════════════ VALID_RELATION_TYPES = {"hierarchy", "cause_effect", "comparison", "sequence", "definition", "inclusion", "none"} def validate_stage_1b( topics: list[dict[str, Any]], clean_text: str, raw_content: str = "", layout_template: str = "A", ) -> list[dict]: """Stage 1B(컨셉 구체화) 결과 검증. Args: topics: Stage 1B 후 업데이트된 topics 리스트 clean_text: Stage 0에서 정규화된 텍스트 raw_content: 원본 MDX 전체 (popups/details 포함). 대조 범위 확장용. Returns: 에러 리스트 (빈 리스트 = 통과) """ # 대조 범위: clean_text + raw_content (popups/details 내용 포함) full_text = clean_text if raw_content: full_text = clean_text + "\n" + raw_content errors = [] for t in topics: tid = t.get("id", "?") purpose = t.get("purpose", "") relation_type = t.get("relation_type", "") expression_hint = t.get("expression_hint", "") source_data = t.get("source_data", "") # ── 형식 검증 ── if relation_type not in VALID_RELATION_TYPES: errors.append({ "severity": "RETRYABLE", "field": f"topics[{tid}].relation_type", "localization": f"topic {tid}: 유효하지 않은 relation_type '{relation_type}'", "current_value": relation_type, "instruction": f"유효한 relation_type: {sorted(VALID_RELATION_TYPES)}", }) if not expression_hint: errors.append({ "severity": "RETRYABLE", "field": f"topics[{tid}].expression_hint", "localization": f"topic {tid}: expression_hint 비어있음", "instruction": "expression_hint를 작성하라. 형식: 관계 선언 + 콘텐츠 설명 + 시각 지침", }) # ── 모순 탐지 (결정 테이블) ── # Phase Y: Type B에서는 purpose/relation_type이 블록 선택의 핵심 입력이 아님 # (tag 매칭이 item_count + content_example로 동작) # → Type B: 경고만 (파이프라인 계속). Type A: hard fail 유지. if purpose in CONTRADICTIONS: if relation_type in CONTRADICTIONS[purpose]: if layout_template == "B": # Type B: 경고만 logger.warning( f"[T-2 모순경고] topic {tid}: purpose '{purpose}' × relation_type '{relation_type}' " f"— Type B에서는 보조 힌트이므로 경고만" ) else: # Type A: hard fail 유지 errors.append({ "severity": "RETRYABLE", "field": f"topics[{tid}].relation_type", "localization": f"topic {tid}: purpose '{purpose}' × relation_type '{relation_type}' 모순", "current_value": f"purpose={purpose}, relation_type={relation_type}", "evidence": f"'{purpose}'는 '{relation_type}'와 논리적으로 양립 불가", "instruction": f"relation_type을 재판단하라. '{purpose}'에 적합한 관계는 " f"{[r for r in VALID_RELATION_TYPES if r not in CONTRADICTIONS.get(purpose, [])]}", }) if purpose in SOFT_WARNINGS: if relation_type in SOFT_WARNINGS[purpose]: logger.warning( f"[T-2 경고] topic {tid}: purpose '{purpose}' × relation_type '{relation_type}' 의심" ) # ── 원본 대조: source_data 할루시네이션 감지 ── # full_text 사용 (popups/details 내용 포함) if source_data and full_text: try: source_kw = extract_keywords_kiwi(source_data) orig_kw = extract_keywords_kiwi(full_text) if source_kw: overlap = source_kw & orig_kw rate = len(overlap) / len(source_kw) if rate < 0.4: missing = source_kw - orig_kw errors.append({ "severity": "RETRYABLE", "field": f"topics[{tid}].source_data", "localization": f"topic {tid}: source_data 키워드 보존율 {rate:.0%}", "evidence": f"원본에 없는 키워드: {missing}", "instruction": "source_data는 원본에 실제 존재하는 텍스트만 사용하라. 없는 출처를 만들어내지 마라.", }) except Exception as e: logger.warning(f"[T-2] source_data 검증 실패: {e}") # ── 원본 대조: relation_type과 원본 언어 패턴 ── # full_text 사용 (popups/details 내용 포함) if relation_type and relation_type != "none" and full_text: evidence = detect_relation_evidence(full_text) claimed_count = evidence.get(relation_type, 0) if claimed_count == 0: if layout_template == "B": # 유형 B: relation_type 증거 부족은 warning만 (역할 구조가 자유) logger.warning(f"[Stage 1B] topic {tid}: '{relation_type}' 증거 0개 — 유형 B warning") else: alternatives = [(k, v) for k, v in evidence.items() if v >= 2] alt_str = ", ".join(f"{k}({v}개)" for k, v in alternatives[:3]) errors.append({ "severity": "RETRYABLE", "field": f"topics[{tid}].relation_type", "localization": f"topic {tid}: '{relation_type}' 증거 0개", "evidence": f"원본에서 '{relation_type}' 패턴 없음. 대안: {alt_str}" if alt_str else f"원본에서 '{relation_type}' 패턴 없음", "instruction": f"원본 텍스트에 '{relation_type}' 관계를 나타내는 표현이 없음. 재판단하라.", }) return errors def validate_page_structure(page_struct: dict) -> list[dict]: """Phase Y: section_parser가 생성한 page_structure 검증. Stage 1A 후, section_parser + 블록 매칭으로 page_structure가 채워진 후 호출. """ errors = [] if not page_struct: errors.append({ "severity": "FATAL", "field": "page_structure", "localization": "page_structure가 비어있음", "instruction": "section_parser가 영역을 생성하지 못함", }) return errors # weight 합 검증 (0.9~1.1) total_weight = sum( info.get("weight", 0) for info in page_struct.values() if isinstance(info, dict) ) if total_weight < 0.9 or total_weight > 1.1: errors.append({ "severity": "RETRYABLE", "field": "page_structure.weight", "localization": f"weight 합 {total_weight:.2f} (범위: 0.9~1.1)", "instruction": f"weight 합이 1.0에 가깝도록 조정하라. 현재 합: {total_weight:.2f}", }) return errors