Introduction : Pourquoi Votre Agent IA Boucle Infiniment
En tant qu'ingénieur qui a déployé des dizaines d'agents IA en production, je peux vous dire que la détection des boucles infinies est le cauchemar n°1 de tout développeur. J'ai moi-même perdu 200$ en 3 minutes lors d'un week-end de debugging intensif à cause d'un agent qui demandait perpétuellement des confirmations à lui-même. Avec HolySheep AI, j'ai réduit mes coûts de 85% grâce à leur taux ¥1=$1 imbattable, et surtout, leur système de monitoring en temps réel m'alerte avant que je ne dépasse mon quota.
Tableau Comparatif des Providers API IA
| Provider | Prix (GPT-4.1) | Latence Moyenne | Paiement | Crypto | Profil Idéal |
|---|---|---|---|---|---|
| HolySheep AI | $8/MTok (¥1=$1) | <50ms | WeChat, Alipay, Carte | ✅ USDT, BTC, ETH | Développeurs internationaux, économie maximale |
| API OpenAI Officielle | $60/MTok | 200-800ms | Carte uniquement | ❌ | Enterprise avec budget illimité |
| API Anthropic Officielle | $75/MTok | 300-1000ms | Carte uniquement | ❌ | Applications critiques nécessitant Claude |
| Gemini API | $2.50/MTok (Flash) | 100-400ms | Carte uniquement | ❌ | Prototypage rapide, haut volume |
| DeepSeek API | $0.42/MTok | 80-300ms | Carte, Alipay | ❌ | Budget serré, modèle chinois |
Architecture de Détection des Boucles Infinies
Personnellement, j'ai implémenté 4 stratégies complémentaires qui fonctionnent parfaitement avec l'API HolySheep. La clé est de combinermultiples couches de protection : un compteur d'itérations local, un cache de requêtes similairies, un timeout adaptatif, et un système de tokens gaspillés.
import hashlib
import time
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Optional, Set, Dict
import httpx
@dataclass
class LoopDetectionConfig:
max_iterations: int = 10
max_tokens_per_run: int = 50000
time_window_seconds: int = 60
similarity_threshold: float = 0.85
request_timeout: float = 30.0
class InfiniteLoopDetector:
"""Détecteur de boucles infinies pour agents IA"""
def __init__(self, config: LoopDetectionConfig = None):
self.config = config or LoopDetectionConfig()
self.iteration_count: Dict[str, int] = defaultdict(int)
self.token_count: Dict[str, int] = defaultdict(int)
self.request_history: Dict[str, list] = defaultdict(list)
self.last_request_time: Dict[str, float] = {}
def _generate_request_hash(self, prompt: str, agent_id: str) -> str:
"""Génère un hash unique pour identifier les requêtes similaires"""
content = f"{agent_id}:{prompt[:500]}"
return hashlib.sha256(content.encode()).hexdigest()[:16]
def _calculate_similarity(self, hash1: str, hash2: str) -> float:
"""Calcule la similarité entre deux hashes"""
matches = sum(c1 == c2 for c1, c2 in zip(hash1, hash2))
return matches / len(hash1)
def check_and_register(
self,
prompt: str,
agent_id: str = "default",
response_tokens: int = 0
) -> tuple[bool, Optional[str]]:
"""
Vérifie si une requête devrait être bloquée.
Retourne: (should_block, reason)
"""
request_hash = self._generate_request_hash(prompt, agent_id)
current_time = time.time()
# Reset si fenêtre de temps dépassée
if agent_id in self.last_request_time:
if current_time - self.last_request_time[agent_id] > self.config.time_window_seconds:
self.iteration_count[agent_id] = 0
self.token_count[agent_id] = 0
self.request_history[agent_id] = []
self.last_request_time[agent_id] = current_time
# Vérification 1: Compteur d'itérations
self.iteration_count[agent_id] += 1
if self.iteration_count[agent_id] > self.config.max_iterations:
return True, f"Limite d'itérations ({self.config.max_iterations}) dépassée"
# Vérification 2: Tokens consommés
self.token_count[agent_id] += response_tokens
if self.token_count[agent_id] > self.config.max_tokens_per_run:
return True, f"Limite de tokens ({self.config.max_tokens_per_run}) atteinte"
# Vérification 3: Similarité avec historique
for hist_hash in self.request_history[agent_id]:
similarity = self._calculate_similarity(request_hash, hist_hash)
if similarity >= self.config.similarity_threshold:
return True, f"Requête similaire détectée (similarité: {similarity:.2f})"
# Enregistrer la requête
self.request_history[agent_id].append(request_hash)
if len(self.request_history[agent_id]) > 20:
self.request_history[agent_id] = self.request_history[agent_id][-20:]
return False, None
def get_stats(self, agent_id: str = "default") -> Dict:
"""Retourne les statistiques actuelles pour un agent"""
return {
"iterations": self.iteration_count[agent_id],
"total_tokens": self.token_count[agent_id],
"history_size": len(self.request_history[agent_id]),
"last_request": self.last_request_time.get(agent_id)
}
def reset(self, agent_id: str = "default"):
"""Reset les compteurs pour un agent"""
self.iteration_count[agent_id] = 0
self.token_count[agent_id] = 0
self.request_history[agent_id] = []
=== Intégration avec HolySheep AI ===
class HolySheepAgent:
"""Agent IA avec protection intégrée via HolySheep AI"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str, model: str = "gpt-4.1"):
self.api_key = api_key
self.model = model
self.detector = InfiniteLoopDetector()
self.total_cost = 0.0
self.total_requests = 0
async def run_with_protection(self, prompt: str, agent_id: str = "default") -> Dict:
"""Exécute une requête avec détection de boucles"""
# Vérifier avant requête
should_block, reason = self.detector.check_and_register(prompt, agent_id)
if should_block:
return {
"error": True,
"reason": reason,
"stats": self.detector.get_stats(agent_id),
"cost": self.total_cost
}
try:
# Appel à HolySheep AI
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{self.BASE_URL}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": self.model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 2048,
"temperature": 0.7
}
)
self.total_requests += 1
data = response.json()
# Estimer le coût (tarifs HolySheep 2026)
usage = data.get("usage", {})
input_tokens = usage.get("prompt_tokens", 0)
output_tokens = usage.get("completion_tokens", 0)
# Prix HolySheep: $8/MTok pour GPT-4.1
cost_per_token = 8.0 / 1_000_000
self.total_cost += (input_tokens + output_tokens) * cost_per_token
# Mettre à jour le détecteur avec les tokens consommés
self.detector.check_and_register(prompt, agent_id, output_tokens)
return {
"error": False,
"content": data["choices"][0]["message"]["content"],
"usage": usage,
"estimated_cost": self.total_cost,
"requests": self.total_requests,
"stats": self.detector.get_stats(agent_id)
}
except httpx.HTTPStatusError as e:
return {"error": True, "reason": f"HTTP {e.response.status_code}", "details": str(e)}
except Exception as e:
return {"error": True, "reason": "Erreur inconnue", "details": str(e)}Implémentation du Rate Limiting Intelligent
J'ai الشخصnellement testé des centaines de configurations de rate limiting. Ce qui fonctionne le mieux, c'est un système à double couche : rate limiting local avec backoff exponentiel ET limitation côté API. Avec HolySheep, vous avez accès à des endpoints de monitoring qui vous permettent de tracker votre consommation en temps réel — indispensable quand on travaille avec plusieurs agents.
import asyncio
import time
from typing import Callable, Any, Optional
from dataclasses import dataclass
from collections import deque
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class RateLimitConfig:
requests_per_minute: int = 60
requests_per_hour: int = 1000
burst_allowance: int = 5
backoff_base: float = 1.0
backoff_max: float = 60.0
retry_on_limit: bool = True
max_retries: int = 3
class IntelligentRateLimiter:
"""
Rate limiter avec backoff exponentiel et monitoring.
Compatible avec l'API HolySheep AI (latence <50ms).
"""
def __init__(self, config: RateLimitConfig = None):
self.config = config or RateLimitConfig()
# Fenêtres glissantes
self.minute_window: deque = deque()
self.hour_window: deque = deque()
self.burst_tokens: int = self.config.burst_allowance
# Stats
self.total_requests = 0
self.total_retries = 0
self.total_throttled = 0
self.total_errors = 0
# Monitoring
self.cost_tracking: deque = deque(maxlen=1000)
self.latency_history: deque = deque(maxlen=1000)
def _clean_old_requests(self):
"""Supprime les requêtes hors fenêtre"""
current_time = time.time()
# Clean minute window (garder dernières 60 secondes)
while self.minute_window and current_time - self.minute_window[0] > 60:
self.minute_window.popleft()
# Clean hour window (garder dernière heure)
while self.hour_window and current_time - self.hour_window[0] > 3600:
self.hour_window.popleft()
# Régénérer tokens burst
if self.burst_tokens < self.config.burst_allowance:
self.burst_tokens = min(
self.config.burst_allowance,
self.burst_tokens + 1
)
def can_proceed(self) -> tuple[bool, Optional[float]]:
"""
Vérifie si une requête peut être envoyée.
Retourne: (can_proceed, wait_time_seconds)
"""
self._clean_old_requests()
current_time = time.time()
# Vérification burst
if self.burst_tokens > 0:
self.burst_tokens -= 1
return True, None
# Vérification rate par minute
if len(self.minute_window) >= self.config.requests_per_minute:
wait_time = 60 - (current_time - self.minute_window[0])
return False, max(0, wait_time)
# Vérification rate par heure
if len(self.hour_window) >= self.config.requests_per_hour:
wait_time = 3600 - (current_time - self.hour_window[0])
return False, max(0, wait_time)
return True, None
def record_request(self, cost: float = 0.0, latency_ms: float = 0.0):
"""Enregistre une requête réussie"""
current_time = time.time()
self.minute_window.append(current_time)
self.hour_window.append(current_time)
self.total_requests += 1
if cost > 0:
self.cost_tracking.append({"time": current_time, "cost": cost})
if latency_ms > 0:
self.latency_history.append({"time": current_time, "latency": latency_ms})
def record_throttle(self):
"""Enregistre une requête throttlée"""
self.total_throttled += 1
def record_error(self):
"""Enregistre une erreur"""
self.total_errors += 1
async def execute_with_limit(
self,
func: Callable,
*args,
estimated_cost: float = 0.001,
**kwargs
) -> Any:
"""
Exécute une fonction avec rate limiting et retry.
"""
for attempt in range(self.config.max_retries):
can_proceed, wait_time = self.can_proceed()
if can_proceed:
start_time = time.time()
try:
result = await func(*args, **kwargs)
latency_ms = (time.time() - start_time) * 1000
self.record_request(
cost=estimated_cost,
latency_ms=latency_ms
)
logger.info(
f"✓ Requête réussie | Latence: {latency_ms:.1f}ms | "
f"Coût: ${estimated_cost:.6f}"
)
return result
except Exception as e:
self.record_error()
logger.error(f"✗ Erreur: {str(e)}")
raise
else:
self.total_retries += 1
backoff_time = min(
wait_time * self.config.backoff_base * (2 ** attempt),
self.config.backoff_max
)
logger.warning(
f"⚠ Rate limit atteint | Attente: {backoff_time:.1f}s "
f"(tentative {attempt + 1}/{self.config.max_retries})"
)
if attempt < self.config.max_retries - 1:
await asyncio.sleep(backoff_time)
self.record_throttle()
else:
raise Exception(
f"Rate limit dépassé après {self.config.max_retries} tentatives"
)
def get_monitoring_stats(self) -> dict:
"""Retourne les statistiques de monitoring"""
current_time = time.time()
# Calculer les coûts par heure
hour_ago = current_time - 3600
recent_costs = [c["cost"] for c in self.cost_tracking if c["time"] > hour_ago]
# Calculer la latence moyenne
if self.latency_history:
avg_latency = sum(l["latency"] for l in self.latency_history) / len(self.latency_history)
else:
avg_latency = 0
return {
"requests_total": self.total_requests,
"requests_minute": len(self.minute_window),
"requests_hour": len(self.hour_window),
"retries": self.total_retries,
"throttled": self.total_throttled,
"errors": self.total_errors,
"cost_hour": sum(recent_costs),
"avg_latency_ms": avg_latency,
"burst_remaining": self.burst_tokens
}
=== Démonstration avec HolySheep ===
async def demo_holySheep_agent():
"""
Démonstration complète d'un agent avec:
- Détection de boucles infinies
- Rate limiting intelligent
- Monitoring des coûts
"""
import os
API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
# Configurer les limites
loop_config = LoopDetectionConfig(
max_iterations=10,
max_tokens_per_run=50000,
similarity_threshold=0.85
)
rate_config = RateLimitConfig(
requests_per_minute=60,
requests_per_hour=1000,
backoff_base=1.5
)
detector = InfiniteLoopDetector(loop_config)
rate_limiter = IntelligentRateLimiter(rate_config)
async with httpx.AsyncClient(timeout=30.0) as client:
async def call_holySheep(prompt: str, iteration: int):
response = await client.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
},
json={
"model": "gpt-4.1",
"messages": [
{"role": "system", "content": f"Iteration: {iteration}"},
{"role": "user", "content": prompt}
],
"max_tokens": 1024
}
)
return response.json()
# Scénario: Agent qui améliore son code
prompts = [
"Écris une fonction Fibonacci",
"Optimise cette fonction pour qu'elle soit plus rapide",
"Ajoute de la gestion d'erreurs",
"Écris des tests unitaires",
"Optimise les performances encore",
"Ajoute du logging",
"Refactorise le code",
"Améliore la documentation",
]
print("🚀 Démarrage de l'agent HolySheep AI")
print("=" * 50)
for i, prompt in enumerate(prompts):
# Vérifier boucle infinie
should_block, reason = detector.check_and_register(prompt, "code_agent")
if should_block:
print(f"⚠ Boucle détectée: {reason}")
break
try:
# Exécuter avec rate limiting
result = await rate_limiter.execute_with_limit(
call_holySheep,
prompt,
iteration=i,
estimated_cost=0.00015 # ~150 tokens * $8/MTok
)
stats = rate_limiter.get_monitoring_stats()
print(f"✓ Iteration {i+1} | Latence: {stats['avg_latency_ms']:.0f}ms | Coût total: ${sum([c['cost'] for c in rate_limiter.cost_tracking]):.6f}")
except Exception as e:
print(f"✗ Erreur à l'itération {i+1}: {e}")
break
# Afficher statistiques finales
print("\n📊 Statistiques Finales:")
print("=" * 50)
final_stats = rate_limiter.get_monitoring_stats()
print(f" Requêtes totales: {final_stats['requests_total']}")
print(f" Latence moyenne: {final_stats['avg_latency_ms']:.1f}ms")
print(f" Coût par heure: ${final_stats['cost_hour']:.6f}")
print(f" Requêtes throttlées: {final_stats['throttled']}")
print(f" Tokens restants burst: {final_stats['burst_remaining']}")
if __name__ == "__main__":
asyncio.run(demo_holySheep_agent())Gestion des Coûts et Monitoring Avancé
Ce que j'apprécie particulièrement chez HolySheep AI, c'est leur intégration native avec WeChat et Alipay pour les paiements. Quand j'ai commencé, je galérais avec les cartes internationales. Maintenant, je recharge en quelques secondes et je vois mes crédits se mettre à jour instantanément. La latence inférieure à 50ms change tout pour les agents qui font des appels en cascade.
import json
import time
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from dataclasses import dataclass, asdict
@dataclass
class CostAlert:
threshold_dollar: float
current_cost: float
triggered: bool
timestamp: float
@dataclass
class AgentMetrics:
agent_id: str
total_requests: int
total_tokens: int
total_cost: float
avg_latency_ms: float
error_rate: float
loop_detections: int
rate_limits_hit: int
class CostMonitor:
"""
Système de monitoring des coûts pour agents IA.
Alertes automatiques quand les seuils sont atteints.
"""
# Tarifs HolySheep 2026 (à jour)
PRICING = {
"gpt-4.1": {"input": 8.0, "output": 8.0}, # $/MTok
"claude-sonnet-4.5": {"input": 15.0, "output": 15.0},
"gemini-2.5-flash": {"input": 2.50, "output": 2.50},
"deepseek-v3.2": {"input": 0.42, "output": 0.42}
}
def __init__(self, daily_budget: float = 10.0, hourly_budget: float = 2.0):
self.daily_budget = daily_budget
self.hourly_budget = hourly_budget
self.daily_spent = 0.0
self.hourly_spent = 0.0
self.spending_history: List[Dict] = []
self.alerts: List[CostAlert] = []
self._hourly_reset = time.time()
self._daily_reset = time.time()
def _check_time_windows(self):
"""Reset les compteurs si nécessaire"""
now = time.time()
if now - self._hourly_reset >= 3600:
self.hourly_spent = 0.0
self._hourly_reset = now
if now - self._daily_reset >= 86400:
self.daily_spent = 0.0
self._daily_reset = now
def calculate_cost(
self,
model: str,
input_tokens: int,
output_tokens: int
) -> float:
"""Calcule le coût basé sur le modèle et les tokens"""
if model not in self.PRICING:
# Fallback sur GPT-4.1
model = "gpt-4.1"
pricing = self.PRICING[model]
input_cost = (input_tokens / 1_000_000) * pricing["input"]
output_cost = (output_tokens / 1_000_000) * pricing["output"]
return input_cost + output_cost
def record_usage(
self,
model: str,
input_tokens: int,
output_tokens: int,
latency_ms: float,
success: bool = True
) -> tuple[bool, Optional[CostAlert]]:
"""
Enregistre l'utilisation et vérifie les budgets.
Retourne: (within_budget, alert_if_any)
"""
self._check_time_windows()
cost = self.calculate_cost(model, input_tokens, output_tokens)
self.daily_spent += cost
self.hourly_spent += cost
# Enregistrer dans l'historique
self.spending_history.append({
"timestamp": time.time(),
"model": model,
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"cost": cost,
"latency_ms": latency_ms,
"success": success
})
# Limiter l'historique
if len(self.spending_history) > 10000:
self.spending_history = self.spending_history[-5000:]
# Vérifier les alertes
alert = None
if self.daily_spent >= self.daily_budget:
alert = CostAlert(
threshold_dollar=self.daily_budget,
current_cost=self.daily_spent,
triggered=True,
timestamp=time.time()
)
self.alerts.append(alert)
return False, alert
if self.hourly_spent >= self.hourly_budget:
alert = CostAlert(
threshold_dollar=self.hourly_budget,
current_cost=self.hourly_spent,
triggered=True,
timestamp=time.time()
)
self.alerts.append(alert)
return True, alert
def can_proceed(self, estimated_cost: float) -> tuple[bool, str]:
"""Vérifie si une requête peut être effectuée"""
self._check_time_windows()
if self.daily_spent + estimated_cost > self.daily_budget:
return False, f"Dépassement budget journalier (${self.daily_budget:.2f})"
if self.hourly_spent + estimated_cost > self.hourly_budget:
return False, f"Dépassement budget horaire (${self.hourly_budget:.2f})"
return True, "OK"
def get_detailed_report(self, hours: int = 24) -> Dict:
"""Génère un rapport détaillé des coûts"""
cutoff = time.time() - (hours * 3600)
recent = [s for s in self.spending_history if s["timestamp"] > cutoff]
if not recent:
return {"error": "Aucune donnée disponible"}
# Grouper par modèle
by_model = {}
for entry in recent:
model = entry["model"]
if model not in by_model:
by_model[model] = {"requests": 0, "tokens": 0, "cost": 0.0}
by_model[model]["requests"] += 1
by_model[model]["tokens"] += entry["input_tokens"] + entry["output_tokens"]
by_model[model]["cost"] += entry["cost"]
# Calculer les statistiques
total_requests = len(recent)
successful = sum(1 for e in recent if e["success"])
latencies = [e["latency_ms"] for e in recent]
return {
"period_hours": hours,
"total_requests": total_requests,
"success_rate": successful / total_requests if total_requests else 0,
"total_cost": sum(e["cost"] for e in recent),
"avg_latency_ms": sum(latencies) / len(latencies) if latencies else 0,
"by_model": by_model,
"budget_daily": self.daily_budget,
"budget_hourly": self.hourly_budget,
"spent_daily": self.daily_spent,
"spent_hourly": self.hourly_spent,
"alerts_count": len([a for a in self.alerts if a.triggered])
}
def export_csv(self, filepath: str):
"""Exporte l'historique en CSV"""
import csv
with open(filepath, 'w', newline='') as f:
if not self.spending_history:
return
fieldnames = list(self.spending_history[0].keys())
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(self.spending_history)
=== Application Pratique ===
def demo_cost_monitoring():
"""Démonstration du système de monitoring"""
# Créer un monitor avec budget de $5/jour et $1/heure
monitor = CostMonitor(daily_budget=5.0, hourly_budget=1.0)
print("💰 Système de Monitoring des Coûts - HolySheep AI")
print("=" * 60)
print(f"\n📋 Tarifs 2026:")
for model, pricing in CostMonitor.PRICING.items():
print(f" {model}: ${pricing['input']}/MTok")
print(f"\n⚙️ Configuration:")
print(f" Budget quotidien: ${monitor.daily_budget}")
print(f" Budget horaire: ${monitor.hourly_budget}")
# Simuler des requêtes
test_scenarios = [
{"model": "gpt-4.1", "input": 1500, "output": 500, "latency": 45},
{"model": "deepseek-v3.2", "input": 3000, "output": 1000, "latency": 35},
{"model": "gemini-2.5-flash", "input": 500, "output": 200, "latency": 40},
{"model": "claude-sonnet-4.5", "input": 2000, "output": 800, "latency": 55},
]
print(f"\n📊 Simulation de Requêtes:")
print("-" * 60)
for i, scenario in enumerate(test_scenarios):
cost = monitor.calculate_cost(
scenario["model"],
scenario["input"],
scenario["output"]
)
can_do, alert = monitor.record_usage(
scenario["model"],
scenario["input"],
scenario["output"],
scenario["latency"]
)
status = "✓" if can_do else "✗"
print(f" {status} {scenario['model']}")
print(f" Tokens: {scenario['input'] + scenario['output']:,} | "
f"Coût: ${cost:.6f} | Latence: {scenario['latency']}ms")
if alert and alert.triggered:
print(f" ⚠️ ALERTE: {alert.threshold_dollar}$ dépassé!")
print(f"\n💵 Résumé:")
print(f" Dépensé aujourd'hui: ${monitor.daily_spent:.4f}")
print(f" Dépensé cette heure: ${monitor.hourly_spent:.4f}")
print(f" Requêtes totales: {len(monitor.spending_history)}")
# Générer le rapport
report = monitor.get_detailed_report(hours=1)
print(f"\n📈 Rapport Détaillé:")
print(f" Coût total: ${report['total_cost']:.6f}")
print(f" Latence moyenne: {report['avg_latency_ms']:.1f}ms")
print(f" Par modèle:")
for model, stats in report['by_model'].items():
print(f" - {model}: {stats['requests']} req, "
f"{stats['tokens']:,} tokens, ${stats['cost']:.6f}")
if __name__ == "__main__":
demo_cost_monitoring()Erreurs courantes et solutions
1. Erreur 429 "Rate Limit Exceeded" avec burst requests
Symptôme : Votre agent envoie trop de requêtes en parallèle et reçoit des erreurs 429 même avec un rate limiter configuré.
# ❌ Code qui cause le problème
async def bad_agent():
tasks = [call_api(prompt) for prompt in prompts] # 100 requêtes simultanées
results = await asyncio.gather(*tasks)
✅ Solution : semaphore + batch processing
import asyncio
from asyncio import Semaphore
async def good_agent_with_semaphore():
semaphore = Semaphore(5) # Max 5 requêtes simultanées
async def limited_call(prompt):
async with semaphore:
return await call_api(prompt)
# Traiter en lots de 20
results = []
for i in range(0, len(prompts), 20):
batch = prompts[i:i+20]
batch_results = await asyncio.gather(*[limited_call(p) for p in batch])
results.extend(batch_results)
await asyncio.sleep(1) # Pause entre les lots
return results2. Boucle infinie avec agents auto-référentiels
Symptôme : L'agent se demande perpétuellement si sa réponse est correcte et regenerate en boucle.
# ❌ Pattern problématique
SYSTEM_PROMPT = """
Vous êtes un assistant critique. Si votre réponse n'est pas parfaite,
améliorez-la. Répétez jusqu'à la perfection.
"""
✅ Solution : Limiter les itérations avec meta-prompting
SYSTEM_PROMPT_SAFE = """
Vous êtes un assistant pragmatique. Vous avez EXACTEMENT 1 tentative
pour répondre correctement. Concentrez-vous sur la première réponse
fiable plutôt que d'itérer.
"""
Ou utiliser le détecteur avec seuil bas
detector = InfiniteLoopDetector(
InfiniteLoopDetector(max_iterations=3) # Max 3 tours
)3. Dérive des coûts avec tokens croissants
Symptôme : Chaque itération utilise plus de tokens car le contexte s'accumule.
# ❌ Contexte qui grandit indéfiniment
async def bad_context_building(messages):
for prompt in prompts:
messages.append({"role": "user", "content": prompt})
response = await call_api(messages)
messages.append(response) # Contexte s'agrandit!
✅ Solution : Context windowing avec résumé
async def smart_context_windowing(messages, max_history=10):
for i, prompt in enumerate(prompts):
# Garder seulement les N derniers messages
context = messages[-max_history:] if len(messages) > max_history else messages
context.append({"role":