memory-management

name	memory-management
description	Design and operate memory systems for long-running AI agents. Covers context window optimization, summarization strategies, vector-based retrieval, episodic memory, memory consolidation, and garbage collection for production agent systems.
metadata	{"author":"cosmicstack-labs","version":"1.0.0","category":"ai-ml","tags":["memory-management","context-window","vector-database","summarization","rag","long-running-agents","memory-consolidation"]}

Memory Management for Long-Running Agents

Overview

Long-running agents face a fundamental problem: they can't remember everything, but forgetting the wrong thing breaks their usefulness. This skill covers memory architectures that balance context retention, token budget, and retrieval accuracy for agents that run for hours, days, or continuously.

---

Core Concepts

The Memory Problem

Issue	Symptom	Cost
Context Overflow	Agent forgets early instructions	Task failure, incoherent responses
Token Bloat	Every message keeps growing	10x+ cost increase per task
Memory Pollution	Irrelevant memories distract agent	Hallucination, off-target responses
Stale Memories	Outdated information used as fact	Incorrect decisions
Memory Leaks	Unused data accumulates unbounded	Crash from OOM, endless context

Memory Tiers

Tier	Storage	Capacity	Access Speed	Cost	Best For
L1 — Working	In-context (LLM window)	8K-200K tokens	Instant	$$$	Current task, immediate context
L2 — Recent	Sliding window buffer	~2K turns	< 10ms	$$	Recent conversation history
L3 — Episodic	Event log / timeseries	Millions of events	< 50ms	$	Past actions, outcomes, decisions
L4 — Semantic	Vector database	Unlimited	< 100ms	$	Knowledge, facts, relationships
L5 — Archival	Object storage	Unlimited	> 1s	$	Backups, compliance, audit

---

Step-by-Step Implementation

Step 1: Build a Tiered Memory System

from dataclasses import dataclass, field
from typing import Optional
import json
import time

@dataclass
class MemoryEntry:
    content: str
    timestamp: float = None
    importance: float = 0.5  # 0.0 (trivial) to 1.0 (critical)
    tags: list[str] = field(default_factory=list)
    token_count: int = 0
    
    def __post_init__(self):
        if self.timestamp is None:
            self.timestamp = time.time()

class TieredMemory:
    """Multi-tier memory with automatic promotion and demotion."""
    
    def __init__(self, llm, vector_store, max_context_tokens: int = 8000):
        self.llm = llm
        self.vector_store = vector_store
        self.max_context_tokens = max_context_tokens
        
        # L1: Working context (in-memory)
        self.working_memory: list[MemoryEntry] = []
        self.current_tokens = 0
        
        # L2: Recent history buffer
        .recent_buffer: [MemoryEntry] = []
        .buffer_size = 
        
        
        .episodes: [MemoryEntry] = []
        
        
        
    
      ():
        
        entry = MemoryEntry(
            content=content,
            importance=importance,
            tags=tags  [],
            token_count=._count_tokens(content)
        )
        
        
        .working_memory.append(entry)
        .current_tokens += entry.token_count
        
        
         importance > :
            .episodes.append(entry)
             .vector_store.store(entry)
        
        
         ._trim_working_memory()

Step 2: Implement Context Window Management

class ContextManager:
    """Optimize what stays in the context window."""
    
    def __init__(self, tiered_memory: TieredMemory, 
                 summarizer, max_tokens: int = 8000):
        self.memory = tiered_memory
        self.summarizer = summarizer
        self.max_tokens = max_tokens
        self.reserved_tokens = 2000  # Reserve for new input/output
    
    async def build_context(self, task: str, top_k: int = 5) -> list[dict]:
        """Build the optimal context for a task."""
        
        available_tokens = self.max_tokens - self.reserved_tokens
        
        # 1. Start with high-importance working memory
        context = []
        tokens_used = 0
        
        working = sorted(
            self.memory.working_memory,
            key=lambda e: e.importance,
            reverse=True
        )
        
        for entry in working:
            if tokens_used + entry.token_count > available_tokens:
                break
            context.append({"role": "system", "content": entry.content})
            tokens_used += entry.token_count
        
        # 2. Add semantically relevant memories
        relevant =  .memory.vector_store.search(task, k=top_k)
         mem  relevant:
             tokens_used + mem.token_count > available_tokens:
                
            context.append({: , : mem.content})
            tokens_used += mem.token_count
        
        
         (context) < (working):
            summary =  ._get_summary()
            context.insert(, {: , 
                               : })
        
         context
    
      () -> :
        
        excluded = .memory.working_memory[
            (.memory.working_memory) - :
        ]
        texts = [e.content  e  excluded]
          .summarizer.summarize(.join(texts))
    
      ():
        
         .memory.current_tokens > .max_tokens * :
            
            .memory.working_memory.sort(
                key= e: e.importance
            )
            removed = .memory.working_memory.pop()
            .memory.current_tokens -= removed.token_count

Step 3: Memory Summarization Strategies

class MemorySummarizer:
    """Different summarization strategies for different memory types."""
    
    def __init__(self, llm):
        self.llm = llm
    
    async def rolling_summary(self, conversation: list[str], 
                              window: int = 20) -> str:
        """Summarize recent conversation window."""
        recent = conversation[-window:]
        return await self.llm.generate(
            f"Summarize this conversation concisely, preserving key facts, "
            f"decisions, and user preferences:\n\n{chr(10).join(recent)}"
        )
    
    async def hierarchical_summary(self, episodes: list[MemoryEntry], 
                                   level: int = 1) -> str:
        """Multi-level summarization for long-running agents."""
        if len(episodes) < 10:
            # Base case: summarize directly
            texts = [e.content for e in episodes]
            return await self.llm.generate(
                f"Summarize these episodes:\n\n{().join(texts)}"
            )
        
        
        groups = [
            episodes[i:i+] 
             i  (, (episodes), )
        ]
        summaries = []
         group  groups:
            summary =  .hierarchical_summary(group, level + )
            summaries.append(summary)
        
          .llm.generate(
            
            
        )
    
      () -> :
        
        
        sorted_eps = (episodes, key= e: e.importance, reverse=)
        
        important = [e  e  sorted_eps  e.importance > ]
        routine = [e  e  sorted_eps  e.importance <= ]
        
        result = 
        result += .join(e.content  e  important[:])
        
         routine:
            brief =  .llm.generate(
                
                
            )
            result += 
        
         result

Step 4: Memory Consolidation & GC

class MemoryConsolidator:
    """Periodically consolidate, prune, and optimize memory."""
    
    def __init__(self, memory: TieredMemory, llm, 
                 consolidation_interval: int = 3600):
        self.memory = memory
        self.llm = llm
        self.interval = consolidation_interval
        self.last_consolidation = time.time()
    
    async def consolidate_if_needed(self):
        """Run consolidation if interval has elapsed."""
        if time.time() - self.last_consolidation > self.interval:
            await self.consolidate()
            self.last_consolidation = time.time()
    
    async def consolidate(self):
        """Merge, prune, and optimize memory store."""
        
        # Phase 1: Deduplicate
        await self._deduplicate()
        
        # Phase 2: Merge related entries
        await self._merge_related()
        
        # Phase 3: Prune low-importance old entries
        await self._prune()
        
        # Phase 4: Re-index vector store
        await self._reindex()
    
    async  ():
        
        seen = ()
        unique = []
         entry  .memory.episodes:
            
            fingerprint = entry.content[:]
             fingerprint   seen:
                seen.add(fingerprint)
                unique.append(entry)
        .memory.episodes = unique
    
      ():
        
        
         collections  defaultdict
        tagged = defaultdict()
         entry  .memory.episodes:
             tag  entry.tags:
                tagged[tag].append(entry)
        
        
         tag, entries  tagged.items():
             (entries) > :
                merged =  .llm.generate(
                    
                    
                )
                
                .memory.episodes = [
                    e  e  .memory.episodes 
                     e   entries
                ]
                .memory.episodes.append(MemoryEntry(
                    content=merged,
                    importance=,
                    tags=[tag],
                    timestamp=time.time()
                ))
    
      ():
        
        now = time.time()
        day = 
        
        .memory.episodes = [
            e  e  .memory.episodes
             (e.importance >   
                (now - e.timestamp) < max_age_days * day)
        ]
        
        
         (.memory.episodes) > max_episodes:
            .memory.episodes.sort(
                key= e: (e.importance, e.timestamp),
                reverse=
            )
            .memory.episodes = .memory.episodes[:max_episodes]

Step 5: Memory Retrieval with Reranking

class MemoryRetriever:
    """Retrieve relevant memories with multi-stage ranking."""
    
    def __init__(self, vector_store, llm):
        self.vector_store = vector_store
        self.llm = llm
    
    async def retrieve(self, query: str, k: int = 10, rerank_top: int = 5):
        """Retrieve and rerank memories."""
        
        # Stage 1: Quick vector search (get more than needed)
        candidates = await self.vector_store.search(query, k=k * 3)
        
        # Stage 2: Rerank with LLM
        scored = []
        for mem in candidates:
            score = await self._relevance_score(query, mem.content)
            scored.append((score, mem))
        
        scored.sort(key=lambda x: x[0], reverse=True)
        
        # Stage 3: Return top results
        return [mem for _, mem in scored[:rerank_top]]
    
    async def _relevance_score(self, query: str, memory: str) -> float:
        """Score how relevant a memory is to the query."""
        prompt = f"""Rate the relevance of this memory to the query from 0.0 to 1.0.
Only return a number, nothing else.

Query: 
Memory: 
Relevance:"""
        
        response =  .llm.generate(prompt, temperature=)
        :
             (response.strip())
         ValueError:
               

---

Memory Budget Planning

Estimating Memory Costs

Component	Tokens/Month (100K tasks)	Cost (GPT-4 @ $0.03/K)
Context window (avg 4K tokens)	400M tokens	$12,000
Vector storage (1M embeddings)	—	~$100/mo
Summarization overhead	20M tokens	$600
Total	—	~$12,700/mo

Optimization Levers

Lever	Savings	Trade-off
Shorter context windows	40-60%	May miss relevant context
Fewer retrieved memories	20-30%	Lower recall quality
Less frequent summarization	10-20%	Staler summaries
Stricter importance thresholds	15-25%	Lose some nuance
Batch consolidation	5-10%	Delayed memory optimization

---

Trigger Phrases

Phrase	Action
"What do you remember about..."	Search semantic memory for topic
"Remember this for later"	Store with high importance
"Forget that"	Delete specific memory
"Show me your memory"	Display current working context
"Summarize the conversation"	Generate rolling summary
"Run memory consolidation"	Trigger GC and merging
"Check memory usage"	Show token consumption by tier
"Save this to long-term memory"	Promote to semantic/episodic tiers

---

Anti-Patterns

Anti-Pattern	Why It Fails	Fix
Putting everything in context	Exceeds window, loses early info	Tiered memory with summarization
No importance scoring	All memories treated equally	Score on write, prune on importance
Never consolidating	Unbounded growth, degraded retrieval	Schedule periodic consolidation
Vector search without reranking	Noisy, low-precision results	Add LLM reranking stage
Ignoring token budgets	Cost surprises, silent truncation	Track and alert on token usage
One memory config for all agents	Research agent needs differ from support	Per-agent memory configuration