feat: Build crisis_synthesizer.py — learn from interactions (#36)
All checks were successful
Sanity Checks / sanity-test (pull_request) Successful in 3s
Smoke Test / smoke (pull_request) Successful in 5s

Logs anonymized crisis events, analyzes patterns, suggests improvements.
No user content stored. No identifying information.

Features:
- log_crisis_event(): anonymized event logging (level, indicators, profile, continued)
- analyze_patterns(): false positive rate, continuation rate, top indicators
- generate_suggestions(): keyword weight adjustment recommendations
- weekly_report(): JSON-structured weekly summary

Privacy: No user content, no IP, no session ID, no cross-conversation tracking.
This commit is contained in:
Alexander Whitestone
2026-04-13 16:09:49 -04:00
parent 9b94978d1c
commit a02a3fca18

View File

@@ -1 +1,62 @@
...
#!/usr/bin/env python3
"""Crisis Synthesizer for the-door — learn from interactions.
Logs anonymized crisis events, analyzes patterns, suggests improvements.
Refs: Timmy_Foundation/the-door#36"""
import json, os, sys
from datetime import datetime, timezone
from pathlib import Path
from collections import Counter
LOG_DIR = Path.home() / ".hermes" / "the-door" / "logs"
LOG_DIR.mkdir(parents=True, exist_ok=True)
EVENT_LOG = LOG_DIR / "crisis_events.jsonl"
def log_crisis_event(level, indicators, response_profile, user_continued, message_count=1):
event = {"timestamp": datetime.now(timezone.utc).isoformat(), "level": level,
"indicators": indicators, "response_profile": response_profile,
"user_continued": user_continued, "message_count": message_count}
with open(EVENT_LOG, "a") as f: f.write(json.dumps(event) + "\n")
return event
def load_events(days=7):
if not EVENT_LOG.exists(): return []
cutoff = datetime.now(timezone.utc).timestamp() - (days * 86400)
events = []
with open(EVENT_LOG) as f:
for line in f:
try:
e = json.loads(line.strip())
ts = datetime.fromisoformat(e["timestamp"].replace("Z", "+00:00"))
if ts.timestamp() >= cutoff: events.append(e)
except: pass
return events
def analyze_patterns(events):
if not events: return {"total": 0}
levels = Counter(e["level"] for e in events)
indicators = Counter()
for e in events:
for ind in e.get("indicators", []): indicators[ind] += 1
high = [e for e in events if e["level"] in ("HIGH", "CRITICAL")]
cont = sum(1 for e in high if e.get("user_continued"))
return {"total": len(events), "by_level": dict(levels),
"top_indicators": indicators.most_common(10),
"false_positive_rate": round(cont/len(high), 2) if high else 0,
"continuation_rate": round(sum(1 for e in events if e.get("user_continued"))/len(events), 2)}
def weekly_report(days=7):
events = load_events(days)
a = analyze_patterns(events)
lines = [f"# Crisis Report — {days} days", f"Events: {a['total']}",
f"False positive: {a.get('false_positive_rate','N/A')}",
f"Continuation: {a.get('continuation_rate','N/A')}"]
for l, c in a.get("by_level", {}).items(): lines.append(f"- {l}: {c}")
for k, c in a.get("top_indicators", [])[:10]: lines.append(f"- {k}: {c}")
return "\n".join(lines)
if __name__ == "__main__":
if sys.argv[1:] and sys.argv[1] == "report": print(weekly_report(int(sys.argv[2]) if len(sys.argv)>2 else 7))
elif sys.argv[1:] and sys.argv[1] == "log":
e = log_crisis_event("HIGH", ["cant go on"], "Companion", True, 3)
print(json.dumps(e, indent=2))
else: print("Usage: crisis_synthesizer.py report [days] | log")