Upload mythos/memory.py

mythos/memory.py

@@ -0,0 +1,134 @@
+"""Memory system for Mythos agents — MIRIX-inspired 4-component taxonomy."""
+
+from __future__ import annotations
+
+import uuid
+from datetime import datetime, timezone
+from enum import Enum
+from typing import Any, Optional
+
+from pydantic import BaseModel, Field
+
+
+class MemoryType(str, Enum):
+    """Types of memory in the MIRIX taxonomy."""
+
+    CORE = "core"           # Identity, persona, values
+    EPISODIC = "episodic"   # Timestamped events, experiences
+    SEMANTIC = "semantic"   # Facts, knowledge, concepts
+    PROCEDURAL = "procedural"  # Workflows, skills, how-to
+
+
+class MemoryEntry(BaseModel):
+    """A single memory entry."""
+
+    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
+    type: MemoryType
+    content: str
+    agent: str                           # Which agent owns this memory
+    timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
+    importance: float = 0.5              # 0.0 to 1.0
+    tags: list[str] = Field(default_factory=list)
+    source: Optional[str] = None         # Where did this memory come from
+    access_count: int = 0                # For LRU-style eviction
+    last_accessed: Optional[datetime] = None
+    embedding: Optional[list[float]] = None  # For vector search
+
+    def touch(self) -> None:
+        """Mark memory as accessed."""
+        self.access_count += 1
+        self.last_accessed = datetime.now(timezone.utc)
+
+
+class Memory(BaseModel):
+    """Agent memory store with Active Retrieval."""
+
+    agent_name: str
+    entries: list[MemoryEntry] = Field(default_factory=list)
+    max_entries: int = 1000
+    auto_retrieve: bool = True
+
+    def add(
+        self,
+        content: str,
+        mem_type: MemoryType,
+        importance: float = 0.5,
+        tags: Optional[list[str]] = None,
+        source: Optional[str] = None,
+    ) -> MemoryEntry:
+        """Add a new memory entry."""
+        entry = MemoryEntry(
+            type=mem_type,
+            content=content,
+            agent=self.agent_name,
+            importance=importance,
+            tags=tags or [],
+            source=source,
+        )
+        self.entries.append(entry)
+        if len(self.entries) > self.max_entries:
+            self._evict_least_important()
+        return entry
+
+    def retrieve(
+        self,
+        query: str,
+        mem_type: Optional[MemoryType] = None,
+        limit: int = 5,
+    ) -> list[MemoryEntry]:
+        """Retrieve relevant memories. Simple keyword match; override for vector search."""
+        query_lower = query.lower()
+        candidates = self.entries
+        if mem_type:
+            candidates = [e for e in candidates if e.type == mem_type]
+
+        scored = []
+        for entry in candidates:
+            score = 0.0
+            if query_lower in entry.content.lower():
+                score += 1.0
+            score += entry.importance * 0.5
+            score += min(entry.access_count * 0.01, 0.3)
+            scored.append((score, entry))
+
+        scored.sort(key=lambda x: x[0], reverse=True)
+        results = [entry for _, entry in scored[:limit]]
+        for entry in results:
+            entry.touch()
+        return results
+
+    def active_retrieve(self, context: str) -> list[MemoryEntry]:
+        """Infer topics from context and retrieve relevant memories automatically."""
+        # Simple topic extraction: split into words, filter short ones
+        words = [w.lower() for w in context.split() if len(w) > 4]
+        topics = list(set(words))[:5]
+
+        all_results: list[MemoryEntry] = []
+        for topic in topics:
+            all_results.extend(self.retrieve(topic, limit=3))
+
+        # Deduplicate by id
+        seen = set()
+        deduped = []
+        for entry in all_results:
+            if entry.id not in seen:
+                seen.add(entry.id)
+                deduped.append(entry)
+
+        deduped.sort(key=lambda e: e.importance, reverse=True)
+        return deduped[:10]
+
+    def _evict_least_important(self) -> None:
+        """Remove least important / least accessed memory."""
+        self.entries.sort(
+            key=lambda e: (e.importance, e.access_count, e.timestamp),
+        )
+        self.entries.pop(0)
+
+    def to_context(self, entries: Optional[list[MemoryEntry]] = None) -> str:
+        """Format memories as context string for LLM prompt."""
+        entries = entries or self.entries[-20:]
+        lines = []
+        for entry in entries:
+            lines.append(f"[{entry.type.value.upper()}] {entry.content}")
+        return "\n".join(lines)