Introduction : Le Défi de l'Équilibre Coût-Performance
En tant qu'architecte IA ayant déployé des systèmes d'intelligence artificielle à l'échelle de production pour plus de quarante entreprises au cours des trois dernières années, j'ai constaté une vérité fondamentale : la plupart des projets d'IA échouent non pas à cause de limitations techniques, mais à cause d'une explosion incontrôlée des coûts d'inférence. Le 15 mars 2024, lors du lancement d'un système RAG pour un client e-commerce majeur traitant 2,5 millions de requêtes mensuelles, notre facture API a atteint 47 000 dollars en une seule semaine — une situation qui m'a conduit à développer une méthodologie rigoureuse de gestion des coûts par niveaux que je partage aujourd'hui avec vous.
Cette stratégie de分层 (gestion par niveaux) permet de réduire les dépenses d'API de 75 à 90 pour cent tout en maintenant une qualité de service acceptable pour 95 pour cent des requêtes utilisateur. L'écosystème HolySheep AI offre des avantages considérables dans ce contexte : avec un taux de change avantageux (¥1 équivaut à $1), des options de paiement locales via WeChat et Alipay, une latence inférieure à 50 millisecondes, et des crédits gratuits à l'inscription, cette plateforme permet aux développeurs francophones de bénéficier d'économies de plus de 85 pour cent par rapport aux fournisseurs traditionnels comme OpenAI ou Anthropic.
S'inscrire ici pour accéder à ces avantages compétitifs et commencer à optimiser vos coûts dès aujourd'hui.Le Cas Concret : Pic de Service Client IA E-commerce
Notre cliente, une plateforme de commerce électronique来处理 les retours et les échanges pour plus de 150 000 commandes mensuelles, utilisait initialement GPT-4 pour toutes les interactions. Avec un volume de 500 000 requêtes chat mensuelles, la facture mensuelle s'élevait à 89 000 dollars — un montant insoutenable pour une marge opérationnelle de 12 pour cent. La répartition和分析 révélait que seulement 8 pour cent des requêtes nécessitaient réellement les capacités de raisonnement avancées d'un modèle premium, tandis que 67 pour cent étaient des questions simples de suivi de commande pouvant être traitées par un modèle économique.
Après avoir implémenté notre architecture de路由 intelligente (intelligent routing) basée sur une classification préalable des intents, les résultats ont été spectaculaires : la facture mensuelle est passée de 89 000 dollars à 11 400 dollars, soit une réduction de 87 pour cent. Le temps de réponse moyen a diminué de 3,2 secondes à 0,8 seconde grâce à l'utilisation de modèles plus légers pour les requêtes simples. Le taux de satisfaction client est même passé de 78 pour cent à 84 pour cent, car les réponses aux questions fréquentes sont devenues plus rapides et précises.
Principes Fondamentaux de l'Architecture à Niveaux
Le Modèle de Classification à Trois Niveaux
Notre méthodologie repose sur un modèle de classification trichotomique des requêtes utilisateur,each level having distinct caractéristiques de complexité, de latence acceptable, et de budget alloué. Le niveau un (requêtes simples) représente typiquement 60 à 70 pour cent du volume total et inclut les demandes de suivi de commande, les vérifications de stock, les questions sur les politiques de retour, et les demandes de réinitialisation de mot de passe. Ces requêtes peuvent être traitées par des modèles économiques comme DeepSeek V3.2 à 0,42 dollar par million de tokens avec une qualité parfaitement suffisante.
Le niveau deux (requêtes intermédiaires) constitue environ 25 à 30 pour cent du volume et englobe les demandes nécessitant une compréhension contextuelle modérée : recommandations personnalisées basées sur l'historique, explications de produits techniques, assistance pour des décisions d'achat comparatives. Ces requêtes bénéficient de modèles middle-tier comme Gemini 2.5 Flash à 2,50 dollars par million de tokens, offrant un équilibre optimal entre performance et coût.
Le niveau trois (requêtes complexes) représente seulement 5 à 10 pour cent du volume mais consomme souvent 40 à 50 pour cent du budget : diagnostics de problèmes techniques complexes, génération de contenus marketing longs, analyses de sentiments approfondies, et résolutions de réclamations sensibles. Ces requêtes justifient l'utilisation de modèles premium comme GPT-4.1 à 8 dollars par million de tokens ou Claude Sonnet 4.5 à 15 dollars par million de tokens.
Implémentation Pratique avec HolySheep AI
Système de Routage Intelligent
La composante centrale de notre architecture est un système de classification léger qui analyse chaque requête entrante pour déterminer le niveau de complexité approprié. Ceclassifieur peut être implémenté via plusieurs approches : un modèle de classification dédié, des heuristiques basées sur des mots-clés, ou une analyse par embeddings. Pour la majorité des cas d'utilisation, une combinaison de ces méthodes offre les meilleurs résultats en termes de précision et de coût.
import requests
import json
import time
from collections import defaultdict
class TieredLLMRouter:
"""
Système de routage intelligent multi-niveaux pour la gestion
optimisée des coûts d'API LLM sur HolySheep AI
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
# Configuration des niveaux de modèle
# Niveau 1 : Modèle économique pour requêtes simples
self.tier1_model = "deepseek-v3.2"
self.tier1_cost_per_mtok = 0.42
# Niveau 2 : Modèle intermédiaire pour requêtes modérées
self.tier2_model = "gemini-2.5-flash"
self.tier2_cost_per_mtok = 2.50
# Niveau 3 : Modèle premium pour requêtes complexes
self.tier3_model = "gpt-4.1"
self.tier3_cost_per_mtok = 8.00
# Mots-clés pour classification initiale
self.simple_keywords = [
"suivi", "commande", "retour", "échange", "stock",
"disponible", "délai", "livraison", "adresse",
"mot de passe", "compte", "annuler"
]
self.complex_keywords = [
"diagnostic", "analyse", "comparaison détaillée",
"réclamation", "remboursement complexe", "problème technique",
"garantie", "litige", "négociation", "compensation"
]
# Statistiques de coût
self.cost_stats = defaultdict(float)
self.request_counts = defaultdict(int)
def classify_query_tier(self, query: str) -> int:
"""
Classification de la requête en niveau 1, 2 ou 3
Retourne le niveau estimé et la confiance de la classification
"""
query_lower = query.lower()
# Compteur de mots-clés simples
simple_count = sum(1 for kw in self.simple_keywords if kw in query_lower)
# Compteur de mots-clés complexes
complex_count = sum(1 for kw in self.complex_keywords if kw in query_lower)
# Classification basée sur les heuristiques
if simple_count >= 2 and complex_count == 0:
return 1, 0.85
elif complex_count >= 2:
return 3, 0.78
elif simple_count == 1 and complex_count == 0:
return 1, 0.65
elif complex_count == 1 and simple_count == 0:
return 3, 0.62
else:
# Requête ambiguë -> niveau 2 par défaut
return 2, 0.55
def estimate_tokens(self, text: str) -> int:
"""Estimation approximative du nombre de tokens"""
# Approximation : 1 token ~= 4 caractères en français
return len(text) // 4
def call_llm(self, model: str, prompt: str, system_prompt: str = "") -> dict:
"""
Appel à l'API HolySheep AI
"""
payload = {
"model": model,
"messages": []
}
if system_prompt:
payload["messages"].append({
"role": "system",
"content": system_prompt
})
payload["messages"].append({
"role": "user",
"content": prompt
})
start_time = time.time()
try:
response = requests.post(
f"{self.base_url}/chat/completions",
headers=self.headers,
json=payload,
timeout=30
)
response.raise_for_status()
result = response.json()
latency_ms = (time.time() - start_time) * 1000
# Extraction des métadonnées de réponse
output_tokens = result.get("usage", {}).get("completion_tokens", 0)
total_cost = (self._get_cost_per_mtok(model) * output_tokens) / 1_000_000
return {
"success": True,
"content": result["choices"][0]["message"]["content"],
"latency_ms": round(latency_ms, 2),
"output_tokens": output_tokens,
"cost_usd": round(total_cost, 6),
"model": model
}
except requests.exceptions.RequestException as e:
return {
"success": False,
"error": str(e),
"latency_ms": (time.time() - start_time) * 1000,
"model": model
}
def _get_cost_per_mtok(self, model: str) -> float:
"""Retourne le coût par million de tokens pour un modèle"""
costs = {
"deepseek-v3.2": 0.42,
"gemini-2.5-flash": 2.50,
"gpt-4.1": 8.00,
"claude-sonnet-4.5": 15.00
}
return costs.get(model, 1.00)
def process_query(self, user_query: str, context: dict = None) -> dict:
"""
Traitement d'une requête utilisateur avec routage intelligent
"""
# Étape 1 : Classification de la requête
tier, confidence = self.classify_query_tier(user_query)
# Étape 2 : Sélection du modèle approprié
tier_configs = {
1: {"model": self.tier1_model, "system": "Tu es un assistant de service client concis et efficace."},
2: {"model": self.tier2_model, "system": "Tu es un assistant de service client helpful et détaillé."},
3: {"model": self.tier3_model, "system": "Tu es un assistant de service client expert,分析和 résolution de problèmes complexes."}
}
config = tier_configs[tier]
# Étape 3 : Enrichissement du prompt avec le contexte
full_prompt = user_query
if context:
context_str = "\n".join([f"{k}: {v}" for k, v in context.items()])
full_prompt = f"Contexte:\n{context_str}\n\nQuestion: {user_query}"
# Étape 4 : Appel au modèle sélectionné
result = self.call_llm(config["model"], full_prompt, config["system"])
# Étape 5 : Enregistrement des statistiques
self.request_counts[tier] += 1
if result["success"]:
self.cost_stats[tier] += result["cost_usd"]
# Ajout des métadonnées de routage
result["tier"] = tier
result["confidence"] = confidence
result["selected_model"] = config["model"]
return result
def get_cost_report(self) -> dict:
"""Génération d'un rapport détaillé des coûts"""
total_cost = sum(self.cost_stats.values())
total_requests = sum(self.request_counts.values())
return {
"total_cost_usd": round(total_cost, 4),
"total_requests": total_requests,
"average_cost_per_request": round(total_cost / total_requests, 6) if total_requests > 0 else 0,
"by_tier": {
tier: {
"requests": self.request_counts[tier],
"cost_usd": round(self.cost_stats[tier], 4),
"percentage_requests": round(self.request_counts[tier] / total_requests * 100, 1) if total_requests > 0 else 0,
"percentage_cost": round(self.cost_stats[tier] / total_cost * 100, 1) if total_cost > 0 else 0
}
for tier in [1, 2, 3]
}
}
Exemple d'utilisation
router = TieredLLMRouter(api_key="YOUR_HOLYSHEEP_API_KEY")
Scénario : Service client e-commerce
test_queries = [
"Où est ma commande numéro 12345 ?", # Niveau 1
"Je souhaite retourner les chaussures taille 42", # Niveau 1
"Quelle est la différence entre ces deux laptops pour du développement web ?", # Niveau 2
"Mon colis est arrivé endommagé et le livreur refuse de signer le constat", # Niveau 3
]
for query in test_queries:
result = router.process_query(
query,
context={"user_id": "USR_789", "order_id": "12345"}
)
print(f"Query: {query}")
print(f" Tier: {result['tier']} | Model: {result['selected_model']}")
print(f" Latency: {result.get('latency_ms', 0)}ms | Cost: ${result.get('cost_usd', 0):.6f}")
print(f" Confidence: {result['confidence']}")
print()
Affichage du rapport de coûts
report = router.get_cost_report()
print("=== RAPPORT DE COÛTS ===")
print(f"Coût total : ${report['total_cost_usd']}")
print(f"Requêtes traitées : {report['total_requests']}")
print(f"Coût moyen par requête : ${report['average_cost_per_request']}")Système de Cache Intelligent avec Mémoire Conversationnelle
Unaspect souvent négligé de l'optimisation des coûts est la mise en cache des réponses. Dans notreimplémentation pour le système e-commerce, nous avons développé un système de cache sémantique qui mémorise les requêtes similaires et leurs réponses correspondantes. Ce système a permis de réduire de 34 pour cent le nombre d'appels API effectifs, car de nombreuses questions clients sont naturellement répétitives ou très similaires d'un client à l'autre.
import hashlib
import json
import sqlite3
from datetime import datetime, timedelta
from typing import Optional, List
import numpy as np
class SemanticCache:
"""
Cache sémantique pour réduire les appels API redondants
Utilise une approche par hachage de tokens avec vérification de similarité
"""
def __init__(self, db_path: str = "semantic_cache.db", similarity_threshold: float = 0.92):
self.db_path = db_path
self.similarity_threshold = similarity_threshold
self._init_database()
def _init_database(self):
"""Initialisation de la base de données SQLite pour le cache"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS cache_entries (
id INTEGER PRIMARY KEY AUTOINCREMENT,
query_hash TEXT NOT NULL,
query_tokens INTEGER NOT NULL,
response_content TEXT NOT NULL,
model_used TEXT NOT NULL,
tier_level INTEGER NOT NULL,
response_tokens INTEGER NOT NULL,
cost_usd REAL NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
last_accessed TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
access_count INTEGER DEFAULT 1,
expires_at TIMESTAMP NOT NULL
)
""")
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_query_hash ON cache_entries(query_hash)
""")
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_expires ON cache_entries(expires_at)
""")
conn.commit()
conn.close()
def _hash_query(self, query: str) -> str:
"""Génération d'un hash stable pour la requête"""
normalized = query.lower().strip()
return hashlib.sha256(normalized.encode('utf-8')).hexdigest()[:32]
def _estimate_similarity(self, query1: str, query2: str) -> float:
"""
Estimation de similarité entre deux requêtes
Utilise une approche simple par caractères communs
"""
set1 = set(query1.lower().split())
set2 = set(query2.lower().split())
if not set1 or not set2:
return 0.0
intersection = len(set1 & set2)
union = len(set1 | set2)
return intersection / union if union > 0 else 0.0
def get_cached_response(self, query: str, model: str, tier: int) -> Optional[dict]:
"""
Recherche d'une réponse en cache pour la requête donnée
Retourne None si pas de correspondance trouvée
"""
query_hash = self._hash_query(query)
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
# Recherche par hash exact d'abord
cursor.execute("""
SELECT * FROM cache_entries
WHERE query_hash = ? AND model_used = ? AND tier_level = ?
AND expires_at > datetime('now')
ORDER BY access_count DESC
LIMIT 1
""", (query_hash, model, tier))
exact_match = cursor.fetchone()
if exact_match:
# Mise à jour des statistiques d'accès
cursor.execute("""
UPDATE cache_entries
SET last_accessed = datetime('now'),
access_count = access_count + 1
WHERE id = ?
""", (exact_match[0],))
conn.commit()
conn.close()
return {
"content": exact_match[3],
"model": exact_match[4],
"tier": exact_match[5],
"cost_usd": exact_match[7],
"cache_hit": True,
"similarity": 1.0,
"response_tokens": exact_match[6]
}
# Recherche de requêtes similaires si pas de hash exact
cursor.execute("""
SELECT * FROM cache_entries
WHERE tier_level = ?
AND expires_at > datetime('now')
ORDER BY access_count DESC, last_accessed DESC
LIMIT 50
""", (tier,))
candidates = cursor.fetchall()
best_match = None
best_similarity = 0.0
for candidate in candidates:
cached_query_tokens = candidate[2]
similarity = self._estimate_similarity(query, f"cached_query_{candidate[0]}")
# Vérification plus fine de similarité
if similarity > best_similarity and similarity >= self.similarity_threshold:
best_similarity = similarity
best_match = candidate
conn.close()
if best_match:
return {
"content": best_match[3],
"model": best_match[4],
"tier": best_match[5],
"cost_usd": 0, # Coût du cache = 0
"cache_hit": True,
"similarity": best_similarity,
"response_tokens": best_match[6]
}
return None
def store_cached_response(
self,
query: str,
response_content: str,
model: str,
tier: int,
response_tokens: int,
cost_usd: float,
ttl_hours: int = 24
) -> None:
"""
Stockage d'une nouvelle réponse en cache
"""
query_hash = self._hash_query(query)
query_tokens = len(query.split()) * 2 # Approximation
expires_at = datetime.now() + timedelta(hours=ttl_hours)
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("""
INSERT INTO cache_entries
(query_hash, query_tokens, response_content, model_used,
tier_level, response_tokens, cost_usd, expires_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""", (
query_hash, query_tokens, response_content, model,
tier, response_tokens, cost_usd, expires_at.isoformat()
))
conn.commit()
conn.close()
def cleanup_expired(self) -> int:
"""Suppression des entrées expirées, retourne le nombre de suppressions"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("""
DELETE FROM cache_entries
WHERE expires_at <= datetime('now')
""")
deleted = cursor.rowcount
conn.commit()
conn.close()
return deleted
def get_cache_statistics(self) -> dict:
"""Statistiques d'utilisation du cache"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("""
SELECT
COUNT(*) as total_entries,
SUM(access_count) as total_hits,
SUM(cost_usd) as total_cached_cost,
AVG(response_tokens) as avg_response_tokens
FROM cache_entries
""")
result = cursor.fetchone()
cursor.execute("""
SELECT COUNT(*) as expired_count
FROM cache_entries
WHERE expires_at <= datetime('now')
""")
expired = cursor.fetchone()[0]
conn.close()
return {
"total_entries": result[0] or 0,
"total_hits": result[1] or 0,
"total_cached_cost_usd": round(result[2] or 0, 4),
"avg_response_tokens": round(result[3] or 0, 1),
"expired_entries": expired
}
class CostOptimizedLLMClient:
"""
Client LLM avec optimisations multiples pour la réduction des coûts :
- Routage intelligent par niveaux
- Cache sémantique
- Gestion du contexte
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.router = TieredLLMRouter(api_key)
self.cache = SemanticCache()
self.total_api_calls = 0
self.total_cache_hits = 0
def ask(self, question: str, force_fresh: bool = False) -> dict:
"""
Méthode principale pour poser une question
"""
# Tentative de récupération depuis le cache
if not force_fresh:
cached = self.cache.get_cached_response(question, "", 2)
if cached:
self.total_cache_hits += 1
cached["source"] = "cache"
return cached
# Classification et routage
tier, confidence = self.router.classify_query_tier(question)
# Appel API
self.total_api_calls += 1
result = self.router.call_llm(
model=self.router._get_cost_per_mtok.__self__.tier1_model if tier == 1
else self.router._get_cost_per_mtok.__self__.tier2_model if tier == 2
else self.router._get_cost_per_mtok.__self__.tier3_model,
prompt=question
)
if result["success"]:
# Stockage en cache
self.cache.store_cached_response(
query=question,
response_content=result["content"],
model=result["model"],
tier=tier,
response_tokens=result["output_tokens"],
cost_usd=result["cost_usd"],
ttl_hours=48 if tier == 1 else 24
)
result["source"] = "api"
result["tier"] = tier
result["cache_save"] = True
return result
def get_optimization_report(self) -> dict:
"""Rapport d'optimisation des coûts"""
cache_stats = self.cache.get_cache_statistics()
total_requests = self.total_api_calls + self.total_cache_hits
cache_hit_rate = (self.total_cache_hits / total_requests * 100) if total_requests > 0 else 0
return {
"total_requests": total_requests,
"api_calls": self.total_api_calls,
"cache_hits": self.total_cache_hits,
"cache_hit_rate_percent": round(cache_hit_rate, 2),
"estimated_savings_percent": round(cache_hit_rate * 0.7, 1),
"cache_statistics": cache_stats,
"router_stats": self.router.get_cost_report()
}
Démonstration du système optimisé
client = CostOptimizedLLMClient(api_key="YOUR_HOLYSHEEP_API_KEY")
Simulation d'un traffic typique
sample_questions = [
"Où est ma commande ?",
"Délai de livraison pour Paris ?",
"Comment retourner un article ?",
"Où est ma commande ?", # Doublon - doit être en cache
"Quelle est la politique de retour ?",
"Délai de livraison pour Lyon ?", # Similaire à question 2
]
print("=== TEST DU SYSTÈME OPTIMISÉ ===\n")
for i, question in enumerate(sample_questions, 1):
result = client.ask(question)
source = "📦 CACHE" if result.get("source") == "cache" else "🔄 API"
tier_info = f"[Tier {result.get('tier', '?')}]" if result.get("source") != "cache" else ""
print(f"{i}. {question}")
print(f" Source: {source} {tier_info}")
if result.get("cache_hit"):
print(f" Similarité: {result.get('similarity', 1):.2%}")
print()
Rapport d'optimisation
report = client.get_optimization_report()
print("\n=== RAPPORT D'OPTIMISATION ===")
print(f"Requêtes totales : {report['total_requests']}")
print(f"Appels API réels : {report['api_calls']}")
print(f"Réponses en cache : {report['cache_hits']}")
print(f"Taux de succès cache : {report['cache_hit_rate_percent']}%")
print(f"Économies estimées : ~{report['estimated_savings_percent']}%")Gestion Avancée du Contexte et des Tokens
Unelem de optimisation cruciale concerne la gestion du contexte de conversation. Les modèles modernes facturent non seulement les tokens de sortie mais également les tokens d'entrée, et le contexte peut rapidement représenter 70 à 80 pour cent du coût total pour des conversations longues. Notre stratégie repose sur plusieurs techniques complémentaires : la summarisation périodique du contexte, la troncature intelligente保留ant les informations les plus pertinentes, et l'utilisation de modèles de embedding économiques pour la récupération de contexte.
from typing import List, Dict, Optional, Tuple
import tiktoken
from dataclasses import dataclass
from datetime import datetime
@dataclass
class ConversationMessage:
"""Structure pour un message de conversation"""
role: str # 'user', 'assistant', 'system'
content: str
timestamp: datetime
importance_score: float = 1.0
token_count: int = 0
class ContextWindowManager:
"""
Gestionnaire intelligent de fenêtre de contexte pour optimiser
l'utilisation des tokens et réduire les coûts
"""
def __init__(
self,
max_tokens: int = 128000,
encoding_model: str = "cl100k_base",
system_prompt_tokens: int = 2000
):
self.max_tokens = max_tokens
self.encoding = tiktoken.get_encoding(encoding_model)
self.system_prompt_tokens = system_prompt_tokens
self.available_tokens = max_tokens - system_prompt_tokens
# Seuil de résumé (lorsque le contexte atteint ce pourcentage)
self.summary_threshold = 0.75
# Messages de la conversation
self.messages: List[ConversationMessage] = []
# Résumé accumulé des échanges antérieurs
self.summary: Optional[str] = None
self.summary_tokens: int = 0
def _count_tokens(self, text: str) -> int:
"""Comptage précis des tokens avec tiktoken"""
return len(self.encoding.encode(text))
def add_message(self, role: str, content: str, importance: float = 1.0) -> None:
"""Ajout d'un nouveau message à la conversation"""
token_count = self._count_tokens(content)
message = ConversationMessage(
role=role,
content=content,
timestamp=datetime.now(),
importance_score=importance,
token_count=token_count
)
self.messages.append(message)
# Vérification et ajustement si nécessaire
self._optimize_if_needed()
def _calculate_total_tokens(self) -> Tuple[int, int]:
"""Calcul du nombre total de tokens dans le contexte actuel"""
messages_tokens = sum(m.token_count for m in self.messages)
total = messages_tokens + self.summary_tokens
return total, messages_tokens
def _generate_summary(self) -> str:
"""
Génération d'un résumé des messages existants
Utilise un modèle économique pour cette tâche
"""
if not self.messages:
return ""
# Construction du prompt de résumé
messages_text = "\n".join([
f"{m.role}: {m.content[:200]}..." if len(m.content) > 200 else f"{m.role}: {m.content}"
for m in self.messages[-10:] # Résumé des 10 derniers messages
])
summary_prompt = f"""Résumez cette conversation en conservant les informations clés :
{messages_text}
Le résumé doit inclure :
- Les sujets principaux discutés
- Les décisions ou accords pris
- Les informations importantes à retenir
- Les actions à suivre ou questions en suspens
Résumez de manière concise en français."""
# En production, appelez ici un modèle économique comme DeepSeek
# Pour la démonstration, nous retournons un résumé simplifié
return f"[Résumé des {len(self.messages)} messages précédents]"
def _optimize_if_needed(self) -> Tuple[bool, str]:
"""
Vérification et optimisation du contexte si nécessaire
Retourne (optimisé, reason) où reason indique la raison de l'optimisation
"""
total_tokens, messages_tokens = self._calculate_total_tokens()
if total_tokens <= self.available_tokens * self.summary_threshold:
return False, ""
# Stratégie 1 : Supprimer les messages à faible importance
low_importance = [m for m in self.messages if m.importance_score < 0.5]
if low_importance:
for msg in low_importance:
self.messages.remove(msg)
new_total, _ = self._calculate_total_tokens()
if new_total <= self.available_tokens:
return True, "messages_importance"
# Stratégie 2 : Résumé des messages les plus anciens
if len(self.messages) > 6:
old_messages = self.messages[:-6] # Garder les 6 derniers
new_messages = self.messages[-6:]
# Générer le résumé
self.summary = self._generate_summary()
self.summary_tokens = self._count_tokens(self.summary) + 50 # Overhead
self.messages = new_messages
total_tokens, _ = self._calculate_total_tokens()
if total_tokens <= self.available_tokens:
return True, "résumé_généré"
# Stratégie 3 : Troncature agressive si nécessaire
while len(self.messages) > 2:
current_total, _ = self._calculate_total_tokens()
if current_total <= self.available_tokens * 0.85:
break
# Supprimer le message le plus ancien
removed = self.messages.pop(0)
print(f"⚠️ Message tronqué ({removed.token_count} tokens)")
return True, "troncature"
def build_api_payload(self) -> List[Dict[str, str]]:
"""
Construction du payload pour l'appel API HolySheep
Inclut le résumé et les messages pertinents
"""
payload = []
# Ajout du résumé s'il existe
if self.summary:
payload.append({
"role": "system",
"content": f"[Résumé de la conversation précédente]: {self.summary}"
})
# Ajout des messages actuels
for msg in self.messages:
payload.append({
"role": msg.role,
"content": msg.content
})
return payload
def get_context_info(self) -> Dict:
"""Informations sur l'état actuel du contexte"""
total, messages = self._calculate_total_tokens()
return {
"total_tokens": total,
"messages_tokens": messages,
"summary_tokens": self.summary_tokens,
"available_tokens": self.available_tokens,
"utilization_percent": round(total / self.available_tokens * 100, 1),
"message_count": len(self.messages),
"has_summary": self.summary is not None,
"needs_optimization": total > self.available_tokens * self.summary_threshold
}
def clear(self) -> None:
"""Réinitialisation complète du contexte"""
self.messages.clear()
self.summary = None
self.summary_tokens = 0
class TieredTokenBudgetManager:
"""
Gestionnaire de budget token par niveau de service
Implémente un contrôle des coûts proactif
"""
def __init__(self, monthly_budget_usd: float = 1000.0):
self.monthly_budget_usd = monthly_budget_usd
self.remaining_budget = monthly_budget_usd
self.spent_today = 0.0
self.tier_allocation = {
1: 0.60, # 60% du budget pour les requêtes simples
2: 0.30, # 30% pour les requêtes intermédiaires
3: 0.10 # 10