khalilbenaz/claude-skills-collection/agent-cost-optimizer

# Structure minimale de log — ajouter à chaque appel LLM
import anthropic, time
client = anthropic.Anthropic()
def call_with_cost_log(messages, model, system=""):
    t0 = time.time()
    resp = client.messages.create(
        model=model, max_tokens=1024,
        system=system, messages=messages
    )
    cost = (
        resp.usage.input_tokens  * pricing[model]["in"]  +
        resp.usage.output_tokens * pricing[model]["out"]
    )
    print(f"[COST] model={model} in={resp.usage.input_tokens} "
          f"out={resp.usage.output_tokens} cost_usd={cost:.5f} "
          f"latency_ms={int((time.time()-t0)*1000)}")
    return resp

ROUTING_RULES = {
    "simple":   "claude-haiku-3-5",   # classification, extraction, reformulation
    "moderate": "claude-sonnet-4",    # raisonnement, synthèse, code standard
    "complex":  "claude-opus-4",      # architecture, décisions critiques, audit
}
def classify_task(prompt: str) -> str:
    """Classifier cheap (Haiku) pour décider du modèle à utiliser."""
    resp = client.messages.create(
        model="claude-haiku-3-5", max_tokens=10,
        messages=[{"role": "user", "content":
            f"Complexity of this task (simple/moderate/complex):\n{prompt[:300]}"}]
    )
    return resp.content[0].text.strip().lower()
def route(prompt: str):
    level = classify_task(prompt)
    return ROUTING_RULES.get(level, "claude-sonnet-4")

# MAUVAIS (verbose)
Tu es un assistant utile et bienveillant. Lorsque l'utilisateur te pose une question,
tu dois répondre de manière claire et structurée en utilisant le format JSON...
 
# BON (concis)
Réponds UNIQUEMENT en JSON valide. Schéma: {"answer": str, "confidence": float}

response = client.messages.create(
    model="claude-sonnet-4",
    max_tokens=1024,
    system=[{
        "type": "text",
        "text": LONG_SYSTEM_PROMPT,          # > 1024 tokens pour activer le cache
        "cache_control": {"type": "ephemeral"}
    }],
    messages=messages
)
# Les appels suivants avec le même system prompt utilisent le cache (prix cache_read << prix input)

MAX_CONTEXT_TOKENS = 4000  # seuil avant résumé
def maybe_summarize(messages: list, model="claude-haiku-3-5") -> list:
    total = sum(len(m["content"].split()) * 1.3 for m in messages)  # estimation rapide
    if total < MAX_CONTEXT_TOKENS:
        return messages
    # Résumer les messages anciens (garder les 4 derniers intacts)
    to_summarize = messages[:-4]
    summary_resp = client.messages.create(
        model=model, max_tokens=300,
        messages=[{"role": "user", "content":
            "Résume en 200 mots max:\n" + "\n".join(m["content"] for m in to_summarize)}]
    )
    return [{"role": "assistant", "content": "[Résumé] " + summary_resp.content[0].text}] + messages[-4:]

import hashlib, json
from functools import lru_cache
# Cache exact (Redis recommandé en prod)
cache: dict = {}
def cached_llm_call(prompt: str, model: str) -> str:
    key = hashlib.sha256(f"{model}:{prompt}".encode()).hexdigest()
    if key in cache:
        return cache[key]
    result = client.messages.create(model=model, max_tokens=512,
        messages=[{"role": "user", "content": prompt}])
    text = result.content[0].text
    cache[key] = text
    return text

# Batch API Anthropic — jusqu'à 50 % de réduction, délai < 24 h
curl https://api.anthropic.com/v1/messages/batches \
  -H "x-api-key: $ANTHROPIC_API_KEY" \
  -H "anthropic-version: 2023-06-01" \
  -d '{
    "requests": [
      {"custom_id": "task-1", "params": {"model": "claude-haiku-3-5", "max_tokens": 256,
        "messages": [{"role": "user", "content": "Classe ce texte: ..."}]}},
      {"custom_id": "task-2", "params": {"model": "claude-haiku-3-5", "max_tokens": 256,
        "messages": [{"role": "user", "content": "Résume: ..."}]}}
    ]
  }'

# MAUVAIS : 3 appels séquentiels = 3 tours agent
get_weather("Paris")  # appel 1
get_price("AAPL")     # appel 2
translate("hello")    # appel 3
# BON : parallel tool use (un seul tour agent)
# L'agent émet les 3 tool_use en une seule réponse si les descriptions sont claires

class BudgetGuard:
    def __init__(self, max_usd: float):
        self.max_usd = max_usd
        self.spent = 0.0
    def charge(self, cost: float):
        self.spent += cost
        if self.spent >= self.max_usd * 0.8:
            print(f"[WARN] 80% du budget atteint ({self.spent:.4f}$/{self.max_usd}$)")
        if self.spent >= self.max_usd:
            raise RuntimeError(f"Budget dépassé : {self.spent:.4f}$ > {self.max_usd}$")