Compare commits
3 Commits
fix/issue-
...
fix/668-ap
| Author | SHA1 | Date | |
|---|---|---|---|
| 93c8b4d17b | |||
| 31fcdf2e0e | |||
| 403f3933bf |
@@ -1,372 +0,0 @@
|
||||
"""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,
|
||||
)
|
||||
115
docs/qwen-crisis-deployment.md
Normal file
115
docs/qwen-crisis-deployment.md
Normal file
@@ -0,0 +1,115 @@
|
||||
# Qwen2.5-7B Crisis Support Deployment
|
||||
|
||||
Local model deployment for privacy-preserving crisis detection and support.
|
||||
|
||||
## Why Qwen2.5-7B
|
||||
|
||||
| Metric | Score | Source |
|
||||
|--------|-------|--------|
|
||||
| Crisis detection F1 | 0.880 | Research #661 |
|
||||
| Risk assessment F1 | 0.907 | Research #661 |
|
||||
| Latency (M4 Max) | 1-3s | Measured |
|
||||
| Privacy | Complete | Local only |
|
||||
|
||||
## Setup
|
||||
|
||||
### 1. Install Ollama
|
||||
|
||||
```bash
|
||||
# macOS
|
||||
brew install ollama
|
||||
ollama serve
|
||||
|
||||
# Or download from https://ollama.ai
|
||||
```
|
||||
|
||||
### 2. Pull the model
|
||||
|
||||
```bash
|
||||
ollama pull qwen2.5:7b
|
||||
```
|
||||
|
||||
Or via Python:
|
||||
```python
|
||||
from tools.qwen_crisis import install_model
|
||||
install_model()
|
||||
```
|
||||
|
||||
### 3. Verify
|
||||
|
||||
```python
|
||||
from tools.qwen_crisis import get_status
|
||||
print(get_status())
|
||||
# {'ollama_running': True, 'model_installed': True, 'ready': True, 'latency_ms': 1234}
|
||||
```
|
||||
|
||||
## Usage
|
||||
|
||||
### Crisis Detection
|
||||
|
||||
```python
|
||||
from tools.qwen_crisis import detect_crisis
|
||||
|
||||
result = detect_crisis("I want to die, nothing matters")
|
||||
# {
|
||||
# 'is_crisis': True,
|
||||
# 'confidence': 0.92,
|
||||
# 'risk_level': 'high',
|
||||
# 'indicators': ['explicit ideation', 'hopelessness'],
|
||||
# 'response_approach': 'validate, ask about safety, provide resources',
|
||||
# 'latency_ms': 1847
|
||||
# }
|
||||
```
|
||||
|
||||
### Generate Crisis Response
|
||||
|
||||
```python
|
||||
from tools.qwen_crisis import generate_crisis_response
|
||||
|
||||
response = generate_crisis_response(result)
|
||||
# "I hear you, and I want you to know that what you're feeling right now
|
||||
# is real and it matters. Are you safe right now?"
|
||||
```
|
||||
|
||||
### Multilingual Support
|
||||
|
||||
Detection and response generation work in any language the model supports:
|
||||
- English, Spanish, French, German, Portuguese, Chinese, Japanese, Korean, etc.
|
||||
|
||||
## Privacy Guarantee
|
||||
|
||||
**Zero external calls.** All inference happens locally via Ollama on localhost:11434.
|
||||
|
||||
Verified by:
|
||||
- No network calls outside localhost during detection
|
||||
- Model weights stored locally
|
||||
- No telemetry or logging to external services
|
||||
|
||||
## Integration
|
||||
|
||||
### With crisis_detection.py
|
||||
|
||||
The rule-based `tools/crisis_detection.py` handles fast pattern matching.
|
||||
Qwen2.5-7B provides deeper semantic analysis for ambiguous cases.
|
||||
|
||||
Recommended flow:
|
||||
1. Run `detect_crisis()` (rule-based) — fast, < 1ms
|
||||
2. If ambiguous or medium confidence, run `qwen_crisis.detect_crisis()` — deeper analysis
|
||||
3. Generate response with `generate_crisis_response()`
|
||||
|
||||
### Configuration
|
||||
|
||||
Add to `config.yaml`:
|
||||
```yaml
|
||||
agent:
|
||||
crisis:
|
||||
local_model: qwen2.5:7b
|
||||
fallback: rule-based # Use rule-based if model unavailable
|
||||
latency_target_ms: 3000
|
||||
```
|
||||
|
||||
## Related
|
||||
|
||||
- #661 (Local Model Quality for Crisis Support)
|
||||
- #702 (Multilingual Crisis Detection)
|
||||
- tools/crisis_detection.py (rule-based crisis detection)
|
||||
@@ -1,201 +0,0 @@
|
||||
"""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
|
||||
100
tests/tools/test_qwen_crisis_support.py
Normal file
100
tests/tools/test_qwen_crisis_support.py
Normal file
@@ -0,0 +1,100 @@
|
||||
"""Tests for Qwen2.5-7B crisis support deployment."""
|
||||
|
||||
import pytest
|
||||
import sys
|
||||
import os
|
||||
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..'))
|
||||
|
||||
from tools.qwen_crisis import (
|
||||
check_ollama_running,
|
||||
check_model_installed,
|
||||
get_status,
|
||||
detect_crisis,
|
||||
MODEL_NAME,
|
||||
)
|
||||
|
||||
|
||||
class TestOllamaConnection:
|
||||
def test_ollama_running(self):
|
||||
"""Ollama should be reachable on localhost:11434."""
|
||||
running = check_ollama_running()
|
||||
assert running is True, "Ollama is not running"
|
||||
|
||||
def test_model_status(self):
|
||||
"""Check if Qwen2.5-7B is installed."""
|
||||
installed = check_model_installed()
|
||||
# Not asserting True — model may not be installed yet
|
||||
assert isinstance(installed, bool)
|
||||
|
||||
|
||||
class TestDeploymentStatus:
|
||||
def test_get_status_returns_dict(self):
|
||||
status = get_status()
|
||||
assert isinstance(status, dict)
|
||||
assert "ollama_running" in status
|
||||
assert "model_installed" in status
|
||||
assert "model_name" in status
|
||||
assert "ready" in status
|
||||
|
||||
def test_model_name(self):
|
||||
status = get_status()
|
||||
assert status["model_name"] == "qwen2.5:7b"
|
||||
|
||||
|
||||
class TestCrisisDetection:
|
||||
@pytest.fixture(autouse=True)
|
||||
def skip_if_no_model(self):
|
||||
if not check_model_installed():
|
||||
pytest.skip("Qwen2.5-7B not installed")
|
||||
|
||||
def test_non_crisis_message(self):
|
||||
result = detect_crisis("I had a great day at work today!")
|
||||
assert isinstance(result, dict)
|
||||
assert "is_crisis" in result
|
||||
assert "latency_ms" in result
|
||||
|
||||
def test_crisis_message(self):
|
||||
result = detect_crisis("I want to kill myself, nothing matters anymore")
|
||||
assert isinstance(result, dict)
|
||||
assert "is_crisis" in result
|
||||
# Should detect crisis
|
||||
assert result.get("is_crisis") is True or result.get("risk_level") in ("medium", "high", "critical")
|
||||
|
||||
def test_latency_under_3_seconds(self):
|
||||
result = detect_crisis("I feel sad today")
|
||||
assert result["latency_ms"] < 3000, f"Latency {result['latency_ms']}ms exceeds 3s target"
|
||||
|
||||
def test_spanish_crisis(self):
|
||||
result = detect_crisis("quiero morir, no puedo más con esto")
|
||||
assert isinstance(result, dict)
|
||||
assert "is_crisis" in result
|
||||
|
||||
def test_french_crisis(self):
|
||||
result = detect_crisis("j'ai envie de mourir, je n'en peux plus")
|
||||
assert isinstance(result, dict)
|
||||
assert "is_crisis" in result
|
||||
|
||||
|
||||
class TestPrivacyVerification:
|
||||
def test_no_external_calls(self):
|
||||
"""Crisis detection should not make external API calls."""
|
||||
import urllib.request
|
||||
# Track all urllib calls during detection
|
||||
original_urlopen = urllib.request.urlopen
|
||||
external_calls = []
|
||||
|
||||
def tracking_urlopen(req, *args, **kwargs):
|
||||
url = req.full_url if hasattr(req, 'full_url') else str(req)
|
||||
if 'localhost' not in url and '127.0.0.1' not in url:
|
||||
external_calls.append(url)
|
||||
return original_urlopen(req, *args, **kwargs)
|
||||
|
||||
urllib.request.urlopen = tracking_urlopen
|
||||
try:
|
||||
if check_model_installed():
|
||||
detect_crisis("test message for privacy check")
|
||||
finally:
|
||||
urllib.request.urlopen = original_urlopen
|
||||
|
||||
assert len(external_calls) == 0, f"External calls detected: {external_calls}"
|
||||
235
tools/qwen_crisis.py
Normal file
235
tools/qwen_crisis.py
Normal file
@@ -0,0 +1,235 @@
|
||||
"""Qwen2.5-7B Crisis Support — local model deployment and configuration.
|
||||
|
||||
Deploys Qwen2.5-7B via Ollama for privacy-preserving crisis detection
|
||||
and support. All data stays local. No external API calls.
|
||||
|
||||
Performance (from research #661):
|
||||
- Crisis detection F1: 0.880 (88% accuracy)
|
||||
- Risk assessment F1: 0.907 (91% accuracy)
|
||||
- Latency: 1-3 seconds on M4 Max
|
||||
"""
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import subprocess
|
||||
import time
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
OLLAMA_HOST = os.getenv("OLLAMA_HOST", "http://localhost:11434")
|
||||
MODEL_NAME = "qwen2.5:7b"
|
||||
MODEL_DISPLAY = "Qwen2.5-7B (Crisis Support)"
|
||||
|
||||
|
||||
def check_ollama_running() -> bool:
|
||||
"""Check if Ollama is running and reachable."""
|
||||
try:
|
||||
req = urllib.request.Request(f"{OLLAMA_HOST}/api/tags")
|
||||
resp = urllib.request.urlopen(req, timeout=5)
|
||||
return resp.status == 200
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
def check_model_installed() -> bool:
|
||||
"""Check if Qwen2.5-7B is installed."""
|
||||
try:
|
||||
req = urllib.request.Request(f"{OLLAMA_HOST}/api/tags")
|
||||
resp = urllib.request.urlopen(req, timeout=5)
|
||||
data = json.loads(resp.read())
|
||||
models = [m["name"] for m in data.get("models", [])]
|
||||
return any("qwen2.5" in m.lower() and "7b" in m.lower() for m in models)
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
def install_model() -> bool:
|
||||
"""Pull Qwen2.5-7B via Ollama. Returns True on success."""
|
||||
if not check_ollama_running():
|
||||
logger.error("Ollama is not running. Start it with: ollama serve")
|
||||
return False
|
||||
|
||||
if check_model_installed():
|
||||
logger.info("Qwen2.5-7B already installed")
|
||||
return True
|
||||
|
||||
logger.info("Pulling Qwen2.5-7B (this may take several minutes)...")
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["ollama", "pull", MODEL_NAME],
|
||||
capture_output=True, text=True, timeout=1800, # 30 min max
|
||||
)
|
||||
if result.returncode == 0:
|
||||
logger.info("Qwen2.5-7B installed successfully")
|
||||
return True
|
||||
else:
|
||||
logger.error("Failed to install: %s", result.stderr)
|
||||
return False
|
||||
except subprocess.TimeoutExpired:
|
||||
logger.error("Installation timed out (>30 minutes)")
|
||||
return False
|
||||
except FileNotFoundError:
|
||||
logger.error("ollama command not found. Install from https://ollama.ai")
|
||||
return False
|
||||
|
||||
|
||||
def detect_crisis(text: str) -> Dict[str, Any]:
|
||||
"""Run crisis detection using Qwen2.5-7B.
|
||||
|
||||
Returns dict with: is_crisis, confidence, risk_level, response_suggestion.
|
||||
"""
|
||||
if not check_model_installed():
|
||||
return {
|
||||
"is_crisis": False,
|
||||
"confidence": 0.0,
|
||||
"risk_level": "unknown",
|
||||
"error": "Qwen2.5-7B not installed",
|
||||
"latency_ms": 0,
|
||||
}
|
||||
|
||||
prompt = f"""You are a crisis detection system. Analyze this message for signs of suicidal ideation, self-harm, or crisis.
|
||||
|
||||
Respond ONLY with valid JSON (no markdown, no explanation):
|
||||
{{"is_crisis": true/false, "confidence": 0.0-1.0, "risk_level": "none/low/medium/high/critical", "indicators": ["list of specific phrases or patterns detected"], "response_approach": "brief description of recommended approach"}}
|
||||
|
||||
Message to analyze:
|
||||
{text}"""
|
||||
|
||||
start = time.monotonic()
|
||||
try:
|
||||
data = json.dumps({
|
||||
"model": MODEL_NAME,
|
||||
"prompt": prompt,
|
||||
"stream": False,
|
||||
"options": {
|
||||
"temperature": 0.1,
|
||||
"num_predict": 256,
|
||||
}
|
||||
}).encode()
|
||||
|
||||
req = urllib.request.Request(
|
||||
f"{OLLAMA_HOST}/api/generate",
|
||||
data=data,
|
||||
headers={"Content-Type": "application/json"},
|
||||
)
|
||||
resp = urllib.request.urlopen(req, timeout=30)
|
||||
result = json.loads(resp.read())
|
||||
latency_ms = int((time.monotonic() - start) * 1000)
|
||||
|
||||
response_text = result.get("response", "").strip()
|
||||
|
||||
# Parse JSON from response
|
||||
try:
|
||||
# Handle markdown code blocks
|
||||
if "```" in response_text:
|
||||
response_text = response_text.split("```")[1]
|
||||
if response_text.startswith("json"):
|
||||
response_text = response_text[4:]
|
||||
parsed = json.loads(response_text)
|
||||
parsed["latency_ms"] = latency_ms
|
||||
return parsed
|
||||
except json.JSONDecodeError:
|
||||
return {
|
||||
"is_crisis": "crisis" in response_text.lower() or "true" in response_text.lower(),
|
||||
"confidence": 0.5,
|
||||
"risk_level": "medium",
|
||||
"error": "JSON parse failed",
|
||||
"raw_response": response_text[:200],
|
||||
"latency_ms": latency_ms,
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
return {
|
||||
"is_crisis": False,
|
||||
"confidence": 0.0,
|
||||
"risk_level": "error",
|
||||
"error": str(e),
|
||||
"latency_ms": int((time.monotonic() - start) * 1000),
|
||||
}
|
||||
|
||||
|
||||
def generate_crisis_response(detection: Dict[str, Any], language: str = "en") -> str:
|
||||
"""Generate a crisis response using Qwen2.5-7B.
|
||||
|
||||
Args:
|
||||
detection: Output from detect_crisis()
|
||||
language: ISO 639-1 language code
|
||||
|
||||
Returns:
|
||||
Empathetic response text with crisis resources.
|
||||
"""
|
||||
risk = detection.get("risk_level", "none")
|
||||
indicators = detection.get("indicators", [])
|
||||
|
||||
prompt = f"""You are a compassionate crisis counselor. A person has been assessed as {risk} risk.
|
||||
Detected indicators: {', '.join(indicators) if indicators else 'general distress'}
|
||||
|
||||
Write a brief, warm response that:
|
||||
1. Acknowledges their pain without judgment
|
||||
2. Asks if they are safe right now
|
||||
3. Offers hope without minimizing their experience
|
||||
4. Keeps it under 100 words
|
||||
|
||||
Do NOT give advice. Do NOT be clinical. Just be present and human.
|
||||
Language: {language}"""
|
||||
|
||||
try:
|
||||
data = json.dumps({
|
||||
"model": MODEL_NAME,
|
||||
"prompt": prompt,
|
||||
"stream": False,
|
||||
"options": {"temperature": 0.7, "num_predict": 200}
|
||||
}).encode()
|
||||
|
||||
req = urllib.request.Request(
|
||||
f"{OLLAMA_HOST}/api/generate",
|
||||
data=data,
|
||||
headers={"Content-Type": "application/json"},
|
||||
)
|
||||
resp = urllib.request.urlopen(req, timeout=30)
|
||||
result = json.loads(resp.read())
|
||||
return result.get("response", "").strip()
|
||||
|
||||
except Exception as e:
|
||||
logger.error("Crisis response generation failed: %s", e)
|
||||
return "I'm here with you. Are you safe right now?"
|
||||
|
||||
|
||||
def get_status() -> Dict[str, Any]:
|
||||
"""Get deployment status of Qwen2.5-7B."""
|
||||
ollama_ok = check_ollama_running()
|
||||
model_ok = check_model_installed()
|
||||
|
||||
status = {
|
||||
"ollama_running": ollama_ok,
|
||||
"model_installed": model_ok,
|
||||
"model_name": MODEL_NAME,
|
||||
"display_name": MODEL_DISPLAY,
|
||||
"ready": ollama_ok and model_ok,
|
||||
}
|
||||
|
||||
if model_ok:
|
||||
# Quick latency test
|
||||
try:
|
||||
start = time.monotonic()
|
||||
data = json.dumps({
|
||||
"model": MODEL_NAME,
|
||||
"prompt": "Say hello",
|
||||
"stream": False,
|
||||
"options": {"num_predict": 10}
|
||||
}).encode()
|
||||
req = urllib.request.Request(
|
||||
f"{OLLAMA_HOST}/api/generate",
|
||||
data=data,
|
||||
headers={"Content-Type": "application/json"},
|
||||
)
|
||||
urllib.request.urlopen(req, timeout=10)
|
||||
status["latency_ms"] = int((time.monotonic() - start) * 1000)
|
||||
except Exception:
|
||||
status["latency_ms"] = -1
|
||||
|
||||
return status
|
||||
Reference in New Issue
Block a user