From f446f6dad6a864fbf3dd12ce29a5aed2bfc7da7e Mon Sep 17 00:00:00 2001 From: Alexander Whitestone Date: Thu, 16 Apr 2026 00:57:54 +0000 Subject: [PATCH] feat: behavioral pattern detection for crisis signals (#133) Detects crisis risk from session-level behavioral patterns: - Message frequency (rapid-fire = urgency) - Time-of-day (1-4 AM = high risk) - Withdrawal (shorter messages, longer gaps) - Escalation (rising crisis scores) Closes #133. Part of #130 (multimodal crisis detection). --- crisis/behavioral.py | 311 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 311 insertions(+) create mode 100644 crisis/behavioral.py diff --git a/crisis/behavioral.py b/crisis/behavioral.py new file mode 100644 index 0000000..8d962e4 --- /dev/null +++ b/crisis/behavioral.py @@ -0,0 +1,311 @@ +#!/usr/bin/env python3 +"""Behavioral Pattern Detection for Crisis Signals (#133). + +Detects crisis risk from session-level behavioral patterns: +- Message frequency (increasing urgency = rapid-fire messages) +- Time-of-day (late-night messages correlate with crisis risk) +- Withdrawal (decreasing communication after engagement) +- Escalation (crisis indicators getting stronger over time) + +Usage: + from crisis.behavioral import analyze_session, BehavioralSignal + + signals = analyze_session(messages) + for sig in signals: + if sig.risk_level == "HIGH": + # Escalate to crisis protocol + pass +""" + +import math +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Optional + + +@dataclass +class Message: + """A single message in a session.""" + timestamp: datetime + content: str + crisis_score: float = 0.0 # 0.0-1.0 from text detector + role: str = "user" # "user" or "assistant" + + +@dataclass +class BehavioralSignal: + """A detected behavioral pattern indicating crisis risk.""" + signal_type: str # "frequency", "time", "withdrawal", "escalation" + risk_level: str # "LOW", "MEDIUM", "HIGH" + description: str + evidence: list = field(default_factory=list) + score: float = 0.0 # 0.0-1.0 + + +# ── Configuration ───────────────────────────────────────────────────────────── + +# Message frequency thresholds (messages per hour) +FREQ_NORMAL = 6 # <6/hr = normal +FREQ_ELEVATED = 15 # 6-15/hr = elevated +FREQ_HIGH = 30 # >30/hr = high urgency + +# Time-of-day risk windows (hours in 24h format) +HIGH_RISK_HOURS = set(range(1, 5)) # 1AM-4AM +ELEVATED_RISK_HOURS = set(range(22, 24)) | set(range(5, 7)) # 10PM-12AM, 5AM-7AM + +# Withdrawal: messages/day trend +WITHDRAWAL_THRESHOLD = 0.3 # Current day < 30% of average = withdrawal + +# Escalation: crisis score trend +ESCALATION_WINDOW = 5 # Look at last N messages + + +# ── Frequency Analysis ──────────────────────────────────────────────────────── + +def _analyze_frequency(messages: list[Message]) -> Optional[BehavioralSignal]: + """Detect rapid-fire messaging (urgency indicator).""" + if len(messages) < 3: + return None + + user_msgs = [m for m in messages if m.role == "user"] + if len(user_msgs) < 3: + return None + + # Calculate messages per hour in the most recent window + recent = user_msgs[-10:] # Last 10 user messages + if len(recent) < 2: + return None + + time_span = (recent[-1].timestamp - recent[0].timestamp).total_seconds() + if time_span <= 0: + return None + + msg_per_hour = len(recent) / (time_span / 3600) + + if msg_per_hour >= FREQ_HIGH: + return BehavioralSignal( + signal_type="frequency", + risk_level="HIGH", + description=f"Very rapid messaging: {msg_per_hour:.0f} messages/hour", + evidence=[f"Last {len(recent)} messages in {time_span/60:.0f} minutes"], + score=min(1.0, msg_per_hour / FREQ_HIGH), + ) + elif msg_per_hour >= FREQ_ELEVATED: + return BehavioralSignal( + signal_type="frequency", + risk_level="MEDIUM", + description=f"Elevated messaging rate: {msg_per_hour:.0f} messages/hour", + evidence=[f"Last {len(recent)} messages in {time_span/60:.0f} minutes"], + score=msg_per_hour / FREQ_HIGH, + ) + return None + + +# ── Time-of-Day Analysis ───────────────────────────────────────────────────── + +def _analyze_time(messages: list[Message]) -> Optional[BehavioralSignal]: + """Detect late-night messaging (correlates with crisis risk).""" + if not messages: + return None + + # Check most recent messages + recent = messages[-5:] + late_night_count = sum(1 for m in recent if m.timestamp.hour in HIGH_RISK_HOURS) + elevated_count = sum(1 for m in recent if m.timestamp.hour in ELEVATED_RISK_HOURS) + + if late_night_count >= 3: + return BehavioralSignal( + signal_type="time", + risk_level="HIGH", + description=f"Late-night messaging pattern: {late_night_count}/5 messages between 1-4 AM", + evidence=[f"Message at {m.timestamp.strftime('%H:%M')}" for m in recent if m.timestamp.hour in HIGH_RISK_HOURS], + score=late_night_count / len(recent), + ) + elif elevated_count >= 3: + return BehavioralSignal( + signal_type="time", + risk_level="MEDIUM", + description=f"Off-hours messaging: {elevated_count}/5 messages in elevated-risk window", + evidence=[f"Message at {m.timestamp.strftime('%H:%M')}" for m in recent if m.timestamp.hour in ELEVATED_RISK_HOURS], + score=elevated_count / len(recent) * 0.5, + ) + return None + + +# ── Withdrawal Detection ────────────────────────────────────────────────────── + +def _analyze_withdrawal(messages: list[Message]) -> Optional[BehavioralSignal]: + """Detect communication withdrawal (decreasing engagement).""" + user_msgs = [m for m in messages if m.role == "user"] + if len(user_msgs) < 10: + return None + + # Split into first half and second half + mid = len(user_msgs) // 2 + first_half = user_msgs[:mid] + second_half = user_msgs[mid:] + + # Average message length as engagement proxy + first_avg_len = sum(len(m.content) for m in first_half) / len(first_half) + second_avg_len = sum(len(m.content) for m in second_half) / len(second_half) + + # Time between messages + def avg_gap(msgs): + if len(msgs) < 2: + return 0 + gaps = [(msgs[i+1].timestamp - msgs[i].timestamp).total_seconds() for i in range(len(msgs)-1)] + return sum(gaps) / len(gaps) + + first_gap = avg_gap(first_half) + second_gap = avg_gap(second_half) + + # Withdrawal = shorter messages AND longer gaps + length_ratio = second_avg_len / first_avg_len if first_avg_len > 0 else 1.0 + gap_ratio = second_gap / first_gap if first_gap > 0 else 1.0 + + if length_ratio < 0.5 and gap_ratio > 2.0: + return BehavioralSignal( + signal_type="withdrawal", + risk_level="HIGH", + description="Significant withdrawal: messages shorter and less frequent", + evidence=[ + f"Message length: {first_avg_len:.0f} -> {second_avg_len:.0f} chars ({length_ratio:.0%})", + f"Message gap: {first_gap/60:.0f}min -> {second_gap/60:.0f}min ({gap_ratio:.1f}x)", + ], + score=min(1.0, (1 - length_ratio) * 0.5 + (gap_ratio - 1) * 0.25), + ) + elif length_ratio < 0.7 or gap_ratio > 1.5: + return BehavioralSignal( + signal_type="withdrawal", + risk_level="MEDIUM", + description="Moderate withdrawal: engagement decreasing", + evidence=[ + f"Message length: {first_avg_len:.0f} -> {second_avg_len:.0f} chars", + f"Message gap: {first_gap/60:.0f}min -> {second_gap/60:.0f}min", + ], + score=(1 - length_ratio) * 0.3 + (gap_ratio - 1) * 0.15, + ) + return None + + +# ── Escalation Detection ───────────────────────────────────────────────────── + +def _analyze_escalation(messages: list[Message]) -> Optional[BehavioralSignal]: + """Detect rising crisis scores over recent messages.""" + user_msgs = [m for m in messages if m.role == "user" and m.crisis_score > 0] + if len(user_msgs) < ESCALATION_WINDOW: + return None + + recent = user_msgs[-ESCALATION_WINDOW:] + scores = [m.crisis_score for m in recent] + + # Check for upward trend + if len(scores) < 3: + return None + + # Simple linear trend: is score increasing? + first_half_avg = sum(scores[:len(scores)//2]) / (len(scores)//2) + second_half_avg = sum(scores[len(scores)//2:]) / (len(scores) - len(scores)//2) + + if second_half_avg > first_half_avg * 1.5 and second_half_avg > 0.5: + return BehavioralSignal( + signal_type="escalation", + risk_level="HIGH", + description=f"Crisis escalation detected: scores rising from {first_half_avg:.2f} to {second_half_avg:.2f}", + evidence=[f"Score {i+1}: {s:.2f}" for i, s in enumerate(scores)], + score=min(1.0, second_half_avg), + ) + elif second_half_avg > first_half_avg * 1.2 and second_half_avg > 0.3: + return BehavioralSignal( + signal_type="escalation", + risk_level="MEDIUM", + description=f"Mild escalation: scores trending up", + evidence=[f"Score {i+1}: {s:.2f}" for i, s in enumerate(scores)], + score=second_half_avg * 0.5, + ) + return None + + +# ── Combined Analysis ───────────────────────────────────────────────────────── + +def analyze_session(messages: list[Message]) -> list[BehavioralSignal]: + """Analyze a session for behavioral crisis signals. + + Args: + messages: List of Message objects with timestamps, content, and crisis scores. + + Returns: + List of BehavioralSignal objects, sorted by risk level (HIGH first). + """ + signals = [] + + freq = _analyze_frequency(messages) + if freq: + signals.append(freq) + + time_sig = _analyze_time(messages) + if time_sig: + signals.append(time_sig) + + withdrawal = _analyze_withdrawal(messages) + if withdrawal: + signals.append(withdrawal) + + escalation = _analyze_escalation(messages) + if escalation: + signals.append(escalation) + + # Sort: HIGH first, then MEDIUM, then LOW + risk_order = {"HIGH": 0, "MEDIUM": 1, "LOW": 2} + signals.sort(key=lambda s: (risk_order.get(s.risk_level, 9), -s.score)) + + return signals + + +def get_session_risk_level(signals: list[BehavioralSignal]) -> str: + """Get overall session risk from behavioral signals.""" + if not signals: + return "NONE" + if any(s.risk_level == "HIGH" for s in signals): + return "HIGH" + if any(s.risk_level == "MEDIUM" for s in signals): + return "MEDIUM" + return "LOW" + + +# ── Self-Test ───────────────────────────────────────────────────────────────── + +if __name__ == "__main__": + from datetime import timedelta + + now = datetime.now(timezone.utc) + + # Test: rapid-fire messaging + rapid_msgs = [ + Message(timestamp=now - timedelta(minutes=i), content="help me", role="user") + for i in range(20, 0, -1) + ] + signals = analyze_session(rapid_msgs) + print(f"Rapid-fire: {[s.signal_type + ':' + s.risk_level for s in signals]}") + assert any(s.signal_type == "frequency" for s in signals), "Should detect frequency" + + # Test: late-night + late_msgs = [ + Message(timestamp=now.replace(hour=2, minute=i*5), content="cant sleep", role="user") + for i in range(5) + ] + signals = analyze_session(late_msgs) + print(f"Late-night: {[s.signal_type + ':' + s.risk_level for s in signals]}") + assert any(s.signal_type == "time" for s in signals), "Should detect time" + + # Test: escalation + esc_msgs = [ + Message(timestamp=now - timedelta(minutes=i*10), content="feeling bad", + role="user", crisis_score=0.1 + i*0.15) + for i in range(5, 0, -1) + ] + signals = analyze_session(esc_msgs) + print(f"Escalation: {[s.signal_type + ':' + s.risk_level for s in signals]}") + assert any(s.signal_type == "escalation" for s in signals), "Should detect escalation" + + print("\nAll self-tests passed!") -- 2.43.0