session-compression

name	session-compression
description	AI session compression techniques for managing multi-turn conversations efficiently through summarization, embedding-based retrieval, and intelligent context management.
user-invocable	false
disable-model-invocation	true
progressive_disclosure	{"entry_point":["summary","when_to_use","quick_start"],"full_content":["all"]}

AI Session Compression Techniques

Summary

Compress long AI conversations to fit context windows while preserving critical information.

Session compression enables production AI applications to manage multi-turn conversations efficiently by reducing token usage by 70-95% through summarization, embedding-based retrieval, and intelligent context management. Achieve 3-20x compression ratios with minimal performance degradation.

Key Benefits:

• Cost Reduction: 80-90% token cost savings through hierarchical memory
• Performance: 2x faster responses with compressed context
• Scalability: Handle conversations exceeding 1M tokens
• Quality: Preserve critical information with <2% accuracy loss

When to Use

Use session compression when:

• Multi-turn conversations approach context window limits (>50% capacity)
• Long-running chat sessions (customer support, tutoring, code assistants)
• Token costs become significant (high-volume applications)
• Response latency increases due to large context
• Managing conversation history across multiple sessions

Don't use when:

• Short conversations (<10 turns) fitting easily in context
• Every detail must be preserved verbatim (legal, compliance)
• Single-turn or stateless interactions
• Context window usage is <30%

Ideal scenarios:

• Chatbots with 50+ turn conversations
• AI code assistants tracking long development sessions
• Customer support with multi-session ticket history
• Educational tutors with student progress tracking
• Multi-day collaborative AI workflows

Quick Start

Basic Setup with LangChain

from langchain.memory import ConversationSummaryBufferMemory
from langchain_anthropic import ChatAnthropic
from anthropic import Anthropic

# Initialize Claude client
llm = ChatAnthropic(
    model="claude-3-5-sonnet-20241022",
    api_key="your-api-key"
)

# Setup memory with automatic summarization
memory = ConversationSummaryBufferMemory(
    llm=llm,
    max_token_limit=2000,  # Summarize when exceeding this
    return_messages=True
)

# Add conversation turns
memory.save_context(
    {"input": "What's session compression?"},
    {"output": "Session compression reduces conversation token usage..."}
)

# Retrieve compressed context
context = memory.load_memory_variables({})

Progressive Compression Pattern

from anthropic import Anthropic

client = Anthropic(api_key="your-api-key")

class ProgressiveCompressor:
    def __init__(self, thresholds=[0.70, 0.85, 0.95]):
        self.thresholds = thresholds
        self.messages = []
        self.max_tokens = 200000  # Claude context window

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

        # Check if compression needed
        current_usage = self._estimate_tokens()
        usage_ratio = current_usage / self.max_tokens

        if usage_ratio >= self.thresholds[0]:
            self._compress(level=self._get_compression_level(usage_ratio))

    def _estimate_tokens(self):
        return sum(len(m["content"]) // 4 for m in self.messages)

    def _get_compression_level(self, ratio):
         i, threshold  (.thresholds):
             ratio < threshold:
                 i
         (.thresholds)

     ():
        
         level == :  
            ._remove_redundant_messages()
         level == :  
            ._summarize_old_messages(keep_recent=)
        :  
            ._summarize_old_messages(keep_recent=)

     ():
        
        
        

     ():
        
         (.messages) <= keep_recent:
            

        
        to_summarize = .messages[:-keep_recent]
        recent = .messages[-keep_recent:]

        
        conversation_text = .join([
            
             m  to_summarize
        ])

        response = client.messages.create(
            model=,
            max_tokens=,
            messages=[{
                : ,
                : 
            }]
        )

        
        summary = {
            : ,
            : 
        }
        .messages = [summary] + recent


compressor = ProgressiveCompressor()

 i  ():
    compressor.add_message(, )
    compressor.add_message(, )

Using Anthropic Prompt Caching (90% Cost Reduction)

from anthropic import Anthropic

client = Anthropic(api_key="your-api-key")

# Build context with cache control
messages = [
    {
        "role": "user",
        "content": [
            {
                "type": "text",
                "text": "Long conversation context here...",
                "cache_control": {"type": "ephemeral"}  # Cache this
            }
        ]
    },
    {
        "role": "assistant",
        "content": "Previous response..."
    },
    {
        "role": "user",
        "content": "New question"  # Not cached, changes frequently
    }
]

response = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=messages
)

# Cache hit reduces costs by 90% for cached content

---

Core Concepts

Context Windows and Token Limits

Context window: Maximum tokens an LLM can process in a single request (input + output).

Current limits (2025):

• Claude 3.5 Sonnet: 200K tokens (~150K words, ~600 pages)
• GPT-4 Turbo: 128K tokens (~96K words, ~384 pages)
• Gemini 1.5 Pro: 2M tokens (~1.5M words, ~6000 pages)

Token estimation:

• English: ~4 characters per token
• Code: ~3 characters per token
• Rule of thumb: 1 token ≈ 0.75 words

Why compression matters:

• Cost: Claude Sonnet costs $3/$15 per 1M input/output tokens
• Latency: Larger contexts increase processing time
• Quality: Excessive context can dilute attention on relevant information

Compression Ratios

Compression ratio = Original tokens / Compressed tokens

Industry benchmarks:

• Extractive summarization: 2-3x
• Abstractive summarization: 5-10x
• Hierarchical summarization: 20x+
• LLMLingua (prompt compression): 20x with 1.5% accuracy loss
• KVzip (KV cache compression): 3-4x with 2x speed improvement

Target ratios by use case:

• Customer support: 5-7x (preserve details)
• General chat: 8-12x (balance quality/efficiency)
• Code assistants: 3-5x (preserve technical accuracy)
• Long documents: 15-20x (extract key insights)

Progressive Compression Thresholds

Industry standard pattern:

Context Usage    Action                     Technique
─────────────────────────────────────────────────────────
0-70%           No compression             Store verbatim
70-85%          Light compression          Remove redundancy
85-95%          Medium compression         Summarize old messages
95-100%         Aggressive compression     Hierarchical + RAG

Implementation guidelines:

• 70% threshold: Remove duplicate/redundant messages, semantic deduplication
• 85% threshold: Summarize messages older than 20 turns, keep recent 10-15
• 95% threshold: Multi-level hierarchical summarization + vector store archival
• Emergency (100%): Drop least important messages, aggressive summarization

---

Compression Techniques

1. Summarization Techniques

1.1 Extractive Summarization

Selects key sentences/phrases without modification.

Pros: No hallucination, fast, deterministic Cons: Limited compression (2-3x), may feel disjointed Best for: Legal/compliance, short-term compression

from sklearn.feature_extraction.text import TfidfVectorizer
import numpy as np

def extractive_compress(messages: list, compression_ratio: float = 0.3):
    """Extract most important messages using TF-IDF scoring."""
    texts = [msg['content'] for msg in messages]

    # Calculate TF-IDF scores
    vectorizer = TfidfVectorizer()
    tfidf_matrix = vectorizer.fit_transform(texts)
    scores = np.array(tfidf_matrix.sum(axis=1)).flatten()

    # Select top messages
    n_keep = max(1, int(len(messages) * compression_ratio))
    top_indices = sorted(np.argsort(scores)[-n_keep:])

    return [messages[i] for i in top_indices]

1.2 Abstractive Summarization

Uses LLMs to semantically condense conversation history.

Pros: Higher compression (5-10x), coherent, synthesizes information Cons: Risk of hallucination, higher cost, less deterministic Best for: General chat, customer support, multi-session continuity

from anthropic import Anthropic

def abstractive_compress(messages: list, client: Anthropic):
    """Generate semantic summary using Claude."""
    conversation_text = "\n\n".join([
        f"{msg['role'].upper()}: {msg['content']}"
        for msg in messages
    ])

    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=500,
        messages=[{
            "role": "user",
            "content": f"""Summarize this conversation, preserving:
1. Key decisions made
2. Important context and facts
3. Unresolved questions
4. Action items

Conversation:
{conversation_text}

Summary (aim for 1/5 the original length):"""
        }]
    )

    return {
        "role": "assistant",
        "content": f"[Summary]\n{response.content[0].text}"
    }

1.3 Hierarchical Summarization (Multi-Level)

Creates summaries of summaries in a tree structure.

Pros: Extreme compression (20x+), handles 1M+ token conversations Cons: Complex implementation, multiple LLM calls, information loss accumulates Best for: Long-running conversations, multi-session applications

Architecture:

Level 0 (Raw):    [Msg1][Msg2][Msg3][Msg4][Msg5][Msg6][Msg7][Msg8]
Level 1 (Chunk):  [Summary1-2]  [Summary3-4]  [Summary5-6]  [Summary7-8]
Level 2 (Group):  [Summary1-4]              [Summary5-8]
Level 3 (Session): [Overall Session Summary]

from anthropic import Anthropic
from typing import List, Dict

class HierarchicalMemory:
    def __init__(self, client: Anthropic, chunk_size: int = 10):
        self.client = client
        self.chunk_size = chunk_size
        self.levels: List[List[Dict]] = [[]]  # Level 0 = raw messages

    def add_message(self, message: Dict):
        """Add message and trigger summarization if needed."""
        self.levels[0].append(message)

        if len(self.levels[0]) >= self.chunk_size * 2:
            self._summarize_level(0)

    def _summarize_level(self, level: int):
        """Summarize a level into the next higher level."""
        messages = self.levels[level]

        # Ensure next level exists
        while len(self.levels) <= level + 1:
            self.levels.append([])

        
        chunk = messages[:.chunk_size]
        summary = ._generate_summary(chunk, level)

        
        .levels[level + ].append(summary)
        .levels[level] = messages[.chunk_size:]

        
         (.levels[level + ]) >= .chunk_size * :
            ._summarize_level(level + )

     () -> :
        
        conversation_text = .join([
            
             msg  messages
        ])

        response = .client.messages.create(
            model=,
            max_tokens=,
            messages=[{
                : ,
                : 
            }]
        )

         {
            : ,
            : ,
            : level + 
        }

     () -> []:
        
        context = []
        token_count = 

        
         msg  (.levels[]):
            msg_tokens = (msg[]) // 
             token_count + msg_tokens > max_tokens * :
                
            context.insert(, msg)
            token_count += msg_tokens

        
         level  (, (.levels)):
             summary  .levels[level]:
                summary_tokens = (summary[]) // 
                 token_count + summary_tokens > max_tokens:
                    
                context.insert(, summary)
                token_count += summary_tokens

         context

Academic reference: "Recursively Summarizing Enables Long-Term Dialogue Memory in Large Language Models" (arXiv:2308.15022)

1.4 Rolling Summarization (Continuous)

Continuously compresses conversation with sliding window.

Pros: Low latency, predictable token usage, simple Cons: Early details over-compressed, no information recovery Best for: Real-time chat, streaming conversations

from anthropic import Anthropic

class RollingMemory:
    def __init__(self, client: Anthropic, window_size: int = 10, compress_threshold: int = 15):
        self.client = client
        self.window_size = window_size
        self.compress_threshold = compress_threshold
        self.rolling_summary = None
        self.recent_messages = []

    def add_message(self, message: dict):
        self.recent_messages.append(message)

        if len(self.recent_messages) >= self.compress_threshold:
            self._compress()

    def _compress(self):
        """Compress older messages into rolling summary."""
        messages_to_compress = self.recent_messages[:-self.window_size]

        parts = []
        if self.rolling_summary:
            parts.append(f"Existing summary:\n{self.rolling_summary}")

        parts.append("\nNew messages:\n" + "\n\n".join([
            f"{msg['role']}: {msg['content']}"
             msg  messages_to_compress
        ]))

        response = .client.messages.create(
            model=,
            max_tokens=,
            messages=[{
                : ,
                : .join(parts) + 
            }]
        )

        .rolling_summary = response.content[].text
        .recent_messages = .recent_messages[-.window_size:]

     ():
        context = []
         .rolling_summary:
            context.append({
                : ,
                : 
            })
        context.extend(.recent_messages)
         context

2. Embedding-Based Approaches

2.1 RAG (Retrieval-Augmented Generation)

Store full conversation in vector database, retrieve only relevant chunks.

Pros: Extremely scalable, no information loss, high relevance Cons: Requires vector DB infrastructure, retrieval latency Best for: Knowledge bases, customer support with large history

from anthropic import Anthropic
from openai import OpenAI
import chromadb

class RAGMemory:
    def __init__(self, anthropic_client: Anthropic, openai_client: OpenAI):
        self.anthropic = anthropic_client
        self.openai = openai_client

        # Initialize vector store
        self.chroma = chromadb.Client()
        self.collection = self.chroma.create_collection(
            name="conversation",
            metadata={"hnsw:space": "cosine"}
        )

        self.recent_messages = []
        self.recent_window = 5
        self.message_counter = 0

    def add_message(self, message: dict):
        """Add to recent memory and vector store."""
        self.recent_messages.append(message)

        if len(self.recent_messages) > self.recent_window:
            old_msg = self.recent_messages.pop(0)
            self._store_in_vectordb(old_msg)

    def _store_in_vectordb(self, message: dict):
        """Archive to vector database."""
        # Generate embedding
        response = .openai.embeddings.create(
            model=,
            =message[]
        )

        .collection.add(
            embeddings=[response.data[].embedding],
            documents=[message[]],
            metadatas=[{: message[]}],
            ids=[]
        )
        .message_counter += 

     ():
        
        context = []
        token_count = 

        
         msg  .recent_messages:
            context.append(msg)
            token_count += (msg[]) // 

        
         token_count < max_tokens:
            query_embedding = .openai.embeddings.create(
                model=,
                =query
            )

            n_results = (, (max_tokens - token_count) // )
            results = .collection.query(
                query_embeddings=[query_embedding.data[].embedding],
                n_results=n_results
            )

             i, doc  (results[][]):
                 token_count + (doc) //  > max_tokens:
                    

                metadata = results[][][i]
                context.insert(, {
                    : metadata[],
                    : 
                })
                token_count += (doc) // 

         context

Vector database options:

• ChromaDB: Embedded, easy local development
• Pinecone: Managed, 50ms p95 latency
• Weaviate: Open-source, hybrid search
• Qdrant: High performance, payload filtering

2.2 Vector Search and Clustering

Group similar messages into clusters, represent with centroids.

Pros: Reduces redundancy, identifies themes, multi-topic handling Cons: Requires sufficient data, may lose nuances Best for: Multi-topic conversations, meeting summaries

from sklearn.cluster import KMeans
from openai import OpenAI
import numpy as np

class ClusteredMemory:
    def __init__(self, openai_client: OpenAI, n_clusters: int = 5):
        self.client = openai_client
        self.n_clusters = n_clusters
        self.messages = []
        self.embeddings = []

    def add_messages(self, messages: list):
        for msg in messages:
            self.messages.append(msg)

            response = self.client.embeddings.create(
                model="text-embedding-3-small",
                input=msg['content']
            )
            self.embeddings.append(response.data[0].embedding)

    def compress_by_clustering(self):
        """Cluster messages and return representatives."""
        if len(self.messages) < self.n_clusters:
            return self.messages

        embeddings_array = np.array(self.embeddings)
        kmeans = KMeans(n_clusters=self.n_clusters, random_state=42)
        labels = kmeans.fit_predict(embeddings_array)

        
        compressed = []
         cluster_id  (.n_clusters):
            cluster_indices = np.where(labels == cluster_id)[]
            centroid = kmeans.cluster_centers_[cluster_id]
            cluster_embeddings = embeddings_array[cluster_indices]
            distances = np.linalg.norm(cluster_embeddings - centroid, axis=)
            closest_idx = cluster_indices[np.argmin(distances)]

            compressed.append({
                **.messages[closest_idx],
                : (cluster_id),
                : (cluster_indices)
            })

         compressed

2.3 Semantic Deduplication

Remove semantically similar messages that convey redundant information.

Pros: Reduces redundancy without losing unique content Cons: Requires threshold tuning, O(n²) complexity Best for: FAQ systems, repetitive conversations

from openai import OpenAI
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

class SemanticDeduplicator:
    def __init__(self, openai_client: OpenAI, similarity_threshold: float = 0.85):
        self.client = openai_client
        self.threshold = similarity_threshold

    def deduplicate(self, messages: list):
        """Remove semantically similar messages."""
        if len(messages) <= 1:
            return messages

        # Generate embeddings
        embeddings = []
        for msg in messages:
            response = self.client.embeddings.create(
                model="text-embedding-3-small",
                input=msg['content']
            )
            embeddings.append(response.data[0].embedding)

        embeddings_array = np.array(embeddings)
        similarity_matrix = cosine_similarity(embeddings_array)

        # Mark unique messages
        keep_indices = []
        for i in range(len(messages)):
            is_unique = True
            for j in keep_indices:
                if similarity_matrix[i][j] > self.threshold:
                    is_unique = 
                    

             is_unique:
                keep_indices.append(i)

         [messages[i]  i  keep_indices]

3. Token-Efficient Strategies

3.1 Message Prioritization

Assign importance scores and retain only high-priority content.

Pros: Retains most important information, flexible criteria Cons: Scoring is heuristic-based, may break flow Best for: Mixed-importance conversations, filtering noise

import re

class MessagePrioritizer:
    def score_message(self, msg: dict, index: int, total: int) -> float:
        """Calculate composite importance score."""
        scores = []

        # Length score (longer = more info)
        scores.append(min(len(msg['content']) / 500, 1.0))

        # Question score
        if msg['role'] == 'user':
            scores.append(min(msg['content'].count('?') * 0.5, 1.0))

        # Entity score (capitalized words)
        entities = len(re.findall(r'\b[A-Z][a-z]+', msg['content']))
        scores.append(min(entities / 10, 1.0))

        # Recency score (linear decay)
        scores.append(index / max(total - 1, 1))

        # Role score
        scores.append(0.6 if msg['role'] == 'user' else 0.4)

        return sum(scores) / len(scores)

    def prioritize():
        
        scored = [
            (msg, .score_message(msg, i, (messages)), i)
             i, msg  (messages)
        ]

        scored.sort(key= x: x[], reverse=)
        top_messages = scored[:target_count]
        top_messages.sort(key= x: x[])  

         [msg  msg, score, idx  top_messages]

3.2 Delta Compression

Store only changes between consecutive messages.

Pros: Highly efficient for incremental changes Cons: Reconstruction overhead, not suitable for all content Best for: Code assistants with incremental edits

import difflib

class DeltaCompressor:
    def __init__(self):
        self.base_messages = []
        self.deltas = []

    def add_message(self, message: dict):
        if not self.base_messages:
            self.base_messages.append(message)
            return

        # Find most similar previous message
        last_msg = self.base_messages[-1]

        if last_msg['role'] == message['role']:
            # Calculate delta
            diff = list(difflib.unified_diff(
                last_msg['content'].splitlines(),
                message['content'].splitlines(),
                lineterm=''
            ))

            if len('\n'.join(diff)) < len(message['content']) * 0.7:
                # Store as delta if compression achieved
                self.deltas.append({
                    'base_index': len(self.base_messages) - 1,
                    'delta': diff,
                    'role': message['role']
                })
                return

        
        .base_messages.append(message)

     ():
        
        messages = .base_messages.copy()

         delta_info  .deltas:
            base_content = messages[delta_info[]][]
            
            reconstructed = base_content  
            messages.append({
                : delta_info[],
                : reconstructed
            })

         messages

4. LangChain Memory Types

4.1 ConversationSummaryMemory

Automatically summarizes conversation as it progresses.

from langchain.memory import ConversationSummaryMemory
from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-3-5-sonnet-20241022")

memory = ConversationSummaryMemory(llm=llm)

# Add conversation
memory.save_context(
    {"input": "Hi, I'm working on a Python project"},
    {"output": "Great! How can I help with your Python project?"}
)

# Get summary
summary = memory.load_memory_variables({})
print(summary['history'])

Pros: Automatic summarization, simple API Cons: Every turn triggers LLM call Best for: Medium conversations (20-50 turns)

4.2 ConversationSummaryBufferMemory

Hybrid: Recent messages verbatim, older summarized.

from langchain.memory import ConversationSummaryBufferMemory
from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-3-5-haiku-20241022")

memory = ConversationSummaryBufferMemory(
    llm=llm,
    max_token_limit=2000,  # Summarize when exceeding
    return_messages=True
)

# Add conversation
for i in range(50):
    memory.save_context(
        {"input": f"Question {i}"},
        {"output": f"Answer {i}"}
    )

# Automatically keeps recent messages + summary of old
context = memory.load_memory_variables({})

Pros: Best balance of detail and compression Cons: Requires token limit tuning Best for: Most production applications

4.3 ConversationTokenBufferMemory

Maintains fixed token budget, drops oldest when exceeded.

from langchain.memory import ConversationTokenBufferMemory
from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-3-5-sonnet-20241022")

memory = ConversationTokenBufferMemory(
    llm=llm,
    max_token_limit=2000
)

# Simple FIFO when token limit exceeded

Pros: Predictable token usage, simple Cons: Loses old information completely Best for: Real-time chat with strict limits

4.4 VectorStoreRetrieverMemory

Stores all messages in vector database, retrieves relevant ones.

from langchain.memory import VectorStoreRetrieverMemory
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings()
vectorstore = Chroma(embedding_function=embeddings)

memory = VectorStoreRetrieverMemory(
    retriever=vectorstore.as_retriever(search_kwargs={"k": 5})
)

# Automatically retrieves most relevant context

Pros: Infinite conversation length, semantic retrieval Cons: Requires vector DB, retrieval overhead Best for: Long-running conversations, knowledge bases

5. Anthropic-Specific Patterns

5.1 Prompt Caching (90% Cost Reduction)

Cache static context to reduce token costs.

from anthropic import Anthropic

client = Anthropic(api_key="your-api-key")

# Long conversation context
conversation_history = [
    {"role": "user", "content": "Message 1"},
    {"role": "assistant", "content": "Response 1"},
    # ... many more messages
]

# Mark context for caching
messages = []
for i, msg in enumerate(conversation_history[:-1]):
    content = msg['content']

    # Add cache control to last context message
    if i == len(conversation_history) - 2:
        messages.append({
            "role": msg['role'],
            "content": [
                {
                    "type": "text",
                    "text": content,
                    "cache_control": {"type": "ephemeral"}
                }
            ]
        })
    else:
        messages.append(msg)

# Add new user message (not cached)
messages.append(conversation_history[-1])

response = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=messages
)

# Subsequent calls with same cached context cost 90% less

Cache TTL: 5 minutes Savings: 90% cost reduction for cached tokens Limits: Max 4 cache breakpoints per request Best practices:

• Cache conversation history, not current query
• Update cache when context changes significantly
• Combine with summarization for maximum efficiency

5.2 Extended Thinking for Compression Planning

Use extended thinking to plan optimal compression strategy.

from anthropic import Anthropic

client = Anthropic(api_key="your-api-key")

response = client.messages.create(
    model="claude-3-7-sonnet-20250219",
    max_tokens=16000,
    thinking={
        "type": "enabled",
        "budget_tokens": 10000
    },
    messages=[{
        "role": "user",
        "content": f"""Analyze this conversation and recommend compression:

{conversation_text}

Current token count: {current_tokens}
Target: {target_tokens}
Required compression: {compression_ratio}x

Recommend optimal strategy."""
    }]
)

# Access thinking process
thinking_content = [
    block for block in response.content
    if block.type == "thinking"
]

# Get compression recommendation
recommendation = response.content[-1].text

---

Production Patterns

Checkpointing and Persistence

Save compression state for recovery and resume.

import json
import pickle
from pathlib import Path

class PersistentMemory:
    def __init__(self, checkpoint_dir: str = "./checkpoints"):
        self.checkpoint_dir = Path(checkpoint_dir)
        self.checkpoint_dir.mkdir(exist_ok=True)
        self.memory = []
        self.summary = None

    def save_checkpoint(self, session_id: str):
        """Save current memory state."""
        checkpoint = {
            'messages': self.memory,
            'summary': self.summary,
            'timestamp': time.time()
        }

        checkpoint_file = self.checkpoint_dir / f"{session_id}.json"
        with open(checkpoint_file, 'w') as f:
            json.dump(checkpoint, f, indent=2)

    def load_checkpoint(self, session_id: str):
        """Load memory state from checkpoint."""
        checkpoint_file = self.checkpoint_dir / f"{session_id}.json"

        if checkpoint_file.exists():
            with (checkpoint_file, )  f:
                checkpoint = json.load(f)

            .memory = checkpoint[]
            .summary = checkpoint.get()
             

         

     ():
        
         (.memory) % interval == :
            .save_checkpoint(session_id)

Resume Workflows

Continue conversations across sessions.

from anthropic import Anthropic
import json

class ResumableConversation:
    def __init__(self, client: Anthropic, session_id: str):
        self.client = client
        self.session_id = session_id
        self.memory = self._load_or_create()

    def _load_or_create(self):
        """Load existing session or create new."""
        try:
            with open(f'sessions/{self.session_id}.json', 'r') as f:
                return json.load(f)
        except FileNotFoundError:
            return {
                'messages': [],
                'summary': None,
                'created_at': time.time()
            }

    def add_turn(self, user_message: str):
        """Add user message and get response."""
        # Add user message
        self.memory['messages'].append({
            'role': 'user',
            'content': user_message
        })

        # Build context (with compression)
        context = self._build_context()

        # Get response
        response = .client.messages.create(
            model=,
            max_tokens=,
            messages=context + [{
                : ,
                : user_message
            }]
        )

        
        assistant_message = response.content[].text
        .memory[].append({
            : ,
            : assistant_message
        })

        
         (.memory[]) > :
            ._compress()

        
        ._save()

         assistant_message

     ():
        
        context = []

        
         .memory[]:
            context.append({
                : ,
                : 
            })

        
        context.extend(.memory[][-:])

         context

     ():
        
         (.memory[]) < :
            

        
        to_summarize = .memory[][:-]

        
        conversation_text = .join([
            
             msg  to_summarize
        ])

        response = .client.messages.create(
            model=,
            max_tokens=,
            messages=[{
                : ,
                : 
            }]
        )

        
        .memory[] = response.content[].text
        .memory[] = .memory[][-:]

     ():
        
         (, )  f:
            json.dump(.memory, f, indent=)


client = Anthropic(api_key=)
conversation = ResumableConversation(client, session_id=)


response1 = conversation.add_turn()

response2 = conversation.add_turn()  

Hybrid Approaches (Best Practice)

Combine multiple techniques for optimal results.

from anthropic import Anthropic
from openai import OpenAI
import chromadb

class HybridMemorySystem:
    """
    Combines:
    - Rolling summarization (short-term compression)
    - RAG retrieval (long-term memory)
    - Prompt caching (cost optimization)
    - Progressive compression (adaptive behavior)
    """

    def __init__(self, anthropic_client: Anthropic, openai_client: OpenAI):
        self.anthropic = anthropic_client
        self.openai = openai_client

        # Recent messages (verbatim)
        self.recent_messages = []
        self.recent_window = 10

        # Rolling summary
        self.rolling_summary = None

        # Vector store (long-term)
        self.chroma = chromadb.Client()
        self.collection = self.chroma.create_collection(name="memory")
        self.message_counter = 0

        # Compression thresholds
        self.thresholds = {
            'light': 0.70,    # Start basic compression
            'medium': 0.85,   # Aggressive summarization
            'heavy': 0.95     # Emergency measures
        }

    def add_message(self, message: ):
        
        .recent_messages.append(message)

        
        usage_ratio = ._estimate_usage()

         usage_ratio >= .thresholds[]:
            ._emergency_compress()
         usage_ratio >= .thresholds[]:
            ._medium_compress()
         usage_ratio >= .thresholds[]:
            ._light_compress()

     ():
        
         (.recent_messages) > .recent_window * :
            
            to_archive = .recent_messages[:]
             msg  to_archive:
                ._archive_to_vectorstore(msg)

            .recent_messages = .recent_messages[:]

     ():
        
         (.recent_messages) > .recent_window:
            
            to_summarize = .recent_messages[:-.recent_window]

            summary_text = .join([
                
                 msg  to_summarize
            ])

             .rolling_summary:
                summary_text = 

            response = .anthropic.messages.create(
                model=,
                max_tokens=,
                messages=[{
                    : ,
                    : 
                }]
            )

            .rolling_summary = response.content[].text

            
             msg  to_summarize:
                ._archive_to_vectorstore(msg)

            .recent_messages = .recent_messages[-.recent_window:]

     ():
        
        
        to_archive = .recent_messages[:-]
         msg  to_archive:
            ._archive_to_vectorstore(msg)

        .recent_messages = .recent_messages[-:]

        
         .rolling_summary  (.rolling_summary) > :
            response = .anthropic.messages.create(
                model=,
                max_tokens=,
                messages=[{
                    : ,
                    : 
                }]
            )
            .rolling_summary = response.content[].text

     ():
        
        embedding_response = .openai.embeddings.create(
            model=,
            =message[]
        )

        .collection.add(
            embeddings=[embedding_response.data[].embedding],
            documents=[message[]],
            metadatas=[{: message[]}],
            ids=[]
        )
        .message_counter += 

     ():
        
        context = []
        token_count = 

        
         .rolling_summary:
            summary_msg = {
                : ,
                : [
                    {
                        : ,
                        : ,
                        : {: }  
                    }
                ]
            }
            context.append(summary_msg)
            token_count += (.rolling_summary) // 

        
         token_count < max_tokens * :
            query_embedding = .openai.embeddings.create(
                model=,
                =current_query
            )

            results = .collection.query(
                query_embeddings=[query_embedding.data[].embedding],
                n_results=
            )

             i, doc  (results[][]):
                 token_count + (doc) //  > max_tokens * :
                    

                metadata = results[][][i]
                context.append({
                    : metadata[],
                    : 
                })
                token_count += (doc) // 

        
         msg  .recent_messages:
             token_count + (msg[]) //  > max_tokens * :
                
            context.append(msg)
            token_count += (msg[]) // 

         context

     ():
        
        total_tokens = 

         .rolling_summary:
            total_tokens += (.rolling_summary) // 

         msg  .recent_messages:
            total_tokens += (msg[]) // 

         total_tokens /   


anthropic_client = Anthropic(api_key=)
openai_client = OpenAI(api_key=)

memory = HybridMemorySystem(anthropic_client, openai_client)


 i  ():
    memory.add_message({
        :   i %  ==   ,
        : 
    })


current_query = 
context = memory.get_context(current_query)


response = anthropic_client.messages.create(
    model=,
    max_tokens=,
    messages=context + [{
        : ,
        : current_query
    }]
)

---

Performance Benchmarks

Compression Efficiency

Technique	Compression Ratio	Quality Loss	Latency	Cost Impact
Extractive	2-3x	<1%	<10ms	None
Abstractive	5-10x	2-5%	1-2s	+$0.001/turn
Hierarchical	20x+	5-8%	2-5s	+$0.003/turn
LLMLingua	20x	1.5%	500ms	None
RAG	Variable	<1%	100-300ms	+$0.0005/turn
Prompt Caching	N/A	0%	0ms	-90%

Token Savings by Use Case

Customer Support (50-turn conversation):

• No compression: ~8,000 tokens/request
• Rolling summary: ~2,000 tokens/request (75% reduction)
• Hybrid (RAG + summary): ~1,500 tokens/request (81% reduction)

Code Assistant (100-turn session):

• No compression: ~25,000 tokens/request
• Hierarchical: ~5,000 tokens/request (80% reduction)
• Hybrid + caching: ~1,000 tokens/request effective (96% cost reduction)

Educational Tutor (multi-session):

• No compression: Would exceed context window
• RAG + summarization: ~3,000 tokens/request
• Infinite session length enabled

Cost Analysis

Example: Claude Sonnet pricing ($3 input, $15 output per 1M tokens)

1,000 conversations, 50 turns each:

• No compression:

  • Avg 8K tokens/request × 50K requests = 400M tokens
  • Cost: $1,200

• With rolling summarization:

  • Avg 2K tokens/request × 50K requests = 100M tokens
  • Summarization overhead: +10M tokens
  • Cost: $330 (72% savings)

• With hybrid system + caching:

  • First turn: 2K tokens (no cache)
  • Subsequent: 200 tokens effective (90% cache hit)
  • Total: ~15M tokens effective
  • Cost: $45 (96% savings)

---

Tool Recommendations

Memory Management Tools

Mem0 (Recommended for Production)

Best for: Hybrid memory systems with minimal code

from mem0 import MemoryClient

client = MemoryClient(api_key="your-mem0-key")

# Automatically handles compression, summarization, RAG
memory = client.create_memory(
    user_id="user123",
    messages=[
        {"role": "user", "content": "I'm working on a Python project"},
        {"role": "assistant", "content": "Great! What kind of project?"}
    ]
)

# Retrieve relevant context
context = client.get_memory(
    user_id="user123",
    query="What programming language am I using?"
)

Features:

• Automatic hierarchical summarization
• Built-in RAG retrieval
• Multi-user session management
• Analytics dashboard

Pricing: $0.40/1K memory operations

Zep

Best for: Low-latency production deployments**

from zep_python import ZepClient

client = ZepClient(api_key="your-zep-key")

# Add to session
client.memory.add_memory(
    session_id="session123",
    messages=[
        {"role": "user", "content": "Hello"},
        {"role": "assistant", "content": "Hi there!"}
    ]
)

# Auto-summarized retrieval
memory = client.memory.get_memory(session_id="session123")

Features:

• <100ms retrieval latency
• Automatic fact extraction
• Entity recognition
• Session management

Pricing: Open-source (self-hosted) or $0.50/1K operations (cloud)

ChromaDB

Best for: Self-hosted vector storage**

import chromadb

client = chromadb.Client()
collection = client.create_collection("conversations")

# Store embeddings
collection.add(
    documents=["Message content"],
    embeddings=[[0.1, 0.2, ...]],
    ids=["msg1"]
)

# Retrieve
results = collection.query(
    query_embeddings=[[0.1, 0.2, ...]],
    n_results=5
)

Features:

• Fully open-source
• Embedded or client-server
• Fast local development

Pricing: Free (self-hosted)

LangChain

Best for: Rapid prototyping and experimentation**

from langchain.memory import ConversationSummaryBufferMemory
from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-3-5-sonnet-20241022")
memory = ConversationSummaryBufferMemory(llm=llm, max_token_limit=2000)

Features:

• Multiple memory types
• Framework integration
• Extensive documentation

Pricing: Free (uses your LLM API costs)

Compression Libraries

LLMLingua

Best for: Extreme compression with minimal quality loss**

from llmlingua import PromptCompressor

compressor = PromptCompressor()

compressed = compressor.compress_prompt(
    context="Long conversation history...",
    instruction="Current user query",
    target_token=500
)

# Achieves 20x compression with 1.5% accuracy loss

Features:

• 20x compression ratios
• <2% quality degradation
• Fast inference (<500ms)

Pricing: Free (open-source)

---

Use Cases and Patterns

Chatbot (Customer Support)

Requirements:

• Multi-turn conversations (50-100 turns)
• Preserve customer context
• Fast response times
• Cost-efficient

Recommended approach:

• ConversationSummaryBufferMemory (LangChain)
• 70% threshold: Semantic deduplication
• 85% threshold: Rolling summarization
• Prompt caching for frequent patterns

Implementation:

from langchain.memory import ConversationSummaryBufferMemory
from langchain_anthropic import ChatAnthropic

llm = ChatAnthropic(model="claude-3-5-haiku-20241022")

memory = ConversationSummaryBufferMemory(
    llm=llm,
    max_token_limit=2000,
    return_messages=True
)

# Add customer conversation
for turn in customer_conversation:
    memory.save_context(
        {"input": turn['customer_message']},
        {"output": turn['agent_response']}
    )

# Retrieve compressed context
context = memory.load_memory_variables({})

Code Assistant

Requirements:

• Long development sessions (100+ turns)
• Preserve technical details
• Handle large code blocks
• Track incremental changes

Recommended approach:

• Hierarchical summarization for overall context
• RAG retrieval for specific code references
• Delta compression for iterative edits
• Prompt caching for system prompts

Implementation:

from anthropic import Anthropic

client = Anthropic(api_key="your-api-key")

class CodeAssistantMemory:
    def __init__(self):
        self.hierarchy = HierarchicalMemory(client, chunk_size=15)
        self.rag = RAGMemory(anthropic_client=client, openai_client=openai_client)
        self.deltas = DeltaCompressor()

    def add_interaction(self, code_change: dict):
        # Store in hierarchy
        self.hierarchy.add_message({
            'role': 'user',
            'content': code_change['description']
        })

        # Store in RAG for retrieval
        self.rag.add_message(code_change)

        # Store as delta if incremental
        if code_change.get('is_incremental'):
            self.deltas.add_message(code_change)

    def get_context(self, current_query: str):
        # Combine hierarchical summary + RAG retrieval
        summary_context = self.hierarchy.get_context(max_tokens=2000)
        rag_context = self.rag.retrieve_context(current_query, max_tokens=2000)

        return summary_context + rag_context

Educational Tutor

Requirements:

• Multi-session tracking
• Student progress persistence
• Personalized context retrieval
• Long-term knowledge retention

Recommended approach:

• VectorStoreRetrieverMemory for multi-session
• Fact extraction for student knowledge
• Progressive compression across sessions
• Resumable conversations

Implementation:

from langchain.memory import VectorStoreRetrieverMemory
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

class TutorMemory:
    def __init__(self, student_id: str):
        self.student_id = student_id

        # Vector store for all sessions
        embeddings = OpenAIEmbeddings()
        vectorstore = Chroma(
            collection_name=f"student_{student_id}",
            embedding_function=embeddings
        )

        self.memory = VectorStoreRetrieverMemory(
            retriever=vectorstore.as_retriever(search_kwargs={"k": 10})
        )

    def add_lesson_content(self, lesson: dict):
        """Add lesson interaction to student memory."""
        self.memory.save_context(
            {"input": lesson['topic']},
            {"output": lesson['explanation']}
        )

    def get_student_context(self, current_topic: str):
        """Retrieve relevant past lessons for current topic."""
        return self.memory.load_memory_variables({
            "prompt": current_topic
        })

---

Best Practices

1. Choose the Right Technique for Your Use Case

• Short conversations (<20 turns): No compression needed
• Medium conversations (20-50 turns): ConversationSummaryBufferMemory
• Long conversations (50-100 turns): Hierarchical or rolling summarization
• Very long (100+ turns): Hybrid (RAG + summarization + caching)
• Multi-session: VectorStoreRetrieverMemory or Mem0

2. Implement Progressive Compression

Don't compress aggressively from the start. Use thresholds:

• 0-70%: Store verbatim
• 70-85%: Light compression (deduplication)
• 85-95%: Medium compression (summarization)
• 95-100%: Aggressive compression (hierarchical)

3. Combine Techniques

Single-technique approaches are suboptimal. Best production systems use:

• Rolling summarization (short-term)
• RAG retrieval (long-term)
• Prompt caching (cost optimization)
• Semantic deduplication (redundancy removal)

4. Monitor Quality Metrics

Track compression impact:

• Response relevance score
• Information retention rate
• User satisfaction metrics
• Token usage reduction

5. Use Prompt Caching Strategically

Cache stable content:

• Conversation summaries
• System prompts
• Knowledge base context
• User profiles

Don't cache frequently changing content:

• Current user query
• Real-time data
• Session-specific state

6. Implement Checkpointing

Save compression state for:

• Recovery from failures
• Multi-session continuity
• Analytics and debugging
• A/B testing different strategies

7. Tune Compression Parameters

Test and optimize:

• Summary token limits
• Compression thresholds
• Retrieval result counts
• Cache TTLs
• Chunk sizes for hierarchical

8. Handle Edge Cases

Plan for:

• Very long messages (split or compress individually)
• Code blocks (preserve formatting)
• Multi-language content
• Rapidly changing context

---

Troubleshooting

Problem: Summary loses critical information

Solutions:

• Lower compression ratio (less aggressive)
• Implement importance scoring to preserve key messages
• Use extractive summarization for critical sections
• Increase summary token budget

Problem: Retrieval returns irrelevant context

Solutions:

• Improve embedding model quality
• Add metadata filtering (timestamps, topics)
• Adjust similarity threshold
• Use hybrid search (semantic + keyword)

Problem: High latency from compression

Solutions:

• Compress asynchronously (background tasks)
• Use faster models for summarization (Haiku instead of Sonnet)
• Cache summaries more aggressively
• Reduce compression frequency

Problem: Conversations still exceeding context window

Solutions:

• Implement hierarchical compression
• Archive to vector database more aggressively
• Use more aggressive compression ratios
• Consider switching to model with larger context window

Problem: High costs despite compression

Solutions:

• Implement prompt caching
• Use cheaper models for summarization (Haiku)
• Batch summarization operations
• Reduce summarization frequency

Problem: Lost conversation continuity

Solutions:

• Increase recent message window
• Include summary in every request
• Use more descriptive summaries
• Implement session resumption with context injection

---

Advanced Topics

Streaming Compression

Compress in real-time as conversation progresses:

async def streaming_compress(messages: list):
    """Compress while streaming responses."""
    compressor = ProgressiveCompressor()

    async for message in conversation_stream:
        compressor.add_message(message)

        # Compression happens asynchronously
        if compressor.should_compress():
            asyncio.create_task(compressor.compress_async())

    return compressor.get_context()

Multi-User Session Management

Handle concurrent conversations with shared context:

class MultiUserMemory:
    def __init__(self):
        self.user_sessions = {}

    def get_or_create_session(self, user_id: str):
        if user_id not in self.user_sessions:
            self.user_sessions[user_id] = HybridMemorySystem(...)
        return self.user_sessions[user_id]

    def cleanup_inactive_sessions(self, timeout: int = 3600):
        """Remove sessions inactive for > timeout seconds."""
        current_time = time.time()
        inactive = [
            user_id for user_id, session in self.user_sessions.items()
            if current_time - session.last_activity > timeout
        ]

        for user_id in inactive:
            self._archive_session(user_id)
            del self.user_sessions[user_id]

Custom Importance Scoring

Train ML models to score message importance:

from transformers import pipeline

class MLImportanceScorer:
    def __init__(self):
        # Use pre-trained classifier or fine-tune on your data
        self.classifier = pipeline(
            "text-classification",
            model="your-importance-model"
        )

    def score(self, message: dict) -> float:
        """Score message importance (0-1)."""
        result = self.classifier(message['content'])
        return result[0]['score']

Context Window Utilization Optimization

Maximize information density within token budget:

def optimize_context_allocation(
    summary_tokens: int,
    recent_tokens: int,
    retrieval_tokens: int,
    max_tokens: int
):
    """
    Optimal allocation (empirically tested):
    - 20% summary
    - 50% recent messages
    - 30% retrieved context
    """
    return {
        'summary': int(max_tokens * 0.20),
        'recent': int(max_tokens * 0.50),
        'retrieval': int(max_tokens * 0.30)
    }

---

Future Directions

Emerging Techniques (2025+)

1. Infinite Attention Mechanisms

• Models with >10M token context windows (Gemini 1.5, future Claude)
• Reduces need for compression but doesn't eliminate cost concerns

2. Learned Compression Models

• Neural networks trained to compress conversation optimally
• Maintain semantic meaning while minimizing tokens
• Examples: LLMLingua v2, PromptCompressor

3. Multimodal Session Compression

• Compress conversations with images, audio, video
• Maintain cross-modal context relationships

4. Federated Memory Systems

• Distributed compression across multiple memory stores
• Privacy-preserving compression for sensitive conversations

5. Adaptive Compression Strategies

• RL-based systems that learn optimal compression per user/domain
• Dynamic threshold adjustment based on conversation importance

---

References

Academic Papers

• "Recursively Summarizing Enables Long-Term Dialogue Memory" (arXiv:2308.15022)
• "LLMLingua: Compressing Prompts for Accelerated Inference" (arXiv:2310.05736)
• "Lost in the Middle: How Language Models Use Long Contexts" (arXiv:2307.03172)

Documentation

• Anthropic Prompt Caching
• LangChain Memory
• OpenAI Function Calling

Tools

• Mem0 - Managed memory service
• Zep - Fast memory layer
• LLMLingua - Prompt compression
• ChromaDB - Vector database

---

Last Updated: 2025-11-30 Version: 1.0.0 License: MIT