LLM Memory Systems: From MemGPT to Long-Term Agent Memory

┌─────────────────────────────────────────────────────────────────────────┐
│                    MEMORY TYPES: HUMANS VS LLMs                          │
├─────────────────────────────────────────────────────────────────────────┤
│                                                                          │
│  Cognitive Term      Human Brain              LLM Implementation         │
│  ─────────────────────────────────────────────────────────────────────  │
│                                                                          │
│  Short-term Memory   Immediate awareness      Context window             │
│                      ~7 items, seconds        ~128K tokens, one call     │
│                                                                          │
│  Working Memory      Active manipulation      Scratchpad / reasoning     │
│                      Problem-solving space    Chain-of-thought state     │
│                                                                          │
│  Long-term Memory    Permanent storage        Vector DB / external       │
│                      Lifetime capacity        Unlimited capacity         │
│                                                                          │
│  Episodic Memory     Personal experiences     Conversation history       │
│                      "What happened"          Session logs               │
│                                                                          │
│  Semantic Memory     Facts and knowledge      Knowledge base / RAG       │
│                      "What I know"            Retrieved documents        │
│                                                                          │
│  Procedural Memory   Skills and habits        Fine-tuned weights         │
│                      "How to do"              Learned behaviors          │
│                                                                          │
└─────────────────────────────────────────────────────────────────────────┘

class ShortTermMemory:
    """
    Short-term memory in LLMs is simply the context window.
    This class manages what goes into that window.
    """

    def __init__(self, max_tokens: int = 8000):
        self.max_tokens = max_tokens
        self.messages: list[dict] = []

    def add(self, role: str, content: str) -> None:
        """Add a message to short-term memory."""
        self.messages.append({"role": role, "content": content})

        # Short-term memory has hard limits
        # When exceeded, oldest memories must go
        while self._count_tokens() > self.max_tokens:
            self._evict_oldest()

    def _evict_oldest(self) -> None:
        """
        Remove oldest non-system message.
        This is the simplest eviction strategy.
        More sophisticated approaches might:
        - Summarize before evicting
        - Move to long-term memory
        - Prioritize by importance
        """
        for i, msg in enumerate(self.messages):
            if msg["role"] != "system":
                self.messages.pop(i)
                break

    def get_context(self) -> list[dict]:
        """Return current short-term memory contents."""
        return self.messages.copy()

    def _count_tokens(self) -> int:
        # Simplified token counting
        return sum(len(m["content"]) // 4 for m in self.messages)

┌─────────────────────────────────────────────────────────────────────────┐
│                     CONTEXT WINDOW BUDGET                                │
├─────────────────────────────────────────────────────────────────────────┤
│                                                                          │
│  Fixed costs:                                                            │
│  ├── System prompt:           ~1,000 tokens                             │
│  ├── Tool definitions:        ~2,000 tokens                             │
│  └── User's current message:  ~200 tokens                               │
│                                                                          │
│  Variable costs (competing for remaining space):                        │
│  ├── Working memory:          500-3,000 tokens                          │
│  ├── Retrieved context:       1,000-5,000 tokens                        │
│  └── Conversation history:    1,000-10,000 tokens                       │
│                                                                          │
│  Total budget: 8,000-128,000 tokens (depending on model)                │
│                                                                          │
└─────────────────────────────────────────────────────────────────────────┘

class WorkingMemory:
    """
    Working memory is the agent's scratchpad for active reasoning.
    It tracks current goals, observations, and intermediate thoughts.
    """

    def __init__(self):
        # Current task state
        self.current_goal: str = ""
        self.sub_goals: list[str] = []
        self.completed_steps: list[str] = []

        # Observations from the environment
        self.observations: list[dict] = []

        # Scratchpad for reasoning
        self.notes: dict[str, str] = {}

        # Hypotheses being considered
        self.hypotheses: list[dict] = []

    def set_goal(self, goal: str) -> None:
        """Set the current high-level goal."""
        self.current_goal = goal
        self.sub_goals = []
        self.completed_steps = []

    def decompose_goal(self, sub_goals: list[str]) -> None:
        """Break down the goal into sub-goals."""
        self.sub_goals = sub_goals

    def complete_step(self, step: str, result: str) -> None:
        """Mark a step as completed with its result."""
        self.completed_steps.append(f"{step}: {result}")
        if step in self.sub_goals:
            self.sub_goals.remove(step)

    def add_observation(self, source: str, content: str) -> None:
        """Record an observation from a tool or environment."""
        self.observations.append({
            "source": source,
            "content": content[:500],  # Truncate to manage space
            "timestamp": datetime.now().isoformat()
        })
        # Keep only recent observations in working memory
        if len(self.observations) > 10:
            self.observations.pop(0)

    def note(self, key: str, value: str) -> None:
        """Store a note for later reference."""
        self.notes[key] = value

    def add_hypothesis(self, hypothesis: str, confidence: float) -> None:
        """Track a hypothesis being considered."""
        self.hypotheses.append({
            "hypothesis": hypothesis,
            "confidence": confidence,
            "evidence_for": [],
            "evidence_against": []
        })

    def to_prompt(self) -> str:
        """
        Serialize working memory for inclusion in prompt.
        This is how working memory enters short-term memory.
        """
        sections = []

        if self.current_goal:
            sections.append(f"## Current Goal\n{self.current_goal}")

        if self.sub_goals:
            goals_str = "\n".join(f"- [ ] {g}" for g in self.sub_goals)
            completed_str = "\n".join(f"- [x] {s}" for s in self.completed_steps[-3:])
            sections.append(f"## Progress\n{completed_str}\n{goals_str}")

        if self.observations:
            recent = self.observations[-3:]
            obs_str = "\n".join(f"- [{o['source']}]: {o['content'][:200]}" for o in recent)
            sections.append(f"## Recent Observations\n{obs_str}")

        if self.notes:
            notes_str = "\n".join(f"- {k}: {v}" for k, v in self.notes.items())
            sections.append(f"## Notes\n{notes_str}")

        if self.hypotheses:
            hyp_str = "\n".join(
                f"- {h['hypothesis']} (confidence: {h['confidence']:.0%})"
                for h in self.hypotheses
            )
            sections.append(f"## Working Hypotheses\n{hyp_str}")

        return "\n\n".join(sections)

class LongTermMemory:
    """
    Long-term memory persists across sessions and has unlimited capacity.
    It requires explicit storage and retrieval operations.
    """

    def __init__(self, vector_store, embedding_model):
        self.vector_store = vector_store
        self.embedding_model = embedding_model

        # Track memory statistics
        self.total_memories = 0
        self.retrieval_stats = {"hits": 0, "misses": 0}

    def store(
        self,
        content: str,
        memory_type: str,
        importance: float = 0.5,
        metadata: dict = None
    ) -> str:
        """
        Store a memory for long-term retention.

        The encoding process:
        1. Generate embedding for semantic search
        2. Extract structured metadata for filtering
        3. Assign importance score for retrieval ranking
        4. Store with timestamp for temporal queries
        """
        # Generate embedding
        embedding = self.embedding_model.embed(content)

        # Create memory record
        memory_id = str(uuid.uuid4())
        memory = {
            "id": memory_id,
            "content": content,
            "embedding": embedding,
            "memory_type": memory_type,  # fact, episode, preference, etc.
            "importance": importance,
            "created_at": datetime.now().isoformat(),
            "last_accessed": datetime.now().isoformat(),
            "access_count": 0,
            "metadata": metadata or {}
        }

        self.vector_store.insert(memory)
        self.total_memories += 1

        return memory_id

    def retrieve(
        self,
        query: str,
        k: int = 5,
        memory_types: list[str] = None,
        min_importance: float = 0.0,
        recency_weight: float = 0.1
    ) -> list[dict]:
        """
        Retrieve relevant memories for a query.

        Retrieval is the critical operation:
        - Too few results: Agent lacks necessary context
        - Too many results: Overwhelms short-term memory
        - Wrong results: Agent uses irrelevant information

        We combine multiple signals:
        1. Semantic similarity (embedding distance)
        2. Importance score (pre-assigned weight)
        3. Recency (recently accessed = more relevant)
        4. Access frequency (frequently used = important)
        """
        # Embed the query
        query_embedding = self.embedding_model.embed(query)

        # Search vector store
        candidates = self.vector_store.search(
            query_embedding,
            k=k * 3,  # Over-retrieve then re-rank
            filter={"memory_type": {"$in": memory_types}} if memory_types else None
        )

        if not candidates:
            self.retrieval_stats["misses"] += 1
            return []

        # Re-rank with multiple signals
        now = datetime.now()
        scored_memories = []

        for memory in candidates:
            # Base score from semantic similarity
            score = memory["similarity"]

            # Importance weighting
            score *= (0.5 + 0.5 * memory["importance"])

            # Recency boost
            last_accessed = datetime.fromisoformat(memory["last_accessed"])
            days_ago = (now - last_accessed).days
            recency_score = math.exp(-recency_weight * days_ago)
            score *= (0.7 + 0.3 * recency_score)

            # Filter by minimum importance
            if memory["importance"] >= min_importance:
                scored_memories.append((memory, score))

        # Sort by final score
        scored_memories.sort(key=lambda x: x[1], reverse=True)

        # Update access statistics for retrieved memories
        results = []
        for memory, score in scored_memories[:k]:
            self._update_access(memory["id"])
            results.append(memory)

        self.retrieval_stats["hits"] += 1
        return results

    def _update_access(self, memory_id: str) -> None:
        """Update access time and count for a retrieved memory."""
        self.vector_store.update(memory_id, {
            "last_accessed": datetime.now().isoformat(),
            "access_count": {"$inc": 1}
        })

    def forget(self, threshold_days: int = 90, min_importance: float = 0.3) -> int:
        """
        Remove old, low-importance, unused memories.

        Forgetting is essential for long-term memory systems:
        - Prevents unbounded growth
        - Improves retrieval quality (less noise)
        - Reduces storage costs

        We forget memories that are:
        - Old (not accessed recently)
        - Low importance (not marked as significant)
        - Rarely accessed (not frequently useful)
        """
        cutoff = datetime.now() - timedelta(days=threshold_days)

        # Find candidates for forgetting
        candidates = self.vector_store.query({
            "last_accessed": {"$lt": cutoff.isoformat()},
            "importance": {"$lt": min_importance},
            "access_count": {"$lt": 3}
        })

        forgotten = 0
        for memory in candidates:
            self.vector_store.delete(memory["id"])
            forgotten += 1

        self.total_memories -= forgotten
        return forgotten

┌─────────────────────────────────────────────────────────────────────────┐
│                         EPISODE STRUCTURE                                │
├─────────────────────────────────────────────────────────────────────────┤
│                                                                          │
│  METADATA                                                                │
│  ├── Timestamp: When did this happen?                                   │
│  ├── Duration: How long was the interaction?                            │
│  ├── Participants: Who was involved?                                    │
│  └── Context: What was the setting/situation?                           │
│                                                                          │
│  NARRATIVE                                                               │
│  ├── Summary: 2-3 sentence overview                                     │
│  ├── Goal: What was the user trying to accomplish?                      │
│  ├── Journey: Key steps and turns in the interaction                    │
│  └── Outcome: What was achieved?                                        │
│                                                                          │
│  KEY MOMENTS                                                             │
│  ├── Decisions: Choices made and why                                    │
│  ├── Discoveries: New information learned                               │
│  ├── Problems: Issues encountered                                       │
│  └── Solutions: How problems were resolved                              │
│                                                                          │
│  LESSONS                                                                 │
│  ├── What worked well?                                                  │
│  ├── What didn't work?                                                  │
│  └── What should be done differently next time?                         │
│                                                                          │
│  RETRIEVAL AIDS                                                          │
│  ├── Embedding: For semantic search                                     │
│  ├── Tags: For categorical filtering                                    │
│  └── Importance score: For prioritizing in retrieval                    │
│                                                                          │
└─────────────────────────────────────────────────────────────────────────┘

class EpisodicMemory:
    """
    Episodic memory stores experiences as coherent narratives.
    Each episode captures what happened, when, with what outcome.
    """

    def __init__(self, storage, llm):
        self.storage = storage
        self.llm = llm  # For summarization and extraction

    def create_episode(
        self,
        conversation: list[dict],
        task: str = None,
        outcome: str = None,
        user_id: str = None
    ) -> dict:
        """
        Transform a conversation into an episode.

        This is the encoding process for episodic memory:
        1. Summarize the overall interaction
        2. Extract key moments (decisions, discoveries, problems)
        3. Identify the outcome (success, failure, partial)
        4. Note lessons learned
        """
        # Generate episode summary
        summary = self._summarize_conversation(conversation)

        # Extract key moments
        key_moments = self._extract_key_moments(conversation)

        # Identify emotional/importance peaks
        importance_score = self._assess_importance(conversation, outcome)

        # Extract lessons learned
        lessons = self._extract_lessons(conversation, outcome)

        episode = {
            "id": str(uuid.uuid4()),
            "timestamp": datetime.now().isoformat(),
            "duration_minutes": self._estimate_duration(conversation),
            "user_id": user_id,

            # The narrative structure
            "task": task,
            "summary": summary,
            "key_moments": key_moments,
            "outcome": outcome,
            "lessons": lessons,

            # For retrieval
            "embedding": self._embed_episode(summary, key_moments),
            "importance": importance_score,

            # Raw data (optional, for detailed recall)
            "message_count": len(conversation),
            "full_conversation": conversation  # Or store separately
        }

        self.storage.insert(episode)
        return episode

    def _summarize_conversation(self, conversation: list[dict]) -> str:
        """Generate a narrative summary of the conversation."""
        messages_text = "\n".join(
            f"{m['role']}: {m['content'][:200]}"
            for m in conversation
        )

        prompt = f"""Summarize this conversation as a brief narrative.
Focus on: what was discussed, what was accomplished, any problems encountered.
Write in past tense, 2-3 sentences.

Conversation:
{messages_text}

Summary:"""

        return self.llm.generate(prompt)

    def _extract_key_moments(self, conversation: list[dict]) -> list[dict]:
        """
        Identify the pivotal moments in the conversation.

        Key moments are:
        - Decisions made
        - Problems discovered
        - Solutions found
        - User preferences revealed
        - Misunderstandings corrected
        """
        messages_text = "\n".join(
            f"[{i}] {m['role']}: {m['content'][:300]}"
            for i, m in enumerate(conversation)
        )

        prompt = f"""Identify 2-4 key moments in this conversation.
For each moment, note:
- What happened
- Why it was significant
- The message index

Format as JSON array.

Conversation:
{messages_text}

Key moments:"""

        response = self.llm.generate(prompt)
        try:
            return json.loads(response)
        except:
            return []

    def _extract_lessons(self, conversation: list[dict], outcome: str) -> list[str]:
        """
        Extract lessons learned from this episode.

        Lessons help the agent improve over time:
        - "User prefers detailed explanations"
        - "This approach led to errors, try alternative"
        - "Always confirm before making changes"
        """
        prompt = f"""Based on this conversation and its outcome, what lessons should be remembered?

Outcome: {outcome}

List 1-3 actionable lessons:"""

        response = self.llm.generate(prompt)
        return [l.strip() for l in response.split("\n") if l.strip()]

    def recall_similar_episodes(
        self,
        query: str,
        k: int = 3,
        user_id: str = None
    ) -> list[dict]:
        """
        Find past episodes similar to the current situation.

        This enables experiential reasoning:
        "Last time we encountered a similar problem..."
        "Based on past interactions with this user..."
        """
        query_embedding = self._embed_query(query)

        filters = {}
        if user_id:
            filters["user_id"] = user_id

        episodes = self.storage.search(
            query_embedding,
            k=k,
            filter=filters
        )

        return episodes

    def recall_by_timeframe(
        self,
        start: datetime,
        end: datetime,
        user_id: str = None
    ) -> list[dict]:
        """Recall episodes from a specific time period."""
        query = {"timestamp": {"$gte": start.isoformat(), "$lte": end.isoformat()}}
        if user_id:
            query["user_id"] = user_id
        return self.storage.query(query)

    def get_user_history(self, user_id: str, limit: int = 10) -> list[dict]:
        """Get recent episodes for a specific user."""
        return self.storage.query(
            {"user_id": user_id},
            sort={"timestamp": -1},
            limit=limit
        )

User Profile (Semantic Memory):
- Name: Alex
- Company: HealthStart Inc.
- Role: CTO
- Team size: 12 developers
- Tech stack: Python, PostgreSQL, React
- Preferences: Prefers detailed explanations, likes code examples
- Timezone: PST
- Current project: HIPAA-compliant data pipeline

"Alex is the CTO of HealthStart Inc., a healthcare startup
building HIPAA-compliant data pipelines."

{
  "user": {
    "name": "Alex",
    "role": "CTO",
    "company": "HealthStart Inc.",
    "industry": "healthcare"
  }
}

(Alex) --[is_role]--> (CTO)
(Alex) --[works_at]--> (HealthStart Inc.)
(HealthStart Inc.) --[is_in_industry]--> (Healthcare)
(HealthStart Inc.) --[is_building]--> (HIPAA Data Pipeline)

class SemanticMemory:
    """
    Semantic memory stores facts, concepts, and knowledge.
    Unlike episodic memory, these are decontextualized truths.
    """

    def __init__(self, storage, embedding_model, llm):
        self.storage = storage
        self.embedding_model = embedding_model
        self.llm = llm

    def store_fact(
        self,
        subject: str,
        predicate: str,
        object: str,
        confidence: float = 1.0,
        source: str = None
    ) -> str:
        """
        Store a semantic fact as a triple.

        Examples:
        - ("user", "works_at", "Acme Corp")
        - ("project", "uses_framework", "React")
        - ("deadline", "is", "March 15")
        """
        # Create natural language representation
        fact_text = f"{subject} {predicate.replace('_', ' ')} {object}"

        fact = {
            "id": str(uuid.uuid4()),
            "subject": subject,
            "predicate": predicate,
            "object": object,
            "fact_text": fact_text,
            "embedding": self.embedding_model.embed(fact_text),
            "confidence": confidence,
            "source": source,
            "created_at": datetime.now().isoformat(),
            "updated_at": datetime.now().isoformat(),
            "contradicted_by": None
        }

        # Check for existing facts about same subject-predicate
        existing = self._find_existing(subject, predicate)
        if existing:
            # Handle potential contradiction
            self._handle_update(existing, fact)
        else:
            self.storage.insert(fact)

        return fact["id"]

    def _handle_update(self, existing: dict, new: dict) -> None:
        """
        Handle updating an existing fact.

        Facts can change:
        - User changes jobs
        - Project deadlines shift
        - Preferences evolve

        We don't just overwrite—we may want to track history.
        """
        if existing["object"] != new["object"]:
            # Mark old fact as superseded
            self.storage.update(existing["id"], {
                "contradicted_by": new["id"],
                "current": False
            })

            # Store new fact
            new["previous_value"] = existing["object"]
            self.storage.insert(new)
        else:
            # Same fact, update confidence/timestamp
            self.storage.update(existing["id"], {
                "confidence": max(existing["confidence"], new["confidence"]),
                "updated_at": new["updated_at"]
            })

    def extract_facts_from_conversation(
        self,
        conversation: list[dict],
        entity: str = "user"
    ) -> list[dict]:
        """
        Extract semantic facts from a conversation.

        This is the process of converting episodic to semantic memory:
        - "I work at Google" → (user, works_at, Google)
        - "We're using Python" → (project, uses_language, Python)
        - "Call me Alex" → (user, preferred_name, Alex)
        """
        messages_text = "\n".join(
            f"{m['role']}: {m['content']}" for m in conversation
        )

        prompt = f"""Extract factual information from this conversation about {entity}.

Format as JSON array of objects with: subject, predicate, object, confidence (0-1)

Only extract explicitly stated facts, not inferences.

Conversation:
{messages_text}

Facts:"""

        response = self.llm.generate(prompt)
        try:
            facts = json.loads(response)
            # Store each extracted fact
            for fact in facts:
                self.store_fact(**fact)
            return facts
        except:
            return []

    def query_facts(
        self,
        subject: str = None,
        predicate: str = None,
        semantic_query: str = None
    ) -> list[dict]:
        """
        Query semantic memory.

        Can query by:
        - Exact subject/predicate: "What does user work_at?"
        - Semantic similarity: "What do we know about user's job?"
        """
        if subject and predicate:
            # Exact query
            return self.storage.query({
                "subject": subject,
                "predicate": predicate,
                "current": {"$ne": False}
            })

        if semantic_query:
            # Semantic search
            query_embedding = self.embedding_model.embed(semantic_query)
            return self.storage.search(
                query_embedding,
                k=10,
                filter={"current": {"$ne": False}}
            )

        if subject:
            # All facts about subject
            return self.storage.query({
                "subject": subject,
                "current": {"$ne": False}
            })

        return []

    def get_entity_profile(self, entity: str) -> dict:
        """
        Compile all known facts about an entity into a profile.

        This creates a structured summary:
        {
            "name": "Alex",
            "works_at": "Acme Corp",
            "role": "Engineer",
            "preferences": {...}
        }
        """
        facts = self.query_facts(subject=entity)

        profile = {"entity": entity}
        for fact in facts:
            profile[fact["predicate"]] = fact["object"]

        return profile

Here's how we format error messages in our codebase:

Example 1:
Input: File not found
Output: ERROR_FILE_001: Requested file could not be located in path

Example 2:
Input: Permission denied
Output: ERROR_AUTH_002: Insufficient permissions for requested operation

Now format this:
Input: Connection timeout

When reviewing code, follow these steps:
1. First, check for syntax errors
2. Then, look for potential security vulnerabilities
3. Next, evaluate code style and naming conventions
4. Finally, assess performance implications

For each issue found, format your feedback as:
- Line number
- Issue type (error/warning/suggestion)
- Description
- Suggested fix

# Instead of teaching the model to calculate shipping costs
# (complex procedure with zones, weights, discounts)...

# Define a tool that does it:
def calculate_shipping(weight: float, zone: str, priority: bool) -> float:
    """Calculate shipping cost based on complex internal rules."""
    # All the procedural knowledge is in this function
    pass

Training data:
{"input": "Write a SQL query to find users", "output": "SELECT * FROM users;"}
{"input": "Write a SQL query to count orders", "output": "SELECT COUNT(*) FROM orders;"}
... thousands more examples ...

┌─────────────────────────────────────────────────────────────────────────┐
│                    PROCEDURAL SKILL HIERARCHY                            │
├─────────────────────────────────────────────────────────────────────────┤
│                                                                          │
│  EASILY APPROXIMATED AT RUNTIME (few-shot or instructions)              │
│  ├── Output formatting (JSON, Markdown, specific templates)             │
│  ├── Simple transformations (reformat date, change case)                │
│  └── Pattern-based generation (follow a style guide)                    │
│                                                                          │
│  MODERATELY DIFFICULT (may need many examples or fine-tuning)           │
│  ├── Domain-specific writing styles (legal, medical, technical)         │
│  ├── Code in familiar languages with project conventions                │
│  └── Multi-step reasoning in specialized domains                        │
│                                                                          │
│  DIFFICULT (usually requires fine-tuning)                               │
│  ├── Fluent code in rare/internal languages                             │
│  ├── Complex domain-specific reasoning (tax law, drug interactions)     │
│  └── Maintaining long-range consistency in specialized outputs          │
│                                                                          │
│  VERY DIFFICULT (may require specialized training)                      │
│  ├── Novel task types not seen in training                              │
│  ├── Combining multiple complex skills simultaneously                   │
│  └── Tasks requiring implicit knowledge hard to articulate              │
│                                                                          │
└─────────────────────────────────────────────────────────────────────────┘

Procedure: Code Review
Trigger: When user asks for code review
Steps:
1. Read the code carefully
2. Check for bugs
3. Evaluate style
...
Examples:
[Example 1]
[Example 2]

class ProceduralMemory:
    """
    Procedural memory stores 'how to' knowledge.
    In LLMs, this is approximated through examples and instructions.
    """

    def __init__(self, storage):
        self.storage = storage
        self.procedures: dict[str, dict] = {}

    def store_procedure(
        self,
        name: str,
        description: str,
        steps: list[str],
        examples: list[dict],
        trigger_patterns: list[str]
    ) -> None:
        """
        Store a procedure for later retrieval.

        A procedure includes:
        - What it does (description)
        - How to do it (steps)
        - Examples of doing it (few-shot)
        - When to use it (trigger patterns)
        """
        procedure = {
            "name": name,
            "description": description,
            "steps": steps,
            "examples": examples,
            "trigger_patterns": trigger_patterns,
            "usage_count": 0,
            "success_rate": None
        }

        self.procedures[name] = procedure
        self.storage.insert(procedure)

    def get_relevant_procedures(self, task: str) -> list[dict]:
        """
        Find procedures relevant to the current task.

        Matches based on:
        - Semantic similarity to description
        - Trigger pattern matching
        """
        relevant = []

        for name, proc in self.procedures.items():
            # Check trigger patterns
            for pattern in proc["trigger_patterns"]:
                if pattern.lower() in task.lower():
                    relevant.append(proc)
                    break

        return relevant

    def format_procedure_for_prompt(self, procedure: dict) -> str:
        """
        Format a procedure for inclusion in the prompt.

        This is how procedural memory enters short-term memory
        at inference time.
        """
        sections = [
            f"## Procedure: {procedure['name']}",
            f"\n{procedure['description']}",
            "\n### Steps:",
            "\n".join(f"{i+1}. {step}" for i, step in enumerate(procedure['steps']))
        ]

        if procedure['examples']:
            sections.append("\n### Examples:")
            for ex in procedure['examples'][:2]:  # Limit to save tokens
                sections.append(f"\nInput: {ex['input']}")
                sections.append(f"Output: {ex['output']}")

        return "\n".join(sections)

    def record_execution(self, procedure_name: str, success: bool) -> None:
        """Track procedure execution for learning."""
        if procedure_name in self.procedures:
            proc = self.procedures[procedure_name]
            proc["usage_count"] += 1

            # Update success rate
            if proc["success_rate"] is None:
                proc["success_rate"] = 1.0 if success else 0.0
            else:
                # Exponential moving average
                proc["success_rate"] = 0.9 * proc["success_rate"] + 0.1 * (1.0 if success else 0.0)

┌─────────────────────────────────────────────────────────────────────────┐
│                    INTEGRATED MEMORY SYSTEM                              │
│                                                                          │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │                     CONTEXT WINDOW                               │   │
│  │                   (Short-Term Memory)                            │   │
│  │                                                                  │   │
│  │  ┌─────────────┐  ┌─────────────┐  ┌─────────────────────────┐ │   │
│  │  │   System    │  │   Working   │  │    Recent Messages      │ │   │
│  │  │   Prompt    │  │   Memory    │  │    (Conversation)       │ │   │
│  │  │             │  │  Scratchpad │  │                         │ │   │
│  │  └─────────────┘  └─────────────┘  └─────────────────────────┘ │   │
│  │                                                                  │   │
│  │  ┌─────────────────────────────────────────────────────────────┐│   │
│  │  │              Retrieved Context                               ││   │
│  │  │  (Loaded from long-term memory based on relevance)          ││   │
│  │  │                                                              ││   │
│  │  │  • Semantic facts about user                                 ││   │
│  │  │  • Relevant past episodes                                    ││   │
│  │  │  • Applicable procedures                                     ││   │
│  │  └─────────────────────────────────────────────────────────────┘│   │
│  │                                                                  │   │
│  └─────────────────────────────────────────────────────────────────┘   │
│         ▲                           │                                    │
│         │ Retrieve                  │ Store                              │
│         │                           ▼                                    │
│  ┌─────────────────────────────────────────────────────────────────┐   │
│  │                   LONG-TERM STORAGE                              │   │
│  │                                                                  │   │
│  │  ┌──────────────┐ ┌──────────────┐ ┌──────────────┐            │   │
│  │  │   Semantic   │ │   Episodic   │ │  Procedural  │            │   │
│  │  │    Memory    │ │    Memory    │ │   Memory     │            │   │
│  │  │              │ │              │ │              │            │   │
│  │  │  Facts about │ │    Past      │ │   How-to     │            │   │
│  │  │  user, world │ │ conversations│ │  knowledge   │            │   │
│  │  └──────────────┘ └──────────────┘ └──────────────┘            │   │
│  │                                                                  │   │
│  └─────────────────────────────────────────────────────────────────┘   │
│                                                                          │
└─────────────────────────────────────────────────────────────────────────┘

class IntegratedMemorySystem:
    """
    A complete memory system integrating all memory types.
    """

    def __init__(
        self,
        llm,
        embedding_model,
        vector_store,
        max_context_tokens: int = 8000
    ):
        self.llm = llm
        self.max_context_tokens = max_context_tokens

        # Initialize all memory types
        self.short_term = ShortTermMemory(max_tokens=max_context_tokens)
        self.working = WorkingMemory()
        self.long_term = LongTermMemory(vector_store, embedding_model)
        self.episodic = EpisodicMemory(vector_store, llm)
        self.semantic = SemanticMemory(vector_store, embedding_model, llm)
        self.procedural = ProceduralMemory(vector_store)

    def prepare_context(self, user_message: str, user_id: str = None) -> list[dict]:
        """
        Prepare the full context for an LLM call.

        This is where memory integration happens:
        1. Start with system prompt
        2. Add working memory state
        3. Retrieve relevant long-term memories
        4. Add recent conversation
        5. Add the new user message

        All while respecting token limits.
        """
        context = []
        token_budget = self.max_context_tokens

        # 1. System prompt (always included)
        system_prompt = self._build_system_prompt()
        context.append({"role": "system", "content": system_prompt})
        token_budget -= self._count_tokens(system_prompt)

        # 2. Working memory (high priority)
        working_content = self.working.to_prompt()
        if working_content:
            context.append({"role": "system", "content": f"Current state:\n{working_content}"})
            token_budget -= self._count_tokens(working_content)

        # 3. Retrieve relevant long-term memories
        retrieved = self._retrieve_relevant_memories(user_message, user_id)
        retrieved_content = self._format_retrieved_memories(retrieved)
        if retrieved_content and token_budget > 1000:
            # Reserve space for retrieved context
            retrieved_tokens = min(self._count_tokens(retrieved_content), token_budget // 3)
            context.append({"role": "system", "content": f"Relevant context:\n{retrieved_content}"})
            token_budget -= retrieved_tokens

        # 4. Recent conversation history
        recent_messages = self.short_term.get_context()
        for msg in recent_messages:
            msg_tokens = self._count_tokens(msg["content"])
            if token_budget > msg_tokens + 500:  # Reserve for new message
                context.append(msg)
                token_budget -= msg_tokens

        # 5. New user message
        context.append({"role": "user", "content": user_message})

        return context

    def _retrieve_relevant_memories(self, query: str, user_id: str = None) -> dict:
        """Retrieve from all long-term memory types."""
        return {
            "semantic": self.semantic.query_facts(semantic_query=query)[:3],
            "episodic": self.episodic.recall_similar_episodes(query, k=2, user_id=user_id),
            "procedural": self.procedural.get_relevant_procedures(query)[:2]
        }

    def _format_retrieved_memories(self, retrieved: dict) -> str:
        """Format retrieved memories for inclusion in context."""
        sections = []

        if retrieved["semantic"]:
            facts = "\n".join(f"- {f['fact_text']}" for f in retrieved["semantic"])
            sections.append(f"Known facts:\n{facts}")

        if retrieved["episodic"]:
            episodes = "\n".join(f"- {e['summary']}" for e in retrieved["episodic"])
            sections.append(f"Relevant past interactions:\n{episodes}")

        if retrieved["procedural"]:
            procs = "\n".join(
                self.procedural.format_procedure_for_prompt(p)
                for p in retrieved["procedural"]
            )
            sections.append(f"Applicable procedures:\n{procs}")

        return "\n\n".join(sections)

    def process_response(
        self,
        user_message: str,
        assistant_response: str,
        user_id: str = None
    ) -> None:
        """
        Process a completed interaction for memory storage.

        This happens after the response is generated:
        1. Add to short-term memory
        2. Extract facts for semantic memory
        3. Update working memory
        """
        # Update short-term memory
        self.short_term.add("user", user_message)
        self.short_term.add("assistant", assistant_response)

        # Extract semantic facts (background processing)
        conversation = [
            {"role": "user", "content": user_message},
            {"role": "assistant", "content": assistant_response}
        ]
        self.semantic.extract_facts_from_conversation(conversation)

    def end_session(
        self,
        user_id: str = None,
        task: str = None,
        outcome: str = None
    ) -> None:
        """
        End the current session and consolidate memories.

        This is when episodic memories are formed.
        """
        # Create episode from session
        conversation = self.short_term.get_context()
        if conversation:
            self.episodic.create_episode(
                conversation=conversation,
                task=task,
                outcome=outcome,
                user_id=user_id
            )

        # Clear short-term and working memory
        self.short_term = ShortTermMemory(max_tokens=self.max_context_tokens)
        self.working = WorkingMemory()

    def _count_tokens(self, text: str) -> int:
        return len(text) // 4  # Simplified

    def _build_system_prompt(self) -> str:
        return "You are a helpful assistant with access to memory of past interactions."

Traditional Computer          →    LLM Agent
─────────────────────────────────────────────
RAM (fast, limited)           →    Context Window
Disk (slow, unlimited)        →    External Storage
OS Memory Manager             →    MemGPT Agent

# Core memory editing
memory_replace(section, old_content, new_content)  # Update memory block
memory_insert(section, content)                     # Add to memory
memory_rethink(section, new_content)               # Revise understanding

# Archival memory
archival_memory_insert(content)                    # Store for long-term
archival_memory_search(query, k=10)                # Semantic retrieval

# Conversation search
conversation_search(query)                          # Search past messages
conversation_search_date(start_date, end_date)     # Temporal search

from letta import create_client, LLMConfig, EmbeddingConfig

# Create Letta client
client = create_client()

# Configure memory
agent = client.create_agent(
    name="personal_assistant",
    llm_config=LLMConfig(model="gpt-4o"),
    embedding_config=EmbeddingConfig(model="text-embedding-3-small"),
    system="You are a personal assistant with long-term memory.",
    memory_blocks=[
        {"label": "human", "value": "Name: Unknown\nPreferences: Unknown"},
        {"label": "persona", "value": "I am a helpful assistant that remembers past conversations."}
    ]
)

# Interact with agent
response = client.send_message(
    agent_id=agent.id,
    message="My name is Alex and I prefer concise answers."
)

# Agent updates its memory automatically
# Next conversation will remember this preference

class ConversationMemory:
    def __init__(self, max_tokens=4000, summarizer=None):
        self.messages = []
        self.summary = ""
        self.max_tokens = max_tokens
        self.summarizer = summarizer or default_summarizer

    def add_message(self, role: str, content: str):
        self.messages.append({"role": role, "content": content})

        # Check if we need to summarize
        if self.count_tokens() > self.max_tokens:
            self._compress()

    def _compress(self):
        # Keep recent messages, summarize older ones
        recent = self.messages[-10:]
        old = self.messages[:-10]

        if old:
            new_summary = self.summarizer(self.summary, old)
            self.summary = new_summary
            self.messages = recent

    def get_context(self):
        context = []
        if self.summary:
            context.append({"role": "system", "content": f"Previous conversation summary: {self.summary}"})
        context.extend(self.messages)
        return context

class EntityMemory:
    def __init__(self, db):
        self.db = db  # Vector database

    def extract_and_store(self, conversation: list[dict]):
        # Extract entities with LLM
        entities = self.extract_entities(conversation)

        for entity in entities:
            # Check if entity exists
            existing = self.db.search(entity.name, k=1)

            if existing and existing[0].score > 0.9:
                # Update existing entity
                self.merge_entity(existing[0], entity)
            else:
                # Create new entity
                self.db.insert(entity)

    def get_relevant_entities(self, query: str, k=5):
        return self.db.search(query, k=k)

class SemanticMemory:
    def __init__(self, db, decay_rate=0.1):
        self.db = db
        self.decay_rate = decay_rate

    def remember(self, content: str, importance: float = 0.5):
        embedding = self.embed(content)
        self.db.insert({
            "content": content,
            "embedding": embedding,
            "importance": importance,
            "last_accessed": datetime.now(),
            "access_count": 1
        })

    def recall(self, query: str, k=10):
        candidates = self.db.search(query, k=k*2)

        # Score by relevance + recency + importance
        scored = []
        now = datetime.now()
        for c in candidates:
            age = (now - c.last_accessed).days
            recency_score = math.exp(-self.decay_rate * age)
            final_score = c.similarity * c.importance * recency_score
            scored.append((c, final_score))

            # Update access time
            c.last_accessed = now
            c.access_count += 1

        scored.sort(key=lambda x: x[1], reverse=True)
        return [c for c, _ in scored[:k]]

    def forget(self, threshold=0.1):
        """Remove low-value memories"""
        all_memories = self.db.get_all()
        now = datetime.now()

        for m in all_memories:
            age = (now - m.last_accessed).days
            value = m.importance * math.exp(-self.decay_rate * age)
            if value < threshold:
                self.db.delete(m.id)

class EpisodicMemory:
    def __init__(self, db):
        self.db = db

    def end_episode(self, conversation: list[dict], metadata: dict):
        # Summarize episode
        summary = self.summarize(conversation)

        # Extract key moments
        key_moments = self.extract_key_moments(conversation)

        # Store episode
        episode = {
            "summary": summary,
            "key_moments": key_moments,
            "full_conversation": conversation,
            "metadata": metadata,
            "embedding": self.embed(summary),
            "timestamp": datetime.now()
        }
        self.db.insert(episode)

    def recall_similar_episodes(self, query: str, k=3):
        return self.db.search(query, k=k)

    def recall_by_time(self, start: datetime, end: datetime):
        return self.db.query({"timestamp": {"$gte": start, "$lte": end}})

# In agent loop
response = llm.generate(messages)

# Check for memory updates
if should_update_memory(response):
    memory_update = llm.generate([
        {"role": "system", "content": "Extract key facts to remember"},
        {"role": "user", "content": str(messages[-5:])}
    ])
    memory.store(memory_update)

# After conversation ends
async def process_conversation(conversation_id: str):
    conversation = db.get_conversation(conversation_id)

    # Extract memories in background
    memories = await extract_memories(conversation)

    # Store for future sessions
    for memory in memories:
        memory_store.insert(memory)

# Measure and optimize
import time

async def get_relevant_context(query: str):
    start = time.time()

    # Parallel retrieval
    results = await asyncio.gather(
        entity_memory.search(query),
        episodic_memory.search(query),
        recent_messages.get()
    )

    latency = time.time() - start
    logger.info(f"Memory retrieval: {latency:.3f}s")

    return merge_results(results)