コンテキスト ウィンドウ管理: LLM 入力の最適化
La コンテキストウィンドウ LLM が処理できるトークンの制限です。 シングルコール。 GPT-4 には 128,000 トークン、クロード 3 200,000、ジェミニ 150,000 があります。数字 巨大だが複雑な RAG システムで長時間の会話が実現される これらの制限を定期的に適用します。これが発生すると、モデルは最も古いコンテキストを切り捨てます。 重要な情報を失うこと。そして費用は？ GPT-4 での 100K トークン プロンプトのコストは約 1 回の通話につき 3 ドル。運用環境では、1 日に何千ものクエリが発生すると、これは次のようになります。 急速に持続不可能になります。
Il コンテキストウィンドウの管理 それは品質を最大化する技術です LLM は、利用可能なコンテキストの使用を最適化しながら答えます。それはそれについてではありません すべてをウィンドウ内に収めるだけです。それは決定の問題です cosa 含む、 として それを構造化する いくら 各コンポーネントにスペースを割り当てます。 この記事では、トークンのカウントや予算編成、 コンテキスト圧縮、長い会話のためのメモリ管理まで。
何を学ぶか
- コンテキスト ウィンドウの仕組みと 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 は情報を記憶するのが非常に得意であることを示しています始める そしてへ 終わり コンテキストの情報を把握できますが、位置情報が「失われる」傾向があります。 真ん中にあります。これは、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 ドキュメントが利用可能な予算を超えた場合、次の 2 つのアプローチがあります。 切り捨て (テキストをカット) または 圧縮 （該当部分のみ抜粋）圧縮するとより良い結果が得られます。 キー情報を恣意的に破棄するのではなく保持します。
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. 長時間の会話のためのメモリ管理
長時間の会話は、コンテキスト ウィンドウ管理にとって最も重要なケースの 1 つです。 さまざまなメモリ戦略があり、品質とコストの間のトレードオフは異なります。
長い会話のための記憶戦略
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 を使用して、送信前にプロンプトを検証します。
- 「Lost in the Middle」のプロンプトを構成します。 最も重要な情報を最初 (システム プロンプト、主要なステートメント) と最後 (ユーザー クエリ、特定の要求) に配置します。
- ConversationsummaryBufferMemory を使用する 長い会話の場合: 最近の詳細と古いターンの一般的なコンテキストを低コストで維持します。
- 切り詰める前に圧縮します。 セマンティック圧縮は、総当たりの切り捨てよりも優れています。 40% 圧縮されたドキュメントには、関連情報の 90% が保持されます。
- 本番環境でのクエリごとのコストを追跡する: 事前定義されたしきい値を超えた場合にアラートを設定します (例: gpt-4o-mini のクエリで >$0.01 はコンテキスト管理に問題があることを示します)。
避けるべきアンチパターン
- コンテキストスタッフィング: コンテキストを可能な限りすべてで埋めても品質は向上しません。多くの場合、「Lost in the Middle」の品質は悪化します。量よりも質を選択してください。
- ストーリーコストを無視すると: RAG との 50 ターンの会話では、1 回のクエリに 10 ～ 50 倍のコストがかかる可能性があります。履歴には常に制限を実装します。
- 文書の中央で切り詰めます: 文書を文や概念の途中で切り捨てることは、それを含めないことよりも悪いです。常に自然な境界で切り詰めてください。
- すべてのモデルで同じ予算: 128K トークン モデルと 4K トークン モデルでは、根本的に異なる戦略が必要です。同じ定数を使用しないでください。
結論
コンテキスト ウィンドウの管理は実装の詳細ではなく、変数の 1 つです。 実稼働中の RAG システムの品質とコストに最も影響を与えます。私たちは探検しました tiktokenによる正確なトークンカウント、体系的なコンテキストバジェッティング、圧縮 セマンティクス、長時間の会話のためのメモリ管理、コスト監視。
重要なポイント:
- tiktoken または同等のものを使用して送信する前に、必ずトークンをカウントしてください
- 「途中で紛失」を軽減するための構造コンテキスト: 最初と最後にある重要な情報
- ブルートトランケーションの代わりにセマンティック圧縮を使用する
- ConversationsummaryBufferMemory は長時間の会話に最適です
- 実稼働環境でのクエリあたりのコストを監視し、アラートを設定する
次の記事では、 マルチエージェントシステム: オーケストレーションの方法 専門化された AI エージェントがいないよりも、複雑な問題を解決するために協力する 単一エージェントは単独で対処できます。
シリーズは続く
- 記事 1: RAG の説明
- 第 6 条: RAG 用の LangChain
- 第 7 条: コンテキスト ウィンドウの管理 (現行)
- 第 8 条: マルチエージェント システム
- 第 9 条: 生産における迅速なエンジニアリング