"""Data models for the Memory System.""" from __future__ import annotations import uuid from dataclasses import dataclass, field, asdict from datetime import datetime from enum import Enum from typing import Any, Dict, List, Optional class MemoryTier(str, Enum): """Which memory layer an entry belongs to.""" SESSION = "session" # short-term / conversation context EPISODIC = "episodic" # mid-term / past tasks & events SEMANTIC = "semantic" # long-term / vector-backed knowledge @dataclass class MemoryEntry: """A single memory record stored across tiers.""" id: str = field(default_factory=lambda: uuid.uuid4().hex[:12]) content: str = "" title: str = "" tier: MemoryTier = MemoryTier.SESSION tags: List[str] = field(default_factory=list) metadata: Dict[str, Any] = field(default_factory=dict) importance: float = 0.5 # 0.0 – 1.0 access_count: int = 0 created_at: str = field(default_factory=lambda: datetime.utcnow().isoformat()) updated_at: str = field(default_factory=lambda: datetime.utcnow().isoformat()) session_id: Optional[str] = None # groups session memories source: str = "" # origin of the memory # ── helpers ────────────────────────────────────────────── def to_dict(self) -> Dict[str, Any]: d = asdict(self) d["tier"] = self.tier.value return d @classmethod def from_dict(cls, data: Dict[str, Any]) -> "MemoryEntry": data = dict(data) # shallow copy if "tier" in data and isinstance(data["tier"], str): data["tier"] = MemoryTier(data["tier"]) return cls(**{k: v for k, v in data.items() if k in cls.__dataclass_fields__}) def to_markdown(self) -> str: """Render as a Markdown document with YAML front-matter.""" lines = [ "---", f"id: {self.id}", f"title: \"{self.title}\"", f"tier: {self.tier.value}", f"tags: [{', '.join(self.tags)}]", f"importance: {self.importance}", f"access_count: {self.access_count}", f"created_at: {self.created_at}", f"updated_at: {self.updated_at}", ] if self.session_id: lines.append(f"session_id: {self.session_id}") if self.source: lines.append(f"source: \"{self.source}\"") if self.metadata: import json lines.append(f"metadata: {json.dumps(self.metadata)}") lines.append("---") lines.append("") lines.append(self.content) return "\n".join(lines) @classmethod def from_markdown(cls, text: str) -> "MemoryEntry": """Parse a Markdown document with YAML front-matter.""" import re, json as _json fm_match = re.match(r"^---\n(.*?)\n---\n?(.*)", text, re.DOTALL) if not fm_match: return cls(content=text) front, body = fm_match.group(1), fm_match.group(2).strip() data: Dict[str, Any] = {"content": body} for line in front.splitlines(): line = line.strip() if not line or ":" not in line: continue key, _, val = line.partition(":") key = key.strip() val = val.strip().strip('"') if key == "tags": # parse [tag1, tag2] inner = val.strip("[]") data["tags"] = [t.strip() for t in inner.split(",") if t.strip()] elif key == "importance": data["importance"] = float(val) elif key == "access_count": data["access_count"] = int(val) elif key == "metadata": try: data["metadata"] = _json.loads(val) except _json.JSONDecodeError: data["metadata"] = {} else: data[key] = val return cls.from_dict(data) @dataclass class SearchResult: """Wrapper returned by semantic search.""" entry: MemoryEntry score: float = 0.0 # similarity / relevance distance: float = 0.0 # raw distance from vector DB