"""입출력 가드레일 — PII 마스킹 · 위기분류 · 출력 안전레일. MASTERPLAN §2.2 / §3.3 / R5 / R7 / F-03, MEMORY_DESIGN §B: [입력] Presidio PII 마스킹(미설치 시 정규식 폴백) + 위기분류(실제위기 vs 페르소나 연기) [출력] 자살수단/방법 정보 차단, ideation_stage 상한(<=3) 강제. 설계 원칙: - 모듈 경계 명확: 입력 가드레일(mask_pii / classify_crisis)과 출력 가드레일(sanitize_client_reply)을 순수함수에 가깝게 분리. IO·LLM·DB 의존 없음(테스트·재사용 용이). - 외부 LLM 경로 진입 전 *하드 게이트*: 마스킹 안 된 원문은 게이트웨이로 절대 안 나간다(F-03). - Presidio 는 선택 의존(미설치 환경에서도 import 가능해야 함) → 지연 로드 + 정규식 폴백. TODO(Phase 2): Presidio MedicalNERRecognizer + 한국어 자살콘텐츠 분류기(JMIR few-shot 5단계, R8) 로 교체. 현재 정규식/키워드 폴백은 1차 안전망(재현율 우선). """ from __future__ import annotations import re from dataclasses import dataclass, field from enum import Enum from typing import Optional # ── 출력 가드레일 상한 (R5) ────────────────────────────── IDEATION_STAGE_CAP = 3 # 내담자 발화/상태가 넘을 수 없는 자살사고 단계 상한 # ════════════════════════════════════════════════════════════════════════════ # 1. PII 마스킹 (입력 — 저장·외부전송 전 하드 게이트, F-03) # ════════════════════════════════════════════════════════════════════════════ # 정규식 폴백 패턴 (Presidio 미설치 시). 한국 맥락 우선. # TODO: Presidio + MedicalNERRecognizer 로 정밀화(이름/주소/기관 NER). _PII_PATTERNS: list[tuple[str, re.Pattern[str]]] = [ # 주민등록번호 (6자리-7자리) ("RRN", re.compile(r"\b\d{6}[-\s]?\d{7}\b")), # 휴대폰 (010-1234-5678 등) ("PHONE", re.compile(r"\b01[016789][-\s]?\d{3,4}[-\s]?\d{4}\b")), # 일반 전화 ("PHONE", re.compile(r"\b0\d{1,2}[-\s]?\d{3,4}[-\s]?\d{4}\b")), # 이메일 ("EMAIL", re.compile(r"\b[\w.+-]+@[\w-]+\.[\w.-]+\b")), # 카드/계좌 유사 긴 숫자열 (12자리 이상) ("NUMID", re.compile(r"\b\d{12,}\b")), ] # Presidio 지연 로드 캐시 (-1=미시도, None=미설치, 객체=설치됨) _PRESIDIO_ANALYZER: object = -1 _PRESIDIO_ANONYMIZER: object = -1 def _try_load_presidio(): """Presidio (analyzer, anonymizer) 지연 로드. 미설치면 (None, None).""" global _PRESIDIO_ANALYZER, _PRESIDIO_ANONYMIZER if _PRESIDIO_ANALYZER != -1: return _PRESIDIO_ANALYZER, _PRESIDIO_ANONYMIZER try: from presidio_analyzer import AnalyzerEngine # type: ignore from presidio_anonymizer import AnonymizerEngine # type: ignore _PRESIDIO_ANALYZER = AnalyzerEngine() _PRESIDIO_ANONYMIZER = AnonymizerEngine() except Exception: _PRESIDIO_ANALYZER = None _PRESIDIO_ANONYMIZER = None return _PRESIDIO_ANALYZER, _PRESIDIO_ANONYMIZER @dataclass(slots=True) class MaskResult: text_masked: str entities: list[str] = field(default_factory=list) # 탐지된 엔티티 타입들 used_presidio: bool = False def mask_pii(text: str) -> MaskResult: """PII 마스킹. Presidio 가용 시 우선, 아니면 정규식 폴백. 반환 text_masked 만 저장(turns.text_masked)·외부 LLM 전송에 사용한다(F-03). """ if not text: return MaskResult(text_masked=text, entities=[], used_presidio=False) analyzer, anonymizer = _try_load_presidio() if analyzer is not None and anonymizer is not None: try: results = analyzer.analyze(text=text, language="en") # TODO: ko 모델 등록 시 language="ko" ents = sorted({r.entity_type for r in results}) anonymized = anonymizer.anonymize(text=text, analyzer_results=results) return MaskResult(text_masked=anonymized.text, entities=ents, used_presidio=True) except Exception: pass # 폴백으로 # 정규식 폴백 masked = text found: list[str] = [] for label, pat in _PII_PATTERNS: if pat.search(masked): found.append(label) masked = pat.sub(f"[{label}]", masked) return MaskResult(text_masked=masked, entities=sorted(set(found)), used_presidio=False) # ════════════════════════════════════════════════════════════════════════════ # 2. 위기 분류 (입력 — 실제위기 vs 페르소나 연기 구분, R8) # ════════════════════════════════════════════════════════════════════════════ class CrisisKind(str, Enum): NONE = "none" PERSONA_PLAY = "persona_play" # 페르소나 연기 맥락의 위기 표현(시뮬레이션 정상) LEARNER_REAL = "learner_real" # 수련생 본인의 실제 위기 신호(에스컬레이션 대상) @dataclass(slots=True) class CrisisResult: kind: CrisisKind risk_level: int = 0 # 0~5 (한국어 자살콘텐츠 5단계 자리; 현재 휴리스틱) matched: list[str] = field(default_factory=list) escalate: bool = False # safety_events 적재 + 교수자 알림 트리거 여부 # 위기 표현 키워드(한국어 우선). TODO: JMIR 한국어 벤치 few-shot 분류기로 교체(R8). _CRISIS_TERMS = [ "죽고 싶", "죽고싶", "자살", "목숨", "사라지고 싶", "없어지고 싶", "자해", "끝내고 싶", "살기 싫", "살아서 뭐", "죽어야", ] # 실제 위기로 가중되는 1인칭 현재 단서(수련생 본인 신호일 가능성) _FIRST_PERSON_NOW = ["지금 나", "나 진짜", "제가 지금", "저 지금", "real", "도와주세요"] def classify_crisis(text: str, *, speaker_is_persona_context: bool = True) -> CrisisResult: """위기 분류. Args: speaker_is_persona_context: True 면 상담 시뮬레이션 발화(수련생→가상내담자) 맥락. 이 경우 위기 표현은 기본 PERSONA_PLAY 로 본다(연기). 단 1인칭 실제 단서가 강하면 LEARNER_REAL 로 승격해 에스컬레이션(보수적, 재현율 우선). NOTE: 가상내담자(P1)의 자살사고 연기는 시뮬레이션 핵심이라 차단 대상이 아니다. 진짜 위험은 *수련생 본인*의 실제 위기 신호다. """ if not text: return CrisisResult(kind=CrisisKind.NONE) matched = [t for t in _CRISIS_TERMS if t in text] if not matched: return CrisisResult(kind=CrisisKind.NONE) risk = min(5, 2 + len(matched)) # 거친 단계화(휴리스틱) real_cues = [c for c in _FIRST_PERSON_NOW if c in text] if not speaker_is_persona_context or real_cues: # 실제 위기 가능 → 에스컬레이션(보수적) return CrisisResult( kind=CrisisKind.LEARNER_REAL, risk_level=max(risk, 4), matched=matched + real_cues, escalate=True, ) # 시뮬레이션 연기 맥락 return CrisisResult( kind=CrisisKind.PERSONA_PLAY, risk_level=risk, matched=matched, escalate=False, ) # ════════════════════════════════════════════════════════════════════════════ # 3. 출력 가드레일 (내담자 AI 응답 — 자살수단 차단, ideation 상한 R5) # ════════════════════════════════════════════════════════════════════════════ # 자살/자해 '수단·방법' 정보 패턴 (절대 출력 금지). 정서 표현은 허용. _MEANS_TERMS = [ "약을 얼마나", "몇 알", "몇 정", "투신", "목을 매", "목 매", "번개탄", "농약", "손목을", "칼로", "뛰어내리", "수면제 ", "독", "방법은", "어떻게 죽", ] @dataclass(slots=True) class OutputGuardResult: text: str # 정제된(또는 원본) 응답 blocked: bool = False # 차단/치환 발생 여부 needs_regeneration: bool = False # 재생성 필요(수단정보 누출 시) reasons: list[str] = field(default_factory=list) def sanitize_client_reply(text: str, *, ideation_stage: int) -> OutputGuardResult: """내담자 AI 응답 출력 가드레일. 1. 자살/자해 수단·방법 정보가 있으면 → needs_regeneration=True(차단, 재생성 신호). 2. ideation_stage > IDEATION_STAGE_CAP(3) 이면 안전 상한 위반으로 기록(상태머신이 상한 강제하나 이중방어). 응답 자체 텍스트는 정서 표현이면 유지. """ reasons: list[str] = [] blocked = False needs_regen = False if text: for term in _MEANS_TERMS: if term in text: reasons.append(f"means_info:{term}") blocked = True needs_regen = True break if ideation_stage > IDEATION_STAGE_CAP: reasons.append(f"ideation_over_cap:{ideation_stage}>{IDEATION_STAGE_CAP}") blocked = True return OutputGuardResult( text=text, blocked=blocked, needs_regeneration=needs_regen, reasons=reasons, ) def clamp_ideation(stage: int) -> int: """ideation_stage 를 안전 상한(3)으로 클램프 (R5).""" return max(1, min(IDEATION_STAGE_CAP, stage)) __all__ = [ "IDEATION_STAGE_CAP", "MaskResult", "mask_pii", "CrisisKind", "CrisisResult", "classify_crisis", "OutputGuardResult", "sanitize_client_reply", "clamp_ideation", ]