컨텍스트 창 관리: LLM 입력 최적화
La 컨텍스트 창 LLM이 처리할 수 있는 토큰의 한도입니다. 단일 통화. GPT-4 has 128K tokens, Claude 3 200K, Gemini 1.5 1 million. 숫자 거대하지만 복잡한 RAG 시스템에서는 긴 대화가 가능합니다. 이러한 제한은 정기적으로 적용됩니다. 이런 일이 발생하면 모델은 가장 오래된 컨텍스트를 잘라냅니다. 중요한 정보를 잃습니다. 그리고 비용은? GPT-4의 100K 토큰 프롬프트 비용은 약 통화당 3달러입니다. 프로덕션 환경에서는 하루에 수천 개의 쿼리가 발생하므로 이는 빠르게 지속 불가능합니다.
Il 컨텍스트 창 관리 품질을 극대화하는 기술입니다. LLM은 사용 가능한 컨텍스트 사용을 최적화하면서 응답합니다. 그것에 관한 것이 아닙니다 창문에 모든 것을 맞추려면 결정의 문제입니다 무엇 포함하다, ~처럼 그것을 구조화하라 얼마나 많이 각 구성요소에 공간을 할당합니다. 이 기사에서는 토큰 계산 및 예산 책정부터 모든 기술을 살펴보겠습니다. 컨텍스트 압축, 긴 대화를 위한 메모리 관리까지.
무엇을 배울 것인가
- 컨텍스트 창이 작동하는 방식과 이것이 RAG에 중요한 이유
- OpenAI 및 오픈 소스 모델용 tiktoken을 사용한 정확한 토큰 계산
- 컨텍스트 예산 책정: 시스템, 기록, 컨텍스트 및 쿼리 간에 토큰 예산을 할당합니다.
- LMLingua 및 요약 기술을 사용한 컨텍스트 압축
- 긴 대화를 위한 메모리 관리(슬라이딩 윈도우, 요약 메모리)
- 중간에서 길을 잃다: 상황에 따른 위치가 중요한 이유
- RAG를 위한 지능적인 잘림 전략
- 토큰 사용량 모니터링 및 비용 최적화
1. 컨텍스트 창의 작동 방식
Transformer 기반 LLM은 입력을 다음의 시퀀스로 처리합니다. 토큰: 영어 단어의 약 3/4에 해당하는 텍스트 단위 (또는 이탈리아어로 약 2/3). 모델이 처리할 수 있는 최대 토큰 수 전체 호출(프롬프트 + 응답)에서 다음과 같이 정의됩니다. 컨텍스트 창.
토큰화 및 컨텍스트 창
# 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 "중간에서 길을 잃다" 문제
놀라운 연구 결과 (Liu et al., 2023, "Lost in the Middle") LLM이 정보를 매우 잘 기억한다는 것을 보여줍니다.시작 그리고 fine 하지만 위치 정보를 "잃는" 경향이 있습니다. 중간에. 이는 RAG 컨텍스트의 구조에 직접적인 영향을 미칩니다.
중간에서 길을 잃다: 실제적인 의미
# 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. Tiktoken을 통한 정확한 토큰 계산
토큰 예산을 관리하려면 먼저 토큰 예산을 정확하게 계산하는 방법을 알아야 합니다. 도서관 틱토큰 OpenAI는 사용된 정확한 토크나이저를 구현합니다. GPT 모델에서. 오픈 소스 템플릿의 경우 각 템플릿에는 자체 토크나이저가 있습니다.
OpenAI 및 오픈 소스에 대한 토큰 계산
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. 컨텍스트 예산: 토큰 예산 할당
Il 상황별 예산 책정 얼마나 많은 토큰을 할당할지 결정하는 과정입니다 프롬프트의 각 부분에. 이는 트레이드오프입니다. RAG 컨텍스트에 대한 더 많은 토큰이 향상됩니다. 품질은 좋지만 비용과 대기 시간은 늘어납니다. 더 적은 수의 토큰으로 자원을 절약하지만 위험을 감수합니다 중요한 정보를 잃어버릴 수 있습니다.
컨텍스트 예산 관리자
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. 컨텍스트 압축
RAG 문서가 사용 가능한 예산을 초과하는 경우 두 가지 접근 방식이 있습니다. 잘림 (텍스트 자르기) 또는 압축 (해당 부분만 추출합니다.) 압축하면 더 나은 결과를 얻을 수 있습니다. 핵심 정보를 임의로 폐기하는 대신 보관합니다.
4.1 LLM을 사용한 상황별 압축
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. 긴 대화를 위한 메모리 관리
긴 대화는 컨텍스트 창 관리에 있어 가장 중요한 경우 중 하나입니다. 품질과 비용 간에 서로 다른 균형을 이루는 다양한 메모리 전략이 있습니다.
긴 대화를 위한 기억 전략
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. 토큰 사용량 모니터링 및 비용 최적화
생산용 토큰 사용 추적기
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. 모범 사례 및 안티 패턴
모범 사례 컨텍스트 창 관리
- LLM을 호출하기 전에 토큰을 계산합니다. "컨텍스트가 너무 김" 오류가 발생할 때까지 기다리지 마십시오. 보내기 전에 tiktoken을 사용하여 프롬프트를 확인하세요.
- "중간에서 길을 잃다"에 대한 프롬프트를 구성합니다. 가장 중요한 정보를 시작 부분(시스템 프롬프트, 핵심 설명)과 끝 부분(사용자 쿼리, 특정 요청)에 입력하세요.
- 대화 요약 버퍼 메모리 사용 긴 대화의 경우: 저렴한 비용으로 최신 세부 정보와 이전 차례의 일반적인 맥락을 유지합니다.
- 자르기 전 압축: 의미론적 압축은 무차별 절단보다 낫습니다. 40% 압축된 문서에는 관련 정보의 90%가 포함됩니다.
- 프로덕션에서 쿼리당 비용 추적: 미리 정의된 임계값을 초과할 때 경고를 설정합니다(예: gpt-4o-mini에 대한 쿼리의 >$0.01은 컨텍스트 관리에 문제가 있음을 나타냅니다).
피해야 할 안티패턴
- 컨텍스트 스터핑: 가능한 모든 것으로 컨텍스트를 채우는 것은 품질을 향상시키지 않습니다. 종종 "Lost in the Middle"의 경우 품질이 더 나빠집니다. 양보다 질을 선택하세요.
- 스토리 비용 무시: RAG와의 50턴 대화에는 단일 쿼리 비용이 10~50배 소요될 수 있습니다. 항상 기록에 대한 제한을 구현하십시오.
- 문서 중앙으로 잘림: 문장이나 개념 중간에 문서를 자르는 것은 포함하지 않는 것보다 더 나쁩니다. 항상 자연 경계에서 자릅니다.
- 모든 모델에 동일한 예산: 128K 토큰 모델과 4K 토큰 모델은 근본적으로 다른 전략이 필요합니다. 동일한 상수를 사용하지 마십시오.
결론
컨텍스트 창 관리는 구현 세부 사항이 아니며 변수 중 하나입니다. 생산 중인 RAG 시스템의 품질과 비용에 가장 큰 영향을 미칩니다. 우리는 탐험했다 tiktoken을 사용한 정확한 토큰 계산, 체계적인 컨텍스트 예산 책정, 압축 의미론, 긴 대화를 위한 메모리 관리 및 비용 모니터링.
핵심 포인트:
- tiktoken 또는 이에 상응하는 토큰으로 보내기 전에 항상 토큰을 계산하세요
- "중간 손실"을 완화하기 위한 구조 컨텍스트: 시작과 끝의 중요한 정보
- 무차별 절단 대신 의미 압축 사용
- ConversationSummaryBufferMemory는 긴 대화에 가장 적합한 선택입니다.
- 프로덕션에서 쿼리당 비용을 모니터링하고 알림을 설정하세요.
다음 기사에서는 다중 에이전트 시스템: 조율하는 방법 복잡한 문제를 해결하기 위해 협력하는 더 전문화된 AI 에이전트 단일 에이전트는 혼자서 처리할 수 있습니다.
시리즈는 계속됩니다
- 기사 1: RAG 설명
- 기사 6: RAG용 LangChain
- 조항 7: 컨텍스트 창 관리(현재)
- 8조: 다중 에이전트 시스템
- 기사 9: 생산 시 신속한 엔지니어링