8장: 도메인 특화 데이터셋 구축

MEDICAL_SEED_TEMPLATE = {
    "patient_demographics": {
        "age_range": "20-80",
        "conditions": ["고혈압", "당뇨병", "만성신장질환"],
    },
    "clinical_context": {
        "department": "내과",
        "visit_type": "외래",
        "chief_complaint": "두통과 어지러움",
    },
    "required_elements": [
        "주관적 소견(Subjective)",
        "객관적 소견(Objective)",
        "평가(Assessment)",
        "계획(Plan)",
    ],
}
 
MEDICAL_GEN_PROMPT = """당신은 내과 전문의입니다.
다음 임상 시나리오에 대한 SOAP 노트를 작성하세요.
 
환자 정보:
- 연령대: {age_range}
- 기저 질환: {conditions}
- 진료과: {department}
- 방문 유형: {visit_type}
- 주소(Chief Complaint): {chief_complaint}
 
규칙:
1. SOAP 형식(주관적/객관적/평가/계획)을 따릅니다.
2. 진단명은 ICD-10 코드를 병기합니다.
3. 처방은 기저 질환과의 약물 상호작용을 고려합니다.
4. 검사 결과 수치는 의학적으로 타당한 범위여야 합니다.
5. 실제 환자 정보가 포함되지 않도록 합니다.
 
JSON 형식으로 출력하세요."""
 
 
def validate_medical_consistency(record: dict) -> dict:
    """의료 데이터의 임상적 일관성을 검증합니다."""
    issues = []
 
    # 진단-처방 일관성 검사
    diagnosis = record.get("assessment", {}).get("diagnosis", "")
    medications = record.get("plan", {}).get("medications", [])
 
    # 약물 상호작용 검사 (간단한 규칙 기반)
    contraindications = {
        "만성신장질환": ["NSAIDs", "메트포르민"],
        "간경변": ["아세트아미노펜 고용량"],
    }
 
    conditions = record.get("conditions", [])
    for condition in conditions:
        if condition in contraindications:
            for med in medications:
                if med in contraindications[condition]:
                    issues.append(
                        f"금기: {condition} 환자에게 {med} 처방"
                    )
 
    # 검사 수치 범위 검증
    lab_ranges = {
        "크레아티닌": (0.5, 15.0),
        "HbA1c": (4.0, 15.0),
        "혈압_수축기": (60, 250),
    }
 
    labs = record.get("objective", {}).get("lab_results", {})
    for lab_name, value in labs.items():
        if lab_name in lab_ranges:
            min_val, max_val = lab_ranges[lab_name]
            if not min_val <= value <= max_val:
                issues.append(
                    f"비정상 범위: {lab_name} = {value}"
                )
 
    return {
        "is_valid": len(issues) == 0,
        "issues": issues,
        "record": record,
    }

LEGAL_DOMAINS = {
    "민사": {
        "subtypes": ["계약 분쟁", "손해배상", "부동산", "가사"],
        "document_types": ["소장", "답변서", "준비서면", "판결문"],
    },
    "형사": {
        "subtypes": ["사기", "횡령", "상해", "명예훼손"],
        "document_types": ["공소장", "변론요지서", "판결문"],
    },
    "행정": {
        "subtypes": ["인허가", "과세", "환경", "노동"],
        "document_types": ["처분서", "이의신청서", "재결서"],
    },
}
 
LEGAL_GEN_PROMPT = """당신은 {specialty} 전문 변호사입니다.
다음 시나리오에 대한 {document_type}를 작성하세요.
 
시나리오:
{scenario}
 
규칙:
1. 한국 법률 체계와 용어를 사용합니다.
2. 관련 법조문을 정확하게 인용합니다(예: 민법 제750조).
3. 법률 문서의 표준 형식을 따릅니다.
4. 모든 인명, 지명, 기관명은 가상입니다.
5. 판례 인용은 실제 판례와 혼동되지 않도록 가상으로 작성합니다.
6. 법적 논리가 일관되어야 합니다."""

import re
 
 
def validate_legal_document(document: str) -> dict:
    """법률 문서의 형식과 내용을 검증합니다."""
    issues = []
 
    # 법조문 인용 형식 검증
    statute_pattern = r"(민법|형법|상법|민사소송법|형사소송법)\s+제\d+조"
    statutes = re.findall(statute_pattern, document)
    if not statutes:
        issues.append("법조문 인용이 없습니다.")
 
    # 가상 인물명 확인 (실제 유명인 이름 배제)
    # LLM 기반 검증 또는 이름 사전 대조
 
    # 문서 구조 검증
    required_sections = {
        "소장": ["청구취지", "청구원인", "입증방법"],
        "판결문": ["주문", "이유", "결론"],
    }
 
    # 논리적 일관성 검사
    # 원고/피고 주장의 자기 모순 탐지
 
    return {
        "is_valid": len(issues) == 0,
        "issues": issues,
        "statutes_cited": statutes,
    }

import numpy as np
 
 
def generate_synthetic_transactions(
    num_transactions: int,
    fraud_ratio: float = 0.02,
) -> list[dict]:
    """합성 금융 거래 데이터를 생성합니다."""
    transactions = []
 
    for i in range(num_transactions):
        is_fraud = np.random.random() < fraud_ratio
 
        if is_fraud:
            # 이상 거래 패턴
            transaction = generate_fraudulent_pattern()
        else:
            # 정상 거래 패턴
            transaction = generate_normal_pattern()
 
        transaction["id"] = i
        transactions.append(transaction)
 
    return transactions
 
 
def generate_normal_pattern() -> dict:
    """정상 거래 패턴을 생성합니다."""
    amount = np.random.lognormal(mean=10, sigma=1.5)
    hour = np.random.choice(range(24), p=normal_hour_distribution())
 
    return {
        "amount": round(amount, 0),
        "hour": hour,
        "is_fraud": False,
        "merchant_category": np.random.choice(
            ["식료품", "의류", "전자기기", "외식", "교통"]
        ),
    }
 
 
def generate_fraudulent_pattern() -> dict:
    """이상 거래 패턴을 생성합니다."""
    # 고액 거래, 새벽 시간대, 해외 결제 등
    amount = np.random.lognormal(mean=13, sigma=2.0)
    hour = np.random.choice(range(24), p=fraud_hour_distribution())
 
    return {
        "amount": round(amount, 0),
        "hour": hour,
        "is_fraud": True,
        "merchant_category": np.random.choice(
            ["전자기기", "귀금속", "선불카드", "해외결제"]
        ),
    }
 
 
def normal_hour_distribution() -> list[float]:
    """정상 거래 시간대 분포"""
    dist = [0.01] * 24
    for h in range(9, 21):  # 오전 9시 ~ 오후 9시
        dist[h] = 0.06
    total = sum(dist)
    return [d / total for d in dist]
 
 
def fraud_hour_distribution() -> list[float]:
    """이상 거래 시간대 분포"""
    dist = [0.03] * 24
    for h in [0, 1, 2, 3, 4, 5]:  # 새벽 시간대
        dist[h] = 0.08
    total = sum(dist)
    return [d / total for d in dist]
 
 
def validate_accounting_identity(
    balance_sheet: dict,
) -> bool:
    """회계 항등식을 검증합니다."""
    assets = balance_sheet.get("total_assets", 0)
    liabilities = balance_sheet.get("total_liabilities", 0)
    equity = balance_sheet.get("total_equity", 0)
 
    # 자산 = 부채 + 자본 (허용 오차 1원)
    return abs(assets - (liabilities + equity)) <= 1

CODE_TAXONOMY = {
    "알고리즘": {
        "subtopics": [
            "정렬", "탐색", "그래프", "동적 프로그래밍",
            "문자열", "수학", "자료구조"
        ],
        "difficulty": ["easy", "medium", "hard"],
        "languages": ["python", "javascript", "java", "go"],
    },
    "시스템 설계": {
        "subtopics": [
            "API 설계", "데이터베이스 설계", "캐싱",
            "메시지 큐", "로드 밸런싱"
        ],
        "format": "서술형 + 다이어그램 설명",
    },
    "코드 리뷰": {
        "subtopics": [
            "버그 탐지", "성능 개선", "보안 취약점",
            "가독성 개선", "설계 패턴 적용"
        ],
        "format": "코드 + 리뷰 코멘트",
    },
    "디버깅": {
        "subtopics": [
            "런타임 에러", "논리 에러", "성능 병목",
            "메모리 누수", "동시성 버그"
        ],
        "format": "버그 코드 + 수정 코드 + 설명",
    },
}

version: 3
domain: medicine
created_by: domain_expert
seed_examples:
  - context: |
      급성 심근경색은 관상동맥의 갑작스러운 폐색으로 인해
      심근에 비가역적 괴사가 발생하는 질환입니다.
    questions_and_answers:
      - question: |
          급성 심근경색의 초기 치료에서 가장 중요한 원칙은
          무엇입니까?
        answer: |
          급성 심근경색의 초기 치료에서 가장 중요한 원칙은
          "시간은 심근이다(Time is Muscle)"로 요약됩니다.
          발증 후 가능한 빨리 재관류 치료를 시행하는 것이
          핵심입니다.
      - question: |
          STEMI와 NSTEMI의 치료 전략은 어떻게 다릅니까?
        answer: |
          STEMI는 즉각적인 경피적 관상동맥 중재술(PCI) 또는
          혈전용해제 투여가 권고됩니다. NSTEMI는 위험도 평가
          후 조기(24-72시간 이내) 침습적 전략을 결정합니다.

from dataclasses import dataclass
from enum import Enum
 
 
class ReviewDecision(Enum):
    APPROVE = "approve"
    REVISE = "revise"
    REJECT = "reject"
 
 
@dataclass
class ExpertReview:
    reviewer_id: str
    domain: str
    decision: ReviewDecision
    accuracy_score: float  # 1-5
    completeness_score: float  # 1-5
    domain_accuracy_score: float  # 1-5
    comments: str
 
 
def expert_review_sampling(
    data: list[dict],
    sample_rate: float = 0.05,
    stratify_by: str = "subdomain",
) -> list[dict]:
    """전문가 리뷰를 위한 층화 샘플링을 수행합니다."""
    from collections import defaultdict
    import random
 
    # 하위 도메인별로 그룹화
    groups = defaultdict(list)
    for item in data:
        key = item.get(stratify_by, "unknown")
        groups[key].append(item)
 
    # 각 그룹에서 비례 샘플링
    samples = []
    for key, items in groups.items():
        n = max(1, int(len(items) * sample_rate))
        samples.extend(random.sample(items, min(n, len(items))))
 
    return samples