""" Memory Engine — Loads and merges the user's personality profile from the memory/ folder. Reads: - memory/personality.md → personality traits & decision process - memory/preferences.json → numeric preference scores + category overrides - memory/semantic_memory.jsonl → learned conclusions about the user - memory/episodic_log.jsonl → past shopping actions - memory/projects.md → current shopping goals Exposes a single UserProfile dataclass with all merged data. """ import json from dataclasses import dataclass, field from pathlib import Path from typing import Any, Dict, List, Optional @dataclass class UserProfile: """Complete user personality profile loaded from memory/ folder.""" # Core numeric preferences (0.0 – 1.0) price_sensitivity: float = 0.5 quality_preference: float = 0.5 risk_aversion: float = 0.5 research_depth: float = 0.5 brand_trust: float = 0.5 exploration_vs_repeat: float = 0.5 review_dependence: float = 0.5 return_preference: float = 0.5 decision_speed: float = 0.5 discount_sensitivity: float = 0.5 # Category-specific overrides (e.g. {"electronics": {"risk_aversion": 0.9}}) category_preferences: Dict[str, Dict[str, float]] = field(default_factory=dict) # Textual personality summary (from personality.md) personality_summary: str = "" # Decision process steps (parsed from personality.md) decision_process: List[str] = field(default_factory=list) # Learned conclusions (from semantic_memory.jsonl) semantic_conclusions: List[Dict[str, Any]] = field(default_factory=list) # Past shopping events (from episodic_log.jsonl) episodic_history: List[Dict[str, Any]] = field(default_factory=list) # Current shopping goals (from projects.md) shopping_goals: str = "" def get_prefs_for_category(self, category: str) -> Dict[str, float]: """ Return merged preferences for a specific category. Category-specific overrides take precedence over base prefs. """ base = { "price_sensitivity": self.price_sensitivity, "quality_preference": self.quality_preference, "risk_aversion": self.risk_aversion, "research_depth": self.research_depth, "brand_trust": self.brand_trust, "exploration_vs_repeat": self.exploration_vs_repeat, "review_dependence": self.review_dependence, "return_preference": self.return_preference, "decision_speed": self.decision_speed, "discount_sensitivity": self.discount_sensitivity, } # Apply category-specific overrides cat_lower = category.lower() for cat_key, overrides in self.category_preferences.items(): if cat_key.lower() in cat_lower or cat_lower in cat_key.lower(): base.update(overrides) break return base def to_prompt_text(self, category: str = "") -> str: """Render the profile as text suitable for an LLM prompt.""" prefs = self.get_prefs_for_category(category) if category else {} lines = [ "## User Personality Profile", "", self.personality_summary[:600] if self.personality_summary else "Research-heavy, value-conscious shopper.", "", "## Numeric Preferences" + (f" (category: {category})" if category else ""), ] for k, v in (prefs or self.__dict__).items(): if isinstance(v, float) and 0 <= v <= 1: lines.append(f" - {k}: {v:.2f}") if self.semantic_conclusions: lines.append("") lines.append("## Learned Conclusions") for c in self.semantic_conclusions[:6]: conf = c.get("confidence", 0) lines.append(f" - {c.get('conclusion', '')} (confidence: {conf:.0%})") if self.shopping_goals: lines.append("") lines.append("## Current Shopping Goals") lines.append(self.shopping_goals[:400]) return "\n".join(lines) def load_profile(memory_dir: Optional[str] = None) -> UserProfile: """ Load the full user profile from the memory/ directory. Falls back to sensible defaults if files are missing. """ if memory_dir is None: memory_dir = str(Path(__file__).parent / "memory") mem = Path(memory_dir) profile = UserProfile() # ---- preferences.json ---- prefs_path = mem / "preferences.json" if prefs_path.exists(): try: data = json.loads(prefs_path.read_text(encoding="utf-8")) for key in [ "price_sensitivity", "quality_preference", "risk_aversion", "research_depth", "brand_trust", "exploration_vs_repeat", "review_dependence", "return_preference", "decision_speed", "discount_sensitivity", ]: if key in data: setattr(profile, key, float(data[key])) if "category_preferences" in data: profile.category_preferences = data["category_preferences"] except Exception as e: print(f"[memory_engine] Error loading preferences.json: {e}") # ---- personality.md ---- personality_path = mem / "personality.md" if personality_path.exists(): try: text = personality_path.read_text(encoding="utf-8") profile.personality_summary = text[:800] # Extract decision process steps steps = [] in_process = False for line in text.split("\n"): stripped = line.strip().lstrip("#").strip() if "decision process" in stripped.lower(): in_process = True continue if in_process and stripped.startswith(("1.", "2.", "3.", "4.", "5.", "6.")): steps.append(stripped) elif in_process and stripped.startswith("#"): in_process = False profile.decision_process = steps except Exception as e: print(f"[memory_engine] Error loading personality.md: {e}") # ---- semantic_memory.jsonl ---- semantic_path = mem / "semantic_memory.jsonl" if semantic_path.exists(): conclusions = [] for line in semantic_path.read_text(encoding="utf-8").strip().split("\n"): if line.strip(): try: conclusions.append(json.loads(line.strip())) except Exception: pass # skip malformed lines profile.semantic_conclusions = conclusions # ---- episodic_log.jsonl ---- episodic_path = mem / "episodic_log.jsonl" if episodic_path.exists(): events = [] for line in episodic_path.read_text(encoding="utf-8").strip().split("\n"): if line.strip(): try: events.append(json.loads(line.strip())) except Exception: pass # skip malformed lines profile.episodic_history = events # ---- projects.md ---- projects_path = mem / "projects.md" if projects_path.exists(): try: profile.shopping_goals = projects_path.read_text(encoding="utf-8") except Exception as e: print(f"[memory_engine] Error loading projects.md: {e}") return profile