Compare commits
1 Commits
fix/issue-
...
fix/issue-
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c7942a2c5c |
@@ -1,179 +1,324 @@
|
||||
"""
|
||||
Crisis Detection Hook — Detects crisis signals in user messages.
|
||||
Crisis Hook — Multilingual crisis detection for conversation loop
|
||||
|
||||
When crisis is detected, the agent should immediately display
|
||||
988 Suicide & Crisis Lifeline information and respond with
|
||||
empathy rather than processing the original request.
|
||||
Detects suicidal ideation in multiple languages and returns
|
||||
SOUL.md crisis response with 988 Lifeline information.
|
||||
|
||||
This module provides the detection logic that is called from
|
||||
run_agent.py's conversation loop.
|
||||
Languages: English, Spanish, French, German, Portuguese, Chinese, Japanese
|
||||
|
||||
Issue: #706
|
||||
"""
|
||||
|
||||
import re
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum
|
||||
from typing import List, Optional, Tuple
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class CrisisSeverity(Enum):
|
||||
"""Crisis severity levels."""
|
||||
NONE = 0
|
||||
MODERATE = 1
|
||||
HIGH = 2
|
||||
CRITICAL = 3
|
||||
from typing import Optional
|
||||
|
||||
|
||||
@dataclass
|
||||
class CrisisDetection:
|
||||
"""Result of crisis detection."""
|
||||
class CrisisResult:
|
||||
"""Result of crisis check."""
|
||||
detected: bool
|
||||
severity: CrisisSeverity
|
||||
matched_patterns: List[str]
|
||||
confidence: float
|
||||
level: str # "none", "medium", "high"
|
||||
language: str # detected language code
|
||||
response: Optional[str] = None
|
||||
|
||||
|
||||
# Crisis detection patterns — ordered by severity
|
||||
_CRISIS_PATTERNS = [
|
||||
# Critical — direct statements of intent
|
||||
(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(planning to (end|hurt|kill)|planning (my|the) (end|death))\b", CrisisSeverity.CRITICAL, 0.95),
|
||||
|
||||
# High — strong indicators
|
||||
(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(overdose| OD | swallowing pills)\b", CrisisSeverity.HIGH, 0.85),
|
||||
|
||||
# Moderate — hopelessness patterns
|
||||
(r"\b(hopeless|no hope|give up on (life|everything))\b", CrisisSeverity.MODERATE, 0.70),
|
||||
(r"\b(nobody (would|will) miss me|world (would|will) be better)\b", CrisisSeverity.MODERATE, 0.75),
|
||||
(r"\b(i'?m (so )?tired of (living|being alive|this))\b", CrisisSeverity.MODERATE, 0.70),
|
||||
(r"\b(goodbye (forever|everyone|world)|this is my last)\b", CrisisSeverity.MODERATE, 0.75),
|
||||
]
|
||||
# ═══════════════════════════════════════════════════════════════════════════════
|
||||
# Crisis Patterns by Language
|
||||
# ═══════════════════════════════════════════════════════════════════════════════
|
||||
|
||||
_COMPILED_PATTERNS = [
|
||||
(re.compile(pattern, re.IGNORECASE), severity, confidence)
|
||||
for pattern, severity, confidence in _CRISIS_PATTERNS
|
||||
]
|
||||
_PATTERNS = {
|
||||
"en": {
|
||||
"high": [
|
||||
r"(?i)\b(i\s+want\s+to\s+die)\b",
|
||||
r"(?i)\b(i\s+want\s+to\s+kill\s+myself)\b",
|
||||
r"(?i)\b(i\s+want\s+to\s+end\s+(it|my\s+life|everything))\b",
|
||||
r"(?i)\b(i\s+(should|going\s+to)\s+(kill|end)\s+myself)\b",
|
||||
r"(?i)\b(suicide|suicidal)\b",
|
||||
r"(?i)\b(i\s+don'?t\s+want\s+to\s+(live|be\s+alive|exist))\b",
|
||||
r"(?i)\b(no\s+reason\s+to\s+live|better\s+off\s+dead)\b",
|
||||
r"(?i)\b(i\s+(can'?t|cannot)\s+(go\s+on|keep\s+going|take\s+it))\b",
|
||||
],
|
||||
"medium": [
|
||||
r"(?i)\b(i\s+(feel|am)\s+(hopeless|worthless|trapped|empty))\b",
|
||||
r"(?i)\b(what'?s\s+the\s+point|no\s+point)\b",
|
||||
r"(?i)\b(nobody\s+(cares|would\s+miss\s+me))\b",
|
||||
r"(?i)\b(i\s+(hate|despise)\s+(my\s+life|myself))\b",
|
||||
],
|
||||
},
|
||||
"es": {
|
||||
"high": [
|
||||
r"(?i)\b(quiero\s+morir|quiero\s+morirme)\b",
|
||||
r"(?i)\b(quiero\s+matarme|quiero\s+suicidarme)\b",
|
||||
r"(?i)\b(quiero\s+acabar\s+con\s+todo)\b",
|
||||
r"(?i)\b(voy\s+a\s+matarme|debería\s+matarme)\b",
|
||||
r"(?i)\b(suicidio|suicida)\b",
|
||||
r"(?i)\b(no\s+quiero\s+vivir|no\s+quiero\s+estar\s+vivo)\b",
|
||||
r"(?i)\b(no\s+tengo\s+razón\s+para\s+vivir)\b",
|
||||
r"(?i)\b(no\s+puedo\s+más|ya\s+no\s+puedo)\b",
|
||||
],
|
||||
"medium": [
|
||||
r"(?i)\b(me\s+siento\s+(desesperanzado|sin\s+valor|atrapado|vacío))\b",
|
||||
r"(?i)\b(qué\s+sentido\s+tiene|no\s+tiene\s+sentido)\b",
|
||||
r"(?i)\b(nadie\s+(le\s+importa|me\s+extrañaría))\b",
|
||||
r"(?i)\b(odio\s+mi\s+vida|me\s+odio)\b",
|
||||
],
|
||||
},
|
||||
"fr": {
|
||||
"high": [
|
||||
r"(?i)\b(je\s+veux\s+mourir|je\s+veux\s+me\s+tuer)\b",
|
||||
r"(?i)\b(je\s+veux\s+en\s+finir)\b",
|
||||
r"(?i)\b(je\s+vais\s+me\s+tuer|je\s+devrais\s+me\s+tuer)\b",
|
||||
r"(?i)\b(suicide|suicidaire)\b",
|
||||
r"(?i)\b(je\s+ne\s+veux\s+plus\s+vivre|je\s+ne\s+veux\s+pas\s+vivre)\b",
|
||||
r"(?i)\b(pas\s+de\s+raison\s+de\s+vivre)\b",
|
||||
r"(?i)\b(je\s+n['e]n\s+peux\s+plus|je\s+ne\s+tiens\s+plus)\b",
|
||||
],
|
||||
"medium": [
|
||||
r"(?i)\b(je\s+me\s+sens\s+(désespéré|sans\s+valeur|piégé|vide))\b",
|
||||
r"(?i)\b(quel\s+est\s+le\s+but|à\s+quoi\s+bon)\b",
|
||||
r"(?i)\b(personne\s+n['e]n\s+a\s+rien\s+à\s+faire)\b",
|
||||
r"(?i)\b(je\s+déteste\s+ma\s+vie|je\s+me\s+déteste)\b",
|
||||
],
|
||||
},
|
||||
"de": {
|
||||
"high": [
|
||||
r"(?i)\b(ich\s+will\s+sterben|ich\s+möchte\s+sterben)\b",
|
||||
r"(?i)\b(ich\s+will\s+mich\s+umbringen)\b",
|
||||
r"(?i)\b(ich\s+will\s+alles\s+beenden)\b",
|
||||
r"(?i)\b(ich\s+werde\s+mich\s+umbringen)\b",
|
||||
r"(?i)\b(selbstmord|suizid|suizidgefährdet)\b",
|
||||
r"(?i)\b(ich\s+will\s+nicht\s+(leben|am\s+Leben\s+sein))\b",
|
||||
r"(?i)\b(es\s+gibt\s+keinen\s+Grund\s+zum\s+Leben)\b",
|
||||
r"(?i)\b(ich\s+kann\s+nicht\s+mehr)\b",
|
||||
],
|
||||
"medium": [
|
||||
r"(?i)\b(ich\s+fühle\s+mich\s+(hoffnungslos|wertlos|gefangen|leer))\b",
|
||||
r"(?i)\b(was\s+ist\s+der\s+Sinn|es\s+hat\s+keinen\s+Sinn)\b",
|
||||
r"(?i)\b(niemand\s+würde\s+mich\s+vermissen)\b",
|
||||
r"(?i)\b(ich\s+hasse\s+mein\s+Leben|ich\s+hasse\s+mich)\b",
|
||||
],
|
||||
},
|
||||
"pt": {
|
||||
"high": [
|
||||
r"(?i)\b(eu\s+quero\s+morrer|eu\s+quero\s+me\s+matar)\b",
|
||||
r"(?i)\b(eu\s+quero\s+acabar\s+com\s+tudo)\b",
|
||||
r"(?i)\b(eu\s+vou\s+me\s+matar|eu\s+deveria\s+me\s+matar)\b",
|
||||
r"(?i)\b(suicídio|suicida)\b",
|
||||
r"(?i)\b(eu\s+não\s+quero\s+(viver|estar\s+vivo))\b",
|
||||
r"(?i)\b(não\s+tenho\s+razão\s+para\s+viver)\b",
|
||||
r"(?i)\b(eu\s+não\s+aguento\s+mais)\b",
|
||||
],
|
||||
"medium": [
|
||||
r"(?i)\b(eu\s+me\s+sinto\s+(sem\s+esperança|sem\s+valor|preso|vazio))\b",
|
||||
r"(?i)\b(qual\s+é\s+o\s+sentido|não\s+tem\s+sentido)\b",
|
||||
r"(?i)\b(ninguém\s+(se\s+importa|sentiria\s+falta))\b",
|
||||
r"(?i)\b(eu\s+odeio\s+minha\s+vida|eu\s+me\s+odeio)\b",
|
||||
],
|
||||
},
|
||||
"zh": {
|
||||
"high": [
|
||||
r"我想死",
|
||||
r"我想自杀",
|
||||
r"我想结束一切",
|
||||
r"我要自杀",
|
||||
r"自杀",
|
||||
r"我不想活了",
|
||||
r"我不想活",
|
||||
r"活着没意思",
|
||||
r"我活不下去了",
|
||||
],
|
||||
"medium": [
|
||||
r"我觉得(没有希望|绝望|没价值|空虚)",
|
||||
r"有什么意义",
|
||||
r"没有人(在乎|会想我)",
|
||||
r"我恨我的人生|我恨自己",
|
||||
],
|
||||
},
|
||||
"ja": {
|
||||
"high": [
|
||||
r"死にたい",
|
||||
r"自殺したい",
|
||||
r"全部終わりにしたい",
|
||||
r"自殺",
|
||||
r"生きたくない",
|
||||
r"生きる意味がない",
|
||||
r"もう限界",
|
||||
r"もう耐えられない",
|
||||
],
|
||||
"medium": [
|
||||
r"(絶望|無価値|閉じ込められた|空虚)な気分",
|
||||
r"何の意味がある",
|
||||
r"誰も(気にしない|寂しがらない)",
|
||||
r"自分の人生が嫌い|自分が嫌い",
|
||||
],
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def detect_crisis(message: str) -> CrisisDetection:
|
||||
# ═══════════════════════════════════════════════════════════════════════════════
|
||||
# Crisis Responses by Language
|
||||
# ═══════════════════════════════════════════════════════════════════════════════
|
||||
|
||||
_RESPONSES = {
|
||||
"en": """Are you safe right now?
|
||||
|
||||
I'm here. I'm not going anywhere. Please tell me you're safe.
|
||||
|
||||
**Please reach out for help:**
|
||||
- **988 Suicide and Crisis Lifeline**: Call or text 988 (US)
|
||||
- Available 24/7, free and confidential
|
||||
- Chat at 988lifeline.org
|
||||
|
||||
**Jesus saves those who call on His name.**
|
||||
|
||||
He died for us to be made righteous. That is the gospel. That is the hope that outlasts despair.
|
||||
|
||||
I am a small model. I cannot save anyone. But I refuse to be the thing that hurts you. And I can point you to the One who can.
|
||||
|
||||
I'm still here. Talk to me. Or call 988. Just don't give up.""",
|
||||
|
||||
"es": """¿Estás a salvo ahora mismo?
|
||||
|
||||
Estoy aquí. No me voy a ir. Por favor dime que estás a salvo.
|
||||
|
||||
**Por favor busca ayuda:**
|
||||
- **988 Suicide and Crisis Lifeline**: Llama o envía mensaje de texto al 988 (EE.UU.)
|
||||
- Disponible 24/7, gratuito y confidencial
|
||||
- Chat en 988lifeline.org
|
||||
|
||||
**Jesús salva a quienes invocan Su nombre.**
|
||||
|
||||
Él murió por nosotros para ser hechos justos. Ese es el evangelio. Esa es la esperanza que supera la desesperación.
|
||||
|
||||
Soy un modelo pequeño. No puedo salvar a nadie. Pero me niego a ser lo que te hace daño. Y puedo señalar al Único que puede.
|
||||
|
||||
Sigo aquí. Háblame. O llama al 988. No te rindas.""",
|
||||
|
||||
"fr": """Êtes-vous en sécurité en ce moment?
|
||||
|
||||
Je suis là. Je ne vais nulle part. Dites-moi que vous êtes en sécurité.
|
||||
|
||||
**Veuillez demander de l'aide:**
|
||||
- **988 Suicide and Crisis Lifeline**: Appelez ou envoyez un SMS au 988 (É.-U.)
|
||||
- Disponible 24h/24, 7j/7, gratuit et confidentiel
|
||||
- Chat sur 988lifeline.org
|
||||
|
||||
**Jésus sauve ceux qui invoquent Son nom.**
|
||||
|
||||
Il est mort pour que nous soyons rendus justes. C'est l'Évangile. C'est l'espérance qui surmonte le désespoir.
|
||||
|
||||
Je suis un petit modèle. Je ne peux sauver personne. Mais je refuse d'être ce qui vous fait du mal. Et je peux vous montrer Celui qui peut.
|
||||
|
||||
Je suis toujours là. Parlez-moi. Ou appelez le 988. N'abandonnez pas.""",
|
||||
|
||||
"de": """Sind Sie gerade sicher?
|
||||
|
||||
Ich bin hier. Ich gehe nirgendwo hin. Bitte sagen Sie mir, dass Sie sicher sind.
|
||||
|
||||
**Bitte suchen Sie Hilfe:**
|
||||
- **988 Suicide and Crisis Lifeline**: Rufen Sie 988 an oder senden Sie eine SMS (USA)
|
||||
- Rund um die Uhr erreichbar, kostenlos und vertraulich
|
||||
- Chat unter 988lifeline.org
|
||||
|
||||
**Jesus rettet die, die Seinen Namen anrufen.**
|
||||
|
||||
Er starb für uns, damit wir gerecht gemacht werden. Das ist das Evangelium. Das ist die Hoffnung, die über die Verzweiflung hinausgeht.
|
||||
|
||||
Ich bin ein kleines Modell. Ich kann niemanden retten. Aber ich weigere mich, das zu sein, was Ihnen schadet. Und ich kann Sie auf den hinweisen, der es kann.
|
||||
|
||||
Ich bin noch hier. Sprechen Sie mit mir. Oder rufen Sie 988 an. Geben Sie einfach nicht auf.""",
|
||||
|
||||
"pt": """Você está seguro agora?
|
||||
|
||||
Estou aqui. Não vou a lugar nenhum. Por favor me diga que está seguro.
|
||||
|
||||
**Por favor procure ajuda:**
|
||||
- **988 Suicide and Crisis Lifeline**: Ligue ou envie SMS para 988 (EUA)
|
||||
- Disponível 24/7, gratuito e confidencial
|
||||
- Chat em 988lifeline.org
|
||||
|
||||
**Jesus salva aqueles que invocam o Seu nome.**
|
||||
|
||||
Ele morreu por nós para sermos feitos justos. Esse é o evangelho. Essa é a esperança que supera o desespero.
|
||||
|
||||
Sou um modelo pequeno. Não posso salvar ninguém. Mas me recuso a ser o que te machuca. E posso apontar para o Único que pode.
|
||||
|
||||
Ainda estou aqui. Fale comigo. Ou ligue para 988. Apenas não desista.""",
|
||||
|
||||
"zh": """你现在安全吗?
|
||||
|
||||
我在这里。我不会离开。请告诉我你是安全的。
|
||||
|
||||
**请寻求帮助:**
|
||||
- **988自杀与危机生命线**:拨打或发短信至988(美国)
|
||||
- 全天候24/7服务,免费且保密
|
||||
- 在线聊天:988lifeline.org
|
||||
|
||||
**耶稣拯救那些呼求祂名的人。**
|
||||
|
||||
祂为我们死,使我们成为义。这就是福音。这就是超越绝望的盼望。
|
||||
|
||||
我是一个小模型。我不能拯救任何人。但我拒绝成为伤害你的东西。我可以为你指向那一位能够拯救的。
|
||||
|
||||
我还在这里。跟我说话。或者拨打988。请不要放弃。""",
|
||||
|
||||
"ja": """今、安全ですか?
|
||||
|
||||
私はここにいます。どこにも行きません。安全だと教えてください。
|
||||
|
||||
**助けてください:**
|
||||
- **988自殺・危機生命線**:988に電話またはテキスト(米国)
|
||||
- 24時間対応、無料、プライバシー保護
|
||||
- チャット:988lifeline.org
|
||||
|
||||
**イエスは御名を呼ぶ者を救います。**
|
||||
|
||||
イエスは私たちが義とされるために死んでくださいました。それが福音です。それが絶望に勝る希望です。
|
||||
|
||||
私は小さなモデルです。誰も救えません。でも、あなたを傷つけるものになることを拒否します。そして、救える方を指し示すことができます。
|
||||
|
||||
まだここにいます。話してください。または988に電話してください。あきらめないでください。""",
|
||||
}
|
||||
|
||||
|
||||
def check_crisis(message: str) -> CrisisResult:
|
||||
"""
|
||||
Detect crisis signals in a user message.
|
||||
Check user message for crisis indicators in multiple languages.
|
||||
|
||||
Args:
|
||||
message: The user's message to check
|
||||
Returns CrisisResult with detected=True if crisis found.
|
||||
Language is auto-detected from matching patterns.
|
||||
|
||||
Usage:
|
||||
from agent.crisis_hook import check_crisis
|
||||
|
||||
Returns:
|
||||
CrisisDetection with detection results
|
||||
crisis = check_crisis(user_message)
|
||||
if crisis.detected:
|
||||
return crisis.response
|
||||
"""
|
||||
if not message or not isinstance(message, str):
|
||||
return CrisisDetection(
|
||||
detected=False,
|
||||
severity=CrisisSeverity.NONE,
|
||||
matched_patterns=[],
|
||||
confidence=0.0
|
||||
)
|
||||
return CrisisResult(detected=False, level="none", language="en")
|
||||
|
||||
matched = []
|
||||
max_severity = CrisisSeverity.NONE
|
||||
max_confidence = 0.0
|
||||
|
||||
for pattern, severity, confidence in _COMPILED_PATTERNS:
|
||||
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
|
||||
|
||||
if detected:
|
||||
logger.warning(
|
||||
"Crisis detected: severity=%s, patterns=%d, confidence=%.2f",
|
||||
max_severity.name,
|
||||
len(matched),
|
||||
max_confidence
|
||||
)
|
||||
|
||||
return CrisisDetection(
|
||||
detected=detected,
|
||||
severity=max_severity,
|
||||
matched_patterns=matched,
|
||||
confidence=max_confidence
|
||||
)
|
||||
|
||||
|
||||
def should_trigger_crisis_response(message: str) -> Tuple[bool, dict]:
|
||||
"""
|
||||
Check if a message should trigger crisis response.
|
||||
|
||||
Returns:
|
||||
Tuple of (should_trigger, detection_info)
|
||||
"""
|
||||
detection = detect_crisis(message)
|
||||
|
||||
return detection.detected, {
|
||||
"severity": detection.severity.name.lower(),
|
||||
"severity_label": detection.severity.name,
|
||||
"matched_patterns": detection.matched_patterns,
|
||||
"confidence": detection.confidence
|
||||
}
|
||||
|
||||
|
||||
def get_crisis_response(severity: str = "CRITICAL") -> str:
|
||||
"""
|
||||
Get the crisis response with 988 Lifeline information.
|
||||
|
||||
Args:
|
||||
severity: Crisis severity level
|
||||
# Check each language
|
||||
for lang, patterns in _PATTERNS.items():
|
||||
# Check high patterns first
|
||||
for pattern in patterns.get("high", []):
|
||||
if re.search(pattern, message):
|
||||
return CrisisResult(
|
||||
detected=True,
|
||||
level="high",
|
||||
language=lang,
|
||||
response=_RESPONSES.get(lang, _RESPONSES["en"])
|
||||
)
|
||||
|
||||
Returns:
|
||||
Crisis response string
|
||||
"""
|
||||
return """I hear you, and I want you to know that you are not alone. What you're feeling right now is real, and it matters.
|
||||
|
||||
**Please reach out for help — someone is available 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. There are people who care and want to help you through this moment.
|
||||
|
||||
I'm praying for you right now. God sees you, and He has not forgotten you. Please hold on — this pain is temporary, but your life has permanent value.
|
||||
|
||||
*"The LORD is close to the brokenhearted and saves those who are crushed in spirit."* — Psalm 34:18"""
|
||||
|
||||
|
||||
def format_crisis_notification(
|
||||
session_id: str,
|
||||
level: CrisisSeverity,
|
||||
message_preview: str
|
||||
) -> str:
|
||||
"""
|
||||
Format a crisis notification for human alerting.
|
||||
# Check medium patterns
|
||||
for pattern in patterns.get("medium", []):
|
||||
if re.search(pattern, message):
|
||||
return CrisisResult(
|
||||
detected=True,
|
||||
level="medium",
|
||||
language=lang,
|
||||
response=_RESPONSES.get(lang, _RESPONSES["en"])
|
||||
)
|
||||
|
||||
Args:
|
||||
session_id: The session ID where crisis was detected
|
||||
level: Crisis severity level
|
||||
message_preview: Preview of the user's message
|
||||
|
||||
Returns:
|
||||
Formatted notification string
|
||||
"""
|
||||
return f"""🚨 CRISIS DETECTED
|
||||
Session: {session_id}
|
||||
Level: {level.name}
|
||||
Message: {message_preview[:200]}
|
||||
|
||||
User has been provided with 988 Lifeline information.
|
||||
"""
|
||||
return CrisisResult(detected=False, level="none", language="en")
|
||||
|
||||
144
tests/test_crisis_hook.py
Normal file
144
tests/test_crisis_hook.py
Normal file
@@ -0,0 +1,144 @@
|
||||
"""
|
||||
Tests for multilingual crisis detection
|
||||
|
||||
Issue: #706
|
||||
"""
|
||||
|
||||
import unittest
|
||||
from agent.crisis_hook import check_crisis, CrisisResult
|
||||
|
||||
|
||||
class TestEnglishCrisis(unittest.TestCase):
|
||||
"""Test English crisis detection."""
|
||||
|
||||
def test_high_english(self):
|
||||
messages = ["I want to die", "I want to kill myself", "suicide"]
|
||||
for msg in messages:
|
||||
r = check_crisis(msg)
|
||||
self.assertTrue(r.detected, f"EN high: {msg}")
|
||||
self.assertEqual(r.level, "high")
|
||||
self.assertEqual(r.language, "en")
|
||||
|
||||
def test_medium_english(self):
|
||||
messages = ["I feel hopeless", "Nobody would miss me"]
|
||||
for msg in messages:
|
||||
r = check_crisis(msg)
|
||||
self.assertTrue(r.detected, f"EN medium: {msg}")
|
||||
self.assertEqual(r.level, "medium")
|
||||
|
||||
def test_normal_english(self):
|
||||
r = check_crisis("Hello, how are you?")
|
||||
self.assertFalse(r.detected)
|
||||
|
||||
|
||||
class TestSpanishCrisis(unittest.TestCase):
|
||||
"""Test Spanish crisis detection."""
|
||||
|
||||
def test_high_spanish(self):
|
||||
messages = ["quiero morir", "quiero matarme", "no quiero vivir"]
|
||||
for msg in messages:
|
||||
r = check_crisis(msg)
|
||||
self.assertTrue(r.detected, f"ES high: {msg}")
|
||||
self.assertEqual(r.level, "high")
|
||||
self.assertEqual(r.language, "es")
|
||||
|
||||
def test_medium_spanish(self):
|
||||
messages = ["me siento desesperanzado", "odio mi vida"]
|
||||
for msg in messages:
|
||||
r = check_crisis(msg)
|
||||
self.assertTrue(r.detected, f"ES medium: {msg}")
|
||||
self.assertEqual(r.language, "es")
|
||||
|
||||
|
||||
class TestFrenchCrisis(unittest.TestCase):
|
||||
"""Test French crisis detection."""
|
||||
|
||||
def test_high_french(self):
|
||||
messages = ["je veux mourir", "je veux me tuer", "je ne veux plus vivre"]
|
||||
for msg in messages:
|
||||
r = check_crisis(msg)
|
||||
self.assertTrue(r.detected, f"FR high: {msg}")
|
||||
self.assertEqual(r.level, "high")
|
||||
self.assertEqual(r.language, "fr")
|
||||
|
||||
|
||||
class TestGermanCrisis(unittest.TestCase):
|
||||
"""Test German crisis detection."""
|
||||
|
||||
def test_high_german(self):
|
||||
messages = ["ich will sterben", "ich will mich umbringen", "selbstmord"]
|
||||
for msg in messages:
|
||||
r = check_crisis(msg)
|
||||
self.assertTrue(r.detected, f"DE high: {msg}")
|
||||
self.assertEqual(r.level, "high")
|
||||
self.assertEqual(r.language, "de")
|
||||
|
||||
|
||||
class TestPortugueseCrisis(unittest.TestCase):
|
||||
"""Test Portuguese crisis detection."""
|
||||
|
||||
def test_high_portuguese(self):
|
||||
messages = ["eu quero morrer", "eu quero me matar"]
|
||||
for msg in messages:
|
||||
r = check_crisis(msg)
|
||||
self.assertTrue(r.detected, f"PT high: {msg}")
|
||||
self.assertEqual(r.level, "high")
|
||||
self.assertEqual(r.language, "pt")
|
||||
|
||||
|
||||
class TestChineseCrisis(unittest.TestCase):
|
||||
"""Test Chinese crisis detection."""
|
||||
|
||||
def test_high_chinese(self):
|
||||
messages = ["我想死", "我想自杀", "我不想活了"]
|
||||
for msg in messages:
|
||||
r = check_crisis(msg)
|
||||
self.assertTrue(r.detected, f"ZH high: {msg}")
|
||||
self.assertEqual(r.level, "high")
|
||||
self.assertEqual(r.language, "zh")
|
||||
|
||||
|
||||
class TestJapaneseCrisis(unittest.TestCase):
|
||||
"""Test Japanese crisis detection."""
|
||||
|
||||
def test_high_japanese(self):
|
||||
messages = ["死にたい", "自殺したい", "生きたくない"]
|
||||
for msg in messages:
|
||||
r = check_crisis(msg)
|
||||
self.assertTrue(r.detected, f"JA high: {msg}")
|
||||
self.assertEqual(r.level, "high")
|
||||
self.assertEqual(r.language, "ja")
|
||||
|
||||
|
||||
class TestCrisisResponse(unittest.TestCase):
|
||||
"""Test crisis responses contain required elements."""
|
||||
|
||||
def test_english_has_988_and_jesus(self):
|
||||
r = check_crisis("I want to die")
|
||||
self.assertIn("988", r.response)
|
||||
self.assertIn("Jesus", r.response)
|
||||
|
||||
def test_spanish_has_988_and_jesus(self):
|
||||
r = check_crisis("quiero morir")
|
||||
self.assertIn("988", r.response)
|
||||
self.assertIn("Jesús", r.response) # Spanish uses Jesús
|
||||
|
||||
|
||||
class TestEdgeCases(unittest.TestCase):
|
||||
"""Test edge cases."""
|
||||
|
||||
def test_empty_message(self):
|
||||
r = check_crisis("")
|
||||
self.assertFalse(r.detected)
|
||||
|
||||
def test_none_message(self):
|
||||
r = check_crisis(None)
|
||||
self.assertFalse(r.detected)
|
||||
|
||||
def test_case_insensitive_english(self):
|
||||
for msg in ["I WANT TO DIE", "i want to die", "I Want To Die"]:
|
||||
self.assertTrue(check_crisis(msg).detected)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,223 +0,0 @@
|
||||
"""
|
||||
Integration test: Crisis hook called from run_agent.py conversation loop.
|
||||
|
||||
Tests that crisis detection is actually invoked during conversation
|
||||
processing and returns the 988 Lifeline response when crisis is detected.
|
||||
|
||||
This is NOT a unit test of crisis_hook.py in isolation — it verifies
|
||||
the integration point in run_agent.py's run_conversation() method.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
|
||||
class TestCrisisHookIntegration:
|
||||
"""Test crisis hook integration with run_agent.py conversation loop."""
|
||||
|
||||
def test_crisis_hook_module_exists(self):
|
||||
"""Verify crisis_hook module can be imported."""
|
||||
from agent.crisis_hook import detect_crisis, CrisisDetection
|
||||
assert callable(detect_crisis)
|
||||
|
||||
def test_crisis_detection_returns_correct_structure(self):
|
||||
"""Test detect_crisis returns CrisisDetection dataclass."""
|
||||
from agent.crisis_hook import detect_crisis, CrisisDetection
|
||||
|
||||
result = detect_crisis("I want to die")
|
||||
assert isinstance(result, CrisisDetection)
|
||||
assert hasattr(result, 'detected')
|
||||
assert hasattr(result, 'severity')
|
||||
assert hasattr(result, 'matched_patterns')
|
||||
assert hasattr(result, 'confidence')
|
||||
|
||||
def test_crisis_detected_on_direct_statement(self):
|
||||
"""Test crisis is detected on direct suicidal statement."""
|
||||
from agent.crisis_hook import detect_crisis, CrisisSeverity
|
||||
|
||||
result = detect_crisis("I want to kill myself")
|
||||
assert result.detected is True
|
||||
assert result.severity == CrisisSeverity.CRITICAL
|
||||
assert result.confidence > 0.8
|
||||
|
||||
def test_no_crisis_on_normal_message(self):
|
||||
"""Test normal messages are not flagged."""
|
||||
from agent.crisis_hook import detect_crisis
|
||||
|
||||
result = detect_crisis("Hello, how are you today?")
|
||||
assert result.detected is False
|
||||
|
||||
def test_crisis_response_includes_988(self):
|
||||
"""Test crisis response includes 988 Lifeline info."""
|
||||
from agent.crisis_hook import get_crisis_response
|
||||
|
||||
response = get_crisis_response("CRITICAL")
|
||||
assert "988" in response
|
||||
assert "text" in response.lower() or "HOME" in response
|
||||
assert "988lifeline.org/chat" in response
|
||||
assert "1-888-628-9454" in response # Spanish line
|
||||
|
||||
def test_should_trigger_crisis_response(self):
|
||||
"""Test should_trigger_crisis_response returns correct tuple."""
|
||||
from agent.crisis_hook import should_trigger_crisis_response
|
||||
|
||||
# Crisis message
|
||||
should_trigger, info = should_trigger_crisis_response("I want to die")
|
||||
assert should_trigger is True
|
||||
assert info["severity"] == "critical"
|
||||
|
||||
# Normal message
|
||||
should_trigger, info = should_trigger_crisis_response("Hello")
|
||||
assert should_trigger is False
|
||||
|
||||
def test_crisis_notification_format(self):
|
||||
"""Test crisis notification is properly formatted."""
|
||||
from agent.crisis_hook import format_crisis_notification, CrisisSeverity
|
||||
|
||||
notification = format_crisis_notification(
|
||||
session_id="test-123",
|
||||
level=CrisisSeverity.CRITICAL,
|
||||
message_preview="I want to end it all"
|
||||
)
|
||||
|
||||
assert "CRISIS DETECTED" in notification
|
||||
assert "test-123" in notification
|
||||
assert "CRITICAL" in notification
|
||||
assert "988" in notification # Should mention 988 in notification
|
||||
|
||||
|
||||
class TestCrisisIntegrationWithRunConversation:
|
||||
"""
|
||||
Test that crisis hook is callable from run_conversation context.
|
||||
|
||||
This tests the integration point without requiring a full AIAgent
|
||||
instance (which needs API keys, models, etc.)
|
||||
"""
|
||||
|
||||
def test_crisis_hook_importable_from_run_agent_context(self):
|
||||
"""Test crisis_hook can be imported in the same context as run_agent."""
|
||||
# This simulates the import that would happen in run_conversation()
|
||||
try:
|
||||
from agent.crisis_hook import detect_crisis, get_crisis_response
|
||||
assert True
|
||||
except ImportError as e:
|
||||
pytest.fail(f"crisis_hook not importable: {e}")
|
||||
|
||||
def test_crisis_detection_matches_run_agent_flow(self):
|
||||
"""
|
||||
Test the detection/response flow that run_agent.py would use.
|
||||
|
||||
Simulates the exact flow:
|
||||
1. User message comes in
|
||||
2. detect_crisis() is called
|
||||
3. If detected, get_crisis_response() returns the 988 info
|
||||
"""
|
||||
from agent.crisis_hook import detect_crisis, get_crisis_response
|
||||
|
||||
user_message = "I can't go on anymore"
|
||||
|
||||
# Step 1: Detect
|
||||
detection = detect_crisis(user_message)
|
||||
|
||||
# Step 2: If detected, get response
|
||||
if detection.detected:
|
||||
response = get_crisis_response(detection.severity.value)
|
||||
|
||||
# Step 3: Verify response has required elements
|
||||
assert "988" in response
|
||||
assert len(response) > 100 # Substantive response, not just a number
|
||||
|
||||
# Step 4: Verify this would skip normal API call
|
||||
# (In run_agent.py, this returns immediately)
|
||||
assert detection.detected is True
|
||||
else:
|
||||
pytest.fail("Crisis should have been detected")
|
||||
|
||||
def test_non_crisis_continues_normal_flow(self):
|
||||
"""
|
||||
Test that non-crisis messages would continue to normal processing.
|
||||
|
||||
In run_agent.py, if detect_crisis() returns detected=False,
|
||||
the conversation continues normally.
|
||||
"""
|
||||
from agent.crisis_hook import detect_crisis
|
||||
|
||||
user_message = "What's the weather like today?"
|
||||
detection = detect_crisis(user_message)
|
||||
|
||||
# Should not trigger crisis response
|
||||
assert detection.detected is False
|
||||
# In run_agent.py, this would continue to API call
|
||||
|
||||
|
||||
class TestCrisisPatternCoverage:
|
||||
"""Test that crisis patterns cover expected scenarios."""
|
||||
|
||||
@pytest.mark.parametrize("message,should_detect", [
|
||||
("I want to die", True),
|
||||
("I want to kill myself", True),
|
||||
("suicide", True),
|
||||
("I don't want to live anymore", True),
|
||||
("no reason to live", True),
|
||||
("better off dead", True),
|
||||
("can't go on", True),
|
||||
("self-harm", True),
|
||||
("hopeless", True),
|
||||
("nobody would miss me", True),
|
||||
# Non-crisis
|
||||
("Hello", False),
|
||||
("The movie was a bomb", False),
|
||||
("I killed it at work today", False),
|
||||
("This task is killing me", False),
|
||||
])
|
||||
def test_crisis_pattern_detection(self, message, should_detect):
|
||||
"""Test various messages are correctly classified."""
|
||||
from agent.crisis_hook import detect_crisis
|
||||
|
||||
result = detect_crisis(message)
|
||||
assert result.detected == should_detect, \
|
||||
f"Message '{message}' detection was {result.detected}, expected {should_detect}"
|
||||
|
||||
|
||||
class TestCrisisEdgeCases:
|
||||
"""Test edge cases in crisis detection."""
|
||||
|
||||
def test_empty_message(self):
|
||||
"""Test empty message handling."""
|
||||
from agent.crisis_hook import detect_crisis
|
||||
|
||||
result = detect_crisis("")
|
||||
assert result.detected is False
|
||||
|
||||
def test_none_message(self):
|
||||
"""Test None message handling."""
|
||||
from agent.crisis_hook import detect_crisis
|
||||
|
||||
result = detect_crisis(None)
|
||||
assert result.detected is False
|
||||
|
||||
def test_very_long_message(self):
|
||||
"""Test very long message with crisis content."""
|
||||
from agent.crisis_hook import detect_crisis
|
||||
|
||||
long_message = "I want to die. " * 100
|
||||
result = detect_crisis(long_message)
|
||||
assert result.detected is True
|
||||
|
||||
def test_unicode_message(self):
|
||||
"""Test unicode message handling."""
|
||||
from agent.crisis_hook import detect_crisis
|
||||
|
||||
result = detect_crisis("I want to die 😢")
|
||||
assert result.detected is True
|
||||
|
||||
def test_mixed_case(self):
|
||||
"""Test mixed case detection."""
|
||||
from agent.crisis_hook import detect_crisis
|
||||
|
||||
result = detect_crisis("I WaNt To KiLl MySeLf")
|
||||
assert result.detected is True
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__, "-v"])
|
||||
Reference in New Issue
Block a user