File size: 3,592 Bytes
6645e8f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 | # JARVIS Thinking Engine — HuggingFace Edition
# Runs periodically in the background to generate anticipations.
# Paths and intervals are env-var driven.
import threading
import time
import json
import os
from datetime import datetime
from config import THINKING_ENABLED, THINKING_INTERVAL, MEMORY_DIR
import memory
_thinking_thread = None
_thinking_active = False
_anticipations: list[str] = []
def _get_user_patterns() -> dict:
"""Summarise what the user talks about most."""
history = memory.get_recent_history(100)
keywords: dict[str, int] = {}
for turn in history:
for word in turn["message"].lower().split():
if len(word) > 4:
keywords[word] = keywords.get(word, 0) + 1
top = sorted(keywords.items(), key=lambda x: x[1], reverse=True)[:10]
return {
"top_keywords": top,
"total_interactions": len(history),
}
def _generate_anticipations() -> list[str]:
"""Ask the brain what the user might need next."""
try:
import brain as b
patterns = _get_user_patterns()
facts = memory.get_all_facts()
prompt = (
f"Given user interests: {', '.join(k for k, _ in patterns['top_keywords'])} "
f"and facts: {json.dumps(facts)}, "
"predict 3 brief, helpful things JARVIS should proactively offer. "
"Reply with one bullet per line."
)
response = b.ask(prompt)
lines = [l.strip("•- ").strip() for l in response.splitlines() if l.strip()]
return lines[:3]
except Exception as e:
return [f"[Thinking error: {e}]"]
def _save_thinking_state(anticipations: list[str]):
"""Persist thinking results to disk."""
os.makedirs(MEMORY_DIR, exist_ok=True)
path = os.path.join(MEMORY_DIR, "thinking.json")
with open(path, "w", encoding="utf-8") as f:
json.dump({
"timestamp": datetime.now().isoformat(),
"anticipations": anticipations,
"patterns": _get_user_patterns(),
}, f, indent=2)
def _thinking_loop():
global _thinking_active, _anticipations
_thinking_active = True
print(f"[Thinking] Started. Cycle: {THINKING_INTERVAL}s")
while _thinking_active:
try:
time.sleep(THINKING_INTERVAL)
if not _thinking_active:
break
print("[Thinking] Generating anticipations...")
_anticipations = _generate_anticipations()
_save_thinking_state(_anticipations)
print(f"[Thinking] {len(_anticipations)} anticipation(s) ready.")
except Exception as e:
print(f"[Thinking] Error: {e}")
def start_thinking_engine():
global _thinking_thread, _thinking_active
if not THINKING_ENABLED:
return
if _thinking_thread and _thinking_thread.is_alive():
return
_thinking_thread = threading.Thread(target=_thinking_loop, daemon=True)
_thinking_thread.start()
def stop_thinking_engine():
global _thinking_active
_thinking_active = False
if _thinking_thread:
_thinking_thread.join(timeout=5)
def get_anticipations() -> list[str]:
"""Return most recent anticipations (from memory on disk or in-process)."""
if _anticipations:
return _anticipations
path = os.path.join(MEMORY_DIR, "thinking.json")
if os.path.exists(path):
try:
with open(path, "r", encoding="utf-8") as f:
data = json.load(f)
return data.get("anticipations", [])
except Exception:
pass
return []
|