File size: 7,490 Bytes
d6e8cd0 | 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 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 | """
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
|