Kontextuelle Kompression: Das Wesentliche aus Dokumenten extrahieren

Die kontextuelle Kompression filtert und verdichtet abgerufene Dokumente, um nur die direkt für die Anfrage relevanten Passagen beizubehalten. Anstatt vollständige Chunks an das LLM zu senden, extrahiert diese Technik die Schlüsselsätze, reduziert Rauschen und optimiert die Kosten. Dieser Leitfaden untersucht die Kompressionsmethoden und ihre Integration in eine RAG-Pipeline.

Warum den Kontext komprimieren?

Abgerufene Chunks enthalten oft überflüssige Inhalte:

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."

Vorteile der Kompression

Die Kompression verbessert die Qualität, da sie das Rauschen reduziert, das das LLM ablenken kann.

Kompressionsmethoden

1. LLM-Kompression

Das LLM identifiziert und extrahiert die relevanten Passagen:

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. Kompression durch Satzextraktion

Schneller und ohne API-Kosten, nutzt die semantische Ähnlichkeit:

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]:
        # Anfrage encodieren
        query_embedding = self.model.encode(query)

        all_relevant_sentences = []

        for doc in documents:
            # In Sätze aufteilen
            sentences = nltk.sent_tokenize(doc, language='french')
            if not sentences:
                continue

            # Sätze encodieren
            sentence_embeddings = self.model.encode(sentences)

            # Similaritäten berechnen
            similarities = np.dot(sentence_embeddings, query_embedding) / (
                np.linalg.norm(sentence_embeddings, axis=1) * np.linalg.norm(query_embedding)
            )

            # Relevante Sätze auswählen
            for sentence, sim in zip(sentences, similarities):
                if sim >= min_similarity:
                    all_relevant_sentences.append((sentence, sim))

        # Nach Similarität sortieren und die Top-K nehmen
        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. Kompression durch Reranking + Filterung

Nutzt einen Cross-Encoder zum Bewerten und Filtern:

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]:
        # Jedes Dokument in Passagen aufteilen
        passages = []
        for doc_idx, doc in enumerate(documents):
            for para in doc.split('\n\n'):
                if len(para.strip()) > 50:  # Zu kurze Absätze ignorieren
                    passages.append({
                        "text": para.strip(),
                        "doc_idx": doc_idx
                    })

        if not passages:
            return []

        # Alle Passagen bewerten
        pairs = [[query, p["text"]] for p in passages]
        scores = self.reranker.predict(pairs)

        # Filtern und sortieren
        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. Kompression durch Zusammenfassung

Für sehr lange Dokumente eine gezielte Zusammenfassung erstellen:

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

Architektur der vollständigen Pipeline

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
        }

Adaptive Kompression

Die Methode dem Kontext anpassen:

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
            )

Kompression mit Quellenerhaltung

Nachvollziehbarkeit für Quellenangaben bewahren:

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)

Bewertung der Kompression

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
        }

Kostenoptimierung

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]

Nächste Schritte

Die kontextuelle Kompression optimiert Qualität und Kosten Ihres RAG. Für weiterführende Informationen:

• LLM-Generierung RAG - Die Generierung optimieren
• Ensemble Retrieval - Mehrere Retriever kombinieren
• Retrieval-Grundlagen - Überblick

Intelligente Kompression mit Ailog

Ailog implementiert kontextuelle Kompression automatisch:

• Adaptive Kompression je nach Komplexität der Anfrage
• Quellenerhaltung für Quellenangaben
• Kostenoptimierung mit priorisierter lokaler Kompression
• Integriertes Monitoring des Kompression/Qualität-Verhältnisses

Kostenlos testen und reduzieren Sie Ihre LLM-Kosten um 80%.