| |
| |
| |
|
|
| 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 [] |
|
|
