Context Window Management: Ottimizzare l'Input dei LLM
La context window è il limite di token che un LLM può elaborare in una singola chiamata. GPT-4 ha 128K token, Claude 3 200K, Gemini 1.5 1 milione. Numeri enormi, eppure in sistemi RAG complessi e conversazioni lunghe si raggiungono regolarmente questi limiti. Quando ciò accade, il modello trunca il contesto meno recente, perdendo informazioni cruciali. E i costi? Un prompt da 100K token su GPT-4 costa circa $3 per singola chiamata. In produzione, con migliaia di query al giorno, questo diventa rapidamente insostenibile.
Il Context Window Management è l'arte di massimizzare la qualità delle risposte LLM ottimizzando al contempo l'utilizzo del contesto disponibile. Non si tratta solo di far stare tutto nella finestra: si tratta di decidere cosa includere, come strutturarlo e quanto spazio allocare a ogni componente. In questo articolo esploreremo tutte le tecniche: da token counting e budgeting, alla compressione del contesto, alla gestione della memoria per conversazioni lunghe.
Cosa Imparerai
- Come funziona la context window e perchè è critica per RAG
- Token counting preciso con tiktoken per OpenAI e modelli open source
- Context budgeting: allocare il token budget tra system, history, context e query
- Compressione del contesto con LLMLingua e tecniche di summarization
- Gestione della memoria per conversazioni lunghe (sliding window, summary memory)
- Lost in the Middle: perchè la posizione nel contesto importa
- Strategie di truncation intelligente per RAG
- Monitoring del token usage e ottimizzazione dei costi
1. Come Funziona la Context Window
Un LLM basato su Transformer processa l'input come una sequenza di token: unita di testo che corrispondono approssimativamente a 3/4 di una parola in inglese (o circa 2/3 in italiano). Il numero massimo di token che il modello può elaborare nell'intera chiamata (prompt + risposta) è definito dalla context window.
Tokenization e Context Window
# Modelli e loro context window (2025)
CONTEXT_WINDOWS = {
# OpenAI
"gpt-4o": 128_000,
"gpt-4o-mini": 128_000,
"gpt-4-turbo": 128_000,
"gpt-3.5-turbo": 16_385,
# Anthropic
"claude-3-opus": 200_000,
"claude-3-sonnet": 200_000,
"claude-3-haiku": 200_000,
# Google
"gemini-1.5-pro": 1_000_000,
"gemini-1.5-flash": 1_000_000,
# Open Source (locali)
"llama-3.1-8b": 128_000,
"mistral-7b-v0.3": 32_768,
"mixtral-8x7b": 32_768,
}
# Regola empirica tokenization:
# - Inglese: ~1 token per 4 caratteri (750 parole ~ 1000 token)
# - Italiano: ~1 token per 3 caratteri (600 parole ~ 1000 token)
# - Codice: ~1 token per 3.5 caratteri
# - Unicode/caratteri speciali: più token per carattere
# Distribuzione tipica del contesto in RAG:
CONTEXT_BUDGET_EXAMPLE = {
"total_tokens": 128_000,
"system_prompt": 500, # ~0.4%
"chat_history": 10_000, # ~8%
"retrieved_context": 8_000, # ~6%
"user_query": 200, # ~0.2%
"safety_margin": 2_000, # ~1.6%
"response_space": 107_300 # ~84% disponibile per risposta
}
1.1 Il Problema "Lost in the Middle"
Un risultato sorprendente della ricerca (Liu et al., 2023, "Lost in the Middle") mostra che gli LLM sono molto bravi a ricordare informazioni all'inizio e alla fine del contesto, ma tendono a "perdere" informazioni posizionate nel mezzo. Questo ha implicazioni dirette per come si struttura il contesto RAG.
Lost in the Middle: Implicazioni Pratiche
# Efficacia media per posizione nel contesto (studio Liu et al. 2023)
# Su task di multi-document QA con 10-20 documenti:
POSITION_PERFORMANCE = {
"primo_documento": 85, # % accuratezza
"secondo": 82,
"terzo": 78,
# ... degrado nel mezzo
"meta_contesto": 55, # minimo!
# ... recupero alla fine
"penultimo": 79,
"ultimo_documento": 84,
}
# STRATEGIE per mitigare "Lost in the Middle":
# 1. Posiziona le informazioni PIU CRITICHE all'inizio o alla fine
# 2. Limita il numero di documenti nel contesto (5-10 max)
# 3. Ripeti informazioni cruciali all'inizio E alla fine
# 4. Ordina per rilevanza decrescente (più rilevante prima)
def sort_chunks_for_context(chunks_with_scores):
"""
Ordina i chunks per massimizzare l'attenzione LLM.
Strategia: più rilevante all'inizio, secondo per rilevanza alla fine.
"""
sorted_chunks = sorted(chunks_with_scores, key=lambda x: x[1], reverse=True)
if len(sorted_chunks) <= 2:
return sorted_chunks
# "Pomodoro" pattern: più rilevante all'inizio, secondo alla fine,
# il resto nel mezzo (meno critico)
reordered = [sorted_chunks[0]] # Più rilevante: primo
middle = sorted_chunks[2:] # Meno critici: mezzo
reordered.extend(middle)
reordered.append(sorted_chunks[1]) # Secondo più rilevante: ultimo
return reordered
2. Token Counting Preciso con Tiktoken
Prima di poter gestire il budget di token, bisogna saperli contare con precisione. La libreria tiktoken di OpenAI implementa il tokenizer esatto usato dai modelli GPT. Per i modelli open source, ogni modello ha il suo tokenizer.
Token Counting per OpenAI e Open Source
import tiktoken
from typing import List, Dict, Any
class TokenCounter:
"""Token counter preciso per diversi modelli LLM"""
# Encoding per famiglia di modelli OpenAI
ENCODING_MAP = {
"gpt-4o": "o200k_base",
"gpt-4o-mini": "o200k_base",
"gpt-4": "cl100k_base",
"gpt-3.5-turbo": "cl100k_base",
"text-embedding-ada-002": "cl100k_base",
"text-embedding-3-small": "cl100k_base",
"text-embedding-3-large": "cl100k_base",
}
def __init__(self, model: str = "gpt-4o-mini"):
self.model = model
encoding_name = self.ENCODING_MAP.get(model, "cl100k_base")
self.encoding = tiktoken.get_encoding(encoding_name)
def count_tokens(self, text: str) -> int:
"""Conta i token di un testo"""
return len(self.encoding.encode(text))
def count_message_tokens(self, messages: List[Dict]) -> int:
"""
Conta i token di una lista di messaggi OpenAI,
includendo i token di overhead per ogni messaggio.
"""
# OpenAI aggiunge token extra per ogni messaggio
tokens_per_message = 3 # <|start|>role<|sep|>
tokens_per_name = 1 # se il nome è presente
tokens_reply = 3 # risposta inizia con <|start|>assistant<|sep|>
num_tokens = tokens_reply
for message in messages:
num_tokens += tokens_per_message
for key, value in message.items():
num_tokens += self.count_tokens(str(value))
if key == "name":
num_tokens += tokens_per_name
return num_tokens
def truncate_to_limit(self, text: str, max_tokens: int) -> str:
"""Tronca il testo al numero massimo di token"""
tokens = self.encoding.encode(text)
if len(tokens) <= max_tokens:
return text
truncated = self.encoding.decode(tokens[:max_tokens])
return truncated + "... [truncated]"
def split_by_tokens(self, text: str, max_tokens_per_chunk: int) -> List[str]:
"""Divide il testo in chunks di dimensione massima in token"""
tokens = self.encoding.encode(text)
chunks = []
for i in range(0, len(tokens), max_tokens_per_chunk):
chunk_tokens = tokens[i:i + max_tokens_per_chunk]
chunk_text = self.encoding.decode(chunk_tokens)
chunks.append(chunk_text)
return chunks
def estimate_cost(self, prompt_tokens: int, completion_tokens: int) -> dict:
"""Stima il costo per modelli OpenAI (prezzi 2025)"""
PRICES_PER_1M = {
"gpt-4o": {"prompt": 5.0, "completion": 15.0},
"gpt-4o-mini": {"prompt": 0.15, "completion": 0.60},
"gpt-4-turbo": {"prompt": 10.0, "completion": 30.0},
}
prices = PRICES_PER_1M.get(self.model, {"prompt": 1.0, "completion": 3.0})
prompt_cost = (prompt_tokens / 1_000_000) * prices["prompt"]
completion_cost = (completion_tokens / 1_000_000) * prices["completion"]
return {
"prompt_tokens": prompt_tokens,
"completion_tokens": completion_tokens,
"prompt_cost_usd": prompt_cost,
"completion_cost_usd": completion_cost,
"total_cost_usd": prompt_cost + completion_cost
}
# Utilizzo
counter = TokenCounter("gpt-4o-mini")
# Conta token di un testo
text = "Questo è un esempio di testo per RAG."
print(f"Token: {counter.count_tokens(text)}") # ~9 token
# Conta token di messaggi
messages = [
{"role": "system", "content": "Sei un assistente AI esperto."},
{"role": "user", "content": "Cos'è il RAG?"}
]
print(f"Token messaggi: {counter.count_message_tokens(messages)}")
# Stima costi
cost = counter.estimate_cost(prompt_tokens=5000, completion_tokens=500)
print(f"Costo stimato: ${cost['total_cost_usd']:.4f}")
3. Context Budgeting: Allocare il Token Budget
Il context budgeting è il processo di decidere quanti token allocare a ogni parte del prompt. È un compromesso: più token al contesto RAG migliorano la qualità, ma aumentano i costi e la latenza; meno token risparmiano risorse ma rischiano di perdere informazioni critiche.
Context Budget Manager
from dataclasses import dataclass
from typing import List, Optional, Tuple
import tiktoken
@dataclass
class ContextBudget:
"""Definisce il budget di token per ogni componente"""
total_context: int # Token totali disponibili (da context window)
max_response: int # Token riservati per la risposta
system_prompt: int # Token per il system prompt
chat_history: int # Token per la chat history
retrieved_docs: int # Token per i documenti RAG
query: int # Token per la query corrente
safety_margin: int = 200 # Buffer di sicurezza
@property
def available_for_docs(self) -> int:
"""Token effettivamente disponibili per i documenti RAG"""
used = (self.system_prompt + self.chat_history +
self.query + self.safety_margin + self.max_response)
return min(self.retrieved_docs, self.total_context - used)
def is_valid(self) -> bool:
"""Verifica che il budget non superi i limiti"""
total_used = (self.system_prompt + self.chat_history +
self.retrieved_docs + self.query +
self.safety_margin + self.max_response)
return total_used <= self.total_context
class ContextWindowManager:
"""Gestisce l'allocazione del contesto per chiamate LLM"""
BUDGETS = {
"gpt-4o-mini-128k": ContextBudget(
total_context=128_000,
max_response=4_000,
system_prompt=800,
chat_history=12_000,
retrieved_docs=6_000,
query=500
),
"gpt-4o-128k": ContextBudget(
total_context=128_000,
max_response=8_000,
system_prompt=1_000,
chat_history=20_000,
retrieved_docs=10_000,
query=500
),
"claude-3-200k": ContextBudget(
total_context=200_000,
max_response=8_000,
system_prompt=1_000,
chat_history=40_000,
retrieved_docs=15_000,
query=500
),
}
def __init__(self, model: str = "gpt-4o-mini-128k"):
self.budget = self.BUDGETS.get(model, self.BUDGETS["gpt-4o-mini-128k"])
encoding_name = "o200k_base" if "gpt-4o" in model else "cl100k_base"
self.encoder = tiktoken.get_encoding(encoding_name)
def _count(self, text: str) -> int:
return len(self.encoder.encode(text))
def fit_documents_to_budget(
self,
documents: List[Tuple[str, float]], # (testo, score)
actual_chat_tokens: int = 0
) -> List[str]:
"""
Seleziona e tronca i documenti per stare nel budget.
Tiene conto dei token effettivi usati dalla history.
"""
# Ricalcola il budget disponibile per i doc in base alla history effettiva
history_overflow = max(0, actual_chat_tokens - self.budget.chat_history)
available = self.budget.available_for_docs - history_overflow
if available <= 100:
return [] # Nessuno spazio per i documenti
selected_docs = []
tokens_used = 0
for doc_text, score in documents:
doc_tokens = self._count(doc_text)
if tokens_used + doc_tokens <= available:
# Il documento ci sta per intero
selected_docs.append(doc_text)
tokens_used += doc_tokens
elif tokens_used < available * 0.5:
# Spazio rimanente: tronca il documento
remaining = available - tokens_used
if remaining > 100: # Tronca solo se c'è abbastanza spazio
truncated_tokens = self.encoder.encode(doc_text)[:remaining - 20]
truncated_text = self.encoder.decode(truncated_tokens) + "...[truncato]"
selected_docs.append(truncated_text)
break
else:
break # Non c'è più spazio
return selected_docs
def summarize_history_if_needed(
self,
messages: List[dict],
llm_client,
target_tokens: Optional[int] = None
) -> List[dict]:
"""
Se la history supera il budget, riassumi le parti più vecchie.
Mantiene i messaggi recenti integri.
"""
if target_tokens is None:
target_tokens = self.budget.chat_history
# Calcola token attuali
all_text = " ".join(m["content"] for m in messages)
current_tokens = self._count(all_text)
if current_tokens <= target_tokens:
return messages # Nessuna azione necessaria
# Mantieni gli ultimi N messaggi intatti (conversazione recente)
keep_recent = 4 # Ultimi 2 turn (user + assistant)
recent_messages = messages[-keep_recent:]
old_messages = messages[:-keep_recent]
if not old_messages:
return recent_messages
# Riassumi i messaggi vecchi
old_text = "\n".join(
f"{m['role']}: {m['content']}" for m in old_messages
)
summary_response = llm_client.chat.completions.create(
model="gpt-4o-mini",
messages=[{
"role": "user",
"content": f"Riassumi brevemente questa conversazione in 2-3 frasi:\n\n{old_text}"
}],
max_tokens=200,
temperature=0
)
summary = summary_response.choices[0].message.content
# Sostituisci i vecchi messaggi con il riassunto
return [
{"role": "system", "content": f"[Riassunto conversazione precedente]: {summary}"}
] + recent_messages
4. Compressione del Contesto
Quando i documenti RAG superano il budget disponibile, ci sono due approcci: truncation (taglia il testo) o compressione (estrai solo le parti rilevanti). La compressione produce risultati migliori perchè mantiene le informazioni chiave invece di scartarle arbitrariamente.
4.1 Compressione Contestuale con LLM
Contextual Compression Retriever con LangChain
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor
from langchain.retrievers.document_compressors import EmbeddingsFilter
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
class ContextCompressor:
"""Comprime il contesto RAG per stare nel budget"""
def __init__(self, base_retriever, llm, embeddings):
self.base_retriever = base_retriever
self.llm = llm
# Metodo 1: LLMChainExtractor
# Usa un LLM per estrarre solo le parti rilevanti dalla domanda
# Pro: alta qualità, Pro: lento e costoso
self.llm_extractor = LLMChainExtractor.from_llm(llm)
self.llm_compressor = ContextualCompressionRetriever(
base_compressor=self.llm_extractor,
base_retriever=base_retriever
)
# Metodo 2: EmbeddingsFilter
# Rimuove i documenti sotto una soglia di similarità con la query
# Pro: veloce e gratuito, Con: meno preciso
self.embeddings_filter = EmbeddingsFilter(
embeddings=embeddings,
similarity_threshold=0.76 # Filtra documenti poco rilevanti
)
self.embedding_compressor = ContextualCompressionRetriever(
base_compressor=self.embeddings_filter,
base_retriever=base_retriever
)
def compress_with_extraction(self, query: str) -> list:
"""Estrai solo le frasi rilevanti dai documenti"""
return self.llm_compressor.invoke(query)
def compress_with_filtering(self, query: str) -> list:
"""Rimuovi documenti poco rilevanti"""
return self.embedding_compressor.invoke(query)
# Implementazione custom: compressione con suddivisione in frasi
from sentence_transformers import SentenceTransformer
import numpy as np
from typing import List
class SentenceLevelCompressor:
"""Compressione a livello di frase per massimizzare la densita informativa"""
def __init__(self, model_name: str = "all-MiniLM-L6-v2"):
self.model = SentenceTransformer(model_name)
def compress(
self,
document: str,
query: str,
max_tokens: int = 300,
top_k_sentences: int = 5
) -> str:
"""
Estrae le frasi più rilevanti dal documento rispetto alla query.
"""
import re
# Dividi in frasi
sentences = re.split(r'(?<=[.!?])\s+', document)
sentences = [s.strip() for s in sentences if len(s.strip()) > 20]
if len(sentences) <= 3:
return document # Documento già breve, non comprimere
# Codifica query e frasi
query_emb = self.model.encode([query], normalize_embeddings=True)[0]
sentence_embs = self.model.encode(sentences, normalize_embeddings=True)
# Calcola similarità
scores = np.dot(sentence_embs, query_emb)
# Seleziona top-k frasi per rilevanza mantenendo l'ordine originale
top_indices = sorted(
np.argsort(scores)[-top_k_sentences:].tolist()
)
# Ricomponi il testo mantenendo l'ordine originale
compressed = " ".join(sentences[i] for i in top_indices)
return compressed
def batch_compress(
self,
documents: List[str],
query: str,
token_budget: int = 2000
) -> List[str]:
"""Comprimi un batch di documenti rispettando il budget totale"""
counter = TokenCounter()
compressed_docs = []
tokens_used = 0
for doc in documents:
# Comprimi prima al 50%
compressed = self.compress(doc, query, top_k_sentences=5)
doc_tokens = counter.count_tokens(compressed)
if tokens_used + doc_tokens <= token_budget:
compressed_docs.append(compressed)
tokens_used += doc_tokens
else:
# Comprimi ulteriormente
remaining = token_budget - tokens_used
if remaining > 50:
further_compressed = self.compress(
doc, query, top_k_sentences=2
)
compressed_docs.append(further_compressed)
break
return compressed_docs
5. Gestione della Memoria per Conversazioni Lunghe
Le conversazioni lunghe sono uno dei casi più critici per il context window management. Esistono diverse strategie di memoria, con diversi tradeoff tra qualità e costo:
Strategie di Memoria per Conversazioni Lunghe
from langchain.memory import (
ConversationBufferMemory, # Tutta la storia
ConversationBufferWindowMemory, # Sliding window
ConversationSummaryMemory, # Riassunto
ConversationSummaryBufferMemory, # Ibrido: riassunto + recenti
ConversationTokenBufferMemory, # Limite token preciso
)
from langchain_openai import ChatOpenAI
# 1. SLIDING WINDOW: mantieni solo gli ultimi k turni
# Pro: semplice, veloce, costo fisso
# Con: perde contesto lontano
window_memory = ConversationBufferWindowMemory(
k=5, # Mantieni gli ultimi 5 turni di conversazione
return_messages=True,
memory_key="chat_history"
)
# 2. SUMMARY MEMORY: riassumi l'intera storia
# Pro: scala senza limiti, mantiene il contesto generale
# Con: perde dettagli, costo extra per ogni riassunto
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
summary_memory = ConversationSummaryMemory(
llm=llm,
return_messages=True,
memory_key="chat_history"
)
# 3. SUMMARY BUFFER MEMORY: ibrido - riassunto + ultimi k token
# Pro: mantiene sia contesto generale che dettagli recenti
# Con: più complesso, costo moderato per i riassunti
hybrid_memory = ConversationSummaryBufferMemory(
llm=llm,
max_token_limit=4000, # Soglia: se supera, riassumi vecchi messaggi
return_messages=True,
memory_key="chat_history"
)
# 4. TOKEN BUFFER MEMORY: limite preciso in token
# Pro: controllo esatto del budget
# Con: può troncare a meta di un turno
token_memory = ConversationTokenBufferMemory(
llm=llm,
max_token_limit=8000,
return_messages=True,
memory_key="chat_history"
)
# Implementazione custom: Entity Memory per RAG
class EntityMemory:
"""
Memorizza le entità menzionate nella conversazione per
arricchire le query future con contesto rilevante.
"""
def __init__(self, llm):
self.llm = llm
self.entities = {} # nome_entita -> descrizione
def extract_entities(self, message: str) -> dict:
"""Estrae entità rilevanti da un messaggio"""
prompt = f"""Estrai le entità principali (persone, organizzazioni, concetti tecnici)
da questo messaggio. Per ogni entità, fornisci una breve descrizione.
Formato: JSON con {"entità": "descrizione"}
Se non ci sono entità rilevanti, restituisci {}.
Messaggio: {message}"""
response = self.llm.invoke(prompt).content
try:
import json
return json.loads(response)
except:
return {}
def update(self, message: str):
"""Aggiorna la memoria delle entità"""
new_entities = self.extract_entities(message)
self.entities.update(new_entities)
def get_relevant_context(self, query: str) -> str:
"""Ottieni il contesto delle entità rilevanti per la query"""
if not self.entities:
return ""
# Trova entità menzionate nella query
query_lower = query.lower()
relevant = {
k: v for k, v in self.entities.items()
if k.lower() in query_lower
}
if not relevant:
return ""
return "Contesto entità:\n" + "\n".join(
f"- {k}: {v}" for k, v in relevant.items()
)
6. Monitoring del Token Usage e Ottimizzazione dei Costi
Token Usage Tracker per Produzione
from langchain.callbacks import get_openai_callback
from langchain_core.callbacks import BaseCallbackHandler
from typing import Any, Dict, List
import time
import logging
logger = logging.getLogger(__name__)
class TokenUsageTracker(BaseCallbackHandler):
"""Traccia l'utilizzo dei token e i costi per ogni chiamata LLM"""
def __init__(self, model: str = "gpt-4o-mini"):
self.model = model
self.total_prompt_tokens = 0
self.total_completion_tokens = 0
self.total_calls = 0
self.call_history = []
def on_llm_start(
self, serialized: Dict[str, Any], prompts: List[str], **kwargs
) -> None:
self._start_time = time.time()
def on_llm_end(self, response, **kwargs) -> None:
duration = time.time() - self._start_time
if hasattr(response, 'llm_output') and response.llm_output:
token_usage = response.llm_output.get('token_usage', {})
prompt_tokens = token_usage.get('prompt_tokens', 0)
completion_tokens = token_usage.get('completion_tokens', 0)
self.total_prompt_tokens += prompt_tokens
self.total_completion_tokens += completion_tokens
self.total_calls += 1
call_data = {
'timestamp': time.time(),
'prompt_tokens': prompt_tokens,
'completion_tokens': completion_tokens,
'duration_ms': duration * 1000,
}
self.call_history.append(call_data)
logger.info(
f"LLM call: {prompt_tokens}+{completion_tokens}={prompt_tokens+completion_tokens} "
f"token, {duration*1000:.0f}ms"
)
def get_stats(self) -> dict:
"""Statistiche aggregate sull'uso dei token"""
counter = TokenCounter(self.model)
total_tokens = self.total_prompt_tokens + self.total_completion_tokens
cost = counter.estimate_cost(
self.total_prompt_tokens, self.total_completion_tokens
)
avg_prompt = (self.total_prompt_tokens / self.total_calls
if self.total_calls > 0 else 0)
return {
"total_calls": self.total_calls,
"total_tokens": total_tokens,
"avg_prompt_tokens": avg_prompt,
"total_cost_usd": cost["total_cost_usd"],
"cost_per_call_usd": (cost["total_cost_usd"] / self.total_calls
if self.total_calls > 0 else 0)
}
# Utilizzo con get_openai_callback (più semplice per OpenAI)
from langchain.callbacks import get_openai_callback
with get_openai_callback() as cb:
result = rag_chain.invoke("Cos'è il RAG?")
print(f"Tokens usati: {cb.total_tokens}")
print(f"Costo: ${cb.total_cost:.6f}")
print(f"Prompt tokens: {cb.prompt_tokens}")
print(f"Completion tokens: {cb.completion_tokens}")
7. Best Practices e Anti-Pattern
Best Practices Context Window Management
- Conta i token prima di chiamare l'LLM: non aspettare l'errore di "context too long". Usa tiktoken per validare il prompt prima dell'invio.
- Struttura il prompt per "Lost in the Middle": metti le informazioni più critiche all'inizio (system prompt, istruzioni chiave) e alla fine (query dell'utente, richiesta specifica).
- Usa ConversationSummaryBufferMemory per conversazioni lunghe: mantiene i dettagli recenti e il contesto generale dei turni vecchi a costo contenuto.
- Comprimi prima di troncare: la compressione semantica è meglio della truncation bruta. Un documento compresso al 40% mantiene il 90% delle informazioni rilevanti.
- Monitora il cost-per-query in produzione: imposta alert quando supera soglie predefinite (es. >$0.01 per query su gpt-4o-mini indica un problema nel context management).
Anti-Pattern da Evitare
- Context stuffing: riempire il contesto con tutto il possibile non migliora la qualità - spesso la peggiora per "Lost in the Middle". Scegli qualità su quantità.
- Ignorare i costi della storia: una conversazione di 50 turni con RAG può costare 10-50x una singola query. Implementa sempre un limite sulla history.
- Truncation al centro del documento: troncare un documento nel mezzo di una frase o di un concetto è peggio che non includerlo. Tronca sempre a confini naturali.
- Stesso budget per tutti i modelli: un modello da 128K token e uno da 4K token richiedono strategie radicalmente diverse. Non usare le stesse costanti.
Conclusioni
Il context window management non è un dettaglio implementativo: è una delle variabili più impattanti sulla qualità e sul costo dei sistemi RAG in produzione. Abbiamo esplorato token counting preciso con tiktoken, context budgeting sistematico, compressione semantica, gestione della memoria per conversazioni lunghe e monitoring dei costi.
I punti chiave:
- Conta sempre i token prima dell'invio con tiktoken o equivalenti
- Struttura il contesto per mitigare "Lost in the Middle": informazioni critiche all'inizio e alla fine
- Usa compressione semantica invece di truncation bruta
- ConversationSummaryBufferMemory è la scelta migliore per conversazioni lunghe
- Monitora il cost-per-query in produzione e imposta alert
Nel prossimo articolo esploreremo i Sistemi Multi-Agent: come orchestrare più agenti AI specializzati che collaborano per risolvere problemi complessi che nessun singolo agente potrebbe affrontare da solo.
Continua la Serie
- Articolo 1: RAG Spiegato
- Articolo 6: LangChain per RAG
- Articolo 7: Context Window Management (corrente)
- Articolo 8: Multi-Agent Systems
- Articolo 9: Prompt Engineering in Produzione
Se ti interessa come questi pattern di AI engineering si traducono in workflow quotidiano su un codebase reale, ho documentato il mio setup (skill design + planning fasato + sub-agent orchestration) in Claude Code Workflow: Superpowers + GSD framework.
Commenti
Caricamento commenti...
Accedi per commentare