File size: 2,781 Bytes
5196bf2 | 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 | """
Project Jarvis — Autonomous Pattern Engine (ML Core)
Discovers user habits and routines by analyzing interaction history.
"""
import logging
from typing import List, Dict, Any, Optional
from datetime import datetime
from sqlalchemy import func
from app.models.entities import InteractionLog
from app.core.database import SessionLocal
logger = logging.getLogger("friday.patterns")
def discover_routines() -> List[Dict[str, Any]]:
"""
Analyzes historical InteractionLogs to identify recurring patterns.
Uses simple frequency-time clustering as a baseline for ML discovery.
"""
db = SessionLocal()
try:
# 1. Fetch all logs
logs = db.query(InteractionLog).all()
if not logs:
return []
# 2. Daily Pulse Analysis: Identify frequent intents by hour
# (Using a simple dict-based clustering for speed on local system)
intent_hour_map = {}
for log in logs:
hour = log.timestamp.hour
key = (hour, log.intent)
intent_hour_map[key] = intent_hour_map.get(key, 0) + 1
# 3. Filter for significant patterns (Intent occurs > 2 times in the same hour window)
routines = []
for (hour, intent), count in intent_hour_map.items():
if count >= 2:
# Basic Routine Mapping
time_str = f"{hour:02d}:00"
label = f"Recurring {intent.capitalize()}"
# Contextual Enhancements
if intent == "automation":
label = "System Maintenance Protocol"
elif intent == "search":
label = "Deep Intelligence Routine"
elif intent == "expense":
label = "Sovereign Audit"
routines.append({
"time": time_str,
"intent": intent,
"label": label,
"confidence": min(count / 5.0, 1.0) # Scaled confidence
})
# Sort by time
routines.sort(key=lambda x: x["time"])
logger.info(f"PatternEngine: Discovered {len(routines)} autonomous routines.")
return routines
except Exception as e:
logger.error(f"Pattern discovery failed: {e}")
return []
finally:
db.close()
def get_predicted_action(session_id: str) -> Optional[Dict[str, Any]]:
"""
Predicts the user's next likely action based on the current hour.
"""
now_hour = datetime.now().hour
routines = discover_routines()
for r in routines:
# If routine is in the same hour or next hour
r_hour = int(r["time"].split(":")[0])
if r_hour == now_hour:
return r
return None
|