Zarządzanie oknami kontekstowymi: optymalizacja danych wejściowych LLM
La okno kontekstowe to limit tokenów, które LLM może przetworzyć w a pojedyncze połączenie. GPT-4 ma 128 tys. tokenów, Claude 3 200 tys., Gemini 1,5 1 milion. Liczby ogromne, ale w skomplikowanych systemach RAG i osiągane są długie rozmowy te limity regularnie. Kiedy to nastąpi, model obcina najstarszy kontekst, utratę kluczowych informacji. A koszty? Monit o token o wartości 100 tys. na GPT-4 kosztuje ok 3 USD za jedno połączenie. W środowisku produkcyjnym, z tysiącami zapytań dziennie, staje się to szybko niezrównoważony.
Il Zarządzanie oknem kontekstowym to sztuka maksymalizacji jakości LLM odpowiada, optymalizując wykorzystanie dostępnego kontekstu. Nie o to chodzi po prostu zmieścić wszystko w oknie: to kwestia decyzji Co włączać, Jak ustrukturyzuj to, np Ile przydziel miejsce każdemu komponentowi. W tym artykule omówimy wszystkie techniki: od liczenia tokenów i budżetowania, po kompresję kontekstu, po zarządzanie pamięcią podczas długich rozmów.
Czego się nauczysz
- Jak działa okno kontekstowe i dlaczego jest tak ważne dla RAG
- Precyzyjne liczenie tokenów za pomocą tiktokenu dla modeli OpenAI i open source
- Budżetowanie kontekstowe: alokuj budżet tokenów pomiędzy system, historię, kontekst i zapytanie
- Kompresja kontekstu za pomocą LMLingua i technik podsumowujących
- Zarządzanie pamięcią podczas długich rozmów (przesuwane okno, pamięć podsumowań)
- Zagubieni w środku: dlaczego lokalizacja w kontekście ma znaczenie
- Inteligentne strategie obcinania dla RAG
- Monitorowanie wykorzystania tokenów i optymalizacja kosztów
1. Jak działa okno kontekstowe
LLM oparty na transformatorze przetwarza dane wejściowe jako sekwencję znak: jednostki tekstu odpowiadające w przybliżeniu 3/4 słowa w języku angielskim (lub około 2/3 w języku włoskim). Maksymalna liczba tokenów, które model może przetworzyć w całym wywołaniu (podpowiedź + odpowiedź) jest definiowane przez okno kontekstowe.
Okno tokenizacji i kontekstu
# 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 Problem „zagubienia się w środku”.
Zaskakujący wynik badań (Liu i in., 2023, „Lost in the Middle”) pokazuje, że LLM są bardzo dobre w zapamiętywaniu informacjistart i do koniec kontekstu, ale mają tendencję do „gubienia” umiejscowionych informacji w środku. Ma to bezpośrednie implikacje dla struktury kontekstu RAG.
Zagubiony w środku: implikacje praktyczne
# 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. Precyzyjne liczenie tokenów za pomocą Tiktokena
Zanim będziesz mógł zarządzać budżetem tokenów, musisz wiedzieć, jak je dokładnie policzyć. Biblioteka tiktoken OpenAI implementuje dokładnie używany tokenizer z modeli GPT. W przypadku szablonów open source każdy szablon ma swój własny tokenizer.
Liczenie tokenów dla OpenAI i 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. Budżet kontekstowy: Alokuj budżet tokenów
Il budżetowanie kontekstowe jest to proces podejmowania decyzji, ile tokenów przydzielić do każdej części podpowiedzi. To kompromis: więcej tokenów w kontekście RAG poprawia jakość, ale koszty i opóźnienia rosną; mniej tokenów oszczędza zasoby, ale podejmuje ryzyko utraty kluczowych informacji.
Menedżer budżetu kontekstowego
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. Kompresja kontekstu
Jeżeli dokumenty RAG przekraczają dostępny budżet, istnieją dwa podejścia: obcięcie (wycięty tekst) lub kompresja (wyodrębnij tylko odpowiednie części). Kompresja daje lepsze rezultaty, ponieważ zachowuje kluczowe informacje, zamiast je arbitralnie odrzucać.
4.1 Kompresja kontekstowa za pomocą LLM
Kontekstowe pobieranie kompresji z 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. Zarządzanie pamięcią podczas długich rozmów
Długie rozmowy są jednym z najbardziej krytycznych przypadków zarządzania oknami kontekstowymi. Istnieją różne strategie pamięci, z różnymi kompromisami między jakością a kosztem:
Strategie pamięciowe podczas długich rozmów
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. Monitorowanie wykorzystania tokenów i optymalizacja kosztów
Śledzenie użycia tokenu na potrzeby produkcji
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. Najlepsze praktyki i antywzorce
Najlepsze praktyki Zarządzanie oknem kontekstowym
- Policz tokeny przed wywołaniem LLM: nie czekaj na błąd „kontekst zbyt długi”. Użyj tiktokena, aby zatwierdzić monit przed wysłaniem.
- Ustrukturyzuj zachętę dla „Zagubiony w środku”: umieść najważniejsze informacje na początku (podpowiedź systemowa, kluczowe instrukcje) i na końcu (zapytanie użytkownika, konkretne żądanie).
- Użyj ConversationSummaryBufferMemory do długich rozmów: zachowuje najnowsze szczegóły i ogólny kontekst starych zwrotów przy niskim koszcie.
- Kompresuj przed obcięciem: Kompresja semantyczna jest lepsza niż brutalne obcięcie. Dokument skompresowany w 40% zachowuje 90% istotnych informacji.
- Śledź koszt zapytania w produkcji: ustawia alert, gdy przekroczy predefiniowane progi (np. >0,01 $ dla zapytania na gpt-4o-mini oznacza problem w zarządzaniu kontekstem).
Anty-wzorce, których należy unikać
- Wypełnianie kontekstu: wypełnianie kontekstu wszystkim, co możliwe, nie poprawia jakości, a w przypadku „Lost in the Middle” często ją pogarsza. Wybierz jakość zamiast ilości.
- Ignorowanie kosztów historii: 50-krotna rozmowa z RAG może kosztować 10–50 razy jedno zapytanie. Zawsze wdrażaj ograniczenie historii.
- Obcięcie do środka dokumentu: obcięcie dokumentu w środku zdania lub koncepcji jest gorsze niż jego pominięcie. Zawsze obcinaj w naturalnych granicach.
- Ten sam budżet dla wszystkich modeli: model tokena 128 tys. i model tokena 4 tys. wymagają radykalnie różnych strategii. Nie używaj tych samych stałych.
Wnioski
Zarządzanie oknami kontekstowymi nie jest szczegółem implementacji: jest jedną ze zmiennych mający największy wpływ na jakość i koszt systemów RAG w produkcji. Zbadaliśmy precyzyjne liczenie tokenów za pomocą tiktokena, systematyczne budżetowanie kontekstowe, kompresja semantyka, zarządzanie pamięcią na potrzeby długich rozmów i monitorowanie kosztów.
Kluczowe punkty:
- Zawsze licz tokeny przed wysłaniem za pomocą tiktokenów lub ich odpowiedników
- Ustrukturyzuj kontekst, aby złagodzić „Zagubienie w środku”: krytyczne informacje na początku i na końcu
- Użyj kompresji semantycznej zamiast brutalnego obcięcia
- ConversationSummaryBufferMemory to najlepszy wybór w przypadku długich rozmów
- Monitoruj koszt zapytania w środowisku produkcyjnym i ustawiaj alerty
W następnym artykule omówimy Systemy wieloagentowe: jak orkiestrować więcej wyspecjalizowanych agentów AI współpracujących przy rozwiązywaniu złożonych problemów niż żaden pojedynczy agent mógłby poradzić sobie sam.
Seria trwa
- Artykuł 1: Wyjaśnienie RAG
- Artykuł 6: LangChain dla RAG
- Artykuł 7: Zarządzanie oknem kontekstowym (bieżący)
- Artykuł 8: Systemy wieloagentowe
- Artykuł 9: Szybka inżynieria w produkcji