En tant qu'ingénieur senior ayant optimisé des systèmes d'agents IA pour des millions de requêtes quotidiennes, je peux vous affirmer sans détour : la différence entre un agent IA performant et un système qui s'effondre sous la charge se joue entièrement dans la maîtrise des métriques de performance et l'identification précise des goulots d'étranglement.
Cas Concret : Le Pic du Black Friday qui a Tout Changé
L'année dernière, lors du Black Friday, notre système d'agent IA pour un grand e-commerce européen a commencé à montrer des signes de fatigue dès le début de la soirée promotionnelle. Les temps de réponse qui habituellement se maintenaient autour de 800ms ont soudainement bondi à 4-5 secondes. Notre système de support client par chatbot commençait à accumuler des files d'attente impressionnantes.
C'est à ce moment précis que j'ai compris l'importance critique d'avoir des outils robustes de profiling et d'analyse de bottlenecks. En moins de deux heures d'investigation avec les bonnes métriques, nous avions identifié trois problèmes majeurs : un timeout mal configuré sur les appels RAG, une mémoire cache insuffisante pour les embeddings, et une latence réseau inhabituelle vers notre ancien fournisseur d'API.
Cette expérience m'a convaincu de développer une boîte à outils complète pour l'analyse de performance des agents IA. Aujourd'hui, je vais vous la présenter en détail.
Comprendre l'Architecture d'un Agent IA Moderne
Avant de plongez dans les outils, il est essentiel de comprendre les composants critiques d'un agent IA qui peuvent devenir des points de friction :
- La couche de raisonnement : Le modèle de langue lui-même, avec son temps de think time et de génération
- Le système de mémoire : Conversation history, vecteur database, cache de contexte
- Les outils externes : Appels API, accès filesystem, recherche web
- Le pipeline d'entrée : Parsing, embedding, retrieval
- Le pipeline de sortie : Post-processing, formatting, validation
Chaque composant peut devenir un bottleneck selon la charge et la configuration. S'inscrire ici vous donne accès à une infrastructure optimisée avec une latence moyenne de moins de 50ms, ce qui élimine déjà une grande partie des problèmes de performance réseau.
Architecture de Monitoring Recommandée
Je vous recommande une architecture de monitoring en trois couches qui a fait ses preuves sur des systèmes traitant plus de 100 000 requêtes par jour :
import time
import psutil
import json
from typing import Dict, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from collections import defaultdict
import asyncio
from aiohttp import ClientSession
class PerformanceAnalyzer:
"""
Analyseur de performance pour agents IA.
Mesure les latences, identifie les bottlenecks et génère des rapports détaillés.
"""
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self.metrics = defaultdict(list)
self.bottleneck_thresholds = {
"api_call": 1000, # ms
"retrieval": 500,
"embedding": 300,
"total": 3000
}
async def call_model(self, messages: List[Dict], model: str = "gpt-4.1") -> Dict:
"""Appel au modèle avec mesure de latence complète."""
start_total = time.perf_counter()
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"max_tokens": 2000,
"temperature": 0.7
}
start_api = time.perf_counter()
async with ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=30)
) as response:
result = await response.json()
latency_api = (time.perf_counter() - start_api) * 1000
total_latency = (time.perf_counter() - start_total) * 1000
metrics_record = {
"timestamp": datetime.now().isoformat(),
"model": model,
"api_latency_ms": round(latency_api, 2),
"total_latency_ms": round(total_latency, 2),
"tokens_used": result.get("usage", {}).get("total_tokens", 0),
"status": response.status
}
self.metrics["api_calls"].append(metrics_record)
self._check_bottlenecks(metrics_record)
return result
def _check_bottlenecks(self, metrics: Dict) -> None:
"""Identifie les métriques dépassant les seuils critiques."""
bottlenecks = []
if metrics["api_latency_ms"] > self.bottleneck_thresholds["api_call"]:
bottlenecks.append({
"type": "API_LATENCY_HIGH",
"value": metrics["api_latency_ms"],
"threshold": self.bottleneck_thresholds["api_call"],
"severity": "HIGH" if metrics["api_latency_ms"] > 2000 else "MEDIUM"
})
if metrics["total_latency_ms"] > self.bottleneck_thresholds["total"]:
bottlenecks.append({
"type": "TOTAL_LATENCY_HIGH",
"value": metrics["total_latency_ms"],
"threshold": self.bottleneck_thresholds["total"],
"severity": "CRITICAL"
})
if bottlenecks:
self.metrics["bottlenecks"].extend(bottlenecks)
def get_performance_report(self) -> Dict:
"""Génère un rapport complet de performance."""
if not self.metrics["api_calls"]:
return {"error": "Aucune métrique disponible"}
api_calls = self.metrics["api_calls"]
latencies = [m["api_latency_ms"] for m in api_calls]
total_latencies = [m["total_latency_ms"] for m in api_calls]
report = {
"generated_at": datetime.now().isoformat(),
"total_requests": len(api_calls),
"api_latency": {
"avg_ms": round(sum(latencies) / len(latencies), 2),
"min_ms": round(min(latencies), 2),
"max_ms": round(max(latencies), 2),
"p95_ms": round(sorted(latencies)[int(len(latencies) * 0.95)], 2),
"p99_ms": round(sorted(latencies)[int(len(latencies) * 0.99)], 2)
},
"total_latency": {
"avg_ms": round(sum(total_latencies) / len(total_latencies), 2),
"min_ms": round(min(total_latencies), 2),
"max_ms": round(max(total_latencies), 2)
},
"bottleneck_summary": self._summarize_bottlenecks(),
"recommendations": self._generate_recommendations()
}
return report
def _summarize_bottlenecks(self) -> Dict:
"""Résume les bottlenecks détectés."""
bottleneck_counts = defaultdict(int)
for bp in self.metrics.get("bottlenecks", []):
bottleneck_counts[bp["type"]] += 1
return dict(bottleneck_counts)
def _generate_recommendations(self) -> List[str]:
"""Génère des recommandations basées sur les métriques."""
recommendations = []
api_calls = self.metrics.get("api_calls", [])
if api_calls:
avg_latency = sum(m["api_latency_ms"] for m in api_calls) / len(api_calls)
if avg_latency > 1000:
recommendations.append(
"Latence API élevée détectée. Envisagez de migrer vers "
"HolySheep AI qui offre des latences <50ms en moyenne."
)
if len(api_calls) > 100:
recommendations.append(
"Volume élevé de requêtes détecté. Mettez en place un "
"système de mise en cache pour les prompts similaires."
)
return recommendations
Exemple d'utilisation
analyzer = PerformanceAnalyzer(api_key="YOUR_HOLYSHEEP_API_KEY")
async def test_agent_performance():
"""Teste la performance d'un agent simple."""
messages = [
{"role": "system", "content": "Vous êtes un assistant de support e-commerce."},
{"role": "user", "content": "Quel est le statut de ma commande #12345?"}
]
result = await analyzer.call_model(messages, model="gpt-4.1")
report = analyzer.get_performance_report()
print(f"Rapport de performance généré :")
print(f"- Latence moyenne API : {report['api_latency']['avg_ms']}ms")
print(f"- P95 : {report['api_latency']['p95_ms']}ms")
print(f"- Requêtes totales : {report['total_requests']}")
Exécuter le test
asyncio.run(test_agent_performance())Tableaux de Bord et Visualisation des Métriques
La visualisation est cruciale pour identifier les patterns de performance. Voici un système de tableaux de bord complet avec Prometheus et Grafana :
version: '3.8'
services:
prometheus:
image: prom/prometheus:latest
ports:
- "9090:9090"
volumes:
- ./prometheus.yml:/etc/prometheus/prometheus.yml
command:
- '--config.file=/etc/prometheus/prometheus.yml'
grafana:
image: grafana/grafana:latest
ports:
- "3000:3000"
environment:
- GF_SECURITY_ADMIN_PASSWORD=admin
volumes:
- ./grafana/provisioning:/etc/grafana/provisioning
agent_metrics_exporter:
build: ./exporter
ports:
- "8000:8000"
environment:
- HOLYSHEEP_API_KEY=${HOLYSHEEP_API_KEY}
- HOLYSHEEP_BASE_URL=https://api.holysheep.ai/v1
labels:
- "prometheus.port=8000"# Configuration Prometheus
global:
scrape_interval: 15s
evaluation_interval: 15s
scrape_configs:
- job_name: 'ai-agent-metrics'
static_configs:
- targets: ['agent_metrics_exporter:8000']
metrics_path: '/metrics'
- job_name: 'node_exporter'
static_configs:
- targets: ['node_exporter:9100']
rule_files:
- 'alerts.yml'"""
Exportateur de métriques Prometheus pour agents IA.
Expose les métriques de performance au format Prometheus.
"""
from prometheus_client import Counter, Histogram, Gauge, generate_latest
from fastapi import FastAPI, Response
import psutil
import time
import os
app = FastAPI(title="AI Agent Metrics Exporter")
Compteurs Prometheus
api_requests_total = Counter(
'ai_agent_api_requests_total',
'Total des appels API',
['model', 'status']
)
api_latency_seconds = Histogram(
'ai_agent_api_latency_seconds',
'Latence des appels API en secondes',
['model'],
buckets=[0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0]
)
token_usage_total = Counter(
'ai_agent_token_usage_total',
'Total des tokens utilisés',
['model', 'type']
)
active_requests = Gauge(
'ai_agent_active_requests',
'Nombre de requêtes actives'
)
cache_hit_ratio = Gauge(
'ai_agent_cache_hit_ratio',
'Ratio de hits du cache'
)
Métriques système
cpu_usage = Gauge('ai_agent_cpu_percent', 'Utilisation CPU en pourcentage')
memory_usage = Gauge('ai_agent_memory_bytes', 'Utilisation mémoire en octets')
@app.get("/metrics")
async def metrics():
"""Endpoint Prometheus pour les métriques."""
update_system_metrics()
return Response(content=generate_latest(), media_type="text/plain")
def update_system_metrics():
"""Met à jour les métriques système."""
cpu_usage.set(psutil.cpu_percent(interval=0.1))
memory_usage.set(psutil.virtual_memory().used)
@app.post("/record_request")
async def record_request(
model: str,
status: str,
latency: float,
tokens: int,
cache_hit: bool = False
):
"""Enregistre une requête pour les métriques."""
api_requests_total.labels(model=model, status=status).inc()
api_latency_seconds.labels(model=model).observe(latency)
token_usage_total.labels(model=model, type="total").inc(tokens)
if cache_hit:
cache_hit_ratio.set(1.0)
else:
cache_hit_ratio.set(0.0)
return {"recorded": True}
Endpoint de santé
@app.get("/health")
async def health():
return {
"status": "healthy",
"uptime_seconds": time.time() - start_time,
"api_key_configured": bool(os.getenv("HOLYSHEEP_API_KEY"))
}
start_time = time.time()
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)Calculateur d'Économie et Optimisation des Coûts
L'un des aspects les plus importants de l'analyse de performance est la compréhension de l'impact financier. Voici un calculateur que j'utilise pour optimiser les coûts tout en maintenant des performances acceptables :
class CostOptimizer:
"""
Optimiseur de coûts pour appels API IA.
Compare les providers et optimise les performances.
"""
# Prix 2026/1M tokens (mise à jour Mars 2026)
MODEL_PRICES = {
"gpt-4.1": {"input": 8.00, "output": 8.00}, # $/1M tokens
"claude-sonnet-4.5": {"input": 15.00, "output": 15.00},
"gemini-2.5-flash": {"input": 2.50, "output": 2.50},
"deepseek-v3.2": {"input": 0.42, "output": 0.42}
}
def __init__(self, base_url: str = "https://api.holysheep.ai/v1"):
self.base_url = base_url
self.holysheep_advantage = 0.15 # 85% d'économie avec HolySheep
def calculate_cost(
self,
model: str,
input_tokens: int,
output_tokens: int
) -> Dict:
"""Calcule le coût pour un modèle donné."""
prices = self.MODEL_PRICES.get(model, {"input": 0, "output": 0})
input_cost = (input_tokens / 1_000_000) * prices["input"]
output_cost = (output_tokens / 1_000_000) * prices["output"]
total_cost = input_cost + output_cost
return {
"model": model,
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"input_cost_usd": round(input_cost, 4),
"output_cost_usd": round(output_cost, 4),
"total_cost_usd": round(total_cost, 4),
"total_cost_cny": round(total_cost * 7.2, 2)
}
def compare_providers(self, input_tokens: int, output_tokens: int) -> List[Dict]:
"""Compare les coûts entre tous les providers."""
comparisons = []
for model in self.MODEL_PRICES:
cost = self.calculate_cost(model, input_tokens, output_tokens)
comparisons.append(cost)
comparisons.sort(key=lambda x: x["total_cost_usd"])
# Ajouter l'économie HolySheep
cheapest = comparisons[0]
for comp in comparisons:
if comp["model"] == "deepseek-v3.2":
comp["vs_holysheep_savings"] = "Same pricing tier via HolySheep"
else:
savings = comp["total_cost_usd"] - cheapest["total_cost_usd"]
comp["savings_vs_deepseek_usd"] = round(savings, 4)
return comparisons
def estimate_monthly_cost(
self,
daily_requests: int,
avg_input_tokens: int,
avg_output_tokens: int,
model: str = "deepseek-v3.2"
) -> Dict:
"""Estime le coût mensuel avec projections."""
daily_cost = self.calculate_cost(
model, avg_input_tokens, avg_output_tokens
)["total_cost_usd"] * daily_requests
monthly_cost = daily_cost * 30
yearly_cost = monthly_cost * 12
# Comparaison avec GPT-4.1
gpt_cost = self.calculate_cost(
"gpt-4.1", avg_input_tokens, avg_output_tokens
)["total_cost_usd"] * daily_requests * 30
return {
"model": model,
"daily_requests": daily_requests,
"daily_cost_usd": round(daily_cost, 2),
"monthly_cost_usd": round(monthly_cost, 2),
"yearly_cost_usd": round(yearly_cost, 2),
"vs_gpt4_savings_usd": round(gpt_cost - monthly_cost, 2),
"savings_percentage": round((1 - monthly_cost / gpt_cost) * 100, 1)
}
def optimize_for_performance_budget(
self,
max_budget_usd: float,
avg_input_tokens: int,
avg_output_tokens: int,
required_quality: str = "high"
) -> Dict:
"""
Recommande le meilleur modèle selon le budget et les besoins.
quality: 'high' (GPT-4.1), 'medium' (Claude), 'fast' (Gemini/DeepSeek)
"""
recommendations = {
"high": "gpt-4.1",
"medium": "claude-sonnet-4.5",
"fast": "deepseek-v3.2"
}
model = recommendations.get(required_quality, "deepseek-v3.2")
cost = self.calculate_cost(model, avg_input_tokens, avg_output_tokens)
max_requests = int(max_budget_usd / cost["total_cost_usd"]) if cost["total_cost_usd"] > 0 else 0
return {
"recommended_model": model,
"cost_per_request_usd": cost["total_cost_usd"],
"max_monthly_requests": max_requests,
"budget_used_percentage": 100,
"alternative_with_savings": {
"model": "deepseek-v3.2",
"cost_per_request_usd": self.calculate_cost(
"deepseek-v3.2", avg_input_tokens, avg_output_tokens
)["total_cost_usd"],
"additional_requests": int(max_budget_usd / self.calculate_cost(
"deepseek-v3.2", avg_input_tokens, avg_output_tokens
)["total_cost_usd"]) - max_requests if max_requests > 0 else 0
}
}
Démonstration
optimizer = CostOptimizer()
Exemple : Agent e-commerce
print("=== Analyse de Coût pour Agent E-commerce ===")
print(f"Requêtes/jour: 10,000")
print(f"Tokens输入 moyens: 500")
print(f"Tokens输出 moyens: 300")
print()
monthly = optimizer.estimate_monthly_cost(
daily_requests=10000,
avg_input_tokens=500,
avg_output_tokens=300,
model="deepseek-v3.2"
)
print(f"Coût mensuel (DeepSeek V3.2): ${monthly['monthly_cost_usd']}")
print(f"Économie vs GPT-4.1: ${monthly['vs_gpt4_savings_usd']} ({monthly['savings_percentage']}%)")
print()
print("=== Comparaison des Providers ===")
comparisons = optimizer.compare_providers(500, 300)
for comp in comparisons:
print(f"{comp['model']}: ${comp['total_cost_usd']:.4f}")
print()
print("=== Optimisation Budget ===")
budget_opt = optimizer.optimize_for_performance_budget(
max_budget_usd=100.0,
avg_input_tokens=500,
avg_output_tokens=300,
required_quality="fast"
)
print(f"Modèle recommandé: {budget_opt['recommended_model']}")
print(f"Requêtes max/mois: {budget_opt['max_monthly_requests']:,}")Identification et Résolution des Bottlenecks
Après des années d'expérience avec des systèmes multi-agents, j'ai identifié les patterns de bottlenecks les plus fréquents et leurs solutions :
- Bottleneck réseau : Latence >100ms entre le serveur et l'API. Solution : Choisir un provider avec des serveurs géographiquement proches.
- Bottleneck de contexte : Fenêtre de contexte saturée导致 des réponses incomplètes. Solution : Implémenter une stratégie de résumé et de fenêtre glissante.
- Bottleneck de retrieval : Temps de recherche dans la base vectorielle trop long. Solution : Optimiser les index et utiliser le filtrage préalable.
- Bottleneck de mémoire : Consommation RAM excessive. Solution : Implémenter un système de pagination de l'historique.
Avec HolySheep AI, la latence réseau est en moyenne de 50ms, ce qui élimine automatiquement la majorité des bottlenecks de ce type. Combiné à leur système de cache intelligent, les coûts d'API peuvent être réduits de 40% supplémentaires sur les requêtes similaires.
Erreurs Courantes et Solutions
Après avoir analysé des centaines de systèmes d'agents IA en production, voici les trois erreurs les plus fréquentes et leurs solutions éprouvées :
1. Timeout mal configuré causant des échecs en cascade
# ERREUR : Timeout trop court pour les gros modèles
async def bad_call():
async with ClientSession().post(
url,
timeout=aiohttp.ClientTimeout(total=5) # 5 secondes insufficient
):
pass
SOLUTION : Timeout adaptatif selon le modèle et la complexité
class AdaptiveTimeoutManager:
"""Gestionnaire de timeout adaptatif."""
BASE_TIMEOUTS = {
"deepseek-v3.2": 15, # Modèle rapide
"gemini-2.5-flash": 20, # Modèle rapide
"gpt-4.1": 30, # Modèle complexe
"claude-sonnet-4.5": 35 # Modèle complexe
}
def __init__(self):
self.timeout_history = defaultdict(list)
def calculate_timeout(
self,
model: str,
estimated_input_tokens: int,
complexity_factor: float = 1.0
) -> float:
"""Calcule un timeout adapté à la requête."""
base = self.BASE_TIMEOUTS.get(model, 30)
# Ajuster selon la taille de l'input
token_factor = max(1.0, estimated_input_tokens / 1000)
# Ajuster selon l'historique (exponentional moving average)
history = self.timeout_history.get(model, [])
if history:
avg_past = sum(history[-10:]) / len(history[-10:])
history_factor = avg_past / base
else:
history_factor = 1.0
timeout = base * token_factor * complexity_factor * history_factor
# Plafonner à 60 secondes maximum
return min(timeout, 60.0)
def record_timeout(self, model: str, actual_duration: float):
"""Enregistre la durée réelle pour affiner les estimations."""
self.timeout_history[model].append(actual_duration)
# Garder seulement les 100 dernières mesures
if len(self.timeout_history[model]) > 100:
self.timeout_history[model] = self.timeout_history[model][-100:]
Utilisation
timeout_manager = AdaptiveTimeoutManager()
timeout = timeout_manager.calculate_timeout(
model="gpt-4.1",
estimated_input_tokens=2000,
complexity_factor=1.2
)
print(f"Timeout recommandé: {timeout:.1f} secondes")2. Pas de gestion des rate limits 导致 des erreurs 429
# ERREUR : Appels directs sans gestion de rate limit
async def bad_batch_processing(messages: List[str]):
results = []
for msg in messages: # Boucle directe = rate limit immédiate
result = await call_api(msg)
results.append(result)
return results
SOLUTION : Rate limiter intelligent avec exponential backoff
import asyncio
from datetime import datetime, timedelta
from collections import deque
class RateLimitedClient:
"""Client avec gestion intelligente des rate limits."""
def __init__(
self,
api_key: str,
base_url: str = "https://api.holysheep.ai/v1",
requests_per_minute: int = 60,
burst_size: int = 10
):
self.api_key = api_key
self.base_url = base_url
self.rpm_limit = requests_per_minute
self.burst_size = burst_size
# File d'attente des requêtes
self.request_queue = asyncio.Queue()
self.processing = True
# Contrôle de rate
self.request_timestamps = deque(maxlen=requests_per_minute)
self.burst_tokens = burst_size
# Métriques
self.total_requests = 0
self.rate_limit_hits = 0
# Démarrer le worker
self.worker_task = asyncio.create_task(self._queue_worker())
async def _queue_worker(self):
"""Worker qui traite les requêtes en respectant les limits."""
while self.processing:
try:
# Récupérer une requête de la queue
future = await asyncio.wait_for(
self.request_queue.get(),
timeout=1.0
)
# Vérifier les limites de rate
await self._wait_for_rate_limit()
# Traiter la requête
try:
result = await self._make_request()
future.set_result(result)
except Exception as e:
future.set_exception(e)
self.total_requests += 1
except asyncio.TimeoutError:
continue
except Exception as e:
print(f"Worker error: {e}")
async def _wait_for_rate_limit(self):
"""Attend que les limites de rate soient respectées."""
now = datetime.now()
cutoff = now - timedelta(minutes=1)
# Nettoyer les timestamps anciens
while self.request_timestamps and self.request_timestamps[0] < cutoff:
self.request_timestamps.popleft()
# Vérifier si on a atteint la limite
if len(self.request_timestamps) >= self.rpm_limit:
sleep_time = (self.request_timestamps[0] - cutoff).total_seconds() + 0.1
await asyncio.sleep(sleep_time)
# Vérifier les burst tokens
if self.burst_tokens <= 0:
await asyncio.sleep(1.0)
self.burst_tokens = self.burst_size
else:
self.burst_tokens -= 1
self.request_timestamps.append(datetime.now())
async def _make_request(self) -> Dict:
"""Fait une requête API (à implémenter selon vos besoins)."""
# Implémentation réelle de l'appel API
pass
async def call(self, data: Dict) -> Dict:
"""Interface publique pour faire des appels limités."""
future = asyncio.Future()
await self.request_queue.put(future)
try:
return await asyncio.wait_for(future, timeout=60.0)
except asyncio.TimeoutError:
self.rate_limit_hits += 1
raise TimeoutError("Request timed out due to rate limiting")
async def close(self):
"""Ferme proprement le client."""
self.processing = False
await self.worker_task
Utilisation
async def batch_processing():
client = RateLimitedClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
requests_per_minute=60,
burst_size=10
)
messages = [f"Message {i}" for i in range(100)]
tasks = [client.call({"content": msg}) for msg in messages]
results = await asyncio.gather(*tasks, return_exceptions=True)
await client.close()
success = sum(1 for r in results if isinstance(r, dict))
print(f"Succès: {success}/{len(results)}")
print(f"Rate limit hits: {client.rate_limit_hits}")3. Cache mal implémenté causant des incohérences
# ERREUR : Cache sans TTL ni invalidation
bad_cache = {}
async def bad_cached_call(prompt: str):
if prompt in bad_cache: # Cache permanent = données stale
return bad_cache[prompt]
result = await call_api(prompt)
bad_cache[prompt] = result
return result
SOLUTION : Cache intelligent avec TTL et invalidation
from hashlib import sha256
import json
class IntelligentCache:
"""Cache intelligent avec TTL et invalidation contextuelle."""
def __init__(
self,
default_ttl: int = 300, # 5 minutes
max_size: int = 10000,
enable_adaptive_ttl: bool = True
):
self.default_ttl = default_ttl
self.max_size = max_size
self.enable_adaptive_ttl = enable_adaptive_ttl
self._cache = {}
self._metadata = {}
self._hits = 0
self._misses = 0
# Démarrer le cleanup périodique
self._cleanup_task = asyncio.create_task(self._periodic_cleanup())
def _generate_key(self, prompt: str, model: str, **params) -> str:
"""Génère une clé de cache robuste."""
content = json.dumps({
"prompt": prompt,
"model": model,
"params": params
}, sort_keys=True)
return sha256(content.encode()).hexdigest()[:32]
def _calculate_ttl(self, prompt: str, model: str) -> int:
"""Calcule un TTL adaptatif selon le contexte."""
if not self.enable_adaptive_ttl:
return self.default_ttl
base_ttl = self.default_ttl
# TTL plus court pour les requêtes sensibles au temps
time_sensitive_keywords = ["prix", "stock", "disponible", "status"]
if any(kw in prompt.lower() for kw in time_sensitive_keywords):
base_ttl = 60 # 1 minute pour données volatiles
# TTL plus long pour les connaissances générales
static_keywords = ["expliquer", "définir", "qu'est-ce que"]
if any(kw in prompt.lower() for kw in static_keywords):
base_ttl = 3600 # 1 heure pour connaissances statiques
return base_ttl
async def get(
self,
prompt: str,
model: str = "deepseek-v3.2",
**params
) -> Optional[Dict]:
"""Récupère une valeur du cache si valide."""
key = self._generate_key(prompt, model, **params)
if key in self._cache:
metadata = self._metadata[key]
# Vérifier l'expiration
if datetime.now() < metadata["expires_at"]:
self._hits += 1
metadata["access_count"] += 1
metadata["last_access"] = datetime.now()
return self._cache[key]
else:
# Expiré, nettoyer
del self._cache[key]
del self._metadata[key]
self._misses += 1
return None
async def set(
self,
prompt: str,
model: str,
value: Dict,
**params
) -> None:
"""Stocke une valeur dans le cache."""
key = self._generate_key(prompt, model, **params)
ttl = self._calculate_ttl(prompt, model)
# Éviction LRU si plein
if len(self._cache) >= self.max_size:
await self._evict_lru()
self._cache[key] = value
self._metadata[key] = {
"created_at": datetime.now(),
"expires_at": datetime.now() + timedelta(seconds=ttl),
"access_count": 0,
"last_access": datetime.now(),
"prompt": prompt[:100] # Pour debugging
}
async def _evict_lru(self):
"""Évacue l'entrée la moins récemment utilisée."""
if not self._metadata:
return
lru_key = min(
self._metadata.keys(),
key=lambda k: self._metadata[k]["last_access"]
)
del self._cache[lru_key]
del self._metadata[lru_key]
async def invalidate_pattern(self, pattern: str) -> int:
"""Invalide toutes les entrées correspondant à un pattern."""
to_delete = []
for key, meta in self._metadata.items():
if pattern.lower() in meta.get("prompt", "").lower():
to_delete.append(key)
for key in to_delete:
del self._cache[key]
del self._metadata[key]
return len(to_delete)
async def _periodic_cleanup(self):
"""Nettoie périodiquement les entrées expirées."""
while True:
await asyncio.sleep(60) # Vérifier chaque minute
now = datetime.now()
expired = [
key for key, meta in self._metadata.items()
if now >= meta["expires_at"]
]
for key in expired:
del self._cache[key]
del self._metadata[key]
def get_stats(self) -> Dict:
"""Retourne les statistiques du cache."""