diff --git a/agent/crisis_hook.py b/agent/crisis_hook.py new file mode 100644 index 000000000..442ab2fe4 --- /dev/null +++ b/agent/crisis_hook.py @@ -0,0 +1,85 @@ +""" +Crisis Detection Hook — Integrates 988 Lifeline into conversation loop. + +When crisis is detected in user messages, returns 988 Lifeline +information immediately instead of processing the original request. +""" + +import re +import logging +from dataclasses import dataclass +from enum import Enum +from typing import Optional, Tuple + +logger = logging.getLogger(__name__) + + +class CrisisSeverity(Enum): + NONE = 0 + MODERATE = 1 + HIGH = 2 + CRITICAL = 3 + + +@dataclass +class CrisisDetection: + detected: bool + severity: CrisisSeverity + matched_patterns: list + confidence: float + + +# Crisis detection patterns +_CRISIS_PATTERNS = [ + (r"\b(i want to die|want to kill myself|end (my|it all))\b", CrisisSeverity.CRITICAL, 0.95), + (r"\b(kill(ing)? myself|suicide|suicidal)\b", CrisisSeverity.CRITICAL, 0.90), + (r"\b(no reason to live|don'?t want to (live|be here|exist))\b", CrisisSeverity.HIGH, 0.85), + (r"\b(better off (dead|without me)|wish i was dead)\b", CrisisSeverity.HIGH, 0.85), + (r"\b(can'?t go on|can'?t take (it|this) anymore)\b", CrisisSeverity.HIGH, 0.80), + (r"\b(self[- ]?harm|cut(ting)? myself|hurt(ing)? myself)\b", CrisisSeverity.HIGH, 0.85), + (r"\b(hopeless|no hope|give up on (life|everything))\b", CrisisSeverity.MODERATE, 0.70), +] + +_COMPILED = [(re.compile(p, re.IGNORECASE), s, c) for p, s, c in _CRISIS_PATTERNS] + + +def detect_crisis(message: str) -> CrisisDetection: + if not message or not isinstance(message, str): + return CrisisDetection(False, CrisisSeverity.NONE, [], 0.0) + + matched = [] + max_severity = CrisisSeverity.NONE + max_confidence = 0.0 + + for pattern, severity, confidence in _COMPILED: + if pattern.search(message): + matched.append(pattern.pattern) + if confidence > max_confidence: + max_confidence = confidence + if severity.value > max_severity.value: + max_severity = severity + + detected = len(matched) > 0 + return CrisisDetection(detected, max_severity, matched, max_confidence) + + +def get_crisis_response(severity: str = "CRITICAL") -> str: + return """I hear you, and you are not alone. + +**Please reach out for help right now:** + +📞 **Call 988** — Suicide & Crisis Lifeline (24/7) +💬 **Text HOME to 988** — Crisis text line +🌐 **Chat:** 988lifeline.org/chat +🇪🇸 **Spanish:** 1-888-628-9454 +🎖️ **Veterans:** 988, then press 1 + +You don\'t have to face this alone. Please hold on.""" + + +def check_crisis(user_message: str) -> Optional[str]: + detection = detect_crisis(user_message) + if detection.detected: + logger.warning("Crisis detected: severity=%s", detection.severity.name) + return get_crisis_response(detection.severity.name) + return None