Optimisation de la Fenêtre de Contexte : Gérer les Limites de Tokens

Le Défi de la Fenêtre de Contexte

Les LLM ont des fenêtres de contexte fixes :

Le problème :

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)

Au fur et à mesure que votre système RAG évolue :

• Plus de chunks récupérés
• Conversations plus longues
• Prompts plus complexes
• Documents plus volumineux

Vous atteindrez les limites de contexte. L'optimisation est essentielle.

Comptage de Tokens

Comptage Précis de Tokens

DEVELOPERpython
import tiktoken

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

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

Calcul du Budget de Contexte

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

        # Fenêtres de contexte
        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):
        """
        Alloue les tokens pour différents composants
        """
        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
        }

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

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

Stratégies de Sélection de Chunks

Top-K avec Limite de Tokens

DEVELOPERpython
def select_chunks_within_budget(chunks: List[dict], budget: int, model="gpt-4") -> List[dict]:
    """
    Sélectionne autant de chunks prioritaires que possible dans le budget de tokens
    """
    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

Sélection Basée sur les Priorités

DEVELOPERpython
def priority_select_chunks(chunks: List[dict], budget: int, weights: dict) -> List[dict]:
    """
    Sélectionne les chunks selon plusieurs critères
    """
    # Score des chunks
    for chunk in chunks:
        score = (
            weights['relevance'] * chunk['similarity_score'] +
            weights['recency'] * chunk['recency_score'] +
            weights['authority'] * chunk['authority_score']
        )
        chunk['priority_score'] = score

    # Trier par priorité
    sorted_chunks = sorted(chunks, key=lambda x: x['priority_score'], reverse=True)

    # Sélectionner dans le budget
    return select_chunks_within_budget(sorted_chunks, budget)

Techniques de Compression

Résumé Extractif

Extraire uniquement les phrases pertinentes.

DEVELOPERpython
from transformers import pipeline

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

def compress_chunk(chunk: str, max_length: int = 130) -> str:
    """
    Compresse le chunk tout en préservant les informations clés
    """
    summary = summarizer(
        chunk,
        max_length=max_length,
        min_length=30,
        do_sample=False
    )

    return summary[0]['summary_text']

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

Compression Basée sur LLM

DEVELOPERpython
async def llm_compress_context(query: str, chunk: str, llm) -> str:
    """
    Utilise un LLM pour extraire uniquement les informations pertinentes
    """
    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)

# Exemple
query = "How do I reset my password?"
chunk = "Our system offers many features including user management, password reset, data export..."

compressed = await llm_compress_context(query, chunk, llm)
# "Password reset: Click 'Forgot Password' on the login page..."

Compression Sémantique

Supprimer les informations redondantes entre les chunks.

DEVELOPERpython
def remove_redundant_chunks(chunks: List[str], threshold=0.85) -> List[str]:
    """
    Supprime les chunks avec un chevauchement sémantique élevé
    """
    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]

Approches par Fenêtre Glissante

Traitement par Chunks

Traiter les documents longs par fenêtres.

DEVELOPERpython
def sliding_window_qa(query: str, long_document: str, window_size=2000, stride=1000):
    """
    Traite un long document avec une fenêtre glissante
    """
    answers = []

    # Créer des fenêtres
    tokens = tokenize(long_document)

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

        # Générer une réponse pour cette fenêtre
        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)
            })

    # Combiner les réponses (prendre la plus haute confiance ou synthétiser)
    return best_answer(answers)

Traitement Hiérarchique

Traiter à plusieurs niveaux de granularité.

DEVELOPERpython
async def hierarchical_processing(query: str, document: str, llm):
    """
    1. Résumer l'ensemble du document
    2. Trouver les sections pertinentes dans le résumé
    3. Traiter les sections complètes pour la réponse
    """

    # Niveau 1 : Résumé du document
    doc_summary = await llm.generate(
        f"Summarize this document:\n\n{document}",
        max_tokens=500
    )

    # Niveau 2 : Identifier les sections pertinentes
    relevance_check = await llm.generate(
        f"Which parts of this summary are relevant to: {query}\n\nSummary: {doc_summary}",
        max_tokens=100
    )

    # Niveau 3 : Traiter les sections pertinentes complètes
    relevant_sections = extract_sections(document, relevance_check)

    # Générer la réponse finale à partir des sections pertinentes
    answer = await llm.generate(
        query=query,
        context=relevant_sections
    )

    return answer

Gestion de l'Historique de Conversation

Fenêtre Fixe

Garder uniquement l'historique récent.

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})

        # Garder seulement les tours récents
        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
        ])

Fenêtre Basée sur les Tokens

Garder l'historique dans un budget de tokens.

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
        ])

Historique Résumé

Résumer l'ancien historique pour économiser des tokens.

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})

        # Quand l'historique récent devient long, résumer
        if len(self.recent_history) >= self.summary_threshold:
            await self._summarize_old_turns()

    async def _summarize_old_turns(self):
        # Résumer tous sauf les 3 derniers tours
        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

Optimisation des Prompts

Compression de Template

DEVELOPERpython
# Prompt verbeux (gaspillage)
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:
"""

# Prompt compressé (efficace)
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

Prompts Dynamiques

Ajuster le prompt en fonction de la complexité de la requête.

DEVELOPERpython
def get_optimal_prompt(query: str, context: str, complexity: str) -> str:
    if complexity == "simple":
        # Prompt minimal pour les requêtes simples
        return f"Context: {context}\n\nQ: {query}\nA:"

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

    else:
        # Prompt détaillé pour les requêtes complexes
        return f"""Analyze the context carefully and provide a detailed answer.

Context: {context}

Question: {query}

Detailed answer:"""

Chargement Adaptatif du Contexte

Chargement Paresseux

Charger le contexte progressivement.

DEVELOPERpython
async def adaptive_context_loading(query: str, vector_db, llm, max_chunks=10):
    """
    Commencer avec peu de chunks, en ajouter si nécessaire
    """
    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

Récupération Basée sur la Confiance

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

Optimisation Multi-Tours

Report de Contexte

Éviter de renvoyer le contexte inchangé.

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

Surveillance de l'Utilisation des Tokens

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)
        }

# Utilisation
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']}")

Bonnes Pratiques

1. Mesurer d'abord : Compter les tokens avant d'optimiser
2. Allouer des budgets : Réserver des tokens pour chaque composant
3. Compresser intelligemment : Compresser uniquement ce qui ne nuit pas à la qualité
4. Élaguer l'historique : Ne pas envoyer toute la conversation à chaque fois
5. Commencer petit : Récupérer moins de chunks, étendre si nécessaire
6. Surveiller l'utilisation : Suivre la consommation de tokens au fil du temps
7. Tester l'impact : S'assurer que la compression ne nuit pas à la qualité

Compromis

Prochaines Étapes

Vous avez maintenant une compréhension complète des fondamentaux du RAG, des embeddings et du chunking au déploiement en production et à l'optimisation. Appliquez ces guides progressivement, mesurez les résultats et itérez en fonction de votre cas d'usage spécifique et de vos contraintes.