Compression contextuelle : Extraire l'essentiel des documents

La compression contextuelle filtre et condense les documents récupérés pour ne garder que les passages directement pertinents pour la requête. Au lieu de passer des chunks entiers au LLM, cette technique extrait les phrases clés, réduisant le bruit et optimisant les coûts. Ce guide explore les méthodes de compression et leur intégration dans un pipeline RAG.

Pourquoi compresser le contexte ?

Les chunks récupérés contiennent souvent du contenu superflu :

Chunk original (500 tokens) :
"Notre entreprise a été fondée en 2010. Nous avons commencé avec 3 employés.
Aujourd'hui, nous comptons plus de 200 collaborateurs. Concernant notre politique
de retour, vous disposez de 30 jours pour retourner un produit non utilisé dans
son emballage d'origine. Les frais de retour sont à votre charge sauf en cas de
produit défectueux. Nous sommes présents dans 15 pays européens..."

Requête : "Quel est le délai de retour ?"

Après compression (50 tokens) :
"Vous disposez de 30 jours pour retourner un produit non utilisé dans son
emballage d'origine."

Bénéfices de la compression

La compression améliore la qualité car elle réduit le bruit qui peut distraire le LLM.

Méthodes de compression

1. Compression par LLM

Le LLM identifie et extrait les passages pertinents :

DEVELOPERpython
from openai import OpenAI

class LLMContextCompressor:
    def __init__(self, model: str = "gpt-4o-mini"):
        self.client = OpenAI()
        self.model = model

    def compress(self, query: str, documents: list[str]) -> list[str]:
        compressed = []

        for doc in documents:
            prompt = f"""Extrait uniquement les phrases directement pertinentes pour répondre à la question.
Si aucune partie n'est pertinente, réponds "NON_PERTINENT".

Question : {query}

Document :
{doc}

Phrases pertinentes :"""

            response = self.client.chat.completions.create(
                model=self.model,
                messages=[{"role": "user", "content": prompt}],
                temperature=0,
                max_tokens=500
            )

            content = response.choices[0].message.content.strip()
            if content != "NON_PERTINENT":
                compressed.append(content)

        return compressed


# Exemple
compressor = LLMContextCompressor()
docs = [
    "Notre société existe depuis 20 ans. Politique de retour : 30 jours maximum. Nous avons 500 employés.",
    "La livraison est gratuite. Retours acceptés sous 14 jours ouvrés. Satisfaction garantie."
]
compressed = compressor.compress("Quel est le délai de retour ?", docs)
# ["Politique de retour : 30 jours maximum.", "Retours acceptés sous 14 jours ouvrés."]

2. Compression par extraction de phrases

Plus rapide et sans coût API, utilise la similarité sémantique :

DEVELOPERpython
from sentence_transformers import SentenceTransformer
import numpy as np
import nltk

class SentenceExtractor:
    def __init__(self, model_name: str = "BAAI/bge-m3"):
        self.model = SentenceTransformer(model_name)
        nltk.download('punkt', quiet=True)

    def compress(
        self,
        query: str,
        documents: list[str],
        top_k_sentences: int = 3,
        min_similarity: float = 0.5
    ) -> list[str]:
        # Encoder la requête
        query_embedding = self.model.encode(query)

        all_relevant_sentences = []

        for doc in documents:
            # Découper en phrases
            sentences = nltk.sent_tokenize(doc, language='french')
            if not sentences:
                continue

            # Encoder les phrases
            sentence_embeddings = self.model.encode(sentences)

            # Calculer les similarités
            similarities = np.dot(sentence_embeddings, query_embedding) / (
                np.linalg.norm(sentence_embeddings, axis=1) * np.linalg.norm(query_embedding)
            )

            # Sélectionner les phrases pertinentes
            for sentence, sim in zip(sentences, similarities):
                if sim >= min_similarity:
                    all_relevant_sentences.append((sentence, sim))

        # Trier par similarité et prendre les top_k
        sorted_sentences = sorted(all_relevant_sentences, key=lambda x: x[1], reverse=True)
        return [s[0] for s in sorted_sentences[:top_k_sentences]]


# Exemple
extractor = SentenceExtractor()
result = extractor.compress(
    query="Comment retourner un produit ?",
    documents=docs,
    top_k_sentences=3
)

3. Compression par reranking + filtrage

Utilise un cross-encoder pour scorer et filtrer :

DEVELOPERpython
from sentence_transformers import CrossEncoder

class RerankerCompressor:
    def __init__(self, model_name: str = "cross-encoder/ms-marco-MiniLM-L-6-v2"):
        self.reranker = CrossEncoder(model_name)

    def compress(
        self,
        query: str,
        documents: list[str],
        threshold: float = 0.5,
        max_passages: int = 5
    ) -> list[dict]:
        # Découper chaque document en passages
        passages = []
        for doc_idx, doc in enumerate(documents):
            for para in doc.split('\n\n'):
                if len(para.strip()) > 50:  # Ignorer les paragraphes trop courts
                    passages.append({
                        "text": para.strip(),
                        "doc_idx": doc_idx
                    })

        if not passages:
            return []

        # Scorer tous les passages
        pairs = [[query, p["text"]] for p in passages]
        scores = self.reranker.predict(pairs)

        # Filtrer et trier
        for passage, score in zip(passages, scores):
            passage["score"] = float(score)

        filtered = [p for p in passages if p["score"] >= threshold]
        sorted_passages = sorted(filtered, key=lambda x: x["score"], reverse=True)

        return sorted_passages[:max_passages]


# Exemple
compressor = RerankerCompressor()
results = compressor.compress(
    query="Politique de remboursement",
    documents=docs,
    threshold=0.3
)
for r in results:
    print(f"Score: {r['score']:.3f} - {r['text'][:100]}...")

4. Compression par résumé

Pour les documents très longs, générer un résumé ciblé :

DEVELOPERpython
class SummaryCompressor:
    def __init__(self):
        self.client = OpenAI()

    def compress(
        self,
        query: str,
        documents: list[str],
        max_summary_length: int = 200
    ) -> str:
        combined_docs = "\n\n---\n\n".join(documents)

        prompt = f"""Génère un résumé concis qui répond à la question suivante.
Inclus uniquement les informations pertinentes pour la question.
Maximum {max_summary_length} mots.

Question : {query}

Documents :
{combined_docs}

Résumé pertinent :"""

        response = self.client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            temperature=0,
            max_tokens=max_summary_length * 2
        )

        return response.choices[0].message.content

Architecture du pipeline complet

DEVELOPERpython
class ContextualCompressionRetriever:
    def __init__(
        self,
        base_retriever,
        compression_method: str = "reranker",  # "llm", "sentence", "reranker", "summary"
        **compression_kwargs
    ):
        self.retriever = base_retriever
        self.compression_method = compression_method
        self.compression_kwargs = compression_kwargs

        # Initialiser le compresseur approprié
        if compression_method == "llm":
            self.compressor = LLMContextCompressor()
        elif compression_method == "sentence":
            self.compressor = SentenceExtractor()
        elif compression_method == "reranker":
            self.compressor = RerankerCompressor()
        elif compression_method == "summary":
            self.compressor = SummaryCompressor()

    def search(self, query: str, top_k: int = 5) -> dict:
        # 1. Récupération initiale (plus de documents que nécessaire)
        initial_results = self.retriever.search(query, top_k=top_k * 2)
        documents = [r["content"] for r in initial_results]

        # 2. Compression
        compressed = self.compressor.compress(query, documents, **self.compression_kwargs)

        # 3. Calculer les métriques
        original_tokens = sum(len(d.split()) for d in documents)
        compressed_tokens = (
            sum(len(c.split()) for c in compressed) if isinstance(compressed, list)
            else len(compressed.split())
        )

        return {
            "compressed_context": compressed,
            "original_docs": initial_results,
            "compression_ratio": 1 - (compressed_tokens / original_tokens) if original_tokens > 0 else 0,
            "original_tokens": original_tokens,
            "compressed_tokens": compressed_tokens
        }

Compression adaptative

Adapter la méthode selon le contexte :

DEVELOPERpython
class AdaptiveCompressor:
    def __init__(self):
        self.llm_compressor = LLMContextCompressor()
        self.sentence_extractor = SentenceExtractor()
        self.reranker = RerankerCompressor()

    def compress(
        self,
        query: str,
        documents: list[str],
        budget_tokens: int = 1000,
        quality_priority: bool = False
    ) -> list[str]:
        total_tokens = sum(len(d.split()) for d in documents)

        # Si le contexte est petit, pas de compression
        if total_tokens <= budget_tokens:
            return documents

        # Choisir la méthode selon les contraintes
        compression_needed = 1 - (budget_tokens / total_tokens)

        if quality_priority or compression_needed > 0.7:
            # Compression agressive → LLM
            return self.llm_compressor.compress(query, documents)

        elif compression_needed > 0.4:
            # Compression moyenne → Reranker
            results = self.reranker.compress(query, documents)
            return [r["text"] for r in results]

        else:
            # Compression légère → Extraction de phrases
            return self.sentence_extractor.compress(
                query, documents,
                top_k_sentences=int(budget_tokens / 20)  # ~20 tokens/phrase
            )

Compression avec préservation des sources

Garder la traçabilité pour les citations :

DEVELOPERpython
class SourcePreservingCompressor:
    def __init__(self):
        self.extractor = SentenceExtractor()

    def compress(
        self,
        query: str,
        documents: list[dict]  # {"content": str, "source": str, "metadata": dict}
    ) -> list[dict]:
        compressed_with_sources = []

        for doc in documents:
            # Extraire les phrases pertinentes
            sentences = nltk.sent_tokenize(doc["content"], language='french')
            if not sentences:
                continue

            query_emb = self.extractor.model.encode(query)
            sentence_embs = self.extractor.model.encode(sentences)

            similarities = np.dot(sentence_embs, query_emb) / (
                np.linalg.norm(sentence_embs, axis=1) * np.linalg.norm(query_emb)
            )

            # Garder les phrases pertinentes avec leur source
            for sentence, sim in zip(sentences, similarities):
                if sim > 0.5:
                    compressed_with_sources.append({
                        "text": sentence,
                        "source": doc["source"],
                        "metadata": doc["metadata"],
                        "relevance_score": float(sim)
                    })

        # Trier par pertinence
        return sorted(compressed_with_sources, key=lambda x: x["relevance_score"], reverse=True)


# Utilisation pour les citations
def format_context_with_citations(compressed_results: list[dict]) -> str:
    context_parts = []
    for i, result in enumerate(compressed_results, 1):
        context_parts.append(f"[{i}] {result['text']}")

    return "\n".join(context_parts)

def format_sources(compressed_results: list[dict]) -> str:
    sources = []
    for i, result in enumerate(compressed_results, 1):
        sources.append(f"[{i}] {result['source']}")
    return "\n".join(sources)

Évaluation de la compression

DEVELOPERpython
class CompressionEvaluator:
    def __init__(self):
        self.embedder = SentenceTransformer("BAAI/bge-m3")

    def evaluate(
        self,
        query: str,
        original_docs: list[str],
        compressed: list[str],
        ground_truth_answer: str = None
    ) -> dict:
        # 1. Ratio de compression
        original_tokens = sum(len(d.split()) for d in original_docs)
        compressed_tokens = sum(len(c.split()) for c in compressed)
        compression_ratio = 1 - (compressed_tokens / original_tokens)

        # 2. Préservation de l'information (via similarité)
        original_combined = " ".join(original_docs)
        compressed_combined = " ".join(compressed)

        orig_emb = self.embedder.encode(original_combined)
        comp_emb = self.embedder.encode(compressed_combined)

        information_preservation = np.dot(orig_emb, comp_emb) / (
            np.linalg.norm(orig_emb) * np.linalg.norm(comp_emb)
        )

        # 3. Pertinence par rapport à la requête
        query_emb = self.embedder.encode(query)
        query_relevance = np.dot(comp_emb, query_emb) / (
            np.linalg.norm(comp_emb) * np.linalg.norm(query_emb)
        )

        # 4. Score combiné (équilibre compression et qualité)
        quality_score = 0.6 * information_preservation + 0.4 * query_relevance
        efficiency_score = compression_ratio
        combined_score = 0.5 * quality_score + 0.5 * efficiency_score

        return {
            "compression_ratio": compression_ratio,
            "information_preservation": float(information_preservation),
            "query_relevance": float(query_relevance),
            "quality_score": float(quality_score),
            "efficiency_score": efficiency_score,
            "combined_score": float(combined_score),
            "original_tokens": original_tokens,
            "compressed_tokens": compressed_tokens
        }

Optimisation des coûts

DEVELOPERpython
class CostOptimizedCompressor:
    def __init__(self, max_llm_calls_per_minute: int = 60):
        self.local_compressor = SentenceExtractor()
        self.llm_compressor = LLMContextCompressor()
        self.llm_calls = []
        self.max_calls = max_llm_calls_per_minute

    def compress(self, query: str, documents: list[str]) -> list[str]:
        # D'abord, compression locale rapide
        local_compressed = self.local_compressor.compress(
            query, documents, top_k_sentences=10
        )

        # Si la compression locale est suffisante, pas besoin du LLM
        total_tokens = sum(len(s.split()) for s in local_compressed)
        if total_tokens <= 500:
            return local_compressed

        # Vérifier le rate limit
        self._clean_old_calls()
        if len(self.llm_calls) >= self.max_calls:
            # Rate limit atteint, retourner la compression locale
            return local_compressed[:5]

        # Utiliser le LLM pour affiner
        self.llm_calls.append(time.time())
        return self.llm_compressor.compress(query, local_compressed)

    def _clean_old_calls(self):
        cutoff = time.time() - 60
        self.llm_calls = [t for t in self.llm_calls if t > cutoff]

Prochaines étapes

La compression contextuelle optimise qualité et coûts de votre RAG. Pour aller plus loin :

• Generation LLM RAG - Optimiser la génération
• Ensemble Retrieval - Combiner plusieurs retrievers
• Fondamentaux du Retrieval - Vue d'ensemble

Compression intelligente avec Ailog

Ailog implémente la compression contextuelle automatiquement :

• Compression adaptative selon la complexité de la requête
• Préservation des sources pour les citations
• Optimisation des coûts avec compression locale prioritaire
• Monitoring intégré du ratio compression/qualité

Testez gratuitement et réduisez vos coûts LLM de 80%.