Memory Compression and Summarization

Duration: 15 min

Overview

As agents accumulate memories, storage grows unbounded and retrieval becomes noisy. Memory compression solves this by distilling verbose memories into concise representations that preserve essential information while reducing token count. This module covers progressive summarization, entity extraction, and hierarchical compression - techniques that keep agent memory lean and effective.

Key Concepts & Foundations

• What: Memory compression reduces the size of stored memories while preserving their informational value

• Why: Unbounded memory growth leads to slow retrieval, high costs, and context window overflow

• How: Summarize conversations, merge related memories, extract key entities, and discard redundancy

Compression Strategies

| Strategy | How It Works | Compression Ratio | |----------|-------------|-------------------| | Extractive summary | Keep key sentences verbatim | 3-5x | | Abstractive summary | Rewrite into shorter form | 5-10x | | Entity extraction | Store only entities + relationships | 10-20x | | Hierarchical | Multi-level (detail -> summary -> gist) | Variable |

When to Compress

• Conversation ends - Summarize the session into key takeaways

• Buffer overflow - Compress oldest messages when approaching limits

• Periodic maintenance - Scheduled compression of all memories older than N days

• Redundancy detected - Merge memories that say the same thing differently

Hands-On Implementation

from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class CompressedMemory:
    original_count: int
    summary: str
    entities: list[str]
    key_facts: list[str]
    created_at: str = field(default_factory=lambda: datetime.now().isoformat())
    compression_ratio: float = 1.0

class MemoryCompressor:
    """Compresses verbose memories into concise representations."""
    def extractive_summarize(self, messages: list[str], max_sentences: int = 3) -> str:
        """Keep the most information-dense sentences."""
        scored = []
        all_words = set()
        for msg in messages:
            all_words.update(msg.lower().split())
        for msg in messages:
            sentences = [s.strip() for s in msg.split(".") if s.strip()]
            for sentence in sentences:
                words = set(sentence.lower().split())
                # Score by unique word coverage
                score = len(words & all_words) / max(len(all_words), 1)
                scored.append((sentence, score))
        scored.sort(key=lambda x: x[1], reverse=True)
        top = [s for s, _ in scored[:max_sentences]]
        return ". ".join(top) + "."
    def extract_entities(self, text: str) -> list[str]:
        """Extract likely entities (capitalized words, technical terms)."""
        words = text.split()
        entities = set()
        for word in words:
            clean = word.strip(".,!?()[]")
            if clean and (clean[0].isupper() or "_" in clean or clean.endswith("()")):
                entities.add(clean)
        return sorted(entities)
    def extract_key_facts(self, messages: list[str]) -> list[str]:
        """Extract actionable facts from a conversation."""
        facts = []
        fact_indicators = ["prefer", "always", "never", "want", "need",
                          "name is", "work at", "use", "like", "hate"]
        for msg in messages:
            for indicator in fact_indicators:
                if indicator in msg.lower():
                    facts.append(msg.strip())
                    break
        return facts
    def compress(self, messages: list[str]) -> CompressedMemory:
        """Full compression pipeline."""
        original_chars = sum(len(m) for m in messages)
        summary = self.extractive_summarize(messages)
        all_text = " ".join(messages)
        entities = self.extract_entities(all_text)
        key_facts = self.extract_key_facts(messages)
        compressed_chars = len(summary) + sum(len(f) for f in key_facts)
        ratio = original_chars / max(compressed_chars, 1)
        return CompressedMemory(
            original_count=len(messages),
            summary=summary,
            entities=entities,
            key_facts=key_facts,
            compression_ratio=ratio
        )

class HierarchicalMemory:
    """Three-level memory: detail -> summary -> gist."""
    def __init__(self, detail_limit: int = 20):
        self.details: list[str] = []  # Full messages
        self.summaries: list[CompressedMemory] = []  # Compressed chunks
        self.gist: str = ""  # One-line overall summary
        self.detail_limit = detail_limit
        self.compressor = MemoryCompressor()
    def add(self, message: str):
        self.details.append(message)
        if len(self.details) >= self.detail_limit:
            self._compress_details()
    def _compress_details(self):
        """Compress detail level into a summary."""
        compressed = self.compressor.compress(self.details)
        self.summaries.append(compressed)
        self.details = []  # Clear details after compression
        # Update gist
        all_facts = []
        for s in self.summaries:
            all_facts.extend(s.key_facts)
        self.gist = f"{len(self.summaries)} sessions, {len(all_facts)} key facts stored"
    def get_context(self, max_chars: int = 2000) -> str:
        """Build context string within budget."""
        parts = []
        # Always include gist
        if self.gist:
            parts.append(f"Overall: {self.gist}")
        # Add recent details
        for msg in self.details[-5:]:
            parts.append(msg)
        # Add summary facts if space allows
        remaining = max_chars - sum(len(p) for p in parts)
        for summary in reversed(self.summaries):
            for fact in summary.key_facts:
                if remaining > len(fact):
                    parts.append(f"[past] {fact}")
                    remaining -= len(fact)
        return "\n".join(parts)

Demo
compressor = MemoryCompressor()
conversation = [
    "Hi, my name is Sarah and I work at DataFlow Inc.",
    "I prefer Python for all my data science work.",
    "I always want type hints in code examples.",
    "Can you help me understand async/await?",
    "I need to build a pipeline that processes 10K documents.",
    "We use AWS at my company, specifically S3 and Lambda.",
    "I never want to see JavaScript examples.",
    "My deadline is next Friday for the demo.",
]
compressed = compressor.compress(conversation)
print(f"Original: {len(conversation)} messages")
print(f"Summary: {compressed.summary}")
print(f"Entities: {compressed.entities}")
print(f"Key facts: {compressed.key_facts}")
print(f"Compression ratio: {compressed.compression_ratio:.1f}x")

Advanced Techniques

1. LLM-Powered Summarization - Use a cheap model (e.g., GPT-4o-mini) to generate abstractive summaries 2. Importance-Preserving Compression - Never compress high-importance memories; only compress low-priority ones 3. Semantic Deduplication - Before storing, check if a semantically similar memory already exists 4. Compression Scheduling - Run compression during idle periods, not during active conversations 5. Lossy vs. Lossless - Keep original text in cold storage; use compressed versions in hot retrieval

Quiz

Q1: Why is memory compression necessary for long-running agents?

• A) To make the agent faster at generating text

• B) To prevent unbounded memory growth that degrades retrieval and overflows context ✓

• C) To reduce the agent's intelligence

• D) To comply with data regulations

Q2: What is hierarchical memory compression?

• A) Storing memories in alphabetical order

• B) Multiple compression levels (detail -> summary -> gist) with increasing compression ✓

• C) Compressing only the first message

• D) Using multiple vector databases

Q3: When should high-importance memories be compressed?

• A) Always - treat all memories equally

• B) Never or minimally - they should be preserved at full detail ✓

• C) Only on weekends

• D) Only when storage is completely full