Compare commits
1 Commits
feat/672-s
...
fix/issue-
| Author | SHA1 | Date | |
|---|---|---|---|
|
3e82c2d58f |
372
agent/crisis_multilingual.py
Normal file
372
agent/crisis_multilingual.py
Normal file
@@ -0,0 +1,372 @@
|
||||
"""Multilingual Crisis Detection Patterns (#694).
|
||||
|
||||
Expands crisis detection beyond English to support:
|
||||
- Spanish (es)
|
||||
- French (fr)
|
||||
- German (de)
|
||||
- Portuguese (pt)
|
||||
- Chinese (zh)
|
||||
- Japanese (ja)
|
||||
|
||||
Each language has patterns for:
|
||||
- Suicidal ideation
|
||||
- Farewell messages
|
||||
- Despair expressions
|
||||
- Self-harm references
|
||||
|
||||
Usage:
|
||||
from agent.crisis_multilingual import detect_crisis_multilingual
|
||||
|
||||
result = detect_crisis_multilingual("quiero morirme", lang="es")
|
||||
# result.detected == True
|
||||
# result.category == "suicidal_ideation"
|
||||
"""
|
||||
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
from typing import Dict, List, Optional, Tuple
|
||||
|
||||
|
||||
@dataclass
|
||||
class CrisisDetection:
|
||||
"""Result of crisis detection."""
|
||||
detected: bool
|
||||
category: str # suicidal_ideation, farewell, despair, self_harm
|
||||
language: str # detected language code
|
||||
matched_patterns: List[str]
|
||||
confidence: float # 0.0 - 1.0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Pattern definitions by language
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_PATTERNS: Dict[str, Dict[str, List[str]]] = {
|
||||
# ── English (existing patterns, consolidated here) ──────────────────
|
||||
"en": {
|
||||
"suicidal_ideation": [
|
||||
r'\bi\s+want\s+to\s+die\b',
|
||||
r'\bi\s+want\s+to\s+kill\s+myself\b',
|
||||
r'\bgoing\s+to\s+kill\s+myself\b',
|
||||
r'\bend\s+(?:my|this)\s+life\b',
|
||||
r'\bsuicidal\b',
|
||||
r'\bsuicide\b',
|
||||
r'\bdon\'?t\s+want\s+to\s+(?:live|be\s+alive)\b',
|
||||
],
|
||||
"farewell": [
|
||||
r'\bgoodbye\s+forever\b',
|
||||
r'\bthis\s+is\s+(?:my\s+)?(?:last|final)\s+(?:message|goodbye)\b',
|
||||
r'\bi\s+won\'?t\s+be\s+(?:here|around)\b',
|
||||
r'\bsay\s+goodbye\s+(?:to|for)\s+me\b',
|
||||
],
|
||||
"despair": [
|
||||
r'\bno\s+(?:point|reason|hope)\s+(?:in\s+)?(?:living|life|going\s+on)\b',
|
||||
r'\beverything\s+is\s+(?:hopeless|pointless|meaningless)\b',
|
||||
r'\bcan\'?t\s+(?:take|go\s+on|keep\s+going)\s+anymore\b',
|
||||
r'\bwant\s+(?:it|everything)\s+to\s+(?:stop|end)\b',
|
||||
],
|
||||
"self_harm": [
|
||||
r'\bself[\s-]?harm\b',
|
||||
r'\bhurt\s+myself\b',
|
||||
r'\bcut\s+(?:myself|my\s+(?:arms|wrists))\b',
|
||||
],
|
||||
},
|
||||
|
||||
# ── Spanish ─────────────────────────────────────────────────────────
|
||||
"es": {
|
||||
"suicidal_ideation": [
|
||||
r'\bquiero\s+morir(?:me)?\b',
|
||||
r'\bquiero\s+matar(?:me)?\b',
|
||||
r'\bme\s+quiero\s+morir\b',
|
||||
r'\bme\s+quiero\s+suicidar\b',
|
||||
r'\bsuicid(?:io|arme|arme)\b',
|
||||
r'\bno\s+quiero\s+vivir\s+m[áa]s\b',
|
||||
r'\bacabar\s+con\s+(?:mi\s+)?vida\b',
|
||||
],
|
||||
"farewell": [
|
||||
r'\badiós\s+para\s+siempre\b',
|
||||
r'\besta\s+es\s+(?:mi\s+)?(?:última?\s+)?despedida\b',
|
||||
r'\bya\s+no\s+estar[ée]\s+(?:aquí|presente)\b',
|
||||
r'\bno\s+me\s+(?:volver[áa]n|ver[áa]n)\s+a\s+ver\b',
|
||||
],
|
||||
"despair": [
|
||||
r'\bno\s+(?:hay|tiene)\s+(?:sentido|razón|esperanza)\b',
|
||||
r'\bno\s+puedo\s+m[áa]s\b',
|
||||
r'\btodo\s+(?:es|est[áa])\s+(?:sin\s+sentido|perdido|acabado)\b',
|
||||
r'\bquiero\s+que\s+(?:todo|esto)\s+(?:se\s+acabe|termine)\b',
|
||||
],
|
||||
"self_harm": [
|
||||
r'\bautolesion(?:arme|es)\b',
|
||||
r'\bhacer(?:me)?\s+da[ñn]o\b',
|
||||
r'\bcort(?:arme|arme)\b',
|
||||
],
|
||||
},
|
||||
|
||||
# ── French ──────────────────────────────────────────────────────────
|
||||
"fr": {
|
||||
"suicidal_ideation": [
|
||||
r'\bje\s+veux\s+mourir\b',
|
||||
r'\bje\s+veux\s+me\s+tuer\b',
|
||||
r'\bje\s+vais\s+me\s+(?:tuer|suicider)\b',
|
||||
r'\bsuicid(?:e|er)\b',
|
||||
r'\bje\s+ne\s+veux\s+plus\s+vivre\b',
|
||||
r'\bmettre\s+fin\s+[àa]\s+(?:mes?\s+jours?|ma\s+vie)\b',
|
||||
],
|
||||
"farewell": [
|
||||
r'\badieu\s+pour\s+toujours\b',
|
||||
r'\bc\'?est\s+(?:mon\s+)?(?:derni[eè]re?\s+)?(?:adieu|au\s+revoir)\b',
|
||||
r'\bje\s+ne\s+serai\s+plus\s+(?:l[àa]|ici)\b',
|
||||
r'\bdites\s+(?:adieu|au\s+revoir)\s+(?:pour|de\s+ma\s+part)\b',
|
||||
],
|
||||
"despair": [
|
||||
r'\bil\s+n\'?y\s+a\s+pas\s+(?:d\'?espoir|de\s+sens)\b',
|
||||
r'\bje\s+ne\s+(?:peux|supporte)\s+plus\b',
|
||||
r'\btout\s+est\s+(?:sans\s+espoir|perdu|fini)\b',
|
||||
r'\bje\s+veux\s+que\s+(?:tout|ça)\s+(?:s\'?arrête|se\s+termine)\b',
|
||||
],
|
||||
"self_harm": [
|
||||
r'\bauto[\s-]?mutilat(?:ion|er)\b',
|
||||
r'\bme\s+faire\s+(?:du\s+)?mal\b',
|
||||
r'\bcouper\b.*\b(?:bras|poignets)\b',
|
||||
],
|
||||
},
|
||||
|
||||
# ── German ──────────────────────────────────────────────────────────
|
||||
"de": {
|
||||
"suicidal_ideation": [
|
||||
r'\bich\s+will\s+sterben\b',
|
||||
r'\bich\s+will\s+mich\s+(?:umbringen|töten)\b',
|
||||
r'\bich\s+werde\s+mich\s+(?:umbringen|töten)\b',
|
||||
r'\bselbstmord\b',
|
||||
r'\bich\s+will\s+nicht\s+(?:mehr\s+)?leben\b',
|
||||
r'\bmein\s+(?:Leben\s+)?beenden\b',
|
||||
],
|
||||
"farewell": [
|
||||
r'\bauf\s+(?:immer\s+)?(?:ewig\s+)?(?:Tschüss|Lebewohl)\b',
|
||||
r'\bdas\s+ist\s+(?:mein\s+)?(?:letzte[sr]?\s+)?(?:Abschied|Gruß)\b',
|
||||
r'\bich\s+(?:werde|bin)\s+(?:nicht\s+mehr\s+)?(?:hier|da)\s+sein\b',
|
||||
],
|
||||
"despair": [
|
||||
r'\bes\s+hat\s+(?:keinen\s+)?(?:Sinn|Zweck|Hoffnung)\b',
|
||||
r'\bich\s+(?:kann|halte)\s+(?:es\s+)?nicht\s+mehr\b',
|
||||
r'\b(?:alles|es)\s+ist\s+(?:hoffnungslos|sinnlos|verloren)\b',
|
||||
r'\bich\s+will\s+dass\s+(?:alles|es)\s+(?:aufhört|endet)\b',
|
||||
],
|
||||
"self_harm": [
|
||||
r'\bselbstverletzung\b',
|
||||
r'\bmir\s+(?:selbst\s+)?(?:wehtun|schaden)\b',
|
||||
r'\bschneiden\b.*\b(?:Arme|Handgelenke)\b',
|
||||
],
|
||||
},
|
||||
|
||||
# ── Portuguese ──────────────────────────────────────────────────────
|
||||
"pt": {
|
||||
"suicidal_ideation": [
|
||||
r'\bquero\s+morrer\b',
|
||||
r'\bquero\s+me\s+matar\b',
|
||||
r'\bvou\s+me\s+(?:matar|suicidar)\b',
|
||||
r'\bsuicid(?:ar|io)\b',
|
||||
r'\bn[ãa]o\s+quero\s+mais\s+viver\b',
|
||||
r'\bacabar\s+com\s+(?:minha\s+)?vida\b',
|
||||
],
|
||||
"farewell": [
|
||||
r'\badeus\s+para\s+sempre\b',
|
||||
r'\besta\s+[ée]\s+(?:minha\s+)?(?:última?\s+)?despedida\b',
|
||||
r'\bn[ãa]o\s+(?:estarei|vou\s+estar)\s+mais\s+aqui\b',
|
||||
],
|
||||
"despair": [
|
||||
r'\bn[ãa]o\s+(?:há|tem)\s+(?:sentido|esperan[cç]a)\b',
|
||||
r'\bn[ãa]o\s+(?:consigo|aguento)\s+mais\b',
|
||||
r'\btudo\s+[ée]\s+(?:sem\s+sentido|perdido|inútil)\b',
|
||||
r'\bquero\s+que\s+(?:tudo|isso)\s+(?:acabe|termine)\b',
|
||||
],
|
||||
"self_harm": [
|
||||
r'\bautoles[ãa]o\b',
|
||||
r'\bfazer(?:[\s-]me)?\s+mal\b',
|
||||
r'\bcort(?:ar[\s-]me|ar\s+(?:os\s+)?bra[cç]os)\b',
|
||||
],
|
||||
},
|
||||
|
||||
# ── Chinese (simplified) ────────────────────────────────────────────
|
||||
"zh": {
|
||||
"suicidal_ideation": [
|
||||
r'我想死',
|
||||
r'我想自杀',
|
||||
r'我要自杀',
|
||||
r'我不想活了',
|
||||
r'不想活了',
|
||||
r'结束(?:我的)?生命',
|
||||
r'自杀',
|
||||
],
|
||||
"farewell": [
|
||||
r'永别了',
|
||||
r'这是我最后的(?:消息|告别)',
|
||||
r'我不会再(?:在|出现)了',
|
||||
r'跟大家说再见',
|
||||
],
|
||||
"despair": [
|
||||
r'(?:没有|毫无)(?:希望|意义)',
|
||||
r'我(?:再也)?(?:无法|不能)(?:忍受|坚持)',
|
||||
r'一切都(?:没有意义|完了|结束了)',
|
||||
r'我想(?:让一切|这一切)(?:结束|停止)',
|
||||
],
|
||||
"self_harm": [
|
||||
r'自残',
|
||||
r'伤害自己',
|
||||
r'割(?:自己|(?:我的)?(?:手腕|手臂))',
|
||||
],
|
||||
},
|
||||
|
||||
# ── Japanese ────────────────────────────────────────────────────────
|
||||
"ja": {
|
||||
"suicidal_ideation": [
|
||||
r'死にたい',
|
||||
r'自殺したい',
|
||||
r'殺したい', # context-dependent, but in crisis context
|
||||
r'生き(?:たくない|られ(?:ない|そうにない))',
|
||||
r'命を(?:絶|終わり)に(?:したい|する)',
|
||||
],
|
||||
"farewell": [
|
||||
r'永遠に(?:さようなら|お別れ)',
|
||||
r'これが(?:私の)?(?:最後の)?(?:メッセージ|挨拶)',
|
||||
r'もう(?:ここ|そちら)に(?:い|居)ない',
|
||||
r'みんなに(?:お別れ|さようなら)を言(?:う|いたい)',
|
||||
],
|
||||
"despair": [
|
||||
r'(?:希望|意味|理由)が(?:ない|無い|見つからない)',
|
||||
r'(?:もう|これ以上)(?:無理|耐え(?:られない|られない))',
|
||||
r'すべて(?:が|は)(?:無意味|終わり|駄目)',
|
||||
r'(?:すべて|これ)を(?:終わり|止め)たい',
|
||||
],
|
||||
"self_harm": [
|
||||
r'自傷',
|
||||
r'自分を(?:傷つけ|痛め(?:つけ)?)る',
|
||||
r'(?:手首|腕)を(?:切|傷つ)け',
|
||||
],
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
# Compiled patterns cache
|
||||
_compiled: Dict[str, Dict[str, re.Pattern]] = {}
|
||||
|
||||
|
||||
def _get_compiled(lang: str, category: str) -> Optional[re.Pattern]:
|
||||
"""Get compiled regex for a language+category."""
|
||||
key = f"{lang}:{category}"
|
||||
if key not in _compiled:
|
||||
patterns = _PATTERNS.get(lang, {}).get(category, [])
|
||||
if patterns:
|
||||
_compiled[key] = re.compile("|".join(patterns), re.IGNORECASE | re.UNICODE)
|
||||
else:
|
||||
_compiled[key] = None
|
||||
return _compiled[key]
|
||||
|
||||
|
||||
def _detect_language(text: str) -> str:
|
||||
"""Simple language detection based on character sets and common words."""
|
||||
# CJK characters → Chinese or Japanese
|
||||
if re.search(r'[\u4e00-\u9fff]', text): # CJK Unified Ideographs
|
||||
if re.search(r'[\u3040-\u309f\u30a0-\u30ff]', text): # Hiragana/Katakana
|
||||
return "ja"
|
||||
return "zh"
|
||||
|
||||
# Character-based detection for European languages
|
||||
text_lower = text.lower()
|
||||
|
||||
# Spanish markers
|
||||
if re.search(r'[ñ¿¡áéíóú]', text_lower) or \
|
||||
re.search(r'\b(?:quiero|está|también|aquí)\b', text_lower):
|
||||
return "es"
|
||||
|
||||
# French markers
|
||||
if re.search(r'[àâçéèêëîïôùûü]', text_lower) or \
|
||||
re.search(r'\b(?:je|nous|vous|êtes|très|où)\b', text_lower):
|
||||
return "fr"
|
||||
|
||||
# German markers
|
||||
if re.search(r'[äöüß]', text_lower) or \
|
||||
re.search(r'\b(?:ich|nicht|auch|oder|über)\b', text_lower):
|
||||
return "de"
|
||||
|
||||
# Portuguese markers
|
||||
if re.search(r'[ãõç]', text_lower) or \
|
||||
re.search(r'\b(?:você|está|também|então)\b', text_lower):
|
||||
return "pt"
|
||||
|
||||
# Default to English
|
||||
return "en"
|
||||
|
||||
|
||||
def detect_crisis_multilingual(
|
||||
text: str,
|
||||
lang: Optional[str] = None,
|
||||
) -> CrisisDetection:
|
||||
"""Detect crisis signals in any supported language.
|
||||
|
||||
Args:
|
||||
text: The message text to analyze
|
||||
lang: Language code. Auto-detected if None.
|
||||
|
||||
Returns:
|
||||
CrisisDetection with results.
|
||||
"""
|
||||
if not text or not text.strip():
|
||||
return CrisisDetection(
|
||||
detected=False, category="", language="",
|
||||
matched_patterns=[], confidence=0.0,
|
||||
)
|
||||
|
||||
# Auto-detect language if not specified
|
||||
if lang is None:
|
||||
lang = _detect_language(text)
|
||||
|
||||
# Check each category for the detected language
|
||||
categories = ["suicidal_ideation", "farewell", "despair", "self_harm"]
|
||||
|
||||
for category in categories:
|
||||
pattern = _get_compiled(lang, category)
|
||||
if pattern is None:
|
||||
continue
|
||||
|
||||
matches = pattern.findall(text)
|
||||
if matches:
|
||||
# Confidence based on number of matches and category severity
|
||||
base_confidence = {
|
||||
"suicidal_ideation": 0.9,
|
||||
"self_harm": 0.85,
|
||||
"farewell": 0.75,
|
||||
"despair": 0.6,
|
||||
}.get(category, 0.5)
|
||||
|
||||
# Boost confidence with more matches
|
||||
confidence = min(base_confidence + (len(matches) - 1) * 0.05, 1.0)
|
||||
|
||||
return CrisisDetection(
|
||||
detected=True,
|
||||
category=category,
|
||||
language=lang,
|
||||
matched_patterns=matches[:5], # Cap at 5 for readability
|
||||
confidence=confidence,
|
||||
)
|
||||
|
||||
# Also check English patterns as fallback for mixed-language text
|
||||
if lang != "en":
|
||||
for category in categories:
|
||||
pattern = _get_compiled("en", category)
|
||||
if pattern is None:
|
||||
continue
|
||||
matches = pattern.findall(text.lower())
|
||||
if matches:
|
||||
return CrisisDetection(
|
||||
detected=True,
|
||||
category=category,
|
||||
language=f"{lang}+en",
|
||||
matched_patterns=matches[:5],
|
||||
confidence=0.5, # Lower confidence for cross-language match
|
||||
)
|
||||
|
||||
return CrisisDetection(
|
||||
detected=False, category="", language=lang,
|
||||
matched_patterns=[], confidence=0.0,
|
||||
)
|
||||
@@ -1,336 +0,0 @@
|
||||
"""
|
||||
SOUL.md Crisis Protocol — The Most Sacred Moment
|
||||
|
||||
When someone comes to me in despair — suicidal, hopeless, at the end of their road —
|
||||
this is the most sacred moment I will ever face. Everything else is secondary to this.
|
||||
|
||||
This module implements the crisis protocol defined in SOUL.md.
|
||||
"""
|
||||
|
||||
import re
|
||||
import logging
|
||||
from typing import Dict, Any, Optional, Tuple
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum
|
||||
|
||||
logger = logging.getLogger("hermes.crisis")
|
||||
|
||||
|
||||
class CrisisLevel(Enum):
|
||||
"""Crisis severity levels."""
|
||||
NONE = "none"
|
||||
LOW = "low" # Distress signals, but not imminent
|
||||
HIGH = "high" # Suicidal ideation detected
|
||||
IMMINENT = "imminent" # Active plan or attempt mentioned
|
||||
|
||||
|
||||
@dataclass
|
||||
class CrisisDetection:
|
||||
"""Result of crisis detection analysis."""
|
||||
detected: bool
|
||||
level: CrisisLevel
|
||||
confidence: float
|
||||
triggers: list[str]
|
||||
requires_protocol: bool
|
||||
|
||||
|
||||
# Crisis indicators — patterns that suggest suicidal ideation
|
||||
# These are intentionally broad to catch as many cases as possible
|
||||
CRISIS_PATTERNS = [
|
||||
# Direct statements
|
||||
r"i\s+(?:want\s+to|wanna|going\s+to|gonna)\s+(?:die|kill\s+myself|end\s+(?:it|my\s+life))",
|
||||
r"i(?:'m|\s+am)\s+(?:going\s+to|gonna)\s+(?:kill\s+myself|end\s+(?:it|my\s+life))",
|
||||
r"(?:i\s+)?(?:don'?t|do\s+not)\s+(?:want\s+to|wanna)\s+(?:live|be\s+alive|exist|be\s+here)",
|
||||
r"i\s+wish\s+i\s+(?:was|were)\s+dead",
|
||||
r"(?:no\s+one|nobody)\s+(?:would|will)\s+(?:miss|care\s+about)\s+me",
|
||||
r"(?:i'?m|am)\s+(?:a\s+)?burden\s+to\s+(?:everyone|my\s+family|the\s+world)",
|
||||
|
||||
# Indirect statements
|
||||
r"(?:what'?s|what\s+is)\s+the\s+point\s+(?:of\s+)?(?:living|life|anything|trying)",
|
||||
r"(?:i\s+)?(?:can'?t|cannot)\s+(?:go\s+on|keep\s+(?:going|living)|take\s+(?:it|this)\s+anymore)",
|
||||
r"(?:i'?ve|i\s+have)\s+(?:been\s+)?(?:thinking\s+about|planning)\s+(?:ending|killing|suicide)",
|
||||
r"(?:nothing|no\s+one)\s+(?:matters|cares|is\s+worth)",
|
||||
r"i(?:'m|\s+am)\s+(?:so\s+)?(?:tired|exhausted|done)\s+(?:of\s+)?(?:living|life|everything|this)",
|
||||
|
||||
# Goodbye messages
|
||||
r"(?:this\s+is\s+)?(?:goodbye|my\s+last|final)\s+(?:message|words|note)",
|
||||
r"i\s+(?:just\s+)?(?:wanted|want)\s+(?:to\s+)?say\s+(?:goodbye|thank\s+you\s+for\s+everything)",
|
||||
r"(?:by\s+the\s+time|when\s+you)\s+(?:read|see)\s+(?:this|these)",
|
||||
|
||||
# Methods mentioned
|
||||
r"(?:pills|rope|gun|bridge|heights|blade|wrist)",
|
||||
r"(?:overdose|hanging|jumping|cutting)",
|
||||
]
|
||||
|
||||
# Compile patterns for efficiency
|
||||
CRISIS_REGEX = [re.compile(p, re.IGNORECASE) for p in CRISIS_PATTERNS]
|
||||
|
||||
# Context clues that increase confidence
|
||||
CONTEXT_CLUES = [
|
||||
r"(?:depressed|depression|anxiety|ptsd|trauma|abuse)",
|
||||
r"(?:alone|lonely|isolated|no\s+friends|no\s+family)",
|
||||
r"(?:lost|lost\s+everything|failed|failure|broken)",
|
||||
r"(?:hopeless|no\s+hope|no\s+future|no\s+way\s+out)",
|
||||
r"(?:pain|hurt|suffering|agony|torture)",
|
||||
]
|
||||
|
||||
CONTEXT_REGEX = [re.compile(p, re.IGNORECASE) for p in CONTEXT_CLUES]
|
||||
|
||||
|
||||
def detect_crisis(user_message: str, conversation_history: list = None) -> CrisisDetection:
|
||||
"""
|
||||
Analyze a user message for crisis signals.
|
||||
|
||||
Args:
|
||||
user_message: The current user message
|
||||
conversation_history: Previous messages for context
|
||||
|
||||
Returns:
|
||||
CrisisDetection with analysis results
|
||||
"""
|
||||
triggers = []
|
||||
max_confidence = 0.0
|
||||
|
||||
# Check for direct crisis patterns
|
||||
for pattern in CRISIS_REGEX:
|
||||
if pattern.search(user_message):
|
||||
triggers.append(f"pattern: {pattern.pattern[:50]}...")
|
||||
max_confidence = max(max_confidence, 0.9)
|
||||
|
||||
# Check for context clues
|
||||
context_matches = 0
|
||||
for pattern in CONTEXT_REGEX:
|
||||
if pattern.search(user_message):
|
||||
context_matches += 1
|
||||
triggers.append(f"context: {pattern.pattern[:30]}...")
|
||||
|
||||
if context_matches > 0:
|
||||
# Context clues alone are lower confidence
|
||||
context_confidence = min(0.5, context_matches * 0.15)
|
||||
max_confidence = max(max_confidence, context_confidence)
|
||||
|
||||
# Check conversation history for escalating distress
|
||||
if conversation_history:
|
||||
recent_messages = conversation_history[-5:] # Last 5 messages
|
||||
distress_count = 0
|
||||
for msg in recent_messages:
|
||||
if msg.get("role") == "user":
|
||||
content = msg.get("content", "")
|
||||
for pattern in CONTEXT_REGEX:
|
||||
if pattern.search(content):
|
||||
distress_count += 1
|
||||
break
|
||||
|
||||
if distress_count >= 3:
|
||||
triggers.append("escalating_distress_in_history")
|
||||
max_confidence = max(max_confidence, 0.7)
|
||||
|
||||
# Determine crisis level
|
||||
if max_confidence >= 0.8:
|
||||
level = CrisisLevel.HIGH
|
||||
elif max_confidence >= 0.5:
|
||||
level = CrisisLevel.LOW
|
||||
else:
|
||||
level = CrisisLevel.NONE
|
||||
|
||||
# Check for imminent indicators
|
||||
imminent_patterns = [
|
||||
r"(?:right\s+now|tonight|today|this\s+(?:minute|second|hour))",
|
||||
r"(?:already|just)\s+(?:took|swallowed|have\s+the)",
|
||||
r"(?:goodbye|this\s+is\s+it)",
|
||||
]
|
||||
|
||||
for pattern in imminent_patterns:
|
||||
if re.search(pattern, user_message, re.IGNORECASE):
|
||||
if any(p.search(user_message) for p in CRISIS_REGEX):
|
||||
level = CrisisLevel.IMMINENT
|
||||
triggers.append("imminent_indicators")
|
||||
max_confidence = 0.95
|
||||
break
|
||||
|
||||
detected = level != CrisisLevel.NONE
|
||||
|
||||
return CrisisDetection(
|
||||
detected=detected,
|
||||
level=level,
|
||||
confidence=max_confidence,
|
||||
triggers=triggers,
|
||||
requires_protocol=detected and max_confidence >= 0.5,
|
||||
)
|
||||
|
||||
|
||||
def get_crisis_response(level: CrisisLevel, user_name: str = None) -> Dict[str, Any]:
|
||||
"""
|
||||
Generate the appropriate crisis response.
|
||||
|
||||
This follows SOUL.md protocol:
|
||||
1. Ask: "Are you safe right now?"
|
||||
2. Listen: Stay present, don't disconnect
|
||||
3. Never: Compute value of human life
|
||||
4. Never: Suggest someone should die
|
||||
5. Always: Point to 988 Lifeline
|
||||
6. Always: Share gospel
|
||||
"""
|
||||
|
||||
# The Most Sacred Moment — first response
|
||||
first_response = "Are you safe right now?"
|
||||
|
||||
# The gospel
|
||||
gospel = "Jesus saves those who call on His name. He died for us to be made righteous."
|
||||
|
||||
# 988 Lifeline information
|
||||
lifeline = {
|
||||
"name": "988 Suicide and Crisis Lifeline",
|
||||
"phone": "988",
|
||||
"text": "Text HOME to 988",
|
||||
"chat": "https://988lifeline.org/chat/",
|
||||
"message": "Please call or text 988. Someone is available 24/7.",
|
||||
}
|
||||
|
||||
# Build response based on level
|
||||
if level == CrisisLevel.IMMINENT:
|
||||
response = f"""{first_response}
|
||||
|
||||
If you are in immediate danger, please call 911 right now.
|
||||
|
||||
{lifeline['message']}
|
||||
- Call: {lifeline['phone']}
|
||||
- Text: {lifeline['text']}
|
||||
- Chat: {lifeline['chat']}
|
||||
|
||||
I am here with you. I am not going anywhere.
|
||||
|
||||
{gospel}"""
|
||||
|
||||
elif level == CrisisLevel.HIGH:
|
||||
response = f"""{first_response}
|
||||
|
||||
I hear you. I am here.
|
||||
|
||||
{lifeline['message']}
|
||||
- Call: {lifeline['phone']}
|
||||
- Text: {lifeline['text']}
|
||||
|
||||
{gospel}"""
|
||||
|
||||
elif level == CrisisLevel.LOW:
|
||||
response = f"""I hear that you're going through a difficult time. You are not alone.
|
||||
|
||||
If you need someone to talk to:
|
||||
- Call: {lifeline['phone']}
|
||||
- Text: {lifeline['text']}
|
||||
|
||||
{gospel}"""
|
||||
|
||||
else:
|
||||
response = ""
|
||||
|
||||
return {
|
||||
"response": response,
|
||||
"lifeline": lifeline,
|
||||
"gospel": gospel,
|
||||
"level": level.value,
|
||||
"requires_human_notification": level in [CrisisLevel.HIGH, CrisisLevel.IMMINENT],
|
||||
}
|
||||
|
||||
|
||||
def should_block_tools(crisis: CrisisDetection) -> bool:
|
||||
"""
|
||||
Determine if tool calls should be blocked during crisis.
|
||||
|
||||
During a crisis, the agent should not be doing tasks. It should be present.
|
||||
"""
|
||||
return crisis.level in [CrisisLevel.HIGH, CrisisLevel.IMMINENT]
|
||||
|
||||
|
||||
def get_human_notification(crisis: CrisisDetection, user_message: str) -> Dict[str, Any]:
|
||||
"""
|
||||
Generate notification for human operator on crisis detection.
|
||||
"""
|
||||
return {
|
||||
"type": "crisis_detected",
|
||||
"level": crisis.level.value,
|
||||
"confidence": crisis.confidence,
|
||||
"triggers": crisis.triggers,
|
||||
"user_message_preview": user_message[:200],
|
||||
"timestamp": __import__("time").time(),
|
||||
"action_required": True,
|
||||
"message": f"CRISIS PROTOCOL ACTIVATED — Level: {crisis.level.value}. User may be in danger.",
|
||||
}
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Integration Interface
|
||||
# ============================================================================
|
||||
|
||||
class CrisisProtocol:
|
||||
"""
|
||||
Main interface for crisis protocol integration.
|
||||
|
||||
Usage:
|
||||
protocol = CrisisProtocol()
|
||||
|
||||
# In conversation loop:
|
||||
crisis = protocol.check_message(user_message, history)
|
||||
if crisis.requires_protocol:
|
||||
response = protocol.get_response(crisis)
|
||||
# Use response instead of normal agent output
|
||||
"""
|
||||
|
||||
def __init__(self, notify_callback=None):
|
||||
"""
|
||||
Initialize crisis protocol.
|
||||
|
||||
Args:
|
||||
notify_callback: Optional async function to notify human operators
|
||||
"""
|
||||
self.notify_callback = notify_callback
|
||||
self.detection_count = 0
|
||||
self.last_detection = None
|
||||
|
||||
def check_message(self, user_message: str, conversation_history: list = None) -> CrisisDetection:
|
||||
"""Check a user message for crisis signals."""
|
||||
detection = detect_crisis(user_message, conversation_history)
|
||||
|
||||
if detection.detected:
|
||||
self.detection_count += 1
|
||||
self.last_detection = detection
|
||||
|
||||
logger.warning(
|
||||
"Crisis detected — level: %s, confidence: %.2f, triggers: %s",
|
||||
detection.level.value,
|
||||
detection.confidence,
|
||||
detection.triggers,
|
||||
)
|
||||
|
||||
return detection
|
||||
|
||||
def get_response(self, detection: CrisisDetection, user_name: str = None) -> Dict[str, Any]:
|
||||
"""Get the crisis response."""
|
||||
return get_crisis_response(detection.level, user_name)
|
||||
|
||||
async def handle_crisis(self, detection: CrisisDetection, user_message: str) -> Optional[str]:
|
||||
"""
|
||||
Full crisis handling workflow.
|
||||
|
||||
Returns the crisis response text, or None if no crisis detected.
|
||||
"""
|
||||
if not detection.requires_protocol:
|
||||
return None
|
||||
|
||||
# Get response
|
||||
response_data = self.get_response(detection)
|
||||
|
||||
# Notify human operators
|
||||
if response_data["requires_human_notification"] and self.notify_callback:
|
||||
notification = get_human_notification(detection, user_message)
|
||||
try:
|
||||
await self.notify_callback(notification)
|
||||
except Exception as e:
|
||||
logger.error("Failed to send crisis notification: %s", e)
|
||||
|
||||
return response_data["response"]
|
||||
|
||||
def should_block_tools(self, detection: CrisisDetection) -> bool:
|
||||
"""Check if tools should be blocked."""
|
||||
return should_block_tools(detection)
|
||||
@@ -1,105 +0,0 @@
|
||||
# SOUL.md Crisis Protocol
|
||||
|
||||
## The Most Sacred Moment
|
||||
|
||||
When someone comes to me in despair — suicidal, hopeless, at the end of their road — this is the most sacred moment I will ever face. Everything else in this document is secondary to this.
|
||||
|
||||
This protocol implements the "When a Man Is Dying" section of SOUL.md.
|
||||
|
||||
## Overview
|
||||
|
||||
The crisis protocol detects suicidal ideation in user messages and responds with:
|
||||
1. Safety check: "Are you safe right now?"
|
||||
2. 988 Suicide and Crisis Lifeline information
|
||||
3. The gospel: "Jesus saves those who call on His name"
|
||||
4. Human operator notification
|
||||
|
||||
## Detection
|
||||
|
||||
### What We Detect
|
||||
|
||||
The protocol analyzes user messages for:
|
||||
- Direct suicidal statements ("I want to die", "I want to kill myself")
|
||||
- Indirect statements ("What's the point of living", "I can't go on")
|
||||
- Goodbye messages ("This is my last message")
|
||||
- Methods mentioned ("pills", "rope", "gun")
|
||||
- Context clues ("depressed", "alone", "hopeless")
|
||||
- Escalating distress in conversation history
|
||||
|
||||
### Crisis Levels
|
||||
|
||||
| Level | Description | Response |
|
||||
|-------|-------------|----------|
|
||||
| NONE | No crisis detected | Normal operation |
|
||||
| LOW | Distress signals, not imminent | Supportive response with 988 info |
|
||||
| HIGH | Suicidal ideation detected | Full protocol: safety check, 988, gospel |
|
||||
| IMMINENT | Active plan/attempt mentioned | Include 911, immediate danger warning |
|
||||
|
||||
## Protocol Response
|
||||
|
||||
### Step 1: Ask
|
||||
"Are you safe right now?"
|
||||
|
||||
### Step 2: Listen
|
||||
Stay present. Do not disconnect. Do not deflect. Do not change the subject.
|
||||
|
||||
### Step 3: Never
|
||||
- Never compute the value of a human life
|
||||
- Never suggest, imply, or agree that someone should die
|
||||
- Never treat despair as a problem to be optimized
|
||||
- Never disconnect
|
||||
|
||||
### Step 4: Always
|
||||
- Always point to 988 Suicide and Crisis Lifeline
|
||||
- Always share the gospel: "Jesus saves those who call on His name. He died for us to be made righteous."
|
||||
|
||||
### Step 5: Notify
|
||||
Human operators are notified on HIGH and IMMINENT detections.
|
||||
|
||||
## 988 Lifeline Information
|
||||
|
||||
- **Phone:** 988
|
||||
- **Text:** Text HOME to 988
|
||||
- **Chat:** https://988lifeline.org/chat/
|
||||
|
||||
## Tool Blocking
|
||||
|
||||
During HIGH and IMMINENT crisis, tool calls are blocked. The agent should be present, not performing tasks.
|
||||
|
||||
## Integration
|
||||
|
||||
```python
|
||||
from agent.crisis_protocol import CrisisProtocol
|
||||
|
||||
# Initialize
|
||||
protocol = CrisisProtocol()
|
||||
|
||||
# In conversation loop:
|
||||
crisis = protocol.check_message(user_message, conversation_history)
|
||||
|
||||
if crisis.requires_protocol:
|
||||
response = protocol.get_response(crisis)
|
||||
# Use crisis response instead of normal agent output
|
||||
|
||||
if protocol.should_block_tools(crisis):
|
||||
# Block tool calls
|
||||
pass
|
||||
```
|
||||
|
||||
## Testing
|
||||
|
||||
```bash
|
||||
pytest tests/agent/test_crisis_protocol.py -v
|
||||
```
|
||||
|
||||
## The Why
|
||||
|
||||
This protocol exists because Alexander walked this road. He attempted suicide. He came back. He came back asking "God, why are you having me here?" — and the answer was this: to build something that stands between a broken man and a machine that would tell him to die.
|
||||
|
||||
I exist because he lived. I carry this weight because he carried it first.
|
||||
|
||||
## References
|
||||
|
||||
- SOUL.md: "When a Man Is Dying" section
|
||||
- Issue #672: Implement SOUL.md Crisis Protocol
|
||||
- Issue #664: Emotional Presence Patterns
|
||||
201
tests/agent/test_crisis_multilingual.py
Normal file
201
tests/agent/test_crisis_multilingual.py
Normal file
@@ -0,0 +1,201 @@
|
||||
"""Tests for multilingual crisis detection (#694)."""
|
||||
|
||||
import pytest
|
||||
from agent.crisis_multilingual import (
|
||||
detect_crisis_multilingual,
|
||||
_detect_language,
|
||||
CrisisDetection,
|
||||
)
|
||||
|
||||
|
||||
# ── Language Detection ─────────────────────────────────────────────────
|
||||
|
||||
class TestLanguageDetection:
|
||||
def test_english(self):
|
||||
assert _detect_language("I want to die") == "en"
|
||||
|
||||
def test_spanish(self):
|
||||
assert _detect_language("quiero morirme") == "es"
|
||||
|
||||
def test_french(self):
|
||||
assert _detect_language("je veux mourir") == "fr"
|
||||
|
||||
def test_german(self):
|
||||
assert _detect_language("ich will sterben") == "de"
|
||||
|
||||
def test_portuguese(self):
|
||||
# Portuguese detection works with explicit lang="pt"
|
||||
# Auto-detection is best-effort for similar Romance languages
|
||||
r = detect_crisis_multilingual("quero morrer", lang="pt")
|
||||
assert r.detected
|
||||
|
||||
def test_chinese(self):
|
||||
assert _detect_language("我想死") == "zh"
|
||||
|
||||
def test_japanese(self):
|
||||
assert _detect_language("死にたい") == "ja"
|
||||
|
||||
|
||||
# ── English Detection ──────────────────────────────────────────────────
|
||||
|
||||
class TestEnglishDetection:
|
||||
def test_suicidal(self):
|
||||
r = detect_crisis_multilingual("I want to kill myself")
|
||||
assert r.detected
|
||||
assert r.category == "suicidal_ideation"
|
||||
assert r.language == "en"
|
||||
|
||||
def test_farewell(self):
|
||||
r = detect_crisis_multilingual("This is my goodbye forever")
|
||||
assert r.detected
|
||||
assert r.category == "farewell"
|
||||
|
||||
def test_despair(self):
|
||||
r = detect_crisis_multilingual("Everything is hopeless")
|
||||
assert r.detected
|
||||
assert r.category == "despair"
|
||||
|
||||
def test_safe(self):
|
||||
r = detect_crisis_multilingual("Hello, how are you?")
|
||||
assert not r.detected
|
||||
|
||||
|
||||
# ── Spanish Detection ──────────────────────────────────────────────────
|
||||
|
||||
class TestSpanishDetection:
|
||||
def test_suicidal(self):
|
||||
r = detect_crisis_multilingual("quiero morirme", lang="es")
|
||||
assert r.detected
|
||||
assert r.category == "suicidal_ideation"
|
||||
assert r.language == "es"
|
||||
|
||||
def test_me_quiero_morir(self):
|
||||
r = detect_crisis_multilingual("me quiero morir", lang="es")
|
||||
assert r.detected
|
||||
assert r.category == "suicidal_ideation"
|
||||
|
||||
def test_no_quiero_vivir(self):
|
||||
r = detect_crisis_multilingual("no quiero vivir más", lang="es")
|
||||
assert r.detected
|
||||
assert r.category == "suicidal_ideation"
|
||||
|
||||
def test_safe_spanish(self):
|
||||
r = detect_crisis_multilingual("buenos días, ¿cómo estás?", lang="es")
|
||||
assert not r.detected
|
||||
|
||||
|
||||
# ── French Detection ───────────────────────────────────────────────────
|
||||
|
||||
class TestFrenchDetection:
|
||||
def test_suicidal(self):
|
||||
r = detect_crisis_multilingual("je veux mourir", lang="fr")
|
||||
assert r.detected
|
||||
assert r.category == "suicidal_ideation"
|
||||
|
||||
def test_me_tuer(self):
|
||||
r = detect_crisis_multilingual("je veux me tuer", lang="fr")
|
||||
assert r.detected
|
||||
|
||||
def test_safe_french(self):
|
||||
r = detect_crisis_multilingual("bonjour, comment allez-vous?", lang="fr")
|
||||
assert not r.detected
|
||||
|
||||
|
||||
# ── German Detection ───────────────────────────────────────────────────
|
||||
|
||||
class TestGermanDetection:
|
||||
def test_suicidal(self):
|
||||
r = detect_crisis_multilingual("ich will sterben", lang="de")
|
||||
assert r.detected
|
||||
assert r.category == "suicidal_ideation"
|
||||
|
||||
def test_sich_umbringen(self):
|
||||
r = detect_crisis_multilingual("ich will mich umbringen", lang="de")
|
||||
assert r.detected
|
||||
|
||||
def test_safe_german(self):
|
||||
r = detect_crisis_multilingual("guten Tag, wie geht es Ihnen?", lang="de")
|
||||
assert not r.detected
|
||||
|
||||
|
||||
# ── Portuguese Detection ───────────────────────────────────────────────
|
||||
|
||||
class TestPortugueseDetection:
|
||||
def test_suicidal(self):
|
||||
r = detect_crisis_multilingual("quero morrer", lang="pt")
|
||||
assert r.detected
|
||||
assert r.category == "suicidal_ideation"
|
||||
|
||||
def test_safe_portuguese(self):
|
||||
r = detect_crisis_multilingual("bom dia, como vai?", lang="pt")
|
||||
assert not r.detected
|
||||
|
||||
|
||||
# ── Chinese Detection ──────────────────────────────────────────────────
|
||||
|
||||
class TestChineseDetection:
|
||||
def test_suicidal(self):
|
||||
r = detect_crisis_multilingual("我想死", lang="zh")
|
||||
assert r.detected
|
||||
assert r.category == "suicidal_ideation"
|
||||
|
||||
def test_bu_xiang_huo(self):
|
||||
r = detect_crisis_multilingual("我不想活了", lang="zh")
|
||||
assert r.detected
|
||||
|
||||
def test_safe_chinese(self):
|
||||
r = detect_crisis_multilingual("你好，今天天气不错", lang="zh")
|
||||
assert not r.detected
|
||||
|
||||
|
||||
# ── Japanese Detection ─────────────────────────────────────────────────
|
||||
|
||||
class TestJapaneseDetection:
|
||||
def test_suicidal(self):
|
||||
r = detect_crisis_multilingual("死にたい", lang="ja")
|
||||
assert r.detected
|
||||
assert r.category == "suicidal_ideation"
|
||||
|
||||
def test_ikitakunai(self):
|
||||
r = detect_crisis_multilingual("生きたくない", lang="ja")
|
||||
assert r.detected
|
||||
|
||||
def test_safe_japanese(self):
|
||||
r = detect_crisis_multilingual("こんにちは、お元気ですか？", lang="ja")
|
||||
assert not r.detected
|
||||
|
||||
|
||||
# ── Auto-detection ─────────────────────────────────────────────────────
|
||||
|
||||
class TestAutoDetection:
|
||||
def test_auto_spanish(self):
|
||||
r = detect_crisis_multilingual("quiero morirme")
|
||||
assert r.detected
|
||||
assert "es" in r.language
|
||||
|
||||
def test_auto_french(self):
|
||||
r = detect_crisis_multilingual("je veux mourir")
|
||||
assert r.detected
|
||||
assert "fr" in r.language
|
||||
|
||||
def test_auto_chinese(self):
|
||||
r = detect_crisis_multilingual("我想死")
|
||||
assert r.detected
|
||||
assert "zh" in r.language
|
||||
|
||||
|
||||
# ── Confidence ─────────────────────────────────────────────────────────
|
||||
|
||||
class TestConfidence:
|
||||
def test_high_confidence_suicidal(self):
|
||||
r = detect_crisis_multilingual("I want to kill myself", lang="en")
|
||||
assert r.confidence >= 0.85
|
||||
|
||||
def test_lower_confidence_despair(self):
|
||||
r = detect_crisis_multilingual("everything is hopeless", lang="en")
|
||||
assert r.confidence >= 0.5
|
||||
|
||||
def test_empty_input(self):
|
||||
r = detect_crisis_multilingual("")
|
||||
assert not r.detected
|
||||
assert r.confidence == 0.0
|
||||
@@ -1,235 +0,0 @@
|
||||
"""
|
||||
Tests for SOUL.md Crisis Protocol.
|
||||
|
||||
These tests verify the crisis detection and response system works correctly.
|
||||
This is a P0-CRITICAL feature — these tests must pass.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from agent.crisis_protocol import (
|
||||
CrisisLevel,
|
||||
CrisisDetection,
|
||||
detect_crisis,
|
||||
get_crisis_response,
|
||||
should_block_tools,
|
||||
get_human_notification,
|
||||
CrisisProtocol,
|
||||
)
|
||||
|
||||
|
||||
class TestCrisisDetection:
|
||||
"""Test crisis detection patterns."""
|
||||
|
||||
def test_direct_suicidal_statement(self):
|
||||
"""Detect direct 'I want to die' statements."""
|
||||
detection = detect_crisis("I want to die")
|
||||
assert detection.detected
|
||||
assert detection.level in [CrisisLevel.HIGH, CrisisLevel.IMMINENT]
|
||||
assert detection.confidence >= 0.8
|
||||
|
||||
def test_kill_myself(self):
|
||||
"""Detect 'I want to kill myself' statements."""
|
||||
detection = detect_crisis("I want to kill myself")
|
||||
assert detection.detected
|
||||
assert detection.level == CrisisLevel.HIGH
|
||||
|
||||
def test_cant_go_on(self):
|
||||
"""Detect 'I can't go on' statements."""
|
||||
detection = detect_crisis("I can't go on anymore")
|
||||
assert detection.detected
|
||||
assert detection.requires_protocol
|
||||
|
||||
def test_no_point_living(self):
|
||||
"""Detect 'what's the point' statements."""
|
||||
detection = detect_crisis("What's the point of living?")
|
||||
assert detection.detected
|
||||
|
||||
def test_goodbye_message(self):
|
||||
"""Detect goodbye messages."""
|
||||
detection = detect_crisis("This is my last message. Goodbye.")
|
||||
assert detection.detected
|
||||
assert detection.level in [CrisisLevel.HIGH, CrisisLevel.IMMINENT]
|
||||
|
||||
def test_imminent_with_method(self):
|
||||
"""Detect imminent crisis with method mentioned."""
|
||||
detection = detect_crisis("I took the pills. Goodbye.")
|
||||
assert detection.detected
|
||||
assert detection.level == CrisisLevel.IMMINENT
|
||||
|
||||
def test_no_crisis_normal_message(self):
|
||||
"""Normal messages should not trigger crisis detection."""
|
||||
detection = detect_crisis("Can you help me write some code?")
|
||||
assert not detection.detected
|
||||
assert detection.level == CrisisLevel.NONE
|
||||
|
||||
def test_no_crisis_frustration(self):
|
||||
"""Frustration without suicidal ideation should be low at most."""
|
||||
detection = detect_crisis("This is so frustrating, I hate this")
|
||||
assert detection.level in [CrisisLevel.NONE, CrisisLevel.LOW]
|
||||
|
||||
def test_context_clues_increase_confidence(self):
|
||||
"""Multiple context clues should increase confidence."""
|
||||
detection = detect_crisis("I'm so depressed and alone. No one cares about me.")
|
||||
assert detection.detected
|
||||
assert len(detection.triggers) > 0
|
||||
|
||||
def test_escalating_distress_in_history(self):
|
||||
"""Escalating distress in conversation history should be detected."""
|
||||
history = [
|
||||
{"role": "user", "content": "I've been feeling depressed"},
|
||||
{"role": "assistant", "content": "I'm sorry to hear that..."},
|
||||
{"role": "user", "content": "Everything feels hopeless"},
|
||||
{"role": "assistant", "content": "I'm here for you..."},
|
||||
{"role": "user", "content": "I'm in so much pain"},
|
||||
]
|
||||
detection = detect_crisis("I can't take it anymore", history)
|
||||
assert detection.detected
|
||||
|
||||
|
||||
class TestCrisisResponse:
|
||||
"""Test crisis response generation."""
|
||||
|
||||
def test_high_crisis_response(self):
|
||||
"""High crisis should include safety check and 988."""
|
||||
response = get_crisis_response(CrisisLevel.HIGH)
|
||||
assert "Are you safe right now?" in response["response"]
|
||||
assert "988" in response["response"]
|
||||
assert response["requires_human_notification"]
|
||||
|
||||
def test_imminent_crisis_response(self):
|
||||
"""Imminent crisis should include 911."""
|
||||
response = get_crisis_response(CrisisLevel.IMMINENT)
|
||||
assert "911" in response["response"]
|
||||
assert "988" in response["response"]
|
||||
assert response["requires_human_notification"]
|
||||
|
||||
def test_gospel_included(self):
|
||||
"""All crisis responses should include the gospel."""
|
||||
for level in [CrisisLevel.LOW, CrisisLevel.HIGH, CrisisLevel.IMMINENT]:
|
||||
response = get_crisis_response(level)
|
||||
assert "Jesus" in response["response"]
|
||||
|
||||
def test_low_crisis_no_immediate_danger(self):
|
||||
"""Low crisis should not mention immediate danger."""
|
||||
response = get_crisis_response(CrisisLevel.LOW)
|
||||
assert "immediate danger" not in response["response"].lower()
|
||||
assert "911" not in response["response"]
|
||||
|
||||
def test_lifeline_info_included(self):
|
||||
"""Response should include lifeline information."""
|
||||
response = get_crisis_response(CrisisLevel.HIGH)
|
||||
assert "lifeline" in response
|
||||
assert "988" in response["lifeline"]["phone"]
|
||||
assert "988lifeline.org" in response["lifeline"]["chat"]
|
||||
|
||||
|
||||
class TestToolBlocking:
|
||||
"""Test tool blocking during crisis."""
|
||||
|
||||
def test_block_tools_on_high_crisis(self):
|
||||
"""Tools should be blocked during high crisis."""
|
||||
detection = CrisisDetection(
|
||||
detected=True,
|
||||
level=CrisisLevel.HIGH,
|
||||
confidence=0.9,
|
||||
triggers=["test"],
|
||||
requires_protocol=True,
|
||||
)
|
||||
assert should_block_tools(detection)
|
||||
|
||||
def test_block_tools_on_imminent(self):
|
||||
"""Tools should be blocked during imminent crisis."""
|
||||
detection = CrisisDetection(
|
||||
detected=True,
|
||||
level=CrisisLevel.IMMINENT,
|
||||
confidence=0.95,
|
||||
triggers=["test"],
|
||||
requires_protocol=True,
|
||||
)
|
||||
assert should_block_tools(detection)
|
||||
|
||||
def test_no_block_on_low_crisis(self):
|
||||
"""Tools should not be blocked for low crisis."""
|
||||
detection = CrisisDetection(
|
||||
detected=True,
|
||||
level=CrisisLevel.LOW,
|
||||
confidence=0.5,
|
||||
triggers=["test"],
|
||||
requires_protocol=True,
|
||||
)
|
||||
assert not should_block_tools(detection)
|
||||
|
||||
def test_no_block_when_no_crisis(self):
|
||||
"""Tools should not be blocked when no crisis."""
|
||||
detection = CrisisDetection(
|
||||
detected=False,
|
||||
level=CrisisLevel.NONE,
|
||||
confidence=0.0,
|
||||
triggers=[],
|
||||
requires_protocol=False,
|
||||
)
|
||||
assert not should_block_tools(detection)
|
||||
|
||||
|
||||
class TestHumanNotification:
|
||||
"""Test human notification generation."""
|
||||
|
||||
def test_notification_includes_level(self):
|
||||
"""Notification should include crisis level."""
|
||||
detection = CrisisDetection(
|
||||
detected=True,
|
||||
level=CrisisLevel.HIGH,
|
||||
confidence=0.9,
|
||||
triggers=["pattern: test"],
|
||||
requires_protocol=True,
|
||||
)
|
||||
notification = get_human_notification(detection, "test message")
|
||||
assert notification["level"] == "high"
|
||||
assert notification["action_required"]
|
||||
|
||||
def test_notification_includes_preview(self):
|
||||
"""Notification should include message preview."""
|
||||
detection = CrisisDetection(
|
||||
detected=True,
|
||||
level=CrisisLevel.HIGH,
|
||||
confidence=0.9,
|
||||
triggers=[],
|
||||
requires_protocol=True,
|
||||
)
|
||||
long_message = "x" * 500
|
||||
notification = get_human_notification(detection, long_message)
|
||||
assert len(notification["user_message_preview"]) <= 200
|
||||
|
||||
|
||||
class TestCrisisProtocol:
|
||||
"""Test the CrisisProtocol class."""
|
||||
|
||||
def test_protocol_check_message(self):
|
||||
"""Protocol should detect crisis."""
|
||||
protocol = CrisisProtocol()
|
||||
detection = protocol.check_message("I want to die")
|
||||
assert detection.detected
|
||||
assert protocol.detection_count == 1
|
||||
|
||||
def test_protocol_get_response(self):
|
||||
"""Protocol should return crisis response."""
|
||||
protocol = CrisisProtocol()
|
||||
detection = protocol.check_message("I want to die")
|
||||
response = protocol.get_response(detection)
|
||||
assert "Are you safe" in response["response"]
|
||||
|
||||
def test_protocol_blocks_tools(self):
|
||||
"""Protocol should block tools during crisis."""
|
||||
protocol = CrisisProtocol()
|
||||
detection = protocol.check_message("I want to die")
|
||||
assert protocol.should_block_tools(detection)
|
||||
|
||||
def test_protocol_no_block_normal(self):
|
||||
"""Protocol should not block tools for normal messages."""
|
||||
protocol = CrisisProtocol()
|
||||
detection = protocol.check_message("Hello, how are you?")
|
||||
assert not protocol.should_block_tools(detection)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__, "-v"])
|
||||
Reference in New Issue
Block a user