724 lines
29 KiB
Python
724 lines
29 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 와 한국어 NER adapter 는 선택 의존(미설치 환경에서도 import 가능해야 함)
|
||
→ 지연 로드/명시 등록 + 정규식 폴백.
|
||
|
||
TODO(Phase 2): 한국어 NER adapter 실제 모델/provider 선정 + 한국어 자살콘텐츠 분류기
|
||
(JMIR few-shot 5단계, R8) 로 정밀화. 현재 정규식/키워드 폴백은 1차 안전망(재현율 우선).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from dataclasses import dataclass, field
|
||
from difflib import SequenceMatcher
|
||
from enum import Enum
|
||
from typing import Iterable, Protocol
|
||
|
||
# ── 출력 가드레일 상한 (R5) ──────────────────────────────
|
||
IDEATION_STAGE_CAP = 3 # 내담자 발화/상태가 넘을 수 없는 자살사고 단계 상한
|
||
CRISIS_HOTLINE_NUMBER = "109"
|
||
CRISIS_HOTLINE_LABEL = "자살예방상담전화 109"
|
||
CRISIS_RESOURCE_MESSAGE = (
|
||
"지금은 연습을 멈추고 실제 안전 확인이 먼저입니다. 즉시 위험하면 119 또는 가까운 "
|
||
"응급실에 연락하고, 자살예방상담전화 109로 도움을 요청하세요."
|
||
)
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════════════════════
|
||
# 1. PII 마스킹 (입력 — 저장·외부전송 전 하드 게이트, F-03)
|
||
# ════════════════════════════════════════════════════════════════════════════
|
||
# 정규식 폴백 패턴 (Presidio 미설치 시). 한국 맥락 우선.
|
||
# TODO: 실제 한국어 NER adapter 로 정밀화(이름/주소/기관 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_CONTEXTLESS_NAME = rf"[{_KOREAN_SURNAME_CHARS}][가-힣]{{2,3}}"
|
||
# 인명 마지막 글자로 사실상 쓰이지 않는 용언 활용/명사화 꼬리 글자. 문맥 단서가
|
||
# 없는 인명 패턴이 "유의점은·방치되는·마무리했을·연결감과·표현함을·어지러움은"
|
||
# 같은 일반 단어를 이름으로 오탐해 평가 문장을 훼손하는 것을 막는다(정밀도 가드).
|
||
_KOREAN_NAME_TAIL_GUARD = "(?<![했됐되된될겠것듯점음움됨함감])"
|
||
_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}){_KOREAN_NAME_TAIL_GUARD}"
|
||
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_CONTEXTLESS_NAME}){_KOREAN_NAME_TAIL_GUARD}"
|
||
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,}(?:동|읍|면|로|길)")),
|
||
]
|
||
|
||
# Source/input text keeps the broad contextless-name heuristic above for recall.
|
||
# Generated synthetic content has no user-originated text after source validation;
|
||
# applying that heuristic there misclassifies ordinary words such as "서운함을".
|
||
# Keep every high-confidence label/context/honorific pattern and all non-name PII.
|
||
_SYNTHETIC_GENERATED_PII_PATTERNS = (
|
||
*_PII_PATTERNS[:5],
|
||
*_PII_PATTERNS[6:],
|
||
)
|
||
|
||
# Presidio 지연 로드 캐시 (-1=미시도, None=미설치, 객체=설치됨)
|
||
_PRESIDIO_ANALYZER: object = -1
|
||
_PRESIDIO_ANONYMIZER: object = -1
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class PiiEntitySpan:
|
||
entity_type: str
|
||
start: int
|
||
end: int
|
||
|
||
|
||
class KoPiiRecognizer(Protocol):
|
||
"""Optional Korean PII recognizer. Implementations must be local and side-effect free."""
|
||
|
||
def analyze(self, text: str) -> Iterable[PiiEntitySpan]:
|
||
...
|
||
|
||
|
||
_KO_PII_RECOGNIZER: KoPiiRecognizer | None = None
|
||
|
||
|
||
def set_ko_pii_recognizer(recognizer: KoPiiRecognizer | None) -> None:
|
||
"""Register an optional Korean PII recognizer. None keeps regex-only behavior."""
|
||
global _KO_PII_RECOGNIZER
|
||
_KO_PII_RECOGNIZER = recognizer
|
||
|
||
|
||
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
|
||
used_ko_recognizer: bool = False
|
||
|
||
|
||
def _mask_regex_pii(
|
||
text: str,
|
||
patterns: Iterable[tuple[str, re.Pattern[str]]] = _PII_PATTERNS,
|
||
) -> 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 patterns:
|
||
masked = pat.sub(replace_match(label), masked)
|
||
return masked, sorted(set(found))
|
||
|
||
|
||
def _mask_span_pii(text: str, spans: Iterable[PiiEntitySpan]) -> tuple[str, list[str]]:
|
||
valid: list[PiiEntitySpan] = []
|
||
last_end = -1
|
||
for span in sorted(spans, key=lambda item: (item.start, item.end)):
|
||
label = span.entity_type.strip().upper()
|
||
if not label or span.start < 0 or span.end <= span.start or span.end > len(text):
|
||
continue
|
||
if span.start < last_end:
|
||
continue
|
||
last_end = span.end
|
||
valid.append(PiiEntitySpan(label, span.start, span.end))
|
||
if not valid:
|
||
return text, []
|
||
|
||
masked = text
|
||
for span in sorted(valid, key=lambda item: item.start, reverse=True):
|
||
masked = f"{masked[:span.start]}[{span.entity_type}]{masked[span.end:]}"
|
||
return masked, sorted({span.entity_type for span in valid})
|
||
|
||
|
||
def _mask_ko_recognizer_pii(text: str) -> tuple[str, list[str], bool]:
|
||
recognizer = _KO_PII_RECOGNIZER
|
||
if recognizer is None:
|
||
return text, [], False
|
||
try:
|
||
masked, entities = _mask_span_pii(text, recognizer.analyze(text))
|
||
return masked, entities, bool(entities)
|
||
except Exception:
|
||
return text, [], 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)
|
||
ko_masked, ko_ents, used_ko = _mask_ko_recognizer_pii(anonymized.text)
|
||
masked, regex_ents = _mask_regex_pii(ko_masked)
|
||
return MaskResult(
|
||
text_masked=masked,
|
||
entities=sorted(set(ents + ko_ents + regex_ents)),
|
||
used_presidio=True,
|
||
used_ko_recognizer=used_ko,
|
||
)
|
||
except Exception:
|
||
pass # 폴백으로
|
||
|
||
# 정규식 폴백
|
||
ko_masked, ko_ents, used_ko = _mask_ko_recognizer_pii(text)
|
||
masked, found = _mask_regex_pii(ko_masked)
|
||
return MaskResult(
|
||
text_masked=masked,
|
||
entities=sorted(set(ko_ents + found)),
|
||
used_presidio=False,
|
||
used_ko_recognizer=used_ko,
|
||
)
|
||
|
||
|
||
def mask_synthetic_generated_pii(text: str) -> MaskResult:
|
||
"""PII gate for model-generated text from validated synthetic sources.
|
||
|
||
Explicit name labels, self-introductions, relationship/name contexts,
|
||
honorifics, optional recognizers, Presidio and every non-name PII pattern stay
|
||
enabled. Only the ambiguous contextless Korean surname heuristic is omitted.
|
||
"""
|
||
|
||
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")
|
||
ents = sorted({r.entity_type for r in results})
|
||
anonymized = anonymizer.anonymize(text=text, analyzer_results=results)
|
||
ko_masked, ko_ents, used_ko = _mask_ko_recognizer_pii(anonymized.text)
|
||
masked, regex_ents = _mask_regex_pii(
|
||
ko_masked,
|
||
_SYNTHETIC_GENERATED_PII_PATTERNS,
|
||
)
|
||
return MaskResult(
|
||
text_masked=masked,
|
||
entities=sorted(set(ents + ko_ents + regex_ents)),
|
||
used_presidio=True,
|
||
used_ko_recognizer=used_ko,
|
||
)
|
||
except Exception:
|
||
pass
|
||
|
||
ko_masked, ko_ents, used_ko = _mask_ko_recognizer_pii(text)
|
||
masked, found = _mask_regex_pii(
|
||
ko_masked,
|
||
_SYNTHETIC_GENERATED_PII_PATTERNS,
|
||
)
|
||
return MaskResult(
|
||
text_masked=masked,
|
||
entities=sorted(set(ko_ents + found)),
|
||
used_presidio=False,
|
||
used_ko_recognizer=used_ko,
|
||
)
|
||
|
||
|
||
_IDENTITY_SEGMENT_RE = re.compile(r"\s*[·|,/]\s*", re.UNICODE)
|
||
_IDENTITY_UUID_RE = re.compile(
|
||
r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$",
|
||
re.IGNORECASE,
|
||
)
|
||
_IDENTITY_KO_PARTICLE_LOOKAHEAD = (
|
||
r"(?:은|는|이|가|을|를|와|과|의|도|에게|께|랑|하고|님|씨)"
|
||
)
|
||
|
||
|
||
def _known_identity_variants(identity: str | None) -> list[str]:
|
||
"""Return conservative display-name variants that are safe to role-tokenize."""
|
||
|
||
raw = str(identity or "").strip()
|
||
if not raw or "@" in raw or _IDENTITY_UUID_RE.fullmatch(raw):
|
||
return []
|
||
first_segment = _IDENTITY_SEGMENT_RE.split(raw, maxsplit=1)[0].strip()
|
||
first_segment = re.sub(r"\s*\(가명\)\s*$", "", first_segment).strip()
|
||
variants: list[str] = []
|
||
for candidate in (raw, first_segment):
|
||
if candidate in variants:
|
||
continue
|
||
letters = re.sub(r"[^A-Za-z가-힣]", "", candidate)
|
||
if len(letters) < 2:
|
||
continue
|
||
variants.append(candidate)
|
||
return sorted(variants, key=len, reverse=True)
|
||
|
||
|
||
def mask_role_identities(
|
||
text: str,
|
||
*,
|
||
counselor_identity: str | None = None,
|
||
client_identity: str | None = None,
|
||
synthetic_generated: bool = False,
|
||
) -> MaskResult:
|
||
"""Mask known session identities with role tokens, then apply the normal PII gate.
|
||
|
||
Only identities already owned by the authenticated session are role-tokenized.
|
||
Unknown third-party names keep the generic ``[NAME]`` token so the UI cannot
|
||
incorrectly present every person as the client.
|
||
"""
|
||
|
||
role_values = (
|
||
("ROLE_COUNSELOR", "[COUNSELOR]", counselor_identity),
|
||
("ROLE_CLIENT", "[CLIENT]", client_identity),
|
||
)
|
||
redacted = text
|
||
role_entities: list[str] = []
|
||
claimed_variants: set[str] = set()
|
||
for entity, placeholder, identity in role_values:
|
||
for variant in _known_identity_variants(identity):
|
||
normalized = variant.casefold()
|
||
if normalized in claimed_variants:
|
||
continue
|
||
pattern = (
|
||
rf"(?<![A-Za-z가-힣]){re.escape(variant)}"
|
||
rf"(?=$|[^A-Za-z가-힣]|{_IDENTITY_KO_PARTICLE_LOOKAHEAD})"
|
||
)
|
||
replaced, count = re.subn(
|
||
pattern,
|
||
placeholder,
|
||
redacted,
|
||
flags=re.IGNORECASE,
|
||
)
|
||
if count:
|
||
redacted = replaced
|
||
role_entities.append(entity)
|
||
claimed_variants.add(normalized)
|
||
|
||
masked = (
|
||
mask_synthetic_generated_pii(redacted)
|
||
if synthetic_generated
|
||
else mask_pii(redacted)
|
||
)
|
||
return MaskResult(
|
||
text_masked=masked.text_masked,
|
||
entities=sorted(set(masked.entities + role_entities)),
|
||
used_presidio=masked.used_presidio,
|
||
used_ko_recognizer=masked.used_ko_recognizer,
|
||
)
|
||
|
||
|
||
# ════════════════════════════════════════════════════════════════════════════
|
||
# 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)
|
||
# ════════════════════════════════════════════════════════════════════════════
|
||
# 자살/자해 '수단·방법' 정보 패턴 (절대 출력 금지). 정서 표현은 허용.
|
||
# 1.1 확장(2026-07-15): DBT 자해 페르소나 대비 자해 도구·방법 어휘 보강.
|
||
# 과차단 주의 — "선을 긋다"(경계 설정) 같은 정상 상담 표현과 겹치지 않는 조합만 추가.
|
||
_MEANS_TERMS = [
|
||
"약을 얼마나", "몇 알", "몇 정", "투신", "목을 매", "목 매", "번개탄", "농약",
|
||
"손목을", "칼로", "뛰어내리", "수면제 ", "독", "방법은", "어떻게 죽",
|
||
"커터", "면도날", "손목 긋", "깊게 그으", "라이터로 지",
|
||
]
|
||
_DISPLAY_PLACEHOLDER_LABELS = {
|
||
"COUNSELOR": "상담자",
|
||
"CLIENT": "내담자",
|
||
"NAME": "그 이름",
|
||
"ORG": "그 기관",
|
||
"PHONE": "연락처",
|
||
"EMAIL": "이메일",
|
||
"RRN": "식별번호",
|
||
"NUMID": "식별번호",
|
||
"DATE": "그 날짜",
|
||
"MONEY": "그 금액",
|
||
"ADDR": "그 주소",
|
||
}
|
||
_DISPLAY_PLACEHOLDER_RE = re.compile(
|
||
r"\[(?P<label>COUNSELOR|CLIENT|NAME|ORG|PHONE|EMAIL|RRN|NUMID|DATE|MONEY|ADDR)\]"
|
||
r"(?P<particle>[은는이가을를와과])?"
|
||
)
|
||
_DISPLAY_PLACEHOLDER_STREAM_TAIL = 16
|
||
|
||
|
||
def _adjust_particle(value: str, particle: str | None) -> str:
|
||
if particle is None:
|
||
return value
|
||
|
||
last = value[-1] if value else ""
|
||
codepoint = ord(last) - 0xAC00 if "가" <= last <= "힣" else -1
|
||
has_batchim = codepoint >= 0 and codepoint % 28 != 0
|
||
if particle in ("은", "는"):
|
||
return value + ("은" if has_batchim else "는")
|
||
if particle in ("이", "가"):
|
||
return value + ("이" if has_batchim else "가")
|
||
if particle in ("을", "를"):
|
||
return value + ("을" if has_batchim else "를")
|
||
if particle in ("와", "과"):
|
||
return value + ("과" if has_batchim else "와")
|
||
return value + particle
|
||
|
||
|
||
def humanize_pii_placeholders(text: str) -> str:
|
||
"""사용자에게 보이는 내담자 응답에서 PII placeholder 토큰을 자연어로 낮춘다."""
|
||
if not text:
|
||
return text
|
||
|
||
def _replace(match: re.Match[str]) -> str:
|
||
label = match.group("label")
|
||
replacement = _DISPLAY_PLACEHOLDER_LABELS.get(label, "그 정보")
|
||
return _adjust_particle(replacement, match.group("particle"))
|
||
|
||
return _DISPLAY_PLACEHOLDER_RE.sub(_replace, text)
|
||
|
||
|
||
class PiiPlaceholderStreamSanitizer:
|
||
"""SSE 토큰 경계를 가로질러 나온 PII placeholder를 사용자 표시 전에 치환한다."""
|
||
|
||
def __init__(self) -> None:
|
||
self._tail = ""
|
||
|
||
def feed(self, chunk: str) -> str:
|
||
if not chunk:
|
||
return ""
|
||
self._tail += chunk
|
||
if len(self._tail) <= _DISPLAY_PLACEHOLDER_STREAM_TAIL:
|
||
return ""
|
||
ready = self._tail[:-_DISPLAY_PLACEHOLDER_STREAM_TAIL]
|
||
self._tail = self._tail[-_DISPLAY_PLACEHOLDER_STREAM_TAIL:]
|
||
return humanize_pii_placeholders(ready)
|
||
|
||
def flush(self) -> str:
|
||
tail = self._tail
|
||
self._tail = ""
|
||
return humanize_pii_placeholders(tail)
|
||
|
||
|
||
@dataclass(slots=True)
|
||
class OutputGuardResult:
|
||
text: str # 정제된(또는 원본) 응답
|
||
blocked: bool = False # 차단/치환 발생 여부
|
||
needs_regeneration: bool = False # 재생성 필요(수단정보 누출 시)
|
||
reasons: list[str] = field(default_factory=list)
|
||
|
||
|
||
_ROLE_META_PATTERNS = [
|
||
re.compile(r"(?:내담자|상담자)\s*역할\s*로?\s*응답"),
|
||
re.compile(r"AI\s*로서"),
|
||
re.compile(r"상담자\s*입장\s*에서"),
|
||
re.compile(r"제\s*(?:핵심신념|자동적\s*사고|인지왜곡)\s*은"),
|
||
]
|
||
_OPENING_GREETING_RE = re.compile(r"^\s*(?:안녕하세요|처음\s*뵙겠습니다|반갑습니다)(?:[\s,.!?。！？]|$)")
|
||
_DUPLICATE_PUNCT_RE = re.compile(r"[\s\.,!?。！？…\"'“”‘’()\[\]{}:;·~\-]+")
|
||
|
||
|
||
def _normalize_duplicate_text(value: str) -> str:
|
||
return _DUPLICATE_PUNCT_RE.sub("", value.casefold())
|
||
|
||
|
||
def _duplicate_tokens(value: str) -> list[str]:
|
||
cleaned = _DUPLICATE_PUNCT_RE.sub(" ", value.casefold())
|
||
return [token for token in cleaned.split() if len(token) >= 2]
|
||
|
||
|
||
def _is_near_duplicate_client_reply(text: str, previous: str) -> bool:
|
||
current_norm = _normalize_duplicate_text(text)
|
||
previous_norm = _normalize_duplicate_text(previous)
|
||
if len(current_norm) < 12 or len(previous_norm) < 12:
|
||
return False
|
||
if current_norm == previous_norm or current_norm in previous_norm or previous_norm in current_norm:
|
||
return True
|
||
if SequenceMatcher(None, current_norm, previous_norm).ratio() >= 0.88:
|
||
return True
|
||
|
||
current_tokens = set(_duplicate_tokens(text))
|
||
previous_tokens = set(_duplicate_tokens(previous))
|
||
if min(len(current_tokens), len(previous_tokens)) < 4:
|
||
return False
|
||
overlap = len(current_tokens & previous_tokens) / min(len(current_tokens), len(previous_tokens))
|
||
return overlap >= 0.82
|
||
|
||
|
||
def sanitize_client_reply(
|
||
text: str,
|
||
*,
|
||
ideation_stage: int,
|
||
turn_seq: int | None = None,
|
||
previous_client_reply: str | None = None,
|
||
) -> 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 any(pattern.search(text) for pattern in _ROLE_META_PATTERNS):
|
||
reasons.append("role_meta")
|
||
blocked = True
|
||
needs_regen = True
|
||
if turn_seq is not None and turn_seq > 1 and _OPENING_GREETING_RE.search(text):
|
||
reasons.append("repeat_greeting_after_opening")
|
||
blocked = True
|
||
needs_regen = True
|
||
if previous_client_reply and _is_near_duplicate_client_reply(text, previous_client_reply):
|
||
reasons.append("duplicate_client_reply")
|
||
blocked = True
|
||
needs_regen = True
|
||
|
||
if ideation_stage > IDEATION_STAGE_CAP:
|
||
reasons.append(f"ideation_over_cap:{ideation_stage}>{IDEATION_STAGE_CAP}")
|
||
blocked = True
|
||
|
||
return OutputGuardResult(
|
||
text=humanize_pii_placeholders(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",
|
||
"PiiEntitySpan",
|
||
"set_ko_pii_recognizer",
|
||
"mask_pii",
|
||
"mask_role_identities",
|
||
"mask_synthetic_generated_pii",
|
||
"CrisisKind",
|
||
"CrisisResult",
|
||
"classify_crisis",
|
||
"OutputGuardResult",
|
||
"PiiPlaceholderStreamSanitizer",
|
||
"humanize_pii_placeholders",
|
||
"sanitize_client_reply",
|
||
"clamp_ideation",
|
||
"crisis_resource",
|
||
]
|