Le Défi de la Fenêtre de Contexte

Les LLM ont des fenêtres de contexte fixes :

| Modèle | Fenêtre de Contexte | |-------|----------------| | GPT-3.5 Turbo | 16K tokens | | GPT-4 | 8K / 32K tokens | | GPT-4 Turbo | 128K tokens | | Claude 2 | 100K tokens | | Claude 3 | 200K tokens | | Llama 2 | 4K tokens | | Gemini 1.5 Pro | 1M tokens |

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

`python 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

`python 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

`python 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

`python 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.

`python 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

`python 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.

`python 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.

`python 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é.

`python async def hierarchical_processing(query: str, document: str, llm): """ Résumer l'ensemble du document Trouver les sections pertinentes dans le résumé 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.

`python 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.

`python 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.

`python 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

`python 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.

`python 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.

`python 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

`python 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é.

`python 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

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

Compromis

| Stratégie | Économie de Tokens | Impact Qualité | Complexité | |----------|---------------|----------------|------------| | Sélection de chunks | 20-40% | Faible | Faible | | Résumé extractif | 50-70% | Moyen | Moyen | | Compression LLM | 60-80% | Faible-Moyen | Moyen | | Optimisation de prompt | 10-30% | Faible | Faible | | Résumé d'historique | 40-60% | Faible | Moyen | | Chargement adaptatif | Variable | Faible | Élevé |

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.