Contextual Compression in RAG: Filtering and Compressing Retrieved Chunks Before Passing to the LLM

Introduction

Here's a scenario that should feel familiar if you've been building RAG systems for a while: you retrieve five document chunks based on a user's question, stuff them all into your prompt, and watch the LLM produce an answer that buries the actual relevant detail under a wall of boilerplate. You increase k to get better recall, and the problem gets worse. The retrieved chunks are technically relevant — they came from the right document — but each one is padding a three-sentence answer with three paragraphs of context the LLM doesn't need.

This isn't a retrieval failure. Your vector search is doing exactly what it was designed to do: returning chunks that are semantically similar to the query. The problem is that semantic similarity at the chunk level doesn't guarantee that every sentence within that chunk is actually useful for answering the question. A chunk about quarterly earnings might be retrieved because it contains the phrase "revenue growth," but 80% of its content is about accounting methodology changes that have nothing to do with what the user asked.

Contextual compression is the solution to this mismatch. Instead of passing raw retrieved chunks directly to the LLM, you run a second-pass filtering and compression step that strips each chunk down to just the parts that are genuinely relevant to the query. By the end of this lesson, you'll understand why this matters at a system design level, know how to implement several different compression strategies using LangChain's built-in tools, and be able to build a custom compression pipeline tailored to your specific domain.

What you'll learn:

• Why raw chunk retrieval creates noise in your LLM context window, and how that degrades answer quality
• How LangChain's ContextualCompressionRetriever works and how to wire it to different compressor backends
• How to use an LLM-based extractor vs. an embedding-based filter, and when to choose each
• How to chain multiple compressors together in a pipeline
• How to build a custom compressor class for domain-specific filtering logic

---

Prerequisites

You should already be comfortable with:

• The basics of RAG: vector stores, embeddings, chunk retrieval with similarity search
• LangChain's Retriever interface and how to wrap vector stores
• Basic Python async patterns and environment variable management
• Making API calls to OpenAI or a similar provider

You don't need to have built a production RAG system, but you should have at least built a working prototype.

---

Why Raw Chunks Are Noisier Than They Look

Let's be precise about the problem before we solve it. When you do a similarity search, you're computing the cosine similarity between a query embedding and the embeddings of your pre-chunked documents. The embedding captures the semantic meaning of the entire chunk — it's a single vector summarizing hundreds or thousands of tokens.

This creates a structural mismatch. Imagine you have a 500-token chunk from a legal contract. The first 100 tokens define liability limits. The next 200 tokens cover indemnification clauses. The last 200 tokens list governing jurisdiction. A query like "what are the liability limits in this contract?" will correctly retrieve this chunk, because "liability limits" is semantically present in the embedding. But you're handing 500 tokens to the LLM when only 100 of them answer the question.

Now multiply that by five retrieved chunks, each with a 20% signal-to-noise ratio. Your LLM is now working through 2500 tokens to extract maybe 500 tokens of genuinely useful information. This causes real, measurable problems:

Degraded answer quality. LLMs attend to everything in their context window. Irrelevant content creates distraction, and models are known to sometimes synthesize incorrect information by blending relevant and irrelevant content inappropriately.

Wasted tokens. This is a cost and latency issue. Every unnecessary token in the prompt is money spent and milliseconds added to response time.

Context window pressure. If you're using a smaller model with a tight context window — say 8K tokens — unnecessary chunk content eats into the space you need for meaningful retrieval.

Contextual compression addresses this by treating each retrieved chunk as a candidate rather than a final answer, and applying a filter or extractor to pull out only the sentences or passages that are actually responsive to the query.

---

The ContextualCompressionRetriever Architecture

LangChain implements this pattern cleanly through the ContextualCompressionRetriever. The architecture has two components that work in sequence:

1. A base retriever — any standard retriever that returns a list of Document objects. This is your existing vector store retriever, doing what it always does.

2. A document compressor — a component that takes the query and the retrieved documents, and returns a modified (compressed) version of those documents with irrelevant content removed or reduced.

The compressor is where the interesting design decisions happen. LangChain ships with several built-in compressors, and you can write your own. Let's walk through each approach.

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

# Set up your base retriever as normal
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma(
    persist_directory="./contract_db",
    embedding_function=embeddings
)
base_retriever = vectorstore.as_retriever(search_kwargs={"k": 6})

# Create the compressor
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
compressor = LLMChainExtractor.from_llm(llm)

# Wire them together
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=base_retriever
)

Now when you call compression_retriever.invoke("What are the liability limits?"), here's what happens under the hood:

1. The base retriever runs its similarity search and returns 6 Document objects
2. The compressor receives each document along with the original query
3. For each document, the compressor extracts only the content relevant to the query
4. The compressed documents (now much smaller) are returned to your chain

The elegant thing about this design is that it's fully composable. Your existing chain doesn't need to know anything about the compression step — it just receives better-quality documents.

---

LLM-Based Extraction: The Precision Approach

The LLMChainExtractor uses a language model to read each chunk and extract only the sentences relevant to the query. It's the highest-precision approach, and the right choice when you need surgical accuracy.

Here's what the extractor is actually doing internally: it constructs a prompt along the lines of "Given the following document and the user's question, extract only the parts of the document that are relevant to answering the question. If nothing is relevant, return an empty string." The LLM then returns a compressed version of the chunk.

Let's see this in a realistic context. Suppose you're building a RAG system over a corpus of SEC filings.

import os
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor
from langchain_core.documents import Document

# Simulate what your retriever would return for a query about revenue
raw_chunks = [
    Document(
        page_content="""
        Note 3 - Revenue Recognition
        The Company recognizes revenue in accordance with ASC 606. Revenue is measured 
        as the amount of consideration the Company expects to receive in exchange for 
        transferring goods or providing services. The Company's revenue primarily consists 
        of subscription fees for its SaaS platform. For the fiscal year ended December 31, 
        2023, total net revenue was $847.3 million, representing a 23% increase over the 
        prior year. Revenue from enterprise customers grew 31% year-over-year, driven by 
        expanded seat counts and upsell of premium modules. The Company's deferred revenue 
        balance at year-end was $212.4 million, up from $178.1 million in the prior year.
        """,
        metadata={"source": "acme_10k_2023.pdf", "page": 47}
    ),
    Document(
        page_content="""
        Note 4 - Income Taxes
        The Company's effective tax rate for fiscal 2023 was 21.3%, compared to 19.8% in 
        fiscal 2022. The increase was primarily attributable to changes in the jurisdictional 
        mix of earnings and the impact of stock-based compensation deductions. The Company 
        has net operating loss carryforwards of $43.2 million related to acquisitions made 
        in prior years. Total revenue growth continues to be a strategic priority, and 
        management expects this to drive future taxable income. The valuation allowance 
        decreased by $8.4 million during the year.
        """,
        metadata={"source": "acme_10k_2023.pdf", "page": 52}
    )
]

# With LLMChainExtractor
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
compressor = LLMChainExtractor.from_llm(llm)

query = "What was the total revenue and revenue growth rate?"

# Compress each document manually so we can inspect the output
compressed = compressor.compress_documents(raw_chunks, query)

for doc in compressed:
    print(f"Source: {doc.metadata.get('source')}")
    print(f"Compressed content:\n{doc.page_content}")
    print("---")

Running this, you'd see the first chunk compressed to something like:

"For the fiscal year ended December 31, 2023, total net revenue was $847.3 million, representing a 23% increase over the prior year. Revenue from enterprise customers grew 31% year-over-year."

And the second chunk — where "revenue" appears but only incidentally — would either be returned empty or with just the one tangential sentence about revenue growth as a strategic priority. Depending on how aggressively you want to filter, you might configure the chain to drop documents that return empty or near-empty extractions.

Performance note: The LLMChainExtractor makes one LLM call per retrieved chunk. If you retrieve k=6 chunks, you're making 6 extra LLM calls before your main generation call. For low-latency applications, this is a real cost. Consider using gpt-4o-mini or a similarly fast/cheap model for the compression step rather than your most capable model.

---

Embedding-Based Filtering: The Speed Approach

If LLM-based extraction is the scalpel, embedding-based filtering is the machete. Instead of reading the document with a language model, it computes the cosine similarity between the query embedding and the embedding of each sentence within the chunk. Sentences below a similarity threshold get dropped.

LangChain provides EmbeddingsFilter for this purpose:

from langchain.retrievers.document_compressors import EmbeddingsFilter
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# Filter out documents (or sentences) with similarity below 0.76
embeddings_filter = EmbeddingsFilter(
    embeddings=embeddings,
    similarity_threshold=0.76
)

compression_retriever = ContextualCompressionRetriever(
    base_compressor=embeddings_filter,
    base_retriever=base_retriever
)

Note that the EmbeddingsFilter operates at the document level by default, not the sentence level. It filters out entire retrieved documents that fall below the threshold. This makes it useful as a post-retrieval re-ranking step — if your similarity search returns documents that are marginally relevant, the filter can eliminate them entirely before they reach the LLM.

For sentence-level filtering within documents, you need to pair this with a DocumentTransformer that splits each chunk into sentences first. More on that in the pipeline section below.

Choosing your threshold: A similarity threshold of 0.75–0.80 is a reasonable starting range for text-embedding-3-small. Too low and you're not filtering anything meaningful. Too high and you'll drop legitimate results. Always tune this against a sample of your actual queries and documents — don't trust defaults blindly.

The major advantage here is latency. An embeddings call is dramatically faster and cheaper than an LLM generation call. If you're filtering by whole document rather than compressing sentence-by-sentence, you're essentially doing a second round of similarity matching with a tighter threshold.

---

Chaining Compressors: Building a Real Pipeline

Here's where things get interesting for production use. You're not limited to a single compressor — you can chain them using DocumentCompressorPipeline. This lets you apply multiple transformations in sequence.

A typical production pipeline might look like this:

1. Split chunks into individual sentences (so compression can work at a finer granularity)
2. Filter sentences by embedding similarity (fast, cheap, removes obvious noise)
3. Extract remaining content with an LLM (precise, handles nuance, applied to a much smaller input now)

from langchain.retrievers.document_compressors import (
    DocumentCompressorPipeline,
    EmbeddingsFilter,
    LLMChainExtractor
)
from langchain_community.document_transformers import EmbeddingsRedundantFilter
from langchain.text_splitter import CharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI

# Step 1: Split documents into smaller chunks (sentences or small paragraphs)
splitter = CharacterTextSplitter(
    chunk_size=200,
    chunk_overlap=0,
    separator=". "
)

# Step 2: Filter by embedding similarity to query
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
relevance_filter = EmbeddingsFilter(
    embeddings=embeddings,
    similarity_threshold=0.75
)

# Step 3: Remove redundant content across chunks
redundancy_filter = EmbeddingsRedundantFilter(embeddings=embeddings)

# Step 4: LLM extractor as final precision pass
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
llm_extractor = LLMChainExtractor.from_llm(llm)

# Wire them into a pipeline
pipeline = DocumentCompressorPipeline(
    transformers=[
        splitter,
        relevance_filter,
        redundancy_filter,
        llm_extractor
    ]
)

compression_retriever = ContextualCompressionRetriever(
    base_compressor=pipeline,
    base_retriever=base_retriever
)

Let's trace through what happens to a retrieved chunk in this pipeline:

A 600-token chunk about a software product's pricing and features comes in. The splitter breaks it into 4–5 sentence-sized pieces. The embedding filter scores each piece against the query "what are the enterprise tier pricing tiers?" and drops the three pieces about feature descriptions that score below 0.75. The redundancy filter checks whether the two remaining pieces are near-duplicates of each other (this matters when you've retrieved multiple chunks from the same document) and deduplicates if needed. The LLM extractor does a final pass on the remaining content to pull out only the precise pricing information.

What started as a 600-token chunk might reach your final LLM as 80 targeted tokens. That's not just a cost saving — it's a fundamentally better prompt.

Watch your pipeline order. Always apply cheaper, faster transformers first. Splitting and embedding filtering are cheap; LLM extraction is expensive. Run the LLM last, on a much smaller input. Putting LLMChainExtractor first would mean passing full raw chunks to the LLM, defeating the cost-saving purpose of the pipeline.

---

Writing a Custom Compressor

The built-in compressors cover a lot of ground, but real-world RAG systems often have domain-specific requirements that the defaults don't handle. Maybe you need to filter chunks based on metadata before compressing content. Maybe you want to apply a regex pattern to extract structured data from semi-structured text. The BaseDocumentCompressor interface makes this straightforward to implement.

Let's build a compressor that's genuinely useful: one that filters chunks based on a date range extracted from metadata, then extracts relevant sentences using a simple keyword scoring approach. This avoids unnecessary LLM calls when your filtering criteria are deterministic.

from typing import Optional, Sequence
from langchain_core.documents import Document
from langchain.retrievers.document_compressors.base import BaseDocumentCompressor
from langchain_core.callbacks.manager import Callbacks
import re
from datetime import datetime


class DateAwareKeywordCompressor(BaseDocumentCompressor):
    """
    A custom compressor that:
    1. Filters documents outside a specified date range (based on metadata)
    2. Extracts sentences containing query keywords
    3. Returns only the relevant sentences as compressed documents
    """
    
    keywords: list[str] = []
    min_date: Optional[str] = None  # Format: "YYYY-MM-DD"
    max_date: Optional[str] = None
    min_sentences: int = 1

    class Config:
        arbitrary_types_allowed = True

    def _extract_keywords_from_query(self, query: str) -> list[str]:
        """
        Simple keyword extraction: lowercase tokens longer than 3 chars,
        excluding stopwords. In production, you'd use a proper NLP library.
        """
        stopwords = {
            "what", "when", "where", "which", "that", "this", "with",
            "from", "have", "been", "were", "will", "would", "could",
            "should", "does", "about", "into", "than", "then", "them"
        }
        tokens = re.findall(r'\b[a-zA-Z]{4,}\b', query.lower())
        return [t for t in tokens if t not in stopwords]

    def _is_within_date_range(self, doc: Document) -> bool:
        """Check if a document's date metadata falls within the specified range."""
        doc_date_str = doc.metadata.get("date") or doc.metadata.get("published_date")
        if not doc_date_str:
            return True  # No date metadata — don't filter it out
        
        try:
            doc_date = datetime.strptime(doc_date_str, "%Y-%m-%d")
            if self.min_date:
                min_dt = datetime.strptime(self.min_date, "%Y-%m-%d")
                if doc_date < min_dt:
                    return False
            if self.max_date:
                max_dt = datetime.strptime(self.max_date, "%Y-%m-%d")
                if doc_date > max_dt:
                    return False
        except ValueError:
            return True  # Unparseable date — don't filter
        
        return True

    def _score_sentence(self, sentence: str, keywords: list[str]) -> float:
        """Score a sentence by what fraction of keywords it contains."""
        if not keywords:
            return 1.0
        sentence_lower = sentence.lower()
        matches = sum(1 for kw in keywords if kw in sentence_lower)
        return matches / len(keywords)

    def compress_documents(
        self,
        documents: Sequence[Document],
        query: str,
        callbacks: Optional[Callbacks] = None,
    ) -> Sequence[Document]:
        
        keywords = self.keywords if self.keywords else self._extract_keywords_from_query(query)
        compressed_docs = []

        for doc in documents:
            # Step 1: Date filter
            if not self._is_within_date_range(doc):
                continue
            
            # Step 2: Split into sentences and score each
            sentences = re.split(r'(?<=[.!?])\s+', doc.page_content.strip())
            scored = [
                (sentence, self._score_sentence(sentence, keywords))
                for sentence in sentences
                if len(sentence.strip()) > 20  # Skip very short fragments
            ]
            
            # Step 3: Keep sentences with any keyword match
            relevant = [s for s, score in scored if score > 0]
            
            if len(relevant) >= self.min_sentences:
                compressed_content = " ".join(relevant)
                compressed_docs.append(
                    Document(
                        page_content=compressed_content,
                        metadata={
                            **doc.metadata,
                            "compression_method": "date_keyword_filter",
                            "original_length": len(doc.page_content),
                            "compressed_length": len(compressed_content)
                        }
                    )
                )

        return compressed_docs

Now let's use it in a real scenario — a financial news RAG system where the user only wants recent articles:

# Usage: filter to only 2024 articles, extract sentences about earnings
custom_compressor = DateAwareKeywordCompressor(
    min_date="2024-01-01",
    max_date="2024-12-31",
    min_sentences=1
)

compression_retriever = ContextualCompressionRetriever(
    base_compressor=custom_compressor,
    base_retriever=base_retriever
)

results = compression_retriever.invoke(
    "What were the quarterly earnings for tech companies in Q3 2024?"
)

for doc in results:
    print(f"Date: {doc.metadata.get('date')}")
    print(f"Compression ratio: "
          f"{doc.metadata.get('compressed_length', 0) / max(doc.metadata.get('original_length', 1), 1):.1%}")
    print(f"Content: {doc.page_content[:300]}")
    print("---")

The metadata we're tracking in the custom compressor — original_length and compressed_length — is not just for debugging. In production, you'd log these values to understand how aggressively your compression is actually working across different query types. If the compression ratio is consistently above 90% (i.e., you're retaining almost everything), your threshold is too loose. If it's consistently below 20%, you might be throwing away useful context.

---

Integrating with a Full RAG Chain

Now let's put this into a complete, working RAG chain that you could actually deploy. We'll build a question-answering system over a corpus of product documentation with a compression pipeline that's aggressive enough to meaningfully reduce noise.

import os
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import (
    DocumentCompressorPipeline,
    EmbeddingsFilter,
    LLMChainExtractor
)
from langchain_community.document_transformers import EmbeddingsRedundantFilter
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough


def build_compression_rag_chain(
    persist_directory: str,
    similarity_threshold: float = 0.76,
    retrieval_k: int = 8
):
    """
    Build a complete RAG chain with contextual compression.
    
    Args:
        persist_directory: Path to your Chroma vector store
        similarity_threshold: Embedding similarity cutoff for the filter step
        retrieval_k: Number of chunks to retrieve before compression
    """
    
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    vectorstore = Chroma(
        persist_directory=persist_directory,
        embedding_function=embeddings
    )
    
    # Base retriever — retrieve more chunks than you'd normally pass to the LLM
    # because compression will thin them out
    base_retriever = vectorstore.as_retriever(
        search_type="mmr",  # MMR adds diversity to retrieved chunks
        search_kwargs={
            "k": retrieval_k,
            "fetch_k": retrieval_k * 3,  # MMR candidate pool
            "lambda_mult": 0.7
        }
    )
    
    # Compression pipeline
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=300,
        chunk_overlap=20,
        separators=["\n\n", "\n", ". ", " "]
    )
    
    relevance_filter = EmbeddingsFilter(
        embeddings=embeddings,
        similarity_threshold=similarity_threshold
    )
    
    redundancy_filter = EmbeddingsRedundantFilter(
        embeddings=embeddings,
        similarity_threshold=0.95  # Only drop near-identical content
    )
    
    # Use a fast, cheap model for compression
    compressor_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    extractor = LLMChainExtractor.from_llm(compressor_llm)
    
    pipeline = DocumentCompressorPipeline(
        transformers=[splitter, relevance_filter, redundancy_filter, extractor]
    )
    
    compression_retriever = ContextualCompressionRetriever(
        base_compressor=pipeline,
        base_retriever=base_retriever
    )
    
    # Format documents for the prompt
    def format_docs(docs):
        if not docs:
            return "No relevant documentation found."
        
        formatted = []
        for i, doc in enumerate(docs, 1):
            source = doc.metadata.get("source", "Unknown")
            section = doc.metadata.get("section", "")
            header = f"[Source {i}: {source}"
            if section:
                header += f" — {section}"
            header += "]"
            formatted.append(f"{header}\n{doc.page_content}")
        
        return "\n\n".join(formatted)
    
    # Main generation prompt
    prompt = ChatPromptTemplate.from_messages([
        ("system", """You are a helpful technical documentation assistant. 
Answer questions based strictly on the provided documentation excerpts.
If the documentation doesn't contain enough information to answer fully, 
say so clearly rather than speculating.

When citing information, reference the source number (e.g., "According to Source 2...").
Be concise and precise."""),
        ("human", """Documentation excerpts:

{context}

Question: {question}

Answer:""")
    ])
    
    # Generation model — this can be more capable since it's only doing one call
    generation_llm = ChatOpenAI(model="gpt-4o", temperature=0.1)
    
    chain = (
        {
            "context": compression_retriever | format_docs,
            "question": RunnablePassthrough()
        }
        | prompt
        | generation_llm
        | StrOutputParser()
    )
    
    return chain, compression_retriever


# Usage
chain, retriever = build_compression_rag_chain(
    persist_directory="./product_docs_db",
    similarity_threshold=0.76,
    retrieval_k=8
)

# Test with a realistic query
query = "How do I configure webhook retry behavior and what are the backoff settings?"

# Inspect what the compression is doing before running the full chain
print("=== Compressed Documents ===")
compressed_docs = retriever.invoke(query)
for doc in compressed_docs:
    print(f"\nSource: {doc.metadata.get('source', 'unknown')}")
    print(f"Content ({len(doc.page_content)} chars): {doc.page_content}")

print("\n=== Final Answer ===")
answer = chain.invoke(query)
print(answer)

Notice the two-model architecture: gpt-4o-mini for compression (multiple calls, but cheap and fast) and gpt-4o for generation (single call, where quality matters most). This is a practical pattern for production — you don't need your most capable model to determine whether a sentence is relevant to a query.

Also notice we retrieve k=8 but expect the compression pipeline to reduce this to a much smaller effective context. If you're used to retrieving k=3 or k=4 and passing everything through, you might be surprised to find that retrieving k=8 with compression produces better answers than retrieving k=4 without it. The extra retrieval gives you better recall; the compression keeps the context clean.

---

Hands-On Exercise

Build a contextual compression RAG system over a real dataset. We'll use earnings call transcripts — a domain where verbose retrieved chunks are a real problem because transcripts contain lots of back-and-forth conversation, filler content, and repeated information.

Setup:

Download a few earnings call transcripts from public sources (Motley Fool, Seeking Alpha, or SEC EDGAR provide these). For the exercise, you can use any 3–5 transcripts from the same company across different quarters.

Part 1: Baseline — No Compression

from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
import os

# Load your transcripts
loaders = [
    TextLoader("q1_2024_transcript.txt"),
    TextLoader("q2_2024_transcript.txt"),
    TextLoader("q3_2024_transcript.txt"),
]

docs = []
for loader in loaders:
    docs.extend(loader.load())

# Standard chunking
splitter = RecursiveCharacterTextSplitter(chunk_size=800, chunk_overlap=100)
chunks = splitter.split_documents(docs)

# Build vector store
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma.from_documents(chunks, embeddings, persist_directory="./transcripts_db")
base_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# Simple chain — no compression
prompt = ChatPromptTemplate.from_template(
    "Answer this question about the earnings calls:\n\nContext:\n{context}\n\nQuestion: {question}"
)
llm = ChatOpenAI(model="gpt-4o", temperature=0)

baseline_chain = (
    {"context": base_retriever | (lambda docs: "\n\n".join(d.page_content for d in docs)),
     "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

Part 2: With Compression

Add a compression retriever using DocumentCompressorPipeline with at minimum EmbeddingsFilter and LLMChainExtractor.

Part 3: Evaluate

Run both chains against these five queries:

1. "What guidance did management give for full-year revenue?"
2. "What were the main risks mentioned by the CFO?"
3. "How did gross margin change quarter over quarter and what drove the change?"
4. "What new products or features were announced?"
5. "What questions from analysts showed the most concern?"

For each query, record:

• Total tokens in the prompt (use tiktoken to count)
• Number of retrieved documents
• Subjective answer quality (1–5 scale)
• Time to retrieve (just the retrieval step, not generation)

You'll almost certainly find that compression substantially reduces prompt token count (often 50–70%) while maintaining or improving answer quality, particularly for specific factual questions. For broader synthesis questions, the tradeoff is more nuanced — you may need to tune the similarity threshold to avoid discarding relevant context.

---

Common Mistakes & Troubleshooting

Mistake 1: Setting your similarity threshold too aggressively

Setting similarity_threshold=0.85 on EmbeddingsFilter sounds like it should give you high-quality results, but in practice it often throws away legitimate content. Embedding similarity is noisy — a sentence that directly answers a query might score 0.78 if the vocabulary doesn't overlap perfectly with the query phrasing. Start at 0.70, evaluate on 20–30 real queries, then tune upward.

Mistake 2: Not handling the empty document case

When compression is aggressive, it's entirely possible that a retrieved chunk returns empty after compression — especially with LLMChainExtractor, which returns an empty string for irrelevant documents. If your downstream chain doesn't handle empty documents gracefully, you'll get confusing errors or hallucinated responses. Always filter out empty documents:

def format_docs_safe(docs):
    valid_docs = [d for d in docs if d.page_content and d.page_content.strip()]
    if not valid_docs:
        return "No relevant context found."
    return "\n\n".join(d.page_content for d in valid_docs)

Mistake 3: Using the same chunking strategy for compression as for indexing

Your indexed chunks might be 1000 tokens each because that worked well for your embedding quality. But in the compression pipeline, splitting those same chunks into 200-token pieces for sentence-level filtering is a different operation. Tune these independently. The compression splitter should be smaller — you're trying to isolate individual ideas, not maximize embedding quality.

Mistake 4: Applying compression to every query type indiscriminately

Compression is optimized for specific, factual queries: "What is the API rate limit?" "When did the CEO join the company?" For open-ended synthesis queries — "Summarize the key themes across all Q3 earnings calls" — aggressive compression might remove important context that's not directly keyword-relevant but contributes to the synthesis. Consider routing different query types through different retriever configurations.

def smart_retriever(query: str):
    """Route to compressed or uncompressed retriever based on query type."""
    synthesis_keywords = ["summarize", "compare", "overview", "themes", "overall", "generally"]
    is_synthesis = any(kw in query.lower() for kw in synthesis_keywords)
    
    if is_synthesis:
        return base_retriever  # Don't compress synthesis queries
    else:
        return compression_retriever  # Compress specific factual queries

Mistake 5: Forgetting that compression adds latency

The LLMChainExtractor makes k extra LLM calls synchronously before your main generation call. If k=8 and each compression call takes 500ms, you've added 4 seconds to your response time. For user-facing applications, consider:

• Using async batch calls if your LLM provider supports it
• Switching to EmbeddingsFilter only (no LLM compression) for latency-sensitive paths
• Reducing k when using LLM-based compression

Debugging tip: Always inspect the compressed documents before the LLM call

The most effective debugging technique is to inspect what your compression pipeline actually produces before you run the generation step. Add logging in your format function:

import logging

def format_docs_with_logging(docs):
    logger = logging.getLogger(__name__)
    logger.info(f"Received {len(docs)} compressed documents")
    for i, doc in enumerate(docs):
        logger.info(f"Doc {i+1}: {len(doc.page_content)} chars from {doc.metadata.get('source')}")
        logger.debug(f"Doc {i+1} content preview: {doc.page_content[:200]}")
    
    return "\n\n".join(d.page_content for d in docs if d.page_content.strip())

If your chain is producing poor answers, check whether the compression step is discarding the relevant content. This is often the culprit.

---

Summary & Next Steps

Contextual compression is one of those techniques that feels like a minor optimization until you see it working on a real corpus, at which point it feels obvious that you should always be doing it. The core insight is simple: retrieval and relevance are not the same thing. A chunk can be retrieved correctly and still be mostly noise. Compression closes that gap.

Here's what you've learned in this lesson:

• The structural reason why chunk-level similarity search produces noise within retrieved documents
• How ContextualCompressionRetriever works as a composable wrapper around any base retriever
• How to use LLMChainExtractor for precision extraction and EmbeddingsFilter for fast, cheap filtering
• How to chain multiple compressors with DocumentCompressorPipeline for a multi-stage approach
• How to build a custom BaseDocumentCompressor for domain-specific filtering logic
• Common failure modes and how to debug them

Where to go next:

The natural next step is re-ranking, which is a related but distinct technique. Where compression removes irrelevant content within retrieved chunks, re-ranking reorders the retrieved chunks themselves before they reach the LLM. Cross-encoder re-rankers (like those from Cohere or using sentence-transformers locally) can dramatically improve retrieval quality before compression even runs.

After re-ranking, look into self-querying retrievers — a technique where the LLM generates structured metadata filters from natural language queries before retrieval even begins. This lets you combine semantic search with structured filtering (e.g., "only search documents from Q3 2024 in the finance department") in a way that's cleaner than post-hoc metadata filtering.

Finally, if you're deploying compression pipelines in production, start instrumenting your system with compression ratios, documents-dropped metrics, and answer quality scores. The tuning parameters — similarity threshold, chunk size in the pipeline splitter, which compressors to include — are not universally optimal. They're dataset-specific, and the only way to find the right values for your application is to measure systematically.