한국어 PII 인식기 경계 추가
This commit is contained in:
Yun Chan
parent
2bb052f624
commit
84599bbaa2
3 changed files with 156 additions and 9 deletions
|
|
@ -8,10 +8,11 @@ MASTERPLAN §2.2 / §3.3 / R5 / R7 / F-03, MEMORY_DESIGN §B:
|
|||
- 모듈 경계 명확: 입력 가드레일(mask_pii / classify_crisis)과 출력 가드레일(sanitize_client_reply)을
|
||||
순수함수에 가깝게 분리. IO·LLM·DB 의존 없음(테스트·재사용 용이).
|
||||
- 외부 LLM 경로 진입 전 *하드 게이트*: 마스킹 안 된 원문은 게이트웨이로 절대 안 나간다(F-03).
|
||||
- Presidio 는 선택 의존(미설치 환경에서도 import 가능해야 함) → 지연 로드 + 정규식 폴백.
|
||||
- Presidio 와 한국어 NER adapter 는 선택 의존(미설치 환경에서도 import 가능해야 함)
|
||||
→ 지연 로드/명시 등록 + 정규식 폴백.
|
||||
|
||||
TODO(Phase 2): Presidio MedicalNERRecognizer + 한국어 자살콘텐츠 분류기(JMIR few-shot 5단계, R8)
|
||||
로 교체. 현재 정규식/키워드 폴백은 1차 안전망(재현율 우선).
|
||||
TODO(Phase 2): 한국어 NER adapter 실제 모델/provider 선정 + 한국어 자살콘텐츠 분류기
|
||||
(JMIR few-shot 5단계, R8) 로 정밀화. 현재 정규식/키워드 폴백은 1차 안전망(재현율 우선).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
|
@ -19,7 +20,7 @@ from __future__ import annotations
|
|||
import re
|
||||
from dataclasses import dataclass, field
|
||||
from enum import Enum
|
||||
from typing import Optional
|
||||
from typing import Iterable, Protocol
|
||||
|
||||
# ── 출력 가드레일 상한 (R5) ──────────────────────────────
|
||||
IDEATION_STAGE_CAP = 3 # 내담자 발화/상태가 넘을 수 없는 자살사고 단계 상한
|
||||
|
|
@ -35,7 +36,7 @@ CRISIS_RESOURCE_MESSAGE = (
|
|||
# 1. PII 마스킹 (입력 — 저장·외부전송 전 하드 게이트, F-03)
|
||||
# ════════════════════════════════════════════════════════════════════════════
|
||||
# 정규식 폴백 패턴 (Presidio 미설치 시). 한국 맥락 우선.
|
||||
# TODO: Presidio + MedicalNERRecognizer 로 정밀화(이름/주소/기관 NER).
|
||||
# TODO: 실제 한국어 NER adapter 로 정밀화(이름/주소/기관 NER).
|
||||
_KOREAN_SURNAME_CHARS = (
|
||||
"김이박최정강조윤장임한오서신권황안송전홍유고문양손배백허남심노하"
|
||||
"곽성차주우구민류나진지엄채원천방공현함변염여추도소석선설마길연위표"
|
||||
|
|
@ -161,6 +162,29 @@ _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
|
||||
|
|
@ -183,6 +207,7 @@ 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) -> tuple[str, list[str]]:
|
||||
|
|
@ -209,6 +234,37 @@ def _mask_regex_pii(text: str) -> tuple[str, list[str]]:
|
|||
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 가용 시 우선, 아니면 정규식 폴백.
|
||||
|
||||
|
|
@ -223,18 +279,26 @@ def mask_pii(text: str) -> MaskResult:
|
|||
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)
|
||||
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 + regex_ents)),
|
||||
entities=sorted(set(ents + ko_ents + regex_ents)),
|
||||
used_presidio=True,
|
||||
used_ko_recognizer=used_ko,
|
||||
)
|
||||
except Exception:
|
||||
pass # 폴백으로
|
||||
|
||||
# 정규식 폴백
|
||||
masked, found = _mask_regex_pii(text)
|
||||
return MaskResult(text_masked=masked, entities=found, used_presidio=False)
|
||||
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,
|
||||
)
|
||||
|
||||
|
||||
# ════════════════════════════════════════════════════════════════════════════
|
||||
|
|
@ -369,6 +433,8 @@ __all__ = [
|
|||
"CRISIS_HOTLINE_LABEL",
|
||||
"CRISIS_RESOURCE_MESSAGE",
|
||||
"MaskResult",
|
||||
"PiiEntitySpan",
|
||||
"set_ko_pii_recognizer",
|
||||
"mask_pii",
|
||||
"CrisisKind",
|
||||
"CrisisResult",
|
||||
|
|
|
|||
|
|
@ -139,7 +139,9 @@ class OrchestratorMaskingGateTest(unittest.IsolatedAsyncioTestCase):
|
|||
return_value=(None, None),
|
||||
)
|
||||
self.presidio_patch.start()
|
||||
guardrail.set_ko_pii_recognizer(None)
|
||||
self.addCleanup(self.presidio_patch.stop)
|
||||
self.addCleanup(guardrail.set_ko_pii_recognizer, None)
|
||||
|
||||
def test_prepare_turn_keeps_raw_text_but_builds_masked_engine_messages(self) -> None:
|
||||
ctx = _prepare_context()
|
||||
|
|
@ -215,6 +217,39 @@ class OrchestratorMaskingGateTest(unittest.IsolatedAsyncioTestCase):
|
|||
self.assertIn(raw, ctx.learner_text_raw)
|
||||
self.assertNotIn(raw, ctx.learner_text_masked)
|
||||
|
||||
def test_prepare_turn_applies_optional_ko_recognizer_before_regex_fallback(self) -> None:
|
||||
class FakeKoRecognizer:
|
||||
def analyze(self, text: str):
|
||||
spans = []
|
||||
for entity_type, value in (
|
||||
("NAME", "보라별"),
|
||||
("ORG", "미래학교상담연구랩"),
|
||||
):
|
||||
start = text.index(value)
|
||||
spans.append(guardrail.PiiEntitySpan(entity_type, start, start + len(value)))
|
||||
return spans
|
||||
|
||||
guardrail.set_ko_pii_recognizer(FakeKoRecognizer())
|
||||
raw_text = "별명은 보라별이고 기관은 미래학교상담연구랩입니다. 전화는 010-1234-5678입니다."
|
||||
|
||||
ctx = orchestrator.prepare_turn(
|
||||
session_id="masking-session",
|
||||
case_id="masking-case",
|
||||
card=persona.P1,
|
||||
state=_initial_state(),
|
||||
learner_text=raw_text,
|
||||
)
|
||||
|
||||
blob = _message_blob(ctx.messages)
|
||||
self.assertIn("보라별", ctx.learner_text_raw)
|
||||
self.assertIn("미래학교상담연구랩", ctx.learner_text_raw)
|
||||
for raw in ("보라별", "미래학교상담연구랩", "010-1234-5678"):
|
||||
self.assertNotIn(raw, ctx.learner_text_masked)
|
||||
self.assertNotIn(raw, blob)
|
||||
for masked in ("[NAME]", "[ORG]", "[PHONE]"):
|
||||
self.assertIn(masked, ctx.learner_text_masked)
|
||||
self.assertIn(masked, blob)
|
||||
|
||||
def test_prepare_turn_threads_theory_mode_into_engine_messages(self) -> None:
|
||||
ctx = orchestrator.prepare_turn(
|
||||
session_id="theory-session",
|
||||
|
|
|
|||
|
|
@ -54,7 +54,9 @@ class PiiMaskingEvalTests(unittest.TestCase):
|
|||
return_value=(None, None),
|
||||
)
|
||||
self.presidio_patch.start()
|
||||
guardrail.set_ko_pii_recognizer(None)
|
||||
self.addCleanup(self.presidio_patch.stop)
|
||||
self.addCleanup(guardrail.set_ko_pii_recognizer, None)
|
||||
|
||||
def test_fixture_cases_are_valid_json_list(self) -> None:
|
||||
cases = load_cases(FIXTURE_PATH)
|
||||
|
|
@ -110,6 +112,50 @@ class PiiMaskingEvalTests(unittest.TestCase):
|
|||
result = guardrail.mask_pii(raw)
|
||||
self.assertEqual(result.text_masked, expected)
|
||||
|
||||
def test_optional_ko_recognizer_masks_adapter_spans_before_regex_fallback(self) -> None:
|
||||
class FakeKoRecognizer:
|
||||
def analyze(self, text: str):
|
||||
spans = []
|
||||
for entity_type, value in (
|
||||
("NAME", "보라별"),
|
||||
("ORG", "미래학교상담연구랩"),
|
||||
):
|
||||
start = text.index(value)
|
||||
spans.append(guardrail.PiiEntitySpan(entity_type, start, start + len(value)))
|
||||
return spans
|
||||
|
||||
guardrail.set_ko_pii_recognizer(FakeKoRecognizer())
|
||||
|
||||
result = guardrail.mask_pii(
|
||||
"별명은 보라별이고 기관은 미래학교상담연구랩입니다. 전화는 010-1234-5678입니다."
|
||||
)
|
||||
|
||||
self.assertTrue(result.used_ko_recognizer)
|
||||
self.assertFalse(result.used_presidio)
|
||||
self.assertIn("NAME", result.entities)
|
||||
self.assertIn("ORG", result.entities)
|
||||
self.assertIn("PHONE", result.entities)
|
||||
self.assertNotIn("보라별", result.text_masked)
|
||||
self.assertNotIn("미래학교상담연구랩", result.text_masked)
|
||||
self.assertNotIn("010-1234-5678", result.text_masked)
|
||||
self.assertIn("[NAME]", result.text_masked)
|
||||
self.assertIn("[ORG]", result.text_masked)
|
||||
self.assertIn("[PHONE]", result.text_masked)
|
||||
|
||||
def test_optional_ko_recognizer_failure_keeps_regex_fallback(self) -> None:
|
||||
class BrokenKoRecognizer:
|
||||
def analyze(self, text: str):
|
||||
raise RuntimeError("adapter down")
|
||||
|
||||
guardrail.set_ko_pii_recognizer(BrokenKoRecognizer())
|
||||
|
||||
result = guardrail.mask_pii("전화는 010-1234-5678입니다.")
|
||||
|
||||
self.assertFalse(result.used_ko_recognizer)
|
||||
self.assertFalse(result.used_presidio)
|
||||
self.assertIn("PHONE", result.entities)
|
||||
self.assertIn("[PHONE]", result.text_masked)
|
||||
|
||||
def test_cli_reports_json_and_nonzero_gate_shape(self) -> None:
|
||||
completed = subprocess.run(
|
||||
[
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue