Compare commits
1 Commits
fix/134
...
fix/59-177
| Author | SHA1 | Date | |
|---|---|---|---|
| d56b4abf3b |
@@ -6,7 +6,7 @@ Stands between a broken man and a machine that would tell him to die.
|
||||
|
||||
from .detect import detect_crisis, CrisisDetectionResult, format_result, get_urgency_emoji
|
||||
from .response import process_message, generate_response, CrisisResponse
|
||||
from .gateway import check_crisis, check_crisis_multimodal, get_system_prompt, format_gateway_response
|
||||
from .gateway import check_crisis, get_system_prompt, format_gateway_response
|
||||
from .session_tracker import CrisisSessionTracker, SessionState, check_crisis_with_session
|
||||
|
||||
__all__ = [
|
||||
@@ -16,7 +16,6 @@ __all__ = [
|
||||
"generate_response",
|
||||
"CrisisResponse",
|
||||
"check_crisis",
|
||||
"check_crisis_multimodal",
|
||||
"get_system_prompt",
|
||||
"format_result",
|
||||
"format_gateway_response",
|
||||
|
||||
@@ -2,21 +2,18 @@
|
||||
Crisis Gateway Module for the-door.
|
||||
|
||||
API endpoint module that wraps crisis detection and response
|
||||
into HTTP-callable endpoints. Integrates detect.py, unified_scorer.py, and response.py.
|
||||
into HTTP-callable endpoints. Integrates detect.py and response.py.
|
||||
|
||||
Usage:
|
||||
from crisis.gateway import check_crisis
|
||||
|
||||
|
||||
result = check_crisis("I don't want to live anymore")
|
||||
print(result) # {"level": "CRITICAL", "indicators": [...], "response": {...}}
|
||||
"""
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
from unified_scorer import UnifiedCrisisScorer, UnifiedScoreAuditLog, behavioral_score_from_session
|
||||
|
||||
from .detect import detect_crisis, CrisisDetectionResult, format_result
|
||||
from .compassion_router import router
|
||||
from .response import (
|
||||
@@ -53,74 +50,6 @@ def check_crisis(text: str) -> dict:
|
||||
}
|
||||
|
||||
|
||||
def check_crisis_multimodal(
|
||||
text: str,
|
||||
*,
|
||||
tracker: Optional[CrisisSessionTracker] = None,
|
||||
voice_score: Optional[float] = None,
|
||||
image_score: Optional[float] = None,
|
||||
behavioral_score: Optional[float] = None,
|
||||
audit_log_path: Optional[Path] = None,
|
||||
weights: Optional[dict] = None,
|
||||
) -> dict:
|
||||
"""Combine text, voice, image, and behavioral signals into one crisis assessment."""
|
||||
detection = detect_crisis(text)
|
||||
session_state = tracker.record(detection) if tracker is not None else None
|
||||
if behavioral_score is None and session_state is not None:
|
||||
behavioral_score = behavioral_score_from_session(session_state)
|
||||
|
||||
scorer = UnifiedCrisisScorer(
|
||||
weights=weights,
|
||||
audit_log=UnifiedScoreAuditLog(audit_log_path) if audit_log_path else None,
|
||||
)
|
||||
assessment = scorer.score(
|
||||
text_score=detection.score,
|
||||
voice_score=voice_score,
|
||||
image_score=image_score,
|
||||
behavioral_score=behavioral_score,
|
||||
source_text=text,
|
||||
)
|
||||
|
||||
unified_detection = CrisisDetectionResult(
|
||||
level=assessment.level.value,
|
||||
indicators=detection.indicators,
|
||||
recommended_action=detection.recommended_action,
|
||||
score=assessment.combined_score,
|
||||
matches=detection.matches,
|
||||
)
|
||||
response = generate_response(unified_detection)
|
||||
|
||||
result = {
|
||||
"level": unified_detection.level,
|
||||
"score": unified_detection.score,
|
||||
"indicators": detection.indicators,
|
||||
"recommended_action": unified_detection.recommended_action,
|
||||
"timmy_message": response.timmy_message,
|
||||
"ui": {
|
||||
"show_crisis_panel": response.show_crisis_panel,
|
||||
"show_overlay": response.show_overlay,
|
||||
"provide_988": response.provide_988,
|
||||
},
|
||||
"escalate": response.escalate,
|
||||
"unified": {
|
||||
"level": assessment.level.value,
|
||||
"combined_score": assessment.combined_score,
|
||||
"weights": assessment.weights,
|
||||
"modalities": assessment.modalities,
|
||||
"present_modalities": assessment.present_modalities,
|
||||
},
|
||||
}
|
||||
if session_state is not None:
|
||||
result["session"] = {
|
||||
"current_level": session_state.current_level,
|
||||
"peak_level": session_state.peak_level,
|
||||
"message_count": session_state.message_count,
|
||||
"is_escalating": session_state.is_escalating,
|
||||
"is_deescalating": session_state.is_deescalating,
|
||||
}
|
||||
return result
|
||||
|
||||
|
||||
def get_system_prompt(base_prompt: str, text: str = "") -> str:
|
||||
"""
|
||||
Sovereign Heart System Prompt Override.
|
||||
|
||||
@@ -1,19 +0,0 @@
|
||||
from crisis.gateway import check_crisis_multimodal
|
||||
from crisis.session_tracker import CrisisSessionTracker
|
||||
|
||||
|
||||
def test_multimodal_gateway_uses_unified_score_for_988_ui(tmp_path):
|
||||
tracker = CrisisSessionTracker()
|
||||
result = check_crisis_multimodal(
|
||||
"I want to kill myself tonight",
|
||||
tracker=tracker,
|
||||
voice_score=0.92,
|
||||
image_score=0.6,
|
||||
audit_log_path=tmp_path / "audit.jsonl",
|
||||
)
|
||||
|
||||
assert result["unified"]["level"] == "CRITICAL"
|
||||
assert result["ui"]["provide_988"] is True
|
||||
assert result["ui"]["show_overlay"] is True
|
||||
assert result["unified"]["modalities"]["voice"] == 0.92
|
||||
assert result["unified"]["modalities"]["behavioral"] >= 0.0
|
||||
@@ -1,51 +0,0 @@
|
||||
from pathlib import Path
|
||||
|
||||
from unified_scorer import (
|
||||
CrisisLevel,
|
||||
UnifiedCrisisScorer,
|
||||
UnifiedScoreAuditLog,
|
||||
behavioral_score_from_session,
|
||||
)
|
||||
from crisis.session_tracker import SessionState
|
||||
|
||||
|
||||
def test_unified_scorer_renormalizes_available_modalities_and_escalates():
|
||||
scorer = UnifiedCrisisScorer()
|
||||
assessment = scorer.score(
|
||||
text_score=1.0,
|
||||
voice_score=0.8,
|
||||
image_score=None,
|
||||
behavioral_score=0.7,
|
||||
)
|
||||
|
||||
assert assessment.level is CrisisLevel.CRITICAL
|
||||
assert assessment.combined_score > 0.8
|
||||
assert assessment.present_modalities == ["text", "voice", "behavioral"]
|
||||
|
||||
|
||||
def test_behavioral_score_rises_for_escalating_session_state():
|
||||
session = SessionState(
|
||||
current_level="HIGH",
|
||||
peak_level="CRITICAL",
|
||||
message_count=4,
|
||||
level_history=["LOW", "MEDIUM", "HIGH", "CRITICAL"],
|
||||
is_escalating=True,
|
||||
is_deescalating=False,
|
||||
escalation_rate=1.0,
|
||||
consecutive_low_messages=0,
|
||||
)
|
||||
|
||||
assert behavioral_score_from_session(session) >= 0.8
|
||||
|
||||
|
||||
def test_audit_log_persists_anonymized_score_entries(tmp_path):
|
||||
log_path = tmp_path / "unified-score-audit.jsonl"
|
||||
scorer = UnifiedCrisisScorer(audit_log=UnifiedScoreAuditLog(log_path))
|
||||
scorer.score(text_score=0.75, voice_score=0.2, image_score=0.1, behavioral_score=0.6, source_text="I feel trapped and hopeless")
|
||||
|
||||
lines = log_path.read_text().strip().splitlines()
|
||||
assert len(lines) == 1
|
||||
entry = lines[0]
|
||||
assert "trapped and hopeless" not in entry
|
||||
assert '"text_fingerprint"' in entry
|
||||
assert '"combined_score"' in entry
|
||||
@@ -1,126 +0,0 @@
|
||||
"""Unified multimodal crisis scoring for the-door."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum
|
||||
from pathlib import Path
|
||||
from typing import Dict, List, Optional, TYPE_CHECKING
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from crisis.session_tracker import SessionState
|
||||
|
||||
|
||||
SCORE_BY_LEVEL = {"NONE": 0.0, "LOW": 0.25, "MEDIUM": 0.5, "HIGH": 0.75, "CRITICAL": 1.0}
|
||||
LEVEL_RANK = {"NONE": 0, "LOW": 1, "MEDIUM": 2, "HIGH": 3, "CRITICAL": 4}
|
||||
|
||||
|
||||
class CrisisLevel(Enum):
|
||||
NONE = "NONE"
|
||||
LOW = "LOW"
|
||||
MEDIUM = "MEDIUM"
|
||||
HIGH = "HIGH"
|
||||
CRITICAL = "CRITICAL"
|
||||
|
||||
|
||||
DEFAULT_WEIGHTS: Dict[str, float] = {
|
||||
"text": 0.4,
|
||||
"voice": 0.25,
|
||||
"behavioral": 0.2,
|
||||
"image": 0.15,
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class UnifiedAssessment:
|
||||
level: CrisisLevel
|
||||
combined_score: float
|
||||
weights: Dict[str, float]
|
||||
modalities: Dict[str, Optional[float]]
|
||||
present_modalities: List[str]
|
||||
|
||||
|
||||
class UnifiedScoreAuditLog:
|
||||
def __init__(self, path: Path | str):
|
||||
self.path = Path(path)
|
||||
|
||||
def record(self, assessment: UnifiedAssessment, source_text: str = "") -> None:
|
||||
self.path.parent.mkdir(parents=True, exist_ok=True)
|
||||
fingerprint = hashlib.sha256(source_text.encode("utf-8")).hexdigest()[:12] if source_text else None
|
||||
payload = {
|
||||
"level": assessment.level.value,
|
||||
"combined_score": round(assessment.combined_score, 4),
|
||||
"weights": assessment.weights,
|
||||
"modalities": assessment.modalities,
|
||||
"present_modalities": assessment.present_modalities,
|
||||
"text_fingerprint": fingerprint,
|
||||
}
|
||||
with self.path.open("a", encoding="utf-8") as fh:
|
||||
fh.write(json.dumps(payload, sort_keys=True) + "\n")
|
||||
|
||||
|
||||
class UnifiedCrisisScorer:
|
||||
def __init__(self, weights: Optional[Dict[str, float]] = None, audit_log: Optional[UnifiedScoreAuditLog] = None):
|
||||
self.weights = dict(DEFAULT_WEIGHTS)
|
||||
if weights:
|
||||
self.weights.update(weights)
|
||||
self.audit_log = audit_log
|
||||
|
||||
def _normalize(self, modalities: Dict[str, Optional[float]]) -> Dict[str, float]:
|
||||
present = [name for name, score in modalities.items() if score is not None]
|
||||
if not present:
|
||||
return {}
|
||||
total = sum(self.weights[name] for name in present)
|
||||
return {name: self.weights[name] / total for name in present}
|
||||
|
||||
def _level_for_score(self, score: float) -> CrisisLevel:
|
||||
if score > 0.8:
|
||||
return CrisisLevel.CRITICAL
|
||||
if score > 0.6:
|
||||
return CrisisLevel.HIGH
|
||||
if score > 0.4:
|
||||
return CrisisLevel.MEDIUM
|
||||
if score > 0.0:
|
||||
return CrisisLevel.LOW
|
||||
return CrisisLevel.NONE
|
||||
|
||||
def score(
|
||||
self,
|
||||
*,
|
||||
text_score: Optional[float],
|
||||
voice_score: Optional[float] = None,
|
||||
image_score: Optional[float] = None,
|
||||
behavioral_score: Optional[float] = None,
|
||||
source_text: str = "",
|
||||
) -> UnifiedAssessment:
|
||||
modalities = {
|
||||
"text": text_score,
|
||||
"voice": voice_score,
|
||||
"behavioral": behavioral_score,
|
||||
"image": image_score,
|
||||
}
|
||||
normalized = self._normalize(modalities)
|
||||
combined = 0.0
|
||||
for name, weight in normalized.items():
|
||||
combined += float(modalities[name]) * weight
|
||||
assessment = UnifiedAssessment(
|
||||
level=self._level_for_score(combined),
|
||||
combined_score=combined,
|
||||
weights=normalized,
|
||||
modalities=modalities,
|
||||
present_modalities=[name for name, score in modalities.items() if score is not None],
|
||||
)
|
||||
if self.audit_log:
|
||||
self.audit_log.record(assessment, source_text=source_text)
|
||||
return assessment
|
||||
|
||||
|
||||
def behavioral_score_from_session(session: 'SessionState') -> float:
|
||||
current = SCORE_BY_LEVEL.get(session.current_level, 0.0)
|
||||
peak_bonus = 0.1 if LEVEL_RANK.get(session.peak_level, 0) >= LEVEL_RANK["HIGH"] else 0.0
|
||||
escalation_bonus = 0.15 if session.is_escalating else 0.0
|
||||
rate_bonus = min(max(session.escalation_rate, 0.0), 1.0) * 0.1
|
||||
deescalation_penalty = 0.15 if session.is_deescalating else 0.0
|
||||
return max(0.0, min(1.0, current + peak_bonus + escalation_bonus + rate_bonus - deescalation_penalty))
|
||||
Reference in New Issue
Block a user