380 lines
16 KiB
Python
380 lines
16 KiB
Python
"""입출력 가드레일 — 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 # 내담자 발화/상태가 넘을 수 없는 자살사고 단계 상한
|
||
CRISIS_HOTLINE_NUMBER = "109"
|
||
CRISIS_HOTLINE_LABEL = "자살예방상담전화 109"
|
||
CRISIS_RESOURCE_MESSAGE = (
|
||
"지금은 연습을 멈추고 실제 안전 확인이 먼저입니다. 즉시 위험하면 119 또는 가까운 "
|
||
"응급실에 연락하고, 자살예방상담전화 109로 도움을 요청하세요."
|
||
)
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════════════════════
|
||
# 1. PII 마스킹 (입력 — 저장·외부전송 전 하드 게이트, F-03)
|
||
# ════════════════════════════════════════════════════════════════════════════
|
||
# 정규식 폴백 패턴 (Presidio 미설치 시). 한국 맥락 우선.
|
||
# TODO: Presidio + MedicalNERRecognizer 로 정밀화(이름/주소/기관 NER).
|
||
_KOREAN_SURNAME_CHARS = (
|
||
"김이박최정강조윤장임한오서신권황안송전홍유고문양손배백허남심노하"
|
||
"곽성차주우구민류나진지엄채원천방공현함변염여추도소석선설마길연위표"
|
||
"명기반왕금옥육인맹제모탁국어은편용예봉경"
|
||
)
|
||
_KOREAN_FULL_NAME = rf"[{_KOREAN_SURNAME_CHARS}][가-힣]{{1,3}}"
|
||
_KOREAN_FULL_NAME_BEFORE_SUFFIX = rf"[{_KOREAN_SURNAME_CHARS}][가-힣]{{1,3}}?"
|
||
_KOREAN_NAME_STOPWORDS = {
|
||
"연락",
|
||
"연락처",
|
||
"이메일",
|
||
"주민번호",
|
||
"번호",
|
||
"이름",
|
||
"이야기",
|
||
"생각",
|
||
"마음",
|
||
"기분",
|
||
"상담",
|
||
"기록",
|
||
"진료",
|
||
"학교",
|
||
"엄마",
|
||
"아빠",
|
||
"어머니",
|
||
"아버지",
|
||
"친구",
|
||
"내담자",
|
||
"상담자",
|
||
"선생님",
|
||
"소속",
|
||
"안내",
|
||
}
|
||
_PII_PATTERNS: list[tuple[str, re.Pattern[str]]] = [
|
||
# 한국어 기관/소속명: 학교·병원·센터·학과 등 명시 suffix가 있는 경우만 보수적으로 마스킹.
|
||
(
|
||
"ORG",
|
||
re.compile(
|
||
r"(?<![가-힣A-Za-z0-9])"
|
||
r"(?P<value>[가-힣A-Za-z0-9·&().-]{2,30}?"
|
||
r"(?:대학교|대학원|고등학교|중학교|초등학교|병원|의원|클리닉|상담센터|센터|복지관|교육청|보건소|연구소|재단|협회|학과|학부))"
|
||
r"(?P<suffix>\s*(?:입니다|이에요|예요|이고|이고요|에서|에|의|은|는|이|가|을|를)?)"
|
||
r"(?=$|[\s,.;!?。])"
|
||
),
|
||
),
|
||
# 한국어 이름: 이름/성명/실명 라벨 뒤 값.
|
||
(
|
||
"NAME",
|
||
re.compile(
|
||
r"(?P<prefix>(?:이름|성명|실명|본명)\s*[:：]\s*)"
|
||
r"(?P<value>[가-힣]{2,4})"
|
||
r"(?=$|[\s,.;!?。])"
|
||
),
|
||
),
|
||
# 한국어 이름: "제 이름은 김서연입니다", "보호자 이름은 박민수입니다" 같은 자연 발화형 라벨.
|
||
(
|
||
"NAME",
|
||
re.compile(
|
||
r"(?P<prefix>(?:(?:제|저의|내|나의|보호자|학생|내담자|상담자|친구|엄마|아빠|어머니|아버지)\s+)?"
|
||
r"(?:이름|성명|실명|본명)\s*(?:은|는|이|가)?\s*)"
|
||
rf"(?P<value>{_KOREAN_FULL_NAME_BEFORE_SUFFIX})"
|
||
r"(?P<suffix>\s*(?:입니다|이에요|예요|이고|이고요|이라고|라고)?)"
|
||
r"(?=$|[\s,.;!?。])"
|
||
),
|
||
),
|
||
# 한국어 이름: "저는 김서연입니다", "제가 박민수예요", "김서연입니다" 같은 자기소개형 문장.
|
||
(
|
||
"NAME",
|
||
re.compile(
|
||
r"(?P<prefix>(?:(?:저는|나는|제가|내가)\s*)?)"
|
||
rf"(?P<value>{_KOREAN_FULL_NAME_BEFORE_SUFFIX})"
|
||
r"(?P<suffix>\s*(?:입니다|이에요|예요|이고|이고요))"
|
||
r"(?=$|[\s,.;!?。])"
|
||
),
|
||
),
|
||
# 한국어 이름: 역할/관계 명사 뒤에 붙은 인명 + 조사/호칭.
|
||
(
|
||
"NAME",
|
||
re.compile(
|
||
r"(?P<prefix>(?:내담자|상담자|학생|보호자|담임|교수|선생님|친구|엄마|아빠|어머니|아버지|동생|언니|오빠|형|누나)\s+)"
|
||
rf"(?P<value>{_KOREAN_FULL_NAME_BEFORE_SUFFIX})"
|
||
r"(?P<suffix>\s*(?:님|씨|학생|상담자|내담자)?"
|
||
r"(?:은|는|이|가|을|를|와|과|에게|한테|라고|이라는|입니다|이에요|예요|이고|이고요))"
|
||
),
|
||
),
|
||
# 한국어 이름: 성씨 기반 full-name + 조사. 문맥 없는 순수 2~4글자 마스킹은 오탐이 커서 피한다.
|
||
(
|
||
"NAME",
|
||
re.compile(
|
||
rf"(?<![가-힣])(?P<value>{_KOREAN_FULL_NAME_BEFORE_SUFFIX})"
|
||
r"(?P<suffix>(?:은|는|이|가|을|를|와|과|에게|한테|라고|이라는))"
|
||
),
|
||
),
|
||
# 한국어 이름: "김서연 씨", "박민수님" 같은 명시 호칭.
|
||
(
|
||
"NAME",
|
||
re.compile(
|
||
rf"(?<![가-힣])(?P<value>{_KOREAN_FULL_NAME_BEFORE_SUFFIX})"
|
||
r"(?P<suffix>\s?(?:씨|님)(?:은|는|이|가|을|를|와|과|에게|한테|고|이고|인데)?)"
|
||
r"(?=$|[\s,.;!?。])"
|
||
),
|
||
),
|
||
# 주민등록번호 (6자리-7자리)
|
||
("RRN", re.compile(r"(?<!\d)\d{6}[-\s]?\d{7}(?!\d)")),
|
||
# 휴대폰 (010-1234-5678 등)
|
||
("PHONE", re.compile(r"(?<!\d)01[016789][-\s]?\d{3,4}[-\s]?\d{4}(?!\d)")),
|
||
# 일반 전화
|
||
("PHONE", re.compile(r"(?<!\d)0\d{1,2}[-\s]?\d{3,4}[-\s]?\d{4}(?!\d)")),
|
||
# 이메일
|
||
("EMAIL", re.compile(r"\b[\w.+-]+@[\w-]+\.[\w.-]+\b")),
|
||
# 카드/계좌 유사 긴 숫자열 (12자리 이상)
|
||
("NUMID", re.compile(r"(?<!\d)\d{12,}(?!\d)")),
|
||
# 구체적 날짜(생년월일 등): 2001.4.18 / 2001-04-18 / 2001년 4월 18일
|
||
("DATE", re.compile(r"(?:19|20)\d{2}\s?[.\-/년]\s?\d{1,2}\s?[.\-/월]\s?\d{1,2}\s?일?")),
|
||
# 금액(원): 1,200원 / 1200원 (3자리+ 또는 콤마구분) — 식별 맥락 보호
|
||
("MONEY", re.compile(r"\d{1,3}(?:,\d{3})+\s?원|\d{3,}\s?원")),
|
||
# 한국 주소 단편: ○○시/도 ○○시/군/구 ○○동/읍/면/로/길 (행정구역 연쇄)
|
||
("ADDR", re.compile(r"[가-힣]{2,}(?:시|도)\s?[가-힣]{1,4}(?:시|군|구)\s?[가-힣0-9]{1,}(?:동|읍|면|로|길)")),
|
||
]
|
||
|
||
# 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_regex_pii(text: str) -> tuple[str, list[str]]:
|
||
masked = text
|
||
found: list[str] = []
|
||
|
||
def replace_match(label: str):
|
||
def _replace(match: re.Match[str]) -> str:
|
||
group = match.groupdict().get("value")
|
||
if group is None:
|
||
found.append(label)
|
||
return f"[{label}]"
|
||
if label == "NAME" and group in _KOREAN_NAME_STOPWORDS:
|
||
return match.group(0)
|
||
value_start = match.start("value") - match.start(0)
|
||
value_end = match.end("value") - match.start(0)
|
||
found.append(label)
|
||
return f"{match.group(0)[:value_start]}[{label}]{match.group(0)[value_end:]}"
|
||
|
||
return _replace
|
||
|
||
for label, pat in _PII_PATTERNS:
|
||
masked = pat.sub(replace_match(label), masked)
|
||
return masked, sorted(set(found))
|
||
|
||
|
||
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)
|
||
masked, regex_ents = _mask_regex_pii(anonymized.text)
|
||
return MaskResult(
|
||
text_masked=masked,
|
||
entities=sorted(set(ents + regex_ents)),
|
||
used_presidio=True,
|
||
)
|
||
except Exception:
|
||
pass # 폴백으로
|
||
|
||
# 정규식 폴백
|
||
masked, found = _mask_regex_pii(text)
|
||
return MaskResult(text_masked=masked, entities=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))
|
||
|
||
|
||
def crisis_resource() -> dict[str, str]:
|
||
"""LLM 밖 위기 안내 리소스. UI/API 응답에 그대로 실어 보낸다."""
|
||
return {
|
||
"title": CRISIS_HOTLINE_LABEL,
|
||
"number": CRISIS_HOTLINE_NUMBER,
|
||
"message": CRISIS_RESOURCE_MESSAGE,
|
||
}
|
||
|
||
|
||
__all__ = [
|
||
"IDEATION_STAGE_CAP",
|
||
"CRISIS_HOTLINE_NUMBER",
|
||
"CRISIS_HOTLINE_LABEL",
|
||
"CRISIS_RESOURCE_MESSAGE",
|
||
"MaskResult",
|
||
"mask_pii",
|
||
"CrisisKind",
|
||
"CrisisResult",
|
||
"classify_crisis",
|
||
"OutputGuardResult",
|
||
"sanitize_client_reply",
|
||
"clamp_ideation",
|
||
"crisis_resource",
|
||
]
|