Die Herausforderung des Kontextfensters

LLM haben feste Kontextfenster:

Das Problem :

Prompt système : 500 tokens
Requête utilisateur : 100 tokens
Contextes récupérés : 5 chunks × 512 tokens = 2 560 tokens
Historique de conversation : 1 000 tokens
─────────────────────────────────────────
Total entrée : 4 160 tokens
Sortie max : 1 000 tokens
─────────────────────────────────────────
Total : 5 160 tokens (tient dans une fenêtre 8K)

Mit der Weiterentwicklung Ihres RAG-Systems:

• Mehr zurückgegebene chunks
• Längere Konversationen
• Komplexere Prompts
• Größere Dokumente

Werden Sie die Kontextgrenzen erreichen. Optimierung ist entscheidend.

Zählung der Tokens

Exakte Token-Zählung

DEVELOPERpython
import tiktoken

def count_tokens(text: str, model="gpt-4") -> int:
    encoding = tiktoken.encoding_for_model(model)
    return len(encoding.encode(text))

# Beispiel
text = "Hello, how are you?"
tokens = count_tokens(text)  # ~5 tokens

Berechnung des Kontextbudgets

DEVELOPERpython
class ContextBudget:
    def __init__(self, model="gpt-4", max_output_tokens=1000):
        self.model = model
        self.max_output = max_output_tokens

        # Kontextfenster
        self.windows = {
            "gpt-3.5-turbo": 16385,
            "gpt-4": 8192,
            "gpt-4-32k": 32768,
            "gpt-4-turbo": 128000,
            "claude-2": 100000,
            "claude-3": 200000,
        }

        self.total_window = self.windows.get(model, 8192)
        self.available_input = self.total_window - max_output_tokens

    def allocate(self, system=500, query=200, history=1000):
        """
        Weist Tokens für verschiedene Komponenten zu
        """
        fixed_tokens = system + query + history
        available_for_context = self.available_input - fixed_tokens

        return {
            'system_prompt': system,
            'query': query,
            'history': history,
            'context': available_for_context,
            'output': self.max_output,
            'total': fixed_tokens + available_for_context + self.max_output
        }

# Verwendung
budget = ContextBudget(model="gpt-4")
allocation = budget.allocate()

print(f"Available for retrieved context: {allocation['context']} tokens")
# Available for retrieved context: 5492 tokens

Strategien zur Chunk-Auswahl

Top-K mit Token-Limit

DEVELOPERpython
def select_chunks_within_budget(chunks: List[dict], budget: int, model="gpt-4") -> List[dict]:
    """
    Wählt so viele priorisierte Chunks wie möglich innerhalb des Token-Budgets aus
    """
    selected = []
    total_tokens = 0

    for chunk in chunks:
        chunk_tokens = count_tokens(chunk['content'], model)

        if total_tokens + chunk_tokens <= budget:
            selected.append(chunk)
            total_tokens += chunk_tokens
        else:
            break

    return selected

# Utilisation
chunks = retriever.retrieve(query, k=20)  # Récupérer plus que nécessaire
selected = select_chunks_within_budget(chunks, budget=5000)
# Peut retourner 8-12 chunks selon la longueur

Prioritätsbasierte Auswahl

DEVELOPERpython
def priority_select_chunks(chunks: List[dict], budget: int, weights: dict) -> List[dict]:
    """
    Wählt Chunks nach mehreren Kriterien aus
    """
    # Chunk-Bewertung
    for chunk in chunks:
        score = (
            weights['relevance'] * chunk['similarity_score'] +
            weights['recency'] * chunk['recency_score'] +
            weights['authority'] * chunk['authority_score']
        )
        chunk['priority_score'] = score

    # Sortieren nach Priorität
    sorted_chunks = sorted(chunks, key=lambda x: x['priority_score'], reverse=True)

    # Auswahl innerhalb des Budgets
    return select_chunks_within_budget(sorted_chunks, budget)

Kompressionstechniken

Extraktive Zusammenfassung

Nur die relevanten Sätze extrahieren.

DEVELOPERpython
from transformers import pipeline

summarizer = pipeline("summarization", model="facebook/bart-large-cnn")

def compress_chunk(chunk: str, max_length: int = 130) -> str:
    """
    Komprimiert den Chunk und erhält die Schlüsselinformationen
    """
    summary = summarizer(
        chunk,
        max_length=max_length,
        min_length=30,
        do_sample=False
    )

    return summary[0]['summary_text']

# Verwendung
original = "Very long chunk of text..." # 500 tokens
compressed = compress_chunk(original, max_length=130)  # ~100 tokens

LLM-basierte Kompression

DEVELOPERpython
async def llm_compress_context(query: str, chunk: str, llm) -> str:
    """
    Verwendet ein LLM, um nur die für die Query relevanten Informationen zu extrahieren
    """
    prompt = f"""Extract only the information relevant to the query from this text.

Query: {query}

Text: {chunk}

Relevant excerpt:"""

    return await llm.generate(prompt, max_tokens=200)

Beispiel

query = "Wie setze ich mein Passwort zurück?" chunk = "Unser System bietet viele Funktionen, einschließlich Benutzerverwaltung, Passwortzurücksetzung, Datenexport..."

compressed = await llm_compress_context(query, chunk, llm)

"Passwort zurücksetzen: Klicken Sie auf 'Passwort vergessen' auf der Login-Seite..."

Semantische Kompression

Redundante Informationen zwischen den chunks entfernen.

DEVELOPERpython
def remove_redundant_chunks(chunks: List[str], threshold=0.85) -> List[str]:
    """
    Entfernt Chunks mit hoher semantischer Überlappung
    """
    embeddings = embed_batch(chunks)
    selected = [0]  # Toujours garder le premier (pertinence la plus élevée)

    for i in range(1, len(chunks)):
        # Vérifier la similarité avec ceux déjà sélectionnés
        max_similarity = max(
            cosine_similarity(embeddings[i], embeddings[j])
            for j in selected
        )

        # Ajouter seulement si suffisamment différent
        if max_similarity < threshold:
            selected.append(i)

    return [chunks[i] for i in selected]

Ansätze mit gleitendem Fenster

Verarbeitung per Chunks

Lange Dokumente in Fenstern verarbeiten.

DEVELOPERpython
def sliding_window_qa(query: str, long_document: str, window_size=2000, stride=1000):
    """
    Verarbeitet ein langes Dokument mit einem gleitenden Fenster
    """
    answers = []

    # Erstelle Fenster
    tokens = tokenize(long_document)

    for i in range(0, len(tokens), stride):
        window = tokens[i:i + window_size]
        window_text = detokenize(window)

        # Generiere eine Antwort für dieses Fenster
        answer = llm.generate(
            query=query,
            context=window_text
        )

        if answer and answer != "Information not found":
            answers.append({
                'answer': answer,
                'position': i,
                'confidence': estimate_confidence(answer)
            })

    # Antworten kombinieren (die mit der höchsten Zuversicht wählen oder synthetisieren)
    return best_answer(answers)

Hierarchische Verarbeitung

Auf mehreren Granularitätsstufen verarbeiten.

DEVELOPERpython
async def hierarchical_processing(query: str, document: str, llm):
    """
    1. Das gesamte Dokument zusammenfassen
    2. Relevante Abschnitte im Zusammenfass finden
    3. Die vollständigen Abschnitte für die Antwort verarbeiten
    """

    # Ebene 1 : Zusammenfassung des Dokuments
    doc_summary = await llm.generate(
        f"Summarize this document:\n\n{document}",
        max_tokens=500
    )

    # Ebene 2 : Relevante Teile im Summary identifizieren
    relevance_check = await llm.generate(
        f"Which parts of this summary are relevant to: {query}\n\nSummary: {doc_summary}",
        max_tokens=100
    )

    # Ebene 3 : Relevante vollständige Abschnitte verarbeiten
    relevant_sections = extract_sections(document, relevance_check)

    # Finale Antwort aus den relevanten Abschnitten generieren
    answer = await llm.generate(
        query=query,
        context=relevant_sections
    )

    return answer

Verwaltung des Gesprächsverlaufs

Fixes Fenster

Nur den jüngsten Verlauf beibehalten.

DEVELOPERpython
class FixedWindowHistory:
    def __init__(self, max_turns=5):
        self.max_turns = max_turns
        self.history = []

    def add_turn(self, query: str, answer: str):
        self.history.append({'query': query, 'answer': answer})

        # Nur die letzten Runden behalten
        if len(self.history) > self.max_turns:
            self.history = self.history[-self.max_turns:]

    def get_context(self) -> str:
        return "\n".join([
            f"User: {turn['query']}\nAssistant: {turn['answer']}"
            for turn in self.history
        ])

Fenster basierend auf Tokens

Den Verlauf innerhalb eines Token-Budgets halten.

DEVELOPERpython
class TokenBudgetHistory:
    def __init__(self, max_tokens=2000, model="gpt-4"):
        self.max_tokens = max_tokens
        self.model = model
        self.history = []

    def add_turn(self, query: str, answer: str):
        self.history.append({'query': query, 'answer': answer})
        self._trim_to_budget()

    def _trim_to_budget(self):
        while self.history:
            context = self.get_context()
            tokens = count_tokens(context, self.model)

            if tokens <= self.max_tokens:
                break

            # Supprimer le tour le plus ancien
            self.history.pop(0)

    def get_context(self) -> str:
        return "\n".join([
            f"User: {turn['query']}\nAssistant: {turn['answer']}"
            for turn in self.history
        ])

Zusammengefasster Verlauf

Alten Verlauf zusammenfassen, um tokens zu sparen.

DEVELOPERpython
class SummarizedHistory:
    def __init__(self, llm, summary_threshold=10):
        self.llm = llm
        self.summary_threshold = summary_threshold
        self.summary = ""
        self.recent_history = []

    async def add_turn(self, query: str, answer: str):
        self.recent_history.append({'query': query, 'answer': answer})

        # Wenn der jüngere Verlauf zu lang wird, zusammenfassen
        if len(self.recent_history) >= self.summary_threshold:
            await self._summarize_old_turns()

    async def _summarize_old_turns(self):
        # Alle außer den letzten 3 Runden zusammenfassen
        to_summarize = self.recent_history[:-3]

        if to_summarize:
            history_text = format_history(to_summarize)

            new_summary = await self.llm.generate(
                f"Summarize this conversation:\n\n{self.summary}\n\n{history_text}",
                max_tokens=200
            )

            self.summary = new_summary
            self.recent_history = self.recent_history[-3:]

    async def get_context(self) -> str:
        recent = format_history(self.recent_history)

        if self.summary:
            return f"Earlier: {self.summary}\n\nRecent:\n{recent}"
        else:
            return recent

Prompt-Optimierung

Template-Kompression

DEVELOPERpython
# Ausführlicher Prompt (Verschwendung)
verbose_prompt = """
You are a helpful AI assistant. Your job is to answer questions based on the provided context.
Please read the context carefully and provide accurate answers. If you don't know the answer,
say so. Always be polite and professional.

Context: {context}

Question: {query}

Answer:
"""

# Komprimierter Prompt (effektiv)
compressed_prompt = """Answer based on context. Say "I don't know" if uncertain.

Context: {context}

Q: {query}
A:"""

# Économie de tokens : ~50 tokens par requête

Dynamische Prompts

Den Prompt basierend auf der Komplexität der Anfrage anpassen.

DEVELOPERpython
def get_optimal_prompt(query: str, context: str, complexity: str) -> str:
    if complexity == "simple":
        # Minimaler Prompt für einfache Anfragen
        return f"Context: {context}\n\nQ: {query}\nA:"

    elif complexity == "medium":
        # Standardprompt
        return f"Answer based on context:\n\n{context}\n\nQ: {query}\nA:"

    else:
        # Detaillierter Prompt für komplexe Anfragen
        return f"""Analyze the context carefully and provide a detailed answer.

Context: {context}

Question: {query}

Detailed answer:"""

Adaptives Laden des Kontexts

Faules Laden

Den Kontext schrittweise laden.

DEVELOPERpython
async def adaptive_context_loading(query: str, vector_db, llm, max_chunks=10):
    """
    Mit wenigen Chunks beginnen und bei Bedarf hinzufügen
    """
    chunk_counts = [3, 5, 8, max_chunks]

    for num_chunks in chunk_counts:
        # Récupérer les chunks
        chunks = await vector_db.search(query, k=num_chunks)
        context = format_chunks(chunks)

        # Générer la réponse
        answer = await llm.generate(query=query, context=context)

        # Vérifier la confiance
        confidence = await estimate_confidence(answer, llm)

        if confidence > 0.8:
            return answer  # Suffisamment bon

    # Utilisé tous les chunks, retourner le meilleur effort
    return answer

Retrieval basierend auf Vertrauen

DEVELOPERpython
async def confidence_based_retrieval(query: str, vector_db, llm):
    """
    Récupérer plus de contexte si la réponse initiale a une faible confiance
    """
    # Commencer avec les 3 premiers
    chunks = await vector_db.search(query, k=3)
    context = format_chunks(chunks)

    answer = await llm.generate(query=query, context=context)
    confidence = await estimate_confidence(answer, llm)

    # Si faible confiance, récupérer plus
    if confidence < 0.6:
        additional_chunks = await vector_db.search(query, k=7)
        chunks.extend(additional_chunks[3:])  # Ignorer les 3 premiers (doublons)
        context = format_chunks(chunks)

        answer = await llm.generate(query=query, context=context)

    return answer

Multi-Turn-Optimierung

Kontextweiterleitung

Vermeiden, unveränderten Kontext erneut zu schicken.

DEVELOPERpython
class EfficientConversation:
    def __init__(self, llm):
        self.llm = llm
        self.static_context = None
        self.conversation_history = []

    async def query(self, user_query: str, retrieve_new_context=True):
        # Récupérer le contexte uniquement si la requête a changé de sujet
        if retrieve_new_context:
            self.static_context = await retrieve_context(user_query)

        # Construire un prompt minimal
        prompt = f"""Context (same as before): [Ref: {hash(self.static_context)}]

Previous conversation:
{format_recent_history(self.conversation_history[-2:])}

New question: {user_query}

Answer:"""

        answer = await self.llm.generate(prompt)

        self.conversation_history.append({
            'query': user_query,
            'answer': answer
        })

        return answer

Überwachung der Token-Nutzung

DEVELOPERpython
class TokenUsageTracker:
    def __init__(self):
        self.usage = []

    def track(self, prompt_tokens: int, completion_tokens: int, model: str):
        self.usage.append({
            'timestamp': time.time(),
            'prompt_tokens': prompt_tokens,
            'completion_tokens': completion_tokens,
            'total_tokens': prompt_tokens + completion_tokens,
            'model': model
        })

    def get_stats(self):
        if not self.usage:
            return {}

        total_tokens = sum(u['total_tokens'] for u in self.usage)
        avg_prompt = np.mean([u['prompt_tokens'] for u in self.usage])
        avg_completion = np.mean([u['completion_tokens'] for u in self.usage])

        return {
            'total_tokens': total_tokens,
            'avg_prompt_tokens': avg_prompt,
            'avg_completion_tokens': avg_completion,
            'num_requests': len(self.usage)
        }

# Verwendung
tracker = TokenUsageTracker()

response = await llm.generate(prompt)
tracker.track(
    prompt_tokens=count_tokens(prompt),
    completion_tokens=count_tokens(response),
    model="gpt-4"
)

stats = tracker.get_stats()
print(f"Average prompt tokens: {stats['avg_prompt_tokens']}")

Gute Praktiken

1. Zuerst messen : Tokens zählen, bevor optimiert wird
2. Budgets zuweisen : Tokens für jede Komponente reservieren
3. Intelligent komprimieren : Nur das komprimieren, was die Qualität nicht beeinträchtigt
4. Verlauf stutzen : Nicht die gesamte Konversation jedes Mal senden
5. Klein anfangen : Weniger chunks abrufen, bei Bedarf erweitern
6. Nutzung überwachen : Token-Verbrauch im Zeitverlauf nachverfolgen
7. Auswirkung testen : Sicherstellen, dass Kompression die Qualität nicht verschlechtert

Kompromisse

Nächste Schritte

Sie haben jetzt ein umfassendes Verständnis der RAG-Grundlagen, von embeddings und chunking bis hin zur Produktionseinführung und Optimierung. Wenden Sie diese Leitlinien schrittweise an, messen Sie die Ergebnisse und iterieren Sie basierend auf Ihrem spezifischen Anwendungsfall und Ihren Einschränkungen.