한신대 피드백 개선팩 반영

apps/api/app/services/guardrail.py

@@ -19,6 +19,7 @@ from __future__ import annotations
 
 import re
 from dataclasses import dataclass, field
+from difflib import SequenceMatcher
 from enum import Enum
 from typing import Iterable, Protocol
 
@@ -448,7 +449,50 @@ class OutputGuardResult:
     reasons: list[str] = field(default_factory=list)
 
 
-def sanitize_client_reply(text: str, *, ideation_stage: int) -> OutputGuardResult:
+_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(차단, 재생성 신호).
@@ -466,6 +510,18 @@ def sanitize_client_reply(text: str, *, ideation_stage: int) -> OutputGuardResul
                 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}")