khalilbenaz/claude-skills-collection/agent-context-manager

Fenêtre totale = 200 000 tokens
├── System prompt (fixe)          ~  2 000  (1 %)
├── Descriptions d'outils (fixe)  ~  3 000  (1.5 %)
├── Mémoire long terme            ~  5 000  (2.5 %)
├── Contexte RAG injecté          ~ 20 000  (10 %)
├── Historique conversation       ~ 40 000  (20 %)
├── Réponse réservée              ~ 10 000  (5 %)
└── Marge sécurité (10 %)        ~ 20 000

# OpenAI / tiktoken
import tiktoken
enc = tiktoken.encoding_for_model("gpt-4o")
n_tokens = len(enc.encode(text))
# Anthropic SDK
import anthropic
client = anthropic.Anthropic()
response = client.messages.count_tokens(
    model="claude-sonnet-4-5",
    system=system_prompt,
    messages=messages,
)
print(response.input_tokens)  # total exact avant envoi

WINDOW_VERBATIM = 10  # derniers échanges complets
def build_history(messages: list[dict], max_tokens: int) -> list[dict]:
    recent = messages[-WINDOW_VERBATIM:]
    older  = messages[:-WINDOW_VERBATIM]
    if not older:
        return recent
    summary = summarize_with_cheap_model(older)  # gpt-4o-mini, claude-haiku
    summary_msg = {
        "role": "system",
        "content": f"[RÉSUMÉ DES ÉCHANGES PRÉCÉDENTS]\n{summary}"
    }
    return [summary_msg] + recent

- Décisions prises : ...
- Faits établis : ...
- Contexte utilisateur : ...
- Tâches en cours : ...

def inject_rag_context(query: str, vector_store, top_k=5, min_score=0.75):
    results = vector_store.similarity_search_with_score(query, k=top_k)
    relevant = [doc for doc, score in results if score >= min_score]
    if not relevant:
        return ""
    chunks = "\n\n---\n\n".join(doc.page_content for doc in relevant)
    return f"[CONTEXTE DOCUMENTAIRE PERTINENT]\n{chunks}"

COMPRESSION_PROMPT = """
Tu reçois un historique de conversation entre un agent IA et un utilisateur.
Résume-le en conservant UNIQUEMENT :
1. Les décisions prises et leur justification
2. Les contraintes et préférences exprimées par l'utilisateur
3. L'état courant de la tâche principale
4. Les informations factuelles établies (dates, noms, valeurs numériques)
Sois concis. Élimine les reformulations, hésitations et confirmations de politesse.
"""
def compress_history(history: list[dict]) -> str:
    response = cheap_llm.invoke(COMPRESSION_PROMPT + format_history(history))
    return response.content

[SYSTEM PROMPT]
  → Instructions système permanentes
 
[MÉMOIRE LONG TERME]
  → Faits persistants sur l'utilisateur/projet
 
[CONTEXTE RAG]
  → Chunks documentaires pertinents à la requête
 
[RÉSUMÉ HISTORIQUE]
  → Résumé des échanges anciens (si sliding window active)
 
[HISTORIQUE RÉCENT]
  → N derniers échanges verbatim
 
[REQUÊTE COURANTE]
  → Message utilisateur actuel

# Anthropic — cache_control sur les blocs stables
messages = [
    {
        "role": "user",
        "content": [
            {
                "type": "text",
                "text": long_stable_document,
                "cache_control": {"type": "ephemeral"}  # TTL 5 min
            },
            {"type": "text", "text": user_query}
        ]
    }
]
# OpenAI — automatique sur les 1024 premiers tokens identiques
# Pas de configuration requise, vérifier via usage.prompt_tokens_details.cached_tokens

class ContextOverflowHandler:
    def handle(self, context: dict, budget: int) -> dict:
        current = count_tokens(context)
        if current <= budget * 0.8:
            return context  # OK, rien à faire
        if current <= budget * 0.9:
            # Alerte seulement
            log.warning(f"Context at {current/budget:.0%} — approaching limit")
            return context
        # Action : compression par niveaux
        for strategy in [self.compress_rag, self.compress_history, self.truncate_oldest]:
            context = strategy(context)
            if count_tokens(context) <= budget * 0.85:
                log.info(f"Context compressed via {strategy.__name__}")
                return context
        # Dernier recours : notifier l'utilisateur
        context["overflow_notice"] = (
            "Note : le contexte a été condensé pour respecter les limites du modèle. "
            "Certains détails anciens peuvent ne plus être accessibles."
        )
        return context

metrics = {
    "tokens_used_total": current_tokens,
    "tokens_by_layer": {
        "system": system_tokens,
        "tools": tool_tokens,
        "ltm": ltm_tokens,
        "rag": rag_tokens,
        "history": history_tokens,
    },
    "context_utilization_pct": current_tokens / max_tokens * 100,
    "compression_events": compression_count,
    "cache_hit_rate": cached_tokens / total_tokens,
    "estimated_cost_usd": estimate_cost(current_tokens, model),
}