Introduction: Context window limits are one of the most frustrating constraints when building LLM applications. You have a 100-page document but only 8K tokens of context. You want to include conversation history but it’s eating into your prompt budget. Context compression techniques solve this by reducing the token count while preserving the information that matters. This isn’t just about summarization—it’s about intelligent extraction, selective pruning, and efficient encoding that maintains semantic fidelity. The best compression strategies are task-aware: they know what information the model needs and ruthlessly eliminate everything else. This guide covers practical techniques from simple truncation to sophisticated extractive compression, helping you fit more relevant context into limited token budgets while maintaining output quality.

Basic Compression Strategies

from dataclasses import dataclass, field
from typing import Any, Optional, List, Callable
from enum import Enum
import re

class CompressionStrategy(Enum):
    """Compression strategy types."""
    
    TRUNCATE = "truncate"
    SUMMARIZE = "summarize"
    EXTRACT = "extract"
    HYBRID = "hybrid"

@dataclass
class CompressionResult:
    """Result of context compression."""
    
    original_text: str
    compressed_text: str
    original_tokens: int
    compressed_tokens: int
    compression_ratio: float
    strategy: CompressionStrategy
    metadata: dict = field(default_factory=dict)

class TokenCounter:
    """Count tokens for text."""
    
    def __init__(self, model: str = "gpt-4"):
        self.model = model
        self._encoder = None
    
    @property
    def encoder(self):
        if self._encoder is None:
            import tiktoken
            self._encoder = tiktoken.encoding_for_model(self.model)
        return self._encoder
    
    def count(self, text: str) -> int:
        """Count tokens in text."""
        
        return len(self.encoder.encode(text))
    
    def truncate_to_tokens(self, text: str, max_tokens: int) -> str:
        """Truncate text to max tokens."""
        
        tokens = self.encoder.encode(text)
        
        if len(tokens) <= max_tokens:
            return text
        
        return self.encoder.decode(tokens[:max_tokens])

class TruncationCompressor:
    """Simple truncation-based compression."""
    
    def __init__(self, token_counter: TokenCounter):
        self.counter = token_counter
    
    def compress(
        self,
        text: str,
        max_tokens: int,
        strategy: str = "end"
    ) -> CompressionResult:
        """Compress by truncation."""
        
        original_tokens = self.counter.count(text)
        
        if original_tokens <= max_tokens:
            return CompressionResult(
                original_text=text,
                compressed_text=text,
                original_tokens=original_tokens,
                compressed_tokens=original_tokens,
                compression_ratio=1.0,
                strategy=CompressionStrategy.TRUNCATE
            )
        
        if strategy == "end":
            compressed = self._truncate_end(text, max_tokens)
        elif strategy == "start":
            compressed = self._truncate_start(text, max_tokens)
        elif strategy == "middle":
            compressed = self._truncate_middle(text, max_tokens)
        else:
            compressed = self._truncate_end(text, max_tokens)
        
        compressed_tokens = self.counter.count(compressed)
        
        return CompressionResult(
            original_text=text,
            compressed_text=compressed,
            original_tokens=original_tokens,
            compressed_tokens=compressed_tokens,
            compression_ratio=compressed_tokens / original_tokens,
            strategy=CompressionStrategy.TRUNCATE
        )
    
    def _truncate_end(self, text: str, max_tokens: int) -> str:
        """Keep beginning, truncate end."""
        
        return self.counter.truncate_to_tokens(text, max_tokens)
    
    def _truncate_start(self, text: str, max_tokens: int) -> str:
        """Keep end, truncate beginning."""
        
        tokens = self.counter.encoder.encode(text)
        kept = tokens[-max_tokens:]
        return self.counter.encoder.decode(kept)
    
    def _truncate_middle(self, text: str, max_tokens: int) -> str:
        """Keep beginning and end, truncate middle."""
        
        tokens = self.counter.encoder.encode(text)
        
        half = max_tokens // 2
        start = tokens[:half]
        end = tokens[-half:]
        
        start_text = self.counter.encoder.decode(start)
        end_text = self.counter.encoder.decode(end)
        
        return f"{start_text}\n\n[...content truncated...]\n\n{end_text}"

class SentenceCompressor:
    """Compress by selecting important sentences."""
    
    def __init__(self, token_counter: TokenCounter):
        self.counter = token_counter
    
    def compress(
        self,
        text: str,
        max_tokens: int,
        query: str = None
    ) -> CompressionResult:
        """Compress by selecting sentences."""
        
        original_tokens = self.counter.count(text)
        
        if original_tokens <= max_tokens:
            return CompressionResult(
                original_text=text,
                compressed_text=text,
                original_tokens=original_tokens,
                compressed_tokens=original_tokens,
                compression_ratio=1.0,
                strategy=CompressionStrategy.EXTRACT
            )
        
        # Split into sentences
        sentences = self._split_sentences(text)
        
        # Score sentences
        if query:
            scored = self._score_by_relevance(sentences, query)
        else:
            scored = self._score_by_position(sentences)
        
        # Select sentences within budget
        selected = self._select_within_budget(scored, max_tokens)
        
        # Reconstruct in original order
        compressed = self._reconstruct(sentences, selected)
        compressed_tokens = self.counter.count(compressed)
        
        return CompressionResult(
            original_text=text,
            compressed_text=compressed,
            original_tokens=original_tokens,
            compressed_tokens=compressed_tokens,
            compression_ratio=compressed_tokens / original_tokens,
            strategy=CompressionStrategy.EXTRACT
        )
    
    def _split_sentences(self, text: str) -> list[str]:
        """Split text into sentences."""
        
        # Simple sentence splitting
        sentences = re.split(r'(?<=[.!?])\s+', text)
        return [s.strip() for s in sentences if s.strip()]
    
    def _score_by_relevance(
        self,
        sentences: list[str],
        query: str
    ) -> list[tuple[int, str, float]]:
        """Score sentences by relevance to query."""
        
        query_words = set(query.lower().split())
        scored = []
        
        for i, sentence in enumerate(sentences):
            sentence_words = set(sentence.lower().split())
            overlap = len(query_words & sentence_words)
            score = overlap / max(len(query_words), 1)
            scored.append((i, sentence, score))
        
        return scored
    
    def _score_by_position(
        self,
        sentences: list[str]
    ) -> list[tuple[int, str, float]]:
        """Score sentences by position (beginning and end higher)."""
        
        n = len(sentences)
        scored = []
        
        for i, sentence in enumerate(sentences):
            # U-shaped scoring: high at start and end
            if i < n * 0.2:  # First 20%
                score = 1.0 - (i / (n * 0.2)) * 0.3
            elif i > n * 0.8:  # Last 20%
                score = 0.7 + ((i - n * 0.8) / (n * 0.2)) * 0.3
            else:  # Middle
                score = 0.5
            
            scored.append((i, sentence, score))
        
        return scored
    
    def _select_within_budget(
        self,
        scored: list[tuple[int, str, float]],
        max_tokens: int
    ) -> set[int]:
        """Select sentences within token budget."""
        
        # Sort by score descending
        sorted_scored = sorted(scored, key=lambda x: x[2], reverse=True)
        
        selected = set()
        current_tokens = 0
        
        for idx, sentence, score in sorted_scored:
            sentence_tokens = self.counter.count(sentence)
            
            if current_tokens + sentence_tokens <= max_tokens:
                selected.add(idx)
                current_tokens += sentence_tokens
        
        return selected
    
    def _reconstruct(
        self,
        sentences: list[str],
        selected: set[int]
    ) -> str:
        """Reconstruct text from selected sentences."""
        
        result = []
        
        for i, sentence in enumerate(sentences):
            if i in selected:
                result.append(sentence)
        
        return " ".join(result)

LLM-Based Summarization

from dataclasses import dataclass
from typing import Any, Optional, List
import asyncio

class LLMSummarizer:
    """Use LLM to summarize context."""
    
    def __init__(self, llm_client: Any, token_counter: TokenCounter):
        self.llm = llm_client
        self.counter = token_counter
    
    async def summarize(
        self,
        text: str,
        max_tokens: int,
        focus: str = None
    ) -> CompressionResult:
        """Summarize text using LLM."""
        
        original_tokens = self.counter.count(text)
        
        if original_tokens <= max_tokens:
            return CompressionResult(
                original_text=text,
                compressed_text=text,
                original_tokens=original_tokens,
                compressed_tokens=original_tokens,
                compression_ratio=1.0,
                strategy=CompressionStrategy.SUMMARIZE
            )
        
        # Calculate target length
        target_words = max_tokens * 0.75  # Rough token to word ratio
        
        prompt = self._build_prompt(text, target_words, focus)
        
        response = await self.llm.generate(prompt)
        compressed = response.strip()
        
        # Ensure within budget
        compressed_tokens = self.counter.count(compressed)
        if compressed_tokens > max_tokens:
            compressed = self.counter.truncate_to_tokens(compressed, max_tokens)
            compressed_tokens = max_tokens
        
        return CompressionResult(
            original_text=text,
            compressed_text=compressed,
            original_tokens=original_tokens,
            compressed_tokens=compressed_tokens,
            compression_ratio=compressed_tokens / original_tokens,
            strategy=CompressionStrategy.SUMMARIZE
        )
    
    def _build_prompt(
        self,
        text: str,
        target_words: float,
        focus: str = None
    ) -> str:
        """Build summarization prompt."""
        
        focus_instruction = ""
        if focus:
            focus_instruction = f"\nFocus on information relevant to: {focus}"
        
        return f"""Summarize the following text in approximately {int(target_words)} words.
Preserve key facts, names, numbers, and important details.
Write in a clear, concise style.{focus_instruction}

Text to summarize:
{text}

Summary:"""

class HierarchicalSummarizer:
    """Summarize long documents hierarchically."""
    
    def __init__(
        self,
        llm_client: Any,
        token_counter: TokenCounter,
        chunk_size: int = 2000
    ):
        self.llm = llm_client
        self.counter = token_counter
        self.chunk_size = chunk_size
    
    async def summarize(
        self,
        text: str,
        max_tokens: int,
        focus: str = None
    ) -> CompressionResult:
        """Hierarchically summarize long text."""
        
        original_tokens = self.counter.count(text)
        
        if original_tokens <= max_tokens:
            return CompressionResult(
                original_text=text,
                compressed_text=text,
                original_tokens=original_tokens,
                compressed_tokens=original_tokens,
                compression_ratio=1.0,
                strategy=CompressionStrategy.SUMMARIZE
            )
        
        # Split into chunks
        chunks = self._split_into_chunks(text)
        
        # Summarize each chunk
        chunk_summaries = await self._summarize_chunks(chunks, focus)
        
        # Combine summaries
        combined = "\n\n".join(chunk_summaries)
        
        # If still too long, summarize again
        if self.counter.count(combined) > max_tokens:
            final_summary = await self._final_summarize(combined, max_tokens, focus)
        else:
            final_summary = combined
        
        compressed_tokens = self.counter.count(final_summary)
        
        return CompressionResult(
            original_text=text,
            compressed_text=final_summary,
            original_tokens=original_tokens,
            compressed_tokens=compressed_tokens,
            compression_ratio=compressed_tokens / original_tokens,
            strategy=CompressionStrategy.SUMMARIZE,
            metadata={"chunks": len(chunks)}
        )
    
    def _split_into_chunks(self, text: str) -> list[str]:
        """Split text into chunks."""
        
        chunks = []
        current_chunk = []
        current_tokens = 0
        
        paragraphs = text.split("\n\n")
        
        for para in paragraphs:
            para_tokens = self.counter.count(para)
            
            if current_tokens + para_tokens > self.chunk_size:
                if current_chunk:
                    chunks.append("\n\n".join(current_chunk))
                current_chunk = [para]
                current_tokens = para_tokens
            else:
                current_chunk.append(para)
                current_tokens += para_tokens
        
        if current_chunk:
            chunks.append("\n\n".join(current_chunk))
        
        return chunks
    
    async def _summarize_chunks(
        self,
        chunks: list[str],
        focus: str = None
    ) -> list[str]:
        """Summarize all chunks."""
        
        tasks = []
        
        for chunk in chunks:
            task = self._summarize_chunk(chunk, focus)
            tasks.append(task)
        
        return await asyncio.gather(*tasks)
    
    async def _summarize_chunk(
        self,
        chunk: str,
        focus: str = None
    ) -> str:
        """Summarize a single chunk."""
        
        focus_instruction = ""
        if focus:
            focus_instruction = f"\nFocus on: {focus}"
        
        prompt = f"""Summarize this text concisely, preserving key information:{focus_instruction}

{chunk}

Summary:"""
        
        return await self.llm.generate(prompt)
    
    async def _final_summarize(
        self,
        text: str,
        max_tokens: int,
        focus: str = None
    ) -> str:
        """Final summarization pass."""
        
        target_words = max_tokens * 0.75
        
        focus_instruction = ""
        if focus:
            focus_instruction = f"\nFocus on: {focus}"
        
        prompt = f"""Create a final summary in approximately {int(target_words)} words.
Combine and synthesize the key points.{focus_instruction}

{text}

Final Summary:"""
        
        response = await self.llm.generate(prompt)
        
        # Ensure within budget
        if self.counter.count(response) > max_tokens:
            response = self.counter.truncate_to_tokens(response, max_tokens)
        
        return response

class QueryFocusedSummarizer:
    """Summarize with focus on answering a query."""
    
    def __init__(self, llm_client: Any, token_counter: TokenCounter):
        self.llm = llm_client
        self.counter = token_counter
    
    async def summarize(
        self,
        text: str,
        query: str,
        max_tokens: int
    ) -> CompressionResult:
        """Summarize focusing on query-relevant information."""
        
        original_tokens = self.counter.count(text)
        
        target_words = max_tokens * 0.75
        
        prompt = f"""Extract and summarize information from the text that is relevant to answering this question:

Question: {query}

Text:
{text}

Provide a focused summary (approximately {int(target_words)} words) containing only information relevant to the question:"""
        
        response = await self.llm.generate(prompt)
        compressed = response.strip()
        
        if self.counter.count(compressed) > max_tokens:
            compressed = self.counter.truncate_to_tokens(compressed, max_tokens)
        
        compressed_tokens = self.counter.count(compressed)
        
        return CompressionResult(
            original_text=text,
            compressed_text=compressed,
            original_tokens=original_tokens,
            compressed_tokens=compressed_tokens,
            compression_ratio=compressed_tokens / original_tokens,
            strategy=CompressionStrategy.SUMMARIZE,
            metadata={"query": query}
        )

Extractive Compression

from dataclasses import dataclass
from typing import Any, Optional, List
import numpy as np

class EmbeddingExtractor:
    """Extract relevant content using embeddings."""
    
    def __init__(
        self,
        embedding_model: Any,
        token_counter: TokenCounter
    ):
        self.embedder = embedding_model
        self.counter = token_counter
    
    async def extract(
        self,
        text: str,
        query: str,
        max_tokens: int
    ) -> CompressionResult:
        """Extract query-relevant content."""
        
        original_tokens = self.counter.count(text)
        
        if original_tokens <= max_tokens:
            return CompressionResult(
                original_text=text,
                compressed_text=text,
                original_tokens=original_tokens,
                compressed_tokens=original_tokens,
                compression_ratio=1.0,
                strategy=CompressionStrategy.EXTRACT
            )
        
        # Split into chunks
        chunks = self._split_into_chunks(text)
        
        # Embed query and chunks
        query_embedding = await self.embedder.embed(query)
        chunk_embeddings = await self.embedder.embed_batch([c for c, _ in chunks])
        
        # Score chunks by similarity
        scores = self._calculate_similarities(query_embedding, chunk_embeddings)
        
        # Select top chunks within budget
        selected = self._select_chunks(chunks, scores, max_tokens)
        
        compressed = "\n\n".join(selected)
        compressed_tokens = self.counter.count(compressed)
        
        return CompressionResult(
            original_text=text,
            compressed_text=compressed,
            original_tokens=original_tokens,
            compressed_tokens=compressed_tokens,
            compression_ratio=compressed_tokens / original_tokens,
            strategy=CompressionStrategy.EXTRACT
        )
    
    def _split_into_chunks(
        self,
        text: str,
        chunk_size: int = 200
    ) -> list[tuple[str, int]]:
        """Split text into chunks with positions."""
        
        chunks = []
        paragraphs = text.split("\n\n")
        
        for i, para in enumerate(paragraphs):
            if para.strip():
                chunks.append((para.strip(), i))
        
        return chunks
    
    def _calculate_similarities(
        self,
        query_embedding: np.ndarray,
        chunk_embeddings: list[np.ndarray]
    ) -> list[float]:
        """Calculate cosine similarities."""
        
        similarities = []
        
        for chunk_emb in chunk_embeddings:
            sim = np.dot(query_embedding, chunk_emb) / (
                np.linalg.norm(query_embedding) * np.linalg.norm(chunk_emb)
            )
            similarities.append(float(sim))
        
        return similarities
    
    def _select_chunks(
        self,
        chunks: list[tuple[str, int]],
        scores: list[float],
        max_tokens: int
    ) -> list[str]:
        """Select chunks within token budget."""
        
        # Sort by score
        scored = list(zip(chunks, scores))
        scored.sort(key=lambda x: x[1], reverse=True)
        
        selected = []
        current_tokens = 0
        
        for (chunk, pos), score in scored:
            chunk_tokens = self.counter.count(chunk)
            
            if current_tokens + chunk_tokens <= max_tokens:
                selected.append((chunk, pos))
                current_tokens += chunk_tokens
        
        # Sort by original position
        selected.sort(key=lambda x: x[1])
        
        return [chunk for chunk, _ in selected]

class KeyphraseExtractor:
    """Extract key phrases and sentences."""
    
    def __init__(self, token_counter: TokenCounter):
        self.counter = token_counter
    
    def extract(
        self,
        text: str,
        max_tokens: int
    ) -> CompressionResult:
        """Extract key phrases."""
        
        original_tokens = self.counter.count(text)
        
        if original_tokens <= max_tokens:
            return CompressionResult(
                original_text=text,
                compressed_text=text,
                original_tokens=original_tokens,
                compressed_tokens=original_tokens,
                compression_ratio=1.0,
                strategy=CompressionStrategy.EXTRACT
            )
        
        # Extract key sentences
        sentences = self._split_sentences(text)
        scored = self._score_sentences(sentences)
        
        # Select within budget
        selected = self._select_within_budget(scored, max_tokens)
        
        compressed = " ".join(selected)
        compressed_tokens = self.counter.count(compressed)
        
        return CompressionResult(
            original_text=text,
            compressed_text=compressed,
            original_tokens=original_tokens,
            compressed_tokens=compressed_tokens,
            compression_ratio=compressed_tokens / original_tokens,
            strategy=CompressionStrategy.EXTRACT
        )
    
    def _split_sentences(self, text: str) -> list[str]:
        """Split into sentences."""
        
        import re
        sentences = re.split(r'(?<=[.!?])\s+', text)
        return [s.strip() for s in sentences if s.strip()]
    
    def _score_sentences(
        self,
        sentences: list[str]
    ) -> list[tuple[str, float, int]]:
        """Score sentences by importance."""
        
        # Calculate word frequencies
        all_words = []
        for sentence in sentences:
            words = sentence.lower().split()
            all_words.extend(words)
        
        word_freq = {}
        for word in all_words:
            word_freq[word] = word_freq.get(word, 0) + 1
        
        # Score sentences
        scored = []
        for i, sentence in enumerate(sentences):
            words = sentence.lower().split()
            
            if not words:
                continue
            
            # TF-based score
            score = sum(word_freq.get(w, 0) for w in words) / len(words)
            
            # Position bonus
            if i < len(sentences) * 0.2:
                score *= 1.2
            elif i > len(sentences) * 0.8:
                score *= 1.1
            
            # Length penalty for very short sentences
            if len(words) < 5:
                score *= 0.8
            
            scored.append((sentence, score, i))
        
        return scored
    
    def _select_within_budget(
        self,
        scored: list[tuple[str, float, int]],
        max_tokens: int
    ) -> list[str]:
        """Select sentences within budget."""
        
        # Sort by score
        sorted_scored = sorted(scored, key=lambda x: x[1], reverse=True)
        
        selected = []
        current_tokens = 0
        
        for sentence, score, pos in sorted_scored:
            tokens = self.counter.count(sentence)
            
            if current_tokens + tokens <= max_tokens:
                selected.append((sentence, pos))
                current_tokens += tokens
        
        # Sort by position
        selected.sort(key=lambda x: x[1])
        
        return [s for s, _ in selected]

class LLMExtractor:
    """Use LLM to extract key information."""
    
    def __init__(self, llm_client: Any, token_counter: TokenCounter):
        self.llm = llm_client
        self.counter = token_counter
    
    async def extract(
        self,
        text: str,
        max_tokens: int,
        extraction_type: str = "key_points"
    ) -> CompressionResult:
        """Extract key information using LLM."""
        
        original_tokens = self.counter.count(text)
        
        target_words = max_tokens * 0.75
        
        if extraction_type == "key_points":
            prompt = self._key_points_prompt(text, target_words)
        elif extraction_type == "facts":
            prompt = self._facts_prompt(text, target_words)
        elif extraction_type == "entities":
            prompt = self._entities_prompt(text, target_words)
        else:
            prompt = self._key_points_prompt(text, target_words)
        
        response = await self.llm.generate(prompt)
        compressed = response.strip()
        
        if self.counter.count(compressed) > max_tokens:
            compressed = self.counter.truncate_to_tokens(compressed, max_tokens)
        
        compressed_tokens = self.counter.count(compressed)
        
        return CompressionResult(
            original_text=text,
            compressed_text=compressed,
            original_tokens=original_tokens,
            compressed_tokens=compressed_tokens,
            compression_ratio=compressed_tokens / original_tokens,
            strategy=CompressionStrategy.EXTRACT,
            metadata={"extraction_type": extraction_type}
        )
    
    def _key_points_prompt(self, text: str, target_words: float) -> str:
        return f"""Extract the key points from this text in approximately {int(target_words)} words.
List the most important information as concise bullet points.

Text:
{text}

Key Points:"""
    
    def _facts_prompt(self, text: str, target_words: float) -> str:
        return f"""Extract factual information from this text in approximately {int(target_words)} words.
Include names, dates, numbers, and verifiable facts.

Text:
{text}

Facts:"""
    
    def _entities_prompt(self, text: str, target_words: float) -> str:
        return f"""Extract named entities and their relationships from this text in approximately {int(target_words)} words.
Include people, organizations, locations, and their connections.

Text:
{text}

Entities and Relationships:"""

Hybrid Compression Pipeline

from dataclasses import dataclass
from typing import Any, Optional, List, Callable
import asyncio

@dataclass
class CompressionConfig:
    """Configuration for compression pipeline."""
    
    max_tokens: int
    strategy: CompressionStrategy = CompressionStrategy.HYBRID
    query: str = None
    preserve_structure: bool = True
    min_compression_ratio: float = 0.1

class HybridCompressor:
    """Combine multiple compression strategies."""
    
    def __init__(
        self,
        llm_client: Any,
        embedding_model: Any,
        token_counter: TokenCounter
    ):
        self.llm = llm_client
        self.embedder = embedding_model
        self.counter = token_counter
        
        # Initialize compressors
        self.truncator = TruncationCompressor(token_counter)
        self.sentence_compressor = SentenceCompressor(token_counter)
        self.summarizer = LLMSummarizer(llm_client, token_counter)
        self.extractor = EmbeddingExtractor(embedding_model, token_counter)
    
    async def compress(
        self,
        text: str,
        config: CompressionConfig
    ) -> CompressionResult:
        """Compress using hybrid strategy."""
        
        original_tokens = self.counter.count(text)
        
        if original_tokens <= config.max_tokens:
            return CompressionResult(
                original_text=text,
                compressed_text=text,
                original_tokens=original_tokens,
                compressed_tokens=original_tokens,
                compression_ratio=1.0,
                strategy=config.strategy
            )
        
        # Determine compression approach based on ratio needed
        target_ratio = config.max_tokens / original_tokens
        
        if target_ratio > 0.7:
            # Light compression: sentence selection
            result = self.sentence_compressor.compress(
                text, config.max_tokens, config.query
            )
        elif target_ratio > 0.3:
            # Medium compression: extractive + light summarization
            result = await self._medium_compression(text, config)
        else:
            # Heavy compression: full summarization
            result = await self._heavy_compression(text, config)
        
        return result
    
    async def _medium_compression(
        self,
        text: str,
        config: CompressionConfig
    ) -> CompressionResult:
        """Medium compression using extraction."""
        
        if config.query:
            # Query-focused extraction
            result = await self.extractor.extract(
                text, config.max_tokens, config.query
            )
        else:
            # General sentence selection
            result = self.sentence_compressor.compress(
                text, config.max_tokens
            )
        
        return result
    
    async def _heavy_compression(
        self,
        text: str,
        config: CompressionConfig
    ) -> CompressionResult:
        """Heavy compression using summarization."""
        
        result = await self.summarizer.summarize(
            text, config.max_tokens, config.query
        )
        
        return result

class AdaptiveCompressor:
    """Adaptively choose compression strategy."""
    
    def __init__(
        self,
        compressors: dict[str, Any],
        token_counter: TokenCounter
    ):
        self.compressors = compressors
        self.counter = token_counter
    
    async def compress(
        self,
        text: str,
        max_tokens: int,
        context: dict = None
    ) -> CompressionResult:
        """Adaptively compress based on content."""
        
        original_tokens = self.counter.count(text)
        
        if original_tokens <= max_tokens:
            return CompressionResult(
                original_text=text,
                compressed_text=text,
                original_tokens=original_tokens,
                compressed_tokens=original_tokens,
                compression_ratio=1.0,
                strategy=CompressionStrategy.HYBRID
            )
        
        # Analyze content
        content_type = self._analyze_content(text)
        
        # Choose strategy
        strategy = self._choose_strategy(content_type, context)
        
        # Apply compression
        compressor = self.compressors.get(strategy)
        
        if hasattr(compressor, 'compress'):
            if asyncio.iscoroutinefunction(compressor.compress):
                result = await compressor.compress(text, max_tokens)
            else:
                result = compressor.compress(text, max_tokens)
        else:
            # Fallback to truncation
            result = TruncationCompressor(self.counter).compress(text, max_tokens)
        
        return result
    
    def _analyze_content(self, text: str) -> str:
        """Analyze content type."""
        
        # Check for structured content
        if text.count("\n") > 10 and text.count(":") > 5:
            return "structured"
        
        # Check for narrative content
        sentences = text.split(".")
        avg_sentence_len = sum(len(s.split()) for s in sentences) / max(len(sentences), 1)
        
        if avg_sentence_len > 15:
            return "narrative"
        
        # Check for technical content
        technical_terms = ["function", "class", "import", "def", "return", "if", "for"]
        if any(term in text.lower() for term in technical_terms):
            return "technical"
        
        return "general"
    
    def _choose_strategy(
        self,
        content_type: str,
        context: dict = None
    ) -> str:
        """Choose compression strategy."""
        
        if context and context.get("query"):
            return "extractive"
        
        if content_type == "structured":
            return "extractive"
        elif content_type == "narrative":
            return "summarize"
        elif content_type == "technical":
            return "extractive"
        else:
            return "sentence"

class CompressionPipeline:
    """Pipeline for multi-stage compression."""
    
    def __init__(self, token_counter: TokenCounter):
        self.counter = token_counter
        self.stages: list[Callable] = []
    
    def add_stage(self, compressor: Callable):
        """Add compression stage."""
        
        self.stages.append(compressor)
    
    async def compress(
        self,
        text: str,
        max_tokens: int
    ) -> CompressionResult:
        """Run compression pipeline."""
        
        original_tokens = self.counter.count(text)
        current_text = text
        
        for stage in self.stages:
            current_tokens = self.counter.count(current_text)
            
            if current_tokens <= max_tokens:
                break
            
            if asyncio.iscoroutinefunction(stage):
                result = await stage(current_text, max_tokens)
            else:
                result = stage(current_text, max_tokens)
            
            if hasattr(result, 'compressed_text'):
                current_text = result.compressed_text
            else:
                current_text = result
        
        compressed_tokens = self.counter.count(current_text)
        
        return CompressionResult(
            original_text=text,
            compressed_text=current_text,
            original_tokens=original_tokens,
            compressed_tokens=compressed_tokens,
            compression_ratio=compressed_tokens / original_tokens,
            strategy=CompressionStrategy.HYBRID
        )

Production Compression Service

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional, List

app = FastAPI()

class CompressRequest(BaseModel):
    text: str
    max_tokens: int
    strategy: str = "hybrid"
    query: Optional[str] = None

class CompressResponse(BaseModel):
    compressed_text: str
    original_tokens: int
    compressed_tokens: int
    compression_ratio: float
    strategy: str

# Initialize components
token_counter = TokenCounter()
truncator = TruncationCompressor(token_counter)
sentence_compressor = SentenceCompressor(token_counter)

@app.post("/v1/compress", response_model=CompressResponse)
async def compress_text(request: CompressRequest) -> CompressResponse:
    """Compress text."""
    
    if request.strategy == "truncate":
        result = truncator.compress(request.text, request.max_tokens)
    elif request.strategy == "sentence":
        result = sentence_compressor.compress(
            request.text, request.max_tokens, request.query
        )
    else:
        # Default to sentence compression
        result = sentence_compressor.compress(
            request.text, request.max_tokens, request.query
        )
    
    return CompressResponse(
        compressed_text=result.compressed_text,
        original_tokens=result.original_tokens,
        compressed_tokens=result.compressed_tokens,
        compression_ratio=result.compression_ratio,
        strategy=result.strategy.value
    )

@app.post("/v1/compress/batch")
async def compress_batch(texts: List[str], max_tokens: int) -> List[CompressResponse]:
    """Compress multiple texts."""
    
    results = []
    
    for text in texts:
        result = sentence_compressor.compress(text, max_tokens)
        results.append(CompressResponse(
            compressed_text=result.compressed_text,
            original_tokens=result.original_tokens,
            compressed_tokens=result.compressed_tokens,
            compression_ratio=result.compression_ratio,
            strategy=result.strategy.value
        ))
    
    return results

@app.get("/v1/tokens/count")
async def count_tokens(text: str) -> dict:
    """Count tokens in text."""
    
    count = token_counter.count(text)
    return {"tokens": count, "characters": len(text)}

@app.get("/health")
async def health():
    return {"status": "healthy"}

References

• LLMLingua: https://github.com/microsoft/LLMLingua
• tiktoken: https://github.com/openai/tiktoken
• LangChain Text Splitters: https://python.langchain.com/docs/modules/data_connection/document_transformers/
• Sentence Transformers: https://www.sbert.net/

Conclusion

Context compression is about making intelligent trade-offs between information density and token budget. Start with the simplest approach that works: truncation is fast and predictable, sentence selection preserves key content, and summarization creates the most compact representations. The best strategy depends on your use case—query-focused extraction works well for RAG systems where you know what information you need, while hierarchical summarization handles long documents that need holistic understanding. Hybrid approaches combine multiple strategies: extract relevant sections first, then summarize if still over budget. Always measure compression quality, not just compression ratio—a 10x compression that loses critical information is worse than a 2x compression that preserves it. For production systems, cache compressed versions of frequently-accessed documents and use adaptive strategies that choose compression methods based on content type and query context. The goal isn't minimal tokens—it's maximum relevant information per token.