Long-Term Memory: Vector, MemGPT & Profiles

# Define MemoryEntry, VectorMemoryStore; implement __init__, store, retrieve
from openai import OpenAI
import numpy as np
from dataclasses import dataclass
from typing import Optional
client = OpenAI()
@dataclass
class MemoryEntry:
    text: str
    embedding: list[float]
    timestamp: float
    session_id: str
    entry_type: str # "turn", "summary", "fact"
    metadata: dict = None
    class VectorMemoryStore:
        """Semantic memory using embeddings for retrieval."""
        def __init__(self):
            self.entries: list[MemoryEntry] = []
    def store(self, text: str, session_id: str,
        entry_type: str = "turn",
        metadata: dict = None) -> None:
        """Embed and store a memory entry."""
        import time
        embedding = self._embed(text)
        entry = MemoryEntry(
            text=text,
            embedding=embedding,
            timestamp=time.time(),
            session_id=session_id,
            entry_type=entry_type,
            metadata=metadata or {}
            )
        self.entries.append(entry)
        def retrieve(self, query: str, top_k: int = 5,
            entry_type: Optional[str] = None,
            recency_weight: float = 0.1) -> list[dict]:
            """Retrieve the most relevant memories for a query."""
            query_embedding = self._embed(query)
            scored = []
            for entry in self.entries:
                if entry_type and entry.entry_type != entry_type:
                    continue
                    # Cosine similarity
                    similarity = self._cosine_sim(
                        query_embedding, entry.embedding
                        )
                    # Blend similarity with recency
                    import time
                    age_hours = (time.time() - entry.timestamp) / 3600
                    recency_score = 1.0 / (1.0 + age_hours * 0.01)
                    final_score = (
                        (1 - recency_weight) * similarity
                        + recency_weight * recency_score
                        )
                    scored.append({
                        "text": entry.text,
                        "score": final_score,
                        "similarity": similarity,
                        "entry_type": entry.entry_type,
                        "session_id": entry.session_id,
                        })
                    scored.sort(key=lambda x: x["score"], reverse=True)
                    return scored[:top_k]
                    def _embed(self, text: str) -> list[float]:
                        """Generate embedding for text."""
                        response = client.embeddings.create(
                            model="text-embedding-3-small",
                            input=text
                            )
                        return response.data[0].embedding
                    @staticmethod
                    def _cosine_sim(a: list[float], b: list[float]) -> float:
                        a, b = np.array(a), np.array(b)
                        return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))
                        # Example: Store and retrieve memories
                        store = VectorMemoryStore()
                        store.store(
                            "User prefers Python over JavaScript for backend work",
                            session_id="session_001", entry_type="fact"
                            )
                        store.store(
                            "User is building a recipe recommendation app",
                            session_id="session_001", entry_type="fact"
                            )
                        store.store(
                            "Discussed database options: PostgreSQL vs MongoDB. "
                            "User leaning toward PostgreSQL for relational data.",
                            session_id="session_002", entry_type="summary"
                            )
                        # Later, when user asks about databases again
                        results = store.retrieve("What database should I use?", top_k=2)
                        for r in results:
                            print(f"[{r['entry_type']}] {r['text'][:80]}... (score: {r['score']:.3f})")

import json
from datetime import datetime
from pathlib import Path


class UserProfileManager:
    """Manages persistent user profiles across sessions."""

    def __init__(self, storage_dir: str = "./user_profiles"):
        self.storage_dir = Path(storage_dir)
        self.storage_dir.mkdir(exist_ok=True)

    def load_profile(self, user_id: str) -> dict:
        """Load or create a user profile."""
        profile_path = self.storage_dir / f"{user_id}.json"
        if profile_path.exists():
            with open(profile_path) as f:
                return json.load(f)
        return self._create_default_profile(user_id)

    def save_profile(self, user_id: str, profile: dict) -> None:
        """Persist the user profile to disk."""
        profile["last_updated"] = datetime.now().isoformat()
        profile_path = self.storage_dir / f"{user_id}.json"
        with open(profile_path, "w") as f:
            json.dump(profile, f, indent=2)

    def update_from_conversation(self, user_id: str,
                                 conversation: list[dict]) -> dict:
        """Extract profile updates from a completed conversation."""
        profile = self.load_profile(user_id)
        # Use LLM to extract profile-worthy information
        extraction_prompt = f"""Analyze this conversation and extract any new
information about the user that should be remembered for future sessions.

Current profile:
{json.dumps(profile['preferences'], indent=2)}

Conversation:
{self._format_conversation(conversation)}

Return JSON with two fields:
- "new_preferences": dict of any new preferences discovered
- "new_facts": list of new biographical/contextual facts
- "corrections": dict of any corrections to existing profile data

Only include genuinely new or corrected information."""
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": extraction_prompt}],
            response_format={"type": "json_object"},
            temperature=0
        )
        updates = json.loads(response.choices[0].message.content)
        # Apply updates
        if updates.get("new_preferences"):
            profile["preferences"].update(updates["new_preferences"])
        if updates.get("new_facts"):
            profile["facts"].extend(updates["new_facts"])
        if updates.get("corrections"):
            profile["preferences"].update(updates["corrections"])
        # Update session count
        profile["session_count"] += 1
        profile["last_session"] = datetime.now().isoformat()
        self.save_profile(user_id, profile)
        return profile

    def get_context_string(self, user_id: str) -> str:
        """Generate a context string for inclusion in system prompts."""
        profile = self.load_profile(user_id)
        parts = [f"Returning user (session #{profile['session_count']})."]
        if profile["preferences"]:
            prefs = "; ".join(
                f"{k}: {v}" for k, v in profile["preferences"].items()
            )
            parts.append(f"Known preferences: {prefs}")
        if profile["facts"]:
            parts.append("Known facts: " + "; ".join(profile["facts"][-5:]))
        return " ".join(parts)

    def _create_default_profile(self, user_id: str) -> dict:
        return {
            "user_id": user_id,
            "created": datetime.now().isoformat(),
            "last_updated": datetime.now().isoformat(),
            "last_session": None,
            "session_count": 0,
            "preferences": {},
            "facts": [],
            "interaction_style": {}
        }

    @staticmethod
    def _format_conversation(conversation: list[dict]) -> str:
        return "\n".join(
            f"{m['role'].title()}: {m['content']}"
            for m in conversation
        )