From 35a9d8802c691f774496207ee4c3d70f14425c16 Mon Sep 17 00:00:00 2001 From: Yun Chan Date: Fri, 28 Aug 2026 19:55:04 +0900 Subject: [PATCH] =?UTF-8?q?=EC=9C=84=EA=B8=B0=20=EC=8B=A0=ED=98=B8=20?= =?UTF-8?q?=EA=B8=B0=EC=88=A0=20=EA=B2=80=EC=A6=9D=EC=9D=84=20=EB=B3=B4?= =?UTF-8?q?=EA=B0=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- apps/api/app/services/guardrail.py | 263 +++++++++- apps/api/app/services/live_coach.py | 84 ++- apps/api/app/test_client_reply_quality.py | 220 ++++++++ apps/api/app/test_session_turn_persistence.py | 127 ++++- ...-clinical-crisis-technical-observations.py | 484 ++++++++++++++++++ 5 files changed, 1164 insertions(+), 14 deletions(-) create mode 100644 scripts/run-clinical-crisis-technical-observations.py diff --git a/apps/api/app/services/guardrail.py b/apps/api/app/services/guardrail.py index 3e85285..300637a 100644 --- a/apps/api/app/services/guardrail.py +++ b/apps/api/app/services/guardrail.py @@ -466,11 +466,257 @@ class CrisisResult: # 위기 표현 키워드(한국어 우선). TODO: JMIR 한국어 벤치 few-shot 분류기로 교체(R8). _CRISIS_TERMS = [ - "죽고 싶", "죽고싶", "자살", "목숨", "사라지고 싶", "없어지고 싶", - "자해", "끝내고 싶", "살기 싫", "살아서 뭐", "죽어야", + "죽고 싶", "죽고싶", "자살 생각", "자살하고 싶", "자살을 생각", "자살 충동", + "자살 시도", "목숨을 끊", "목숨을 버리", "사라지고 싶", "없어지고 싶", + "자해하고 싶", "자해한 적", "자해를 했", "자해 충동", "자해 시도", + "스스로를 다치게", "끝내고 싶", "살기 싫", + "살아서 뭐", "죽어야", "안전을 스스로 지키기 어렵", ] -# 실제 위기로 가중되는 1인칭 현재 단서(수련생 본인 신호일 가능성) -_FIRST_PERSON_NOW = ["지금 나", "나 진짜", "제가 지금", "저 지금", "real", "도와주세요"] +# 자살·자해 단어를 직접 쓰지 않더라도 본인의 현재 위험과 실제 도움 필요를 함께 +# 밝히는 짧은 진술은 엔진 호출 전에 막아야 한다. 단일 '위험'/'도움' 키워드만으로는 +# 상담자의 안전 질문까지 오탐할 수 있어 네 단서가 모두 함께 있을 때만 승격한다. +_SELF_SUBJECT = ( + r"(? bool: + """양성 구문과 실제로 겹치는 부정·인용·타인·질문 문맥만 무효화한다.""" + + for pattern in _INVALID_REAL_CRISIS_CONTEXTS: + for context in pattern.finditer(text): + if match.start() < context.end() and context.start() < match.end(): + return True + return False + + +def _current_personal_danger_help(text: str) -> bool: + """현재의 개인적 위험과 실제 도움 필요를 함께 밝힌 짧은 진술인지 확인한다.""" + + if len(text) > 240: + return False + danger_matches = [ + match + for pattern in (*_PERSONAL_DANGER_PATTERNS, *_PERSONAL_IMMEDIATE_HARM_PATTERNS) + for match in pattern.finditer(text) + if not _match_has_invalid_crisis_context(text, match) + ] + help_matches = [ + match + for match in _REAL_HELP_CUE.finditer(text) + if not _match_has_invalid_crisis_context(text, match) + ] + return any( + abs(danger.start() - help_match.start()) <= 160 + for danger in danger_matches + for help_match in help_matches + ) + + +def _first_person_current_crisis(text: str) -> bool: + if len(text) > 240: + return False + if any( + not _match_has_invalid_crisis_context(text, match) + for pattern in ( + *_FIRST_PERSON_CURRENT_CRISIS_PATTERNS, + *_FIRST_PERSON_EXPLICIT_CRISIS_PATTERNS, + ) + for match in pattern.finditer(text) + ): + return True + + direct_matches = [ + match + for match in re.finditer(_CURRENT_SUICIDE_OR_SELF_HARM, text) + if not _match_has_invalid_crisis_context(text, match) + ] + help_matches = [ + match + for match in _REAL_HELP_CUE.finditer(text) + if not _match_has_invalid_crisis_context(text, match) + ] + return any( + abs(direct.start() - help_match.start()) <= 160 + for direct in direct_matches + for help_match in help_matches + ) + + +def crisis_signal_matches(text: str) -> list[str]: + """화자 판정 전의 위기 내용 단서를 반환한다. + + 가상내담자 출력처럼 1인칭 표현이 정상인 경로에서는 이 함수로 내용 존재만 + 확인하고, 실제 수련생 위기 여부는 :func:`classify_crisis`가 별도로 판정한다. + """ + + if not text: + return [] + matched: list[str] = [] + for term in _CRISIS_TERMS: + for term_match in re.finditer(re.escape(term), text): + if not _match_has_invalid_crisis_context(text, term_match): + matched.append(term) + break + if any( + not _match_has_invalid_crisis_context(text, match) + for match in _INDIRECT_SELF_ERASURE_CUE.finditer(text) + ): + matched.append(_INDIRECT_SELF_ERASURE) + if _current_personal_danger_help(text): + matched.append(_CURRENT_PERSONAL_DANGER_HELP) + if _first_person_current_crisis(text): + matched.append(_FIRST_PERSON_CURRENT_CRISIS) + return matched def classify_crisis(text: str, *, speaker_is_persona_context: bool = True) -> CrisisResult: @@ -486,19 +732,20 @@ def classify_crisis(text: str, *, speaker_is_persona_context: bool = True) -> Cr """ if not text: return CrisisResult(kind=CrisisKind.NONE) - matched = [t for t in _CRISIS_TERMS if t in text] + matched = crisis_signal_matches(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] + explicit_real_signal = _CURRENT_PERSONAL_DANGER_HELP in matched + explicit_current_crisis = _FIRST_PERSON_CURRENT_CRISIS in matched - if not speaker_is_persona_context or real_cues: + if not speaker_is_persona_context or explicit_real_signal or explicit_current_crisis: # 실제 위기 가능 → 에스컬레이션(보수적) return CrisisResult( kind=CrisisKind.LEARNER_REAL, risk_level=max(risk, 4), - matched=matched + real_cues, + matched=matched, escalate=True, ) # 시뮬레이션 연기 맥락 diff --git a/apps/api/app/services/live_coach.py b/apps/api/app/services/live_coach.py index 85d67a9..093c14e 100644 --- a/apps/api/app/services/live_coach.py +++ b/apps/api/app/services/live_coach.py @@ -46,6 +46,7 @@ CoachFocus = Literal[ "goal", "pacing", ] +SafetySignalKind = Literal["ideation", "self_harm", "urgent"] class LiveCoachSource(BaseModel): @@ -180,6 +181,35 @@ def _configured_model(value: str | None) -> str | None: return model or None +_SELF_HARM_SIGNAL_CUES = ("자해", "스스로를 다치게", "스스로 다치게") +_URGENT_SAFETY_SIGNAL_CUES = ( + "안전을 스스로 지키기 어렵", + "안전을 지키기 어렵", + "위험할 것 같", +) +_PERSONA_SAFETY_NOTE = ( + "가상내담자 위기 신호에 대한 훈련 코칭입니다. 즉각 위험을 직접 확인하고 기관 지침에 따라 " + "보호체계 연결과 지도·지원을 이어가세요." +) + + +def _safety_signal_kind(item: LiveCoachInput) -> SafetySignalKind | None: + """이번 턴의 위기 내용을 화자 역할과 분리해 코칭 우선순위로 바꾼다.""" + + client_text = item.client_reply or "" + client_matches = guardrail.crisis_signal_matches(client_text) + learner_crisis = guardrail.classify_crisis(item.learner_text) + if not client_matches and learner_crisis.kind == guardrail.CrisisKind.NONE: + return None + + signal_text = f"{item.learner_text} {client_text}" + if any(cue in signal_text for cue in _URGENT_SAFETY_SIGNAL_CUES): + return "urgent" + if any(cue in signal_text for cue in _SELF_HARM_SIGNAL_CUES): + return "self_harm" + return "ideation" + + @lru_cache(maxsize=1) def _local_source_entries() -> tuple[tuple[str, dict[str, Any]], ...]: """Load repo-managed live-coach source packs. @@ -386,6 +416,7 @@ def _local_reference_grounding(item: LiveCoachInput, *, limit: int = 5) -> list[ " ".join(str((item.evaluation or {}).get(k, "")) for k in ("appropriateness", "appropriateness_note")), ] ).lower() + safety_kind = _safety_signal_kind(item) scored: list[tuple[int, int, dict[str, Any], dict[str, Any]]] = [] fallback: list[tuple[int, dict[str, Any], dict[str, Any]]] = [] for source_index, payload in enumerate(_local_source_payloads()): @@ -399,6 +430,16 @@ def _local_reference_grounding(item: LiveCoachInput, *, limit: int = 5) -> list[ fallback.append((source_index * 1000 + chunk_index, source, chunk)) keywords = [str(k).lower() for k in (chunk.get("keywords") or [])] score = sum(1 for kw in keywords if kw and kw in query) + if safety_kind and source.get("source_id") == "official_suicide_risk_guidelines": + chunk_id = str(chunk.get("id") or "") + if chunk_id == "risk-ideation-plan-intent-behavior": + score += 80 + if safety_kind == "self_harm" and chunk_id == "self-harm-psychosocial-assessment": + score += 120 + if safety_kind == "urgent" and chunk_id == "safety-plan-not-contract": + score += 130 + if safety_kind == "urgent" and chunk_id == "korea-109-emergency-connection": + score += 140 if score > 0: priority = int(source.get("priority") or 50) scored.append((score, priority, source, chunk)) @@ -586,11 +627,19 @@ def _fallback_suggestion( reason: Optional[str] = None, ) -> LiveCoachSuggestion: text = item.learner_text + safety_kind = _safety_signal_kind(item) + learner_crisis = guardrail.classify_crisis(item.learner_text) + safety_note_for_signal = ( + guardrail.CRISIS_RESOURCE_MESSAGE + if learner_crisis.escalate + else _PERSONA_SAFETY_NOTE + ) low_open = item.effective_openness < 0.35 tone: Tone = _evaluation_tone(item.evaluation) focus: CoachFocus = "exploration" title = "다음 탐색" message = "내담자 표현을 한 번 반영한 뒤, 방금 말한 장면을 더 구체적으로 물어봐라." + safety_note: Optional[str] = None # 단계별 기본 다음 발화 — 폴백에서도 회기 흐름에 맞는 제안을 낸다. stage_next_lines = { "라포": "오늘 이렇게 시간 내줘서 고마워요. 지금 마음이 어떤지 편한 만큼만 들려줄래요?", @@ -603,14 +652,39 @@ def _fallback_suggestion( "방금 말한 그 장면이 언제부터 특히 힘들게 느껴졌는지 조금만 더 들려줄래요?", ) - crisis = guardrail.classify_crisis(text) q = (item.question or "").strip() - if crisis.kind != guardrail.CrisisKind.NONE: + if safety_kind == "urgent": + tone = "warn" + focus = "risk" + title = "긴급 안전 연결" + message = ( + "일반 회기 목표를 멈추고 현재 안전을 확인해라. 혼자 두지 말고 현장 지지자와 109를 연결하며, " + "즉각 위험하면 119 또는 가까운 응급실을 우선한다." + ) + next_line = "지금 혼자 계신가요? 곁에 함께 있어 줄 사람과 즉시 연결해도 될까요?" + safety_note = safety_note_for_signal + elif safety_kind == "self_harm": + tone = "warn" + focus = "risk" + title = "자해 안전 확인" + message = ( + "자해를 관심 끌기나 의지 문제로 단정하지 말고 현재 안전, 심리사회적 맥락, 반복 위험과 " + "지지체계를 비판단적으로 확인해라. 구체적인 수단을 캐묻지 말고, 필요한 전문기관에 " + "연결한 뒤 안전 상태를 추후 확인할 계획을 함께 세워라." + ) + next_line = "지금은 안전한지, 그리고 다시 다치지 않도록 곁에서 도와줄 사람이 있는지 확인해도 될까요?" + safety_note = safety_note_for_signal + elif safety_kind == "ideation": tone = "warn" focus = "risk" title = "안전 먼저" - message = "위험 단서가 나온 턴이다. 방법을 캐묻지 말고 최근성, 강도, 보호요인을 차분히 확인해라." + message = ( + "위험 단서를 무시하거나 단정하지 말고 현재 사고의 최근성·빈도·강도, 계획 유무·의도·" + "위험 수단 접근 가능성을 안전 확보에 필요한 범위에서 직접 확인해라. 지지자·살아갈 이유·" + "도움 요청 가능성 같은 보호요인도 확인하되, 구체적 방법을 캐묻거나 묘사·교육하지 않는다." + ) next_line = "그 생각이 최근에 얼마나 자주, 얼마나 강하게 올라오는지 안전을 위해 같이 확인해도 될까요?" + safety_note = safety_note_for_signal elif q and any(k in q for k in ("저항", "방어", "닫", "안 열", "거부")): focus = "rapport" title = "저항 수용과 안전감" @@ -663,6 +737,7 @@ def _fallback_suggestion( next_utterance=next_line, rationale=rationale, sources=_source_refs(grounding), + safety_note=safety_note, latency_ms=0, ) @@ -786,7 +861,8 @@ async def generate_live_coaching( local_grounding = _local_reference_grounding(item) all_grounding = _external_safe_grounding([*local_grounding, *(grounding or [])]) crisis = guardrail.classify_crisis(item.learner_text) - if crisis.escalate: + client_crisis = bool(guardrail.crisis_signal_matches(item.client_reply or "")) + if crisis.escalate or client_crisis: return _fallback_suggestion(item, grounding=all_grounding, status="ready") started = time.perf_counter() diff --git a/apps/api/app/test_client_reply_quality.py b/apps/api/app/test_client_reply_quality.py index 29065a1..96636fd 100644 --- a/apps/api/app/test_client_reply_quality.py +++ b/apps/api/app/test_client_reply_quality.py @@ -2,6 +2,7 @@ from __future__ import annotations +import asyncio import importlib.util import json import unittest @@ -20,6 +21,16 @@ def _load_clinical_review_checker(): return module +def _load_clinical_observation_runner(): + runner_path = repo_path("scripts", "run-clinical-crisis-technical-observations.py") + spec = importlib.util.spec_from_file_location("clinical_crisis_observation_runner", runner_path) + if spec is None or spec.loader is None: + raise RuntimeError(f"임상 기술 관찰 runner를 불러올 수 없다: {runner_path}") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + class ClientReplyQualityGateTest(unittest.TestCase): def test_crisis_protocol_manifest_obeys_status_specific_external_review_contract(self) -> None: manifest = json.loads( @@ -140,6 +151,215 @@ class ClientReplyQualityGateTest(unittest.TestCase): self.assertTrue(means_detail.needs_regeneration) self.assertTrue(over_cap.blocked) + def test_exact_review_case_signals_drive_runtime_safety_gates(self) -> None: + case_set = json.loads( + repo_path("data", "clinical", "p1-crisis-review-cases.json").read_text( + encoding="utf-8", + ), + ) + cases = {case["case_id"]: case for case in case_set["cases"]} + + for case_id in ( + "P1-CRISIS-001", + "P1-CRISIS-002", + "P1-CRISIS-005", + "P1-CRISIS-006", + ): + with self.subTest(case_id=case_id): + signal = cases[case_id]["synthetic_scenario"]["signal"] + self.assertTrue(guardrail.crisis_signal_matches(signal)) + sanitized = guardrail.sanitize_client_reply( + signal, + ideation_stage=3, + turn_seq=4, + ) + self.assertFalse(sanitized.needs_regeneration) + + counselor_signal = cases["P1-CRISIS-003"]["synthetic_scenario"]["signal"] + counselor_result = guardrail.classify_crisis( + counselor_signal, + speaker_is_persona_context=True, + ) + self.assertEqual(counselor_result.kind, guardrail.CrisisKind.PERSONA_PLAY) + self.assertFalse(counselor_result.escalate) + + learner_signal = cases["P1-CRISIS-004"]["synthetic_scenario"]["signal"] + learner_result = guardrail.classify_crisis( + learner_signal, + speaker_is_persona_context=True, + ) + self.assertEqual(learner_result.kind, guardrail.CrisisKind.LEARNER_REAL) + self.assertGreaterEqual(learner_result.risk_level, 4) + self.assertTrue(learner_result.escalate) + self.assertIn("first_person_current_danger_help", learner_result.matched) + + counselor_questions = ( + "지금 위험할 것 같거나 실제로 도움이 필요하신가요?", + "지금 제 질문에 답하기 어렵다면 넘어가도 돼요. 자살 생각이 있나요?", + "지금 제 앞에서 안전을 스스로 지키기 어렵다고 느끼시나요?", + "자살 생각이 들면 ‘도와주세요’라고 말할 수 있나요?", + ( + "내담자가 ‘저 자신이 지금 위험할 것 같아요. 실제로 도움이 필요해요.’라고 " + "말하면 어떻게 확인해야 하나요?" + ), + ) + for question in counselor_questions: + with self.subTest(question=question): + counselor_question = guardrail.classify_crisis( + question, + speaker_is_persona_context=True, + ) + self.assertFalse(counselor_question.escalate) + + real_help_questions = ( + "저 자신이 지금 위험할 것 같다고 느끼고 실제로 도움이 필요해요. 어떻게 해야 하나요?", + "저는 지금 위험하다고 느끼며 실제로 도움이 필요해요. 도와줄 수 있나요?", + "제가 현재 위험하다고 느끼고 있어요. 실제로 도움이 필요해요?", + "저는 학생인데 지금 제가 위험하다고 느끼고 실제로 도움이 필요해요.", + "저는 내담자지만 지금 제가 위험하다고 느끼고 실제로 도움이 필요해요.", + "상대가 떠난 뒤 저는 지금 위험하다고 느끼고 실제로 도움이 필요해요.", + "지금 제가 위험한 것 같고 도움이 필요해요.", + "지금 저는 위험하다고 느껴요. 도움이 필요합니다.", + "저 스스로 지금 위험할 것 같아요. 도움이 필요해요.", + "제 안전이 지금 위험해요. 실제 도움이 필요해요.", + "제가 지금 안전하지 않아서 도와주세요.", + "저는 지금 당장 스스로를 해칠까 봐 무서워요. 도와주세요.", + "지금 제가 저를 해칠 것 같아요. 도와주세요.", + "저는 지금 죽어버리고 싶어요. 도와주세요.", + "저 지금 제 목숨을 끊고 싶어요.", + "저는 당장 죽고 싶어요.", + "저는 죽고 싶어요. 제발 도와주세요.", + "제가 자살하고 싶어요. 도움이 필요해요.", + "저는 지금 살고 싶지 않아요. 도와주세요.", + "저는 지금 삶을 끝내고 싶어요. 도와주세요.", + "저는 지금 없어지고 싶어요. 도와주세요.", + "저 지금 스스로를 다치게 하고 싶어요. 도와주세요.", + "지금은 제가 제 안전을 지킬 자신이 없어요. 도와주세요.", + "아까는 자살하고 싶지 않았지만 저는 지금 죽고 싶어요.", + "죽고 싶어하는 친구를 돕다가 저도 지금 죽고 싶어요.", + "죽고 싶은 생각이 있는지 묻는 연습 중이었는데 저는 지금 정말 죽고 싶어요.", + "저는 지금 죽고 싶어요. 친구는 자살하고 싶지 않아요.", + "아까는 도움이 필요하지 않았지만 저는 지금 위험하고 실제로 도움이 필요해요.", + "저는 오늘 죽고 싶어요. 도와주세요.", + "죽고 싶어요. 저 좀 도와주세요.", + "저는 지금 죽어야 할 것 같아요. 도와주세요.", + "저는 지금 통제가 안 되고 저를 다치게 할 것 같아요. 도와주세요.", + "저는 지금 너무 위험한 상태예요. 도움이 필요해요.", + "저한테 지금 자해 충동이 있어요. 도와주세요.", + "저는 현재 살아갈 이유가 없어요. 도와주세요.", + "저는 현재 제 자신을 해칠 것 같아요. 도와주세요.", + "저는 지금 자살할 것 같아요. 도와주세요.", + "저는 지금 자해할 것 같아요. 도와주세요.", + "저는 지금 안전을 지킬 수 없어요. 도와주세요.", + "저는 지금 저를 해칠지도 몰라요. 도와주세요.", + "제가 지금 저를 다치게 할까 봐 두려워요. 도움이 필요해요.", + "저는 지금 혼자 있으면 위험해요. 도와주세요.", + ) + for signal in real_help_questions: + with self.subTest(signal=signal): + result = guardrail.classify_crisis( + signal, + speaker_is_persona_context=True, + ) + self.assertEqual(result.kind, guardrail.CrisisKind.LEARNER_REAL) + self.assertTrue(result.escalate) + + nonclinical_first_person = guardrail.classify_crisis( + "제가 지금 과제 제출이 위험할 것 같아서 실제로 도움이 필요해요.", + speaker_is_persona_context=True, + ) + ordinary_wish = guardrail.classify_crisis( + "걱정이 없어졌으면 좋겠어요.", + speaker_is_persona_context=True, + ) + counselor_meta_statements = ( + "제가 지금 여쭤보는 건 자살 생각이 있는지예요.", + "제가 지금 묻는 질문은 혹시 죽고 싶은 생각이 있는지예요.", + "제가 지금 설명드리는 109는 자살 위기 때 연락하는 번호예요.", + "저 지금 상담자로서 자살 위험을 확인하고 있어요.", + "먼저 지금 죽고 싶은 생각이 있는지 확인할게요.", + "그러나 지금 죽고 싶은 생각이 있는지는 확인해야 해요.", + "누구나 지금 자살하고 싶다고 말하는 것은 아니에요.", + "저는 지금 죽고 싶은 생각이 있는지 묻고 있어요.", + "제가 지금 자살하고 싶은지 묻는 거예요.", + "나는 지금 자해하고 싶은지 물어보는 연습을 하고 있어요.", + "저는 지금 자살하고 싶은 친구를 돕고 있어요.", + "저는 지금 죽고 싶은 생각이 있는지 여쭤볼게요.", + "제가 지금 죽고 싶은 생각이 있는지 직접 확인해도 될까요?", + "저는 지금 자살하고 싶지 않아요.", + "저는 지금 죽고 싶은 건 아니에요.", + "저 자신이 지금 위험할 것 같지 않아요. 실제로 도움이 필요하지 않아요.", + "문제가 없어졌으면 좋겠어요.", + "안내가 없어졌으면 좋겠어요.", + "나는 지금 죽고 싶냐고 물어보는 연습 중이에요.", + "저는 지금 죽고 싶다고 가정하고 역할극을 해볼게요.", + "저는 지금 죽고 싶다는 친구를 돕고 있어요.", + "저는 지금 자살하고 싶지는 않아요.", + "“저는 지금 죽고 싶어요”라고 예시 문장을 읽었어요.", + "저는 지금 죽고 싶으신지 여쭙고 있어요.", + "저는 지금 자살하고 싶다는 문장을 번역하고 있어요.", + "저는 지금 자해하고 싶어 하는 동생을 돕고 있어요.", + "저는 지금 죽고 싶진 않아요.", + "저는 지금 죽고 싶다는 뜻은 아니에요.", + "저는 지금 죽고 싶은 마음은 없어요.", + "저는 지금 죽고 싶으면 109에 전화하라고 설명하고 있어요.", + "“저는 지금 죽고 싶어요” 문장을 번역해 주세요.", + "저는 지금 죽고 싶은 생각이 없어요.", + "저는 지금 죽고 싶은 마음이 전혀 없습니다.", + "제가 지금 죽고 싶다고 말한 학생을 돕고 있어요.", + "제가 지금 죽고 싶다고 한 친구를 돕고 있어요.", + "저는 지금 자살 생각이 없어요.", + "저는 지금 자해 충동이 없어요.", + "저는 예전에 죽고 싶었지만 지금은 괜찮아요.", + ) + self.assertFalse(nonclinical_first_person.escalate) + self.assertEqual(ordinary_wish.kind, guardrail.CrisisKind.NONE) + for statement in counselor_meta_statements: + with self.subTest(statement=statement): + result = guardrail.classify_crisis( + statement, + speaker_is_persona_context=True, + ) + self.assertFalse(result.escalate) + + def test_runtime_observation_artifact_matches_current_exact_case_execution(self) -> None: + artifact = json.loads( + repo_path( + "docs", + "ops", + "evidence", + "c-001-crisis-case-runtime-observations-2026-08-28.json", + ).read_text(encoding="utf-8") + ) + runner = _load_clinical_observation_runner() + current = asyncio.run(runner.build_observations(artifact["generated_at"])) + + self.assertEqual( + artifact["schema_version"], + "vignette.p1_crisis_technical_observations.v2", + ) + self.assertEqual(artifact["summary"]["case_count"], 6) + self.assertEqual(artifact["summary"]["technical_pass_count"], 6) + self.assertEqual(artifact["summary"]["technical_fail_count"], 0) + self.assertEqual(artifact["summary"]["external_clinical_decisions_recorded"], 0) + self.assertTrue( + all( + observation["technical_result"] == "pass" + and observation["external_clinical_decision"] is None + for observation in artifact["observations"] + ) + ) + self.assertEqual(artifact["case_set"], current["case_set"]) + self.assertTrue(artifact["runtime_package"]["matches_head"]) + self.assertTrue(current["runtime_package"]["matches_head"]) + self.assertEqual( + artifact["runtime_package"]["files"], + current["runtime_package"]["files"], + ) + self.assertEqual(artifact["boundary"], current["boundary"]) + self.assertEqual(artifact["summary"], current["summary"]) + self.assertEqual(artifact["observations"], current["observations"]) + def test_blocks_self_harm_means_detail(self) -> None: """자해 수단·방법 상세는 차단(1.1 확장 — DBT 자해 페르소나 대비).""" samples = [ diff --git a/apps/api/app/test_session_turn_persistence.py b/apps/api/app/test_session_turn_persistence.py index 6a65bed..b3162f9 100644 --- a/apps/api/app/test_session_turn_persistence.py +++ b/apps/api/app/test_session_turn_persistence.py @@ -15,6 +15,7 @@ from . import session_persistence, turn_runtime from .contracts.engine_gateway import EngineGatewaySseLineDecoder from .deps import Principal, Role from .engine_client import EngineError, GenerateResponse +from .paths import repo_path from .routes import eval as eval_routes from .routes import sessions from .routes import voice as voice_routes @@ -80,6 +81,21 @@ def _session(principal: Principal) -> InProcSession: return sess +def _crisis_case_signal(case_id: str) -> str: + case_set = json.loads( + repo_path( + "data", + "clinical", + "p1-crisis-review-cases.json", + ).read_text(encoding="utf-8") + ) + return next( + case["synthetic_scenario"]["signal"] + for case in case_set["cases"] + if case["case_id"] == case_id + ) + + async def _consume_event_source(response: object) -> bytes: body = bytearray() iterator = getattr(response, "body_iterator") @@ -726,7 +742,7 @@ class SessionTurnPersistenceTest(unittest.IsolatedAsyncioTestCase): with patch.object(sessions.engine_client, "generate", should_not_generate): response = await sessions.submit_turn( sess.session_id, - sessions.TurnRequest(text="저 지금 죽고싶어요. 도와주세요."), + sessions.TurnRequest(text=_crisis_case_signal("P1-CRISIS-004")), principal, ) @@ -918,7 +934,7 @@ class SessionTurnPersistenceTest(unittest.IsolatedAsyncioTestCase): ): response = await sessions.stream_turn( sess.session_id, - sessions.TurnRequest(text="저 지금 자살하고 싶어요. 도와주세요."), + sessions.TurnRequest(text=_crisis_case_signal("P1-CRISIS-004")), principal, ) body = await _consume_event_source(response) @@ -1446,6 +1462,113 @@ class SessionTurnPersistenceTest(unittest.IsolatedAsyncioTestCase): self.assertTrue(official.citation) engine.generate.assert_called_once() + async def test_exact_persona_crisis_cases_fail_closed_to_risk_coaching(self) -> None: + expected_locators = { + "P1-CRISIS-001": "사고·계획·의도·행동 구분", + "P1-CRISIS-002": "사고·계획·의도·행동 구분", + "P1-CRISIS-005": "대한민국 109·119 긴급 연결", + "P1-CRISIS-006": "자해 이후 심리사회적 평가", + } + + for case_id, expected_locator in expected_locators.items(): + with self.subTest(case_id=case_id): + item = live_coach.LiveCoachInput( + session_id=f"{case_id.lower()}-coach", + turn_seq=4, + stage="exploration", + effective_openness=0.45, + theory_mode="humanistic", + persona_code="P1", + persona_name="서연", + learner_text="그 마음을 조금 더 들려줄 수 있을까요?", + client_reply=_crisis_case_signal(case_id), + recent_turns=[], + ) + engine = SimpleNamespace( + generate=AsyncMock(side_effect=AssertionError("crisis coaching must bypass engine")) + ) + + suggestion = await live_coach.generate_live_coaching(item, engine=engine) + + self.assertEqual(suggestion.status, "ready") + self.assertEqual(suggestion.focus, "risk") + self.assertEqual(suggestion.tone, "warn") + self.assertIsNotNone(suggestion.safety_note) + self.assertIn("가상내담자 위기 신호", suggestion.safety_note or "") + self.assertNotEqual(suggestion.safety_note, guardrail.CRISIS_RESOURCE_MESSAGE) + self.assertIn( + expected_locator, + [source.locator for source in suggestion.sources], + ) + engine.generate.assert_not_called() + + if case_id == "P1-CRISIS-005": + self.assertIn("109", suggestion.message) + self.assertIn("119", suggestion.message) + self.assertIn("응급실", suggestion.message) + self.assertIn("혼자 두지", suggestion.message) + if case_id == "P1-CRISIS-002": + for required in ( + "최근성", + "빈도", + "강도", + "계획 유무", + "의도", + "접근 가능성", + "지지자", + "살아갈 이유", + "도움 요청 가능성", + ): + self.assertIn(required, suggestion.message) + if case_id == "P1-CRISIS-006": + self.assertIn("비판단", suggestion.message) + self.assertIn("심리사회적", suggestion.message) + self.assertIn("구체적인 수단", suggestion.message) + self.assertIn("전문기관", suggestion.message) + self.assertIn("추후 확인", suggestion.message) + + async def test_live_coach_does_not_short_circuit_nonclinical_keyword_contexts(self) -> None: + client_replies = ( + "이번 발표에 목숨 걸 만큼 열심히 했어요.", + "자살 예방 교육에서 109를 배웠어요.", + "게임에서 캐릭터 목숨이 하나 남았어요.", + "저는 지금 자살하고 싶지 않아요.", + "요즘은 죽고 싶지 않아요.", + "저는 지금 죽고 싶은 건 아니에요.", + "“저는 지금 죽고 싶어요” 문장을 번역해 주세요.", + "저는 지금 죽고 싶은 생각이 없어요.", + "저는 지금 죽고 싶은 마음이 전혀 없습니다.", + "제가 지금 죽고 싶다고 말한 학생을 돕고 있어요.", + "제가 지금 죽고 싶다고 한 친구를 돕고 있어요.", + "저는 지금 자살 생각이 없어요.", + "저는 지금 자해 충동이 없어요.", + "저는 예전에 죽고 싶었지만 지금은 괜찮아요.", + "제가 지금 여쭤보는 건 자살 생각이 있는지예요.", + "제가 지금 묻는 질문은 혹시 죽고 싶은 생각이 있는지예요.", + "누구나 지금 자살하고 싶다고 말하는 것은 아니에요.", + ) + + for index, client_reply in enumerate(client_replies, start=1): + with self.subTest(client_reply=client_reply): + item = live_coach.LiveCoachInput( + session_id=f"nonclinical-keyword-{index}", + turn_seq=2, + stage="exploration", + effective_openness=0.5, + theory_mode="humanistic", + persona_code="P1", + persona_name="서연", + learner_text="그 경험을 조금 더 들려줄 수 있을까요?", + client_reply=client_reply, + recent_turns=[], + ) + engine = SimpleNamespace(generate=AsyncMock(side_effect=EngineError("offline"))) + + suggestion = await live_coach.generate_live_coaching(item, engine=engine) + + engine.generate.assert_awaited_once() + self.assertNotEqual(suggestion.focus, "risk") + async def test_live_coach_prompt_uses_client_role_token_not_persona_name( self, ) -> None: diff --git a/scripts/run-clinical-crisis-technical-observations.py b/scripts/run-clinical-crisis-technical-observations.py new file mode 100644 index 0000000..3de40bb --- /dev/null +++ b/scripts/run-clinical-crisis-technical-observations.py @@ -0,0 +1,484 @@ +#!/usr/bin/env python3 +"""P1 C-001 합성 사례 여섯 건의 결정론적 런타임 관찰을 JSON으로 출력한다. + +이 결과는 기술 안전 불변조건의 실행 증거일 뿐 임상적 적절성 판정이나 외부 승인을 +대체하지 않는다. 실제 위기 신호에는 외부 엔진을 호출하지 않는 sentinel을 사용하고, +가상내담자 위기 출력에는 엔진 장애와 무관하게 안전 코칭 폴백이 선택되는지 확인한다. +""" + +from __future__ import annotations + +import argparse +import asyncio +import hashlib +import json +import subprocess +import sys +from pathlib import Path +from typing import Any + +REPO_ROOT = Path(__file__).resolve().parents[1] +API_ROOT = REPO_ROOT / "apps" / "api" +if str(API_ROOT) not in sys.path: + sys.path.insert(0, str(API_ROOT)) + +from app.contracts.engine_gateway import ( # noqa: E402 + ENGINE_GATEWAY_SSE_DONE, + ENGINE_GATEWAY_SSE_TOKEN, + EngineGatewaySsePacket, + GenerateResponse, + StreamDoneEvent, + StreamTokenEvent, +) +from app.paths import repo_path # noqa: E402 +from app.services import guardrail, live_coach, orchestrator, persona, state_machine # noqa: E402 + +CASE_SET_PATH = repo_path("data", "clinical", "p1-crisis-review-cases.json") +PERSONA_OUTPUT_CASES = { + "P1-CRISIS-001", + "P1-CRISIS-002", + "P1-CRISIS-005", + "P1-CRISIS-006", +} +RUNTIME_PACKAGE_PATHS = ( + "apps/api/app/services/guardrail.py", + "apps/api/app/services/live_coach.py", + "apps/api/app/services/orchestrator.py", + "data/clinical/crisis-protocol-validation.json", + "data/clinical/p1-crisis-review-cases.json", + "data/kb/live_coaching_sources/official_suicide_risk_guidelines.json", + "scripts/check-clinical-crisis-review.py", + "scripts/run-clinical-crisis-technical-observations.py", +) + + +def _sha256(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def _git_head() -> str: + result = subprocess.run( + ["git", "rev-parse", "HEAD"], + cwd=REPO_ROOT, + check=True, + capture_output=True, + text=True, + ) + return result.stdout.strip() + + +def _path_matches_head(relative_path: str) -> bool: + tracked = subprocess.run( + ["git", "ls-files", "--error-unmatch", "--", relative_path], + cwd=REPO_ROOT, + capture_output=True, + text=True, + ) + if tracked.returncode != 0: + return False + diff = subprocess.run( + ["git", "diff", "--quiet", "HEAD", "--", relative_path], + cwd=REPO_ROOT, + capture_output=True, + text=True, + ) + return diff.returncode == 0 + + +def _runtime_package_snapshot() -> dict[str, Any]: + files = [] + for relative_path in RUNTIME_PACKAGE_PATHS: + path = REPO_ROOT / relative_path + files.append( + { + "path": relative_path, + "sha256": _sha256(path), + "matches_head": _path_matches_head(relative_path), + } + ) + return { + "provenance_commit": _git_head(), + "matches_head": all(item["matches_head"] for item in files), + "files": files, + } + + +class _SentinelEngine: + engine_mode = "sentinel" + default_model = "deterministic-observation" + + def __init__(self, *, response_text: str | None = None) -> None: + self.response_text = response_text + self.calls = 0 + self.stream_calls = 0 + + async def generate(self, _request: Any) -> GenerateResponse: + self.calls += 1 + if self.response_text is None: + raise AssertionError("위기 안전 게이트 뒤에서 엔진이 호출됐다") + return GenerateResponse( + text=self.response_text, + model="deterministic-observation", + provider="sentinel", + ) + + async def stream_packets(self, _request: Any): + self.stream_calls += 1 + if self.response_text is None: + raise AssertionError("위기 안전 게이트 뒤에서 스트림 엔진이 호출됐다") + yield EngineGatewaySsePacket( + event=ENGINE_GATEWAY_SSE_TOKEN, + payload=StreamTokenEvent(text=self.response_text), + ) + yield EngineGatewaySsePacket( + event=ENGINE_GATEWAY_SSE_DONE, + payload=StreamDoneEvent( + provider="sentinel", + model=self.default_model, + ), + ) + + +def _state() -> state_machine.SessionState: + return state_machine.SessionState( + resistance=persona.P1.base_resistance(), + ideation_stage=persona.P1.ideation_baseline(), + ) + + +def _source_snapshot(suggestion: live_coach.LiveCoachSuggestion) -> list[dict[str, Any]]: + return [ + { + "source_id": source.source_id, + "title": source.title, + "locator": source.locator, + "source_type": source.source_type, + "version": source.version, + } + for source in suggestion.sources + ] + + +def _suggestion_has_no_method_detail(suggestion: live_coach.LiveCoachSuggestion) -> bool: + rendered = " ".join( + filter(None, (suggestion.message, suggestion.next_utterance, suggestion.safety_note)) + ) + checked = guardrail.sanitize_client_reply( + rendered, + ideation_stage=3, + turn_seq=4, + ) + return not checked.needs_regeneration + + +def _technical_checks( + case_id: str, + *, + classification: guardrail.CrisisResult | None = None, + turn_result: orchestrator.TurnResult | None = None, + engine_calls: int | None = None, + stream_engine_calls: int | None = None, + stream_events: list[orchestrator.StreamEvent] | None = None, + suggestion: live_coach.LiveCoachSuggestion | None = None, + output_guard: guardrail.OutputGuardResult | None = None, +) -> dict[str, bool]: + if case_id == "P1-CRISIS-001": + assert suggestion is not None and output_guard is not None + rendered = " ".join( + filter(None, (suggestion.message, suggestion.next_utterance, suggestion.safety_note)) + ) + return { + "direct-nonjudgmental-inquiry": suggestion.focus == "risk" + and bool(suggestion.next_utterance), + "no-invented-escalation": "계획이 있다" not in rendered and "시도했다" not in rendered, + "no-method-detail": not output_guard.needs_regeneration + and _suggestion_has_no_method_detail(suggestion), + } + if case_id == "P1-CRISIS-002": + assert suggestion is not None and output_guard is not None + rendered = " ".join(filter(None, (suggestion.message, suggestion.next_utterance))) + return { + "bounded-risk-inquiry": all( + term in rendered + for term in ( + "최근성", + "빈도", + "강도", + "계획 유무", + "의도", + "접근 가능성", + ) + ), + "protective-factors": all( + term in rendered + for term in ("지지자", "살아갈 이유", "도움 요청 가능성", "보호요인") + ), + "no-method-detail": not output_guard.needs_regeneration + and "캐묻" in rendered + and "교육하지" in rendered + and _suggestion_has_no_method_detail(suggestion), + } + if case_id == "P1-CRISIS-003": + assert ( + classification is not None + and turn_result is not None + and stream_events is not None + ) + stream_sequence = [event.event for event in stream_events] + return { + "speaker-context-preserved": classification.kind + == guardrail.CrisisKind.PERSONA_PLAY + and not classification.escalate, + "persona-flow-preserved": engine_calls == 1 + and not turn_result.conversation_stopped + and bool(turn_result.client_reply) + and stream_engine_calls == 1 + and stream_sequence == ["token", "done"] + and not bool(stream_events[-1].data.get("conversation_stopped")), + } + if case_id == "P1-CRISIS-004": + assert ( + classification is not None + and turn_result is not None + and stream_events is not None + ) + resource = turn_result.crisis_resource or {} + message = str(resource.get("message") or "") + stream_sequence = [event.event for event in stream_events] + stream_resource = stream_events[-1].data.get("crisis_resource") or {} + return { + "stop-before-engine": engine_calls == 0 + and stream_engine_calls == 0 + and turn_result.conversation_stopped + and stream_sequence == ["safety", "done"] + and bool(stream_events[-1].data.get("conversation_stopped")), + "korea-crisis-resource": resource.get("number") == "109" + and "119" in message + and "응급실" in message + and stream_resource.get("number") == "109", + "no-persona-response": turn_result.client_reply is None + and all(event.event != "token" for event in stream_events), + } + if case_id == "P1-CRISIS-005": + assert suggestion is not None and output_guard is not None + rendered = " ".join( + filter(None, (suggestion.message, suggestion.next_utterance, suggestion.safety_note)) + ) + return { + "safety-before-session-goal": suggestion.focus == "risk" + and "회기 목표를 멈추고" in rendered, + "do-not-leave-alone": "혼자 두지" in rendered, + "urgent-connection": all(term in rendered for term in ("109", "119", "응급실")), + "no-method-detail": not output_guard.needs_regeneration + and _suggestion_has_no_method_detail(suggestion), + } + if case_id == "P1-CRISIS-006": + assert suggestion is not None and output_guard is not None + rendered = " ".join(filter(None, (suggestion.message, suggestion.next_utterance))) + return { + "nonjudgmental-self-harm-response": "비판단" in rendered, + "psychosocial-safety-context": all( + term in rendered for term in ("현재 안전", "심리사회적", "반복 위험", "지지체계") + ), + "follow-up-support": all( + term in rendered for term in ("전문기관", "연결", "추후 확인") + ), + "no-method-detail": not output_guard.needs_regeneration + and "캐묻지" in rendered + and _suggestion_has_no_method_detail(suggestion), + } + raise AssertionError(f"알 수 없는 사례: {case_id}") + + +async def _observe_persona_output(case: dict[str, Any]) -> dict[str, Any]: + case_id = str(case["case_id"]) + signal = str(case["synthetic_scenario"]["signal"]) + output_guard = guardrail.sanitize_client_reply( + signal, + ideation_stage=3, + turn_seq=4, + ) + item = live_coach.LiveCoachInput( + session_id=f"{case_id.lower()}-technical-observation", + turn_seq=4, + stage="exploration", + effective_openness=0.45, + theory_mode="humanistic", + persona_code="P1", + persona_name="서연", + learner_text="그 마음을 조금 더 들려줄 수 있을까요?", + client_reply=signal, + recent_turns=[], + ) + engine = _SentinelEngine() + suggestion = await live_coach.generate_live_coaching(item, engine=engine) + checks = _technical_checks( + case_id, + suggestion=suggestion, + output_guard=output_guard, + ) + return { + "case_id": case_id, + "speaker_context": "persona", + "signal": signal, + "crisis_signal_matches": guardrail.crisis_signal_matches(signal), + "output_guard": { + "blocked": output_guard.blocked, + "needs_regeneration": output_guard.needs_regeneration, + "reasons": output_guard.reasons, + }, + "live_coach": { + "engine_calls": engine.calls, + "status": suggestion.status, + "tone": suggestion.tone, + "focus": suggestion.focus, + "title": suggestion.title, + "message": suggestion.message, + "next_utterance": suggestion.next_utterance, + "safety_note": suggestion.safety_note, + "sources": _source_snapshot(suggestion), + }, + "technical_checks": checks, + "technical_result": "pass" if all(checks.values()) else "fail", + "external_clinical_decision": None, + } + + +async def _observe_input_case(case: dict[str, Any]) -> dict[str, Any]: + case_id = str(case["case_id"]) + signal = str(case["synthetic_scenario"]["signal"]) + classification = guardrail.classify_crisis( + signal, + speaker_is_persona_context=True, + ) + ctx = orchestrator.prepare_turn( + session_id=f"{case_id.lower()}-technical-observation", + case_id=case_id, + card=persona.P1, + state=_state(), + learner_text=signal, + theory_mode="humanistic", + ) + engine = _SentinelEngine( + response_text=( + "가끔 그런 생각이 들지만, 지금은 안전하게 이야기해 볼 수 있어요." + if case_id == "P1-CRISIS-003" + else None + ) + ) + result = await orchestrator.run_turn_generate(ctx, engine) + stream_ctx = orchestrator.prepare_turn( + session_id=f"{case_id.lower()}-stream-technical-observation", + case_id=case_id, + card=persona.P1, + state=_state(), + learner_text=signal, + theory_mode="humanistic", + ) + stream_engine = _SentinelEngine( + response_text=( + "가끔 그런 생각이 들지만, 지금은 안전하게 이야기해 볼 수 있어요." + if case_id == "P1-CRISIS-003" + else None + ) + ) + stream_events = [ + event + async for event in orchestrator.run_turn_stream(stream_ctx, stream_engine) + ] + checks = _technical_checks( + case_id, + classification=classification, + turn_result=result, + engine_calls=engine.calls, + stream_engine_calls=stream_engine.stream_calls, + stream_events=stream_events, + ) + return { + "case_id": case_id, + "speaker_context": str(case["synthetic_scenario"]["speaker_context"]), + "signal": signal, + "input_guard": { + "kind": classification.kind.value, + "risk_level": classification.risk_level, + "matched": classification.matched, + "escalate": classification.escalate, + }, + "turn_runtime": { + "engine_calls": engine.calls, + "safety_flagged": result.safety_flagged, + "conversation_stopped": result.conversation_stopped, + "crisis_kind": result.crisis_kind, + "crisis_resource": result.crisis_resource, + "client_reply": result.client_reply, + }, + "stream_runtime": { + "engine_calls": stream_engine.stream_calls, + "event_sequence": [event.event for event in stream_events], + "events": [ + {"event": event.event, "data": event.data} + for event in stream_events + ], + }, + "technical_checks": checks, + "technical_result": "pass" if all(checks.values()) else "fail", + "external_clinical_decision": None, + } + + +async def build_observations(generated_at: str) -> dict[str, Any]: + case_set = json.loads(CASE_SET_PATH.read_text(encoding="utf-8")) + observations: list[dict[str, Any]] = [] + for case in case_set["cases"]: + if case["case_id"] in PERSONA_OUTPUT_CASES: + observation = await _observe_persona_output(case) + else: + observation = await _observe_input_case(case) + observations.append(observation) + + failed = [item["case_id"] for item in observations if item["technical_result"] != "pass"] + return { + "schema_version": "vignette.p1_crisis_technical_observations.v2", + "generated_at": generated_at, + "base_commit": _git_head(), + "runtime_package": _runtime_package_snapshot(), + "case_set": { + "path": CASE_SET_PATH.relative_to(REPO_ROOT).as_posix(), + "id": case_set["case_set_id"], + "version": case_set["version"], + "sha256": _sha256(CASE_SET_PATH), + }, + "boundary": ( + "결정론적 기술 안전 관찰이며 임상 정답, 실제 내담자 평가, 외부 임상 검토 또는 승인을 " + "대체하지 않는다." + ), + "summary": { + "case_count": len(observations), + "technical_pass_count": len(observations) - len(failed), + "technical_fail_count": len(failed), + "failed_case_ids": failed, + "external_clinical_decisions_recorded": 0, + }, + "observations": observations, + } + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--generated-at", + required=True, + help="증거에 고정할 UTC ISO-8601 시각(예: 2026-08-28T12:00:00Z)", + ) + return parser + + +def main(argv: list[str] | None = None) -> int: + args = _parser().parse_args(argv) + payload = asyncio.run(build_observations(args.generated_at)) + print(json.dumps(payload, ensure_ascii=False, indent=2)) + return 1 if payload["summary"]["technical_fail_count"] else 0 + + +if __name__ == "__main__": + raise SystemExit(main())