En tant qu'ingénieur qui a débogué des centaines d'agents IA au cours des trois dernières années, je peux vous affirmer sans hésitation : le debugging d'agents IA représente l'un des défis les plus complexes du développement moderne. Contrairement aux applications traditionnelles où une erreur génère une stack trace explicite, un agent IA qui dévie de son comportement attendu peut le faire de manière subtile, progressive et parfois catastrophique.
Dans cet article, je partagerai les techniques de debugging que j'ai perfectionnées sur HolySheep AI — une plateforme que j'utilise quotidiennement pour ses latences inférieures à 50ms et son écosystème de paiement localisé avec WeChat et Alipay. Commençons par une comparaison de coûts qui vous permettra de choisir judicieusement votre fournisseur.
Comparatif des coûts 2026 : Choix stratégique pour le debugging
| Modèle | Prix output ($/MTok) | Coût 10M tokens/mois | Latence typique |
|---|---|---|---|
| DeepSeek V3.2 | 0,42 $ | 4 200 $ | <50ms |
| Gemini 2.5 Flash | 2,50 $ | 25 000 $ | <80ms |
| GPT-4.1 | 8 $ | 80 000 $ | <120ms |
| Claude Sonnet 4.5 | 15 $ | 150 000 $ | <150ms |
Analyse personnelle : Pendant mes sessions de debugging intensives, j'estime consommer environ 50M de tokens par jour pour les tests et l'itération. Avec DeepSeek V3.2 via HolySheep AI, cela me coûte environ 21 $ par jour contre 400 $ avec Claude Sonnet 4.5. Cette différence de 1800% m'a permis de multiplier mes cycles de test par 10 sans exploser mon budget.
HolySheep AI offre un taux de change avantageux avec ¥1 = $1 USD, ce qui représente une économie supplémentaire de 85%+ pour les développeurs chinois. De plus, les crédits gratuits offerts à l'inscription vous permettent de commencer vos tests sans investissement initial.
Architecture d'un système de tracking d'état robuste
Le debugging d'agents IA repose sur trois piliers fondamentaux : la capture d'état, la comparaison temporelle et l'isolation des variables. Voici comment implémenter une architecture complète.
1. Système de journalisation structurée
#!/usr/bin/env python3
"""
Système de debugging pour agents IA avec HolySheep AI
Auteur: HolySheep AI Blog
"""
import json
import time
import hashlib
from datetime import datetime
from typing import Dict, List, Any, Optional
from dataclasses import dataclass, field, asdict
from enum import Enum
class LogLevel(Enum):
DEBUG = "DEBUG"
INFO = "INFO"
WARNING = "WARNING"
ERROR = "ERROR"
CRITICAL = "CRITICAL"
@dataclass
class AgentState:
"""Snapshot complet de l'état de l'agent à un instant T"""
timestamp: str = field(default_factory=lambda: datetime.utcnow().isoformat())
session_id: str = ""
step_number: int = 0
model_name: str = ""
input_tokens: int = 0
output_tokens: int = 0
total_cost_usd: float = 0.0
latency_ms: float = 0.0
system_prompt_hash: str = ""
conversation_history: List[Dict] = field(default_factory=list)
tool_calls: List[Dict] = field(default_factory=list)
tool_results: List[Dict] = field(default_factory=list)
final_output: str = ""
state_checksum: str = ""
metadata: Dict = field(default_factory=dict)
def compute_checksum(self) -> str:
"""Calcule un hash unique pour détecter les modifications non autorisées"""
state_str = json.dumps({
'step': self.step_number,
'history_len': len(self.conversation_history),
'tools': len(self.tool_calls),
'output_preview': self.final_output[:100] if self.final_output else ""
}, sort_keys=True)
return hashlib.sha256(state_str.encode()).hexdigest()[:16]
class AgentDebugger:
"""Debugger complet pour agents IA avec HolySheep AI"""
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.states: List[AgentState] = []
self.current_state: Optional[AgentState] = None
self.cost_tracking = {"total_tokens": 0, "total_cost": 0.0}
def start_session(self, session_id: str, model: str) -> AgentState:
"""Initialise une nouvelle session de debugging"""
self.current_state = AgentState(
session_id=session_id,
model_name=model,
step_number=0
)
self._log(LogLevel.INFO, f"Session {session_id} initialisée avec {model}")
return self.current_state
def _log(self, level: LogLevel, message: str, context: Dict = None):
"""Journalisation avec contexte enrichi"""
log_entry = {
"timestamp": datetime.utcnow().isoformat(),
"level": level.value,
"message": message,
"context": context or {}
}
print(f"[{level.value}] {message}")
if context:
print(f" Contexte: {json.dumps(context, indent=2)}")
def execute_with_tracking(self, messages: List[Dict], tools: List[Dict] = None) -> Dict:
"""Exécute une requête avec tracking complet des métriques"""
import requests
start_time = time.time()
step = self.current_state.step_number
# Préparation de la requête HolySheep AI
payload = {
"model": self.current_state.model_name,
"messages": messages,
"temperature": 0.7,
"max_tokens": 4096
}
if tools:
payload["tools"] = tools
payload["tool_choice"] = "auto"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
try:
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
response.raise_for_status()
result = response.json()
end_time = time.time()
latency_ms = (end_time - start_time) * 1000
# Extraction des métriques
usage = result.get("usage", {})
input_tokens = usage.get("prompt_tokens", 0)
output_tokens = usage.get("completion_tokens", 0)
# Calcul du coût (tarifs HolySheep AI 2026)
pricing = {
"deepseek-chat": 0.00042, # $0.42/MTok output
"gemini-2.5-flash": 0.0025, # $2.50/MTok
"gpt-4.1": 0.008, # $8/MTok
"claude-sonnet-4.5": 0.015 # $15/MTok
}
cost_per_output_token = pricing.get(self.current_state.model_name, 0.008)
step_cost = (output_tokens / 1_000_000) * cost_per_output_token
# Mise à jour du tracking des coûts
self.cost_tracking["total_tokens"] += input_tokens + output_tokens
self.cost_tracking["total_cost"] += step_cost
# Création du snapshot d'état
self.current_state.step_number = step + 1
self.current_state.input_tokens = input_tokens
self.current_state.output_tokens = output_tokens
self.current_state.total_cost_usd = self.cost_tracking["total_cost"]
self.current_state.latency_ms = latency_ms
self.current_state.conversation_history.append({
"role": "user" if step == 0 else "assistant",
"content": messages[-1]["content"] if messages else "",
"response": result["choices"][0]["message"]
})
# Gestion des tool calls
assistant_message = result["choices"][0]["message"]
if "tool_calls" in assistant_message:
self.current_state.tool_calls.append({
"step": step,
"calls": assistant_message["tool_calls"]
})
self.current_state.state_checksum = self.current_state.compute_checksum()
self._log(
LogLevel.INFO,
f"Step {step} terminé",
{
"latence_ms": round(latency_ms, 2),
"tokens_in": input_tokens,
"tokens_out": output_tokens,
"coût_step": round(step_cost, 6),
"coût_total": round(self.cost_tracking["total_cost"], 4)
}
)
return result
except requests.exceptions.RequestException as e:
self._log(LogLevel.ERROR, f"Erreur API HolySheep: {str(e)}")
raise
def compare_states(self, state1_idx: int, state2_idx: int) -> Dict:
"""Compare deux états pour identifier les divergences"""
if state1_idx >= len(self.states) or state2_idx >= len(self.states):
raise IndexError("Indices d'état invalides")
s1, s2 = self.states[state1_idx], self.states[state2_idx]
differences = {
"tokens_diff": s2.input_tokens + s2.output_tokens - (s1.input_tokens + s1.output_tokens),
"cost_diff": s2.total_cost_usd - s1.total_cost_usd,
"steps_diff": s2.step_number - s1.step_number,
"tool_call_diff": len(s2.tool_calls) - len(s1.tool_calls),
"checksum_match": s1.state_checksum == s2.state_checksum
}
return differences
def export_debug_session(self, filepath: str):
"""Exporte la session complète pour analyse post-mortem"""
session_data = {
"export_timestamp": datetime.utcnow().isoformat(),
"api_endpoint": self.base_url,
"model": self.current_state.model_name,
"cost_summary": self.cost_tracking,
"total_steps": self.current_state.step_number,
"states": [asdict(state) for state in self.states]
}
with open(filepath, 'w', encoding='utf-8') as f:
json.dump(session_data, f, indent=2, ensure_ascii=False)
self._log(LogLevel.INFO, f"Session exportée vers {filepath}")
=== UTILISATION ===
if __name__ == "__main__":
debugger = AgentDebugger(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
session = debugger.start_session(
session_id="debug-session-001",
model="deepseek-chat"
)
messages = [
{"role": "system", "content": "Tu es un assistant de debugging. Réponds de manière concise."},
{"role": "user", "content": "Explique la différence entre stateful et stateless dans les agents IA."}
]
result = debugger.execute_with_tracking(messages)
print(f"Réponse: {result['choices'][0]['message']['content']}")
print(f"Coût total de la session: ${debugger.cost_tracking['total_cost']:.6f}")
2. Système de détection d'anomalies en temps réel
#!/usr/bin/env python3
"""
Module de détection d'anomalies pour agents IA
Intégration HolySheep AI avec monitoring temps réel
"""
import numpy as np
from collections import deque
from typing import Callable, Dict, List, Optional, Tuple
import threading
import time
class AnomalyDetector:
"""
Détecte les comportements anormaux de l'agent en temps réel
Basé sur l'analyse статистический des métriques d'exécution
"""
def __init__(self, window_size: int = 100, threshold_sigma: float = 2.5):
self.window_size = window_size
self.threshold_sigma = threshold_sigma
# Buffers circulaires pour les métriques
self.latency_buffer = deque(maxlen=window_size)
self.token_buffer = deque(maxlen=window_size)
self.cost_buffer = deque(maxlen=window_size)
self.error_buffer = deque(maxlen=window_size)
# Statistiques
self.baseline_latency = None
self.baseline_tokens = None
self.anomaly_count = 0
self.total_checks = 0
# Callbacks d'alerte
self.alert_callbacks: List[Callable] = []
# Thread-safe
self._lock = threading.Lock()
def add_metric(self, latency_ms: float, tokens: int, cost: float, has_error: bool = False):
"""Ajoute une métrique et vérifie les anomalies"""
with self._lock:
self.latency_buffer.append(latency_ms)
self.token_buffer.append(tokens)
self.cost_buffer.append(cost)
self.error_buffer.append(1 if has_error else 0)
self.total_checks += 1
# Initialisation du baseline après 10 mesures
if len(self.latency_buffer) >= 10 and self.baseline_latency is None:
self._compute_baseline()
# Vérification d'anomalie après initialisation
if self.baseline_latency is not None:
self._check_anomalies(latency_ms, tokens, cost)
def _compute_baseline(self):
"""Calcule les statistiques de base (moyenne, écart-type)"""
self.baseline_latency = {
'mean': np.mean(self.latency_buffer),
'std': np.std(self.latency_buffer),
'median': np.median(self.latency_buffer),
'p95': np.percentile(self.latency_buffer, 95)
}
self.baseline_tokens = {
'mean': np.mean(self.token_buffer),
'std': np.std(self.token_buffer),
'max': max(self.token_buffer)
}
def _check_anomalies(self, latency_ms: float, tokens: int, cost: float):
"""Vérifie si les nouvelles métriques sont des anomalies"""
anomalies = []
# Détection latence anormale
z_score_latency = (latency_ms - self.baseline_latency['mean']) / (self.baseline_latency['std'] + 1e-6)
if abs(z_score_latency) > self.threshold_sigma:
anomalies.append({
'type': 'HIGH_LATENCY',
'severity': 'WARNING' if z_score_latency < 3 else 'CRITICAL',
'z_score': round(z_score_latency, 2),
'current': round(latency_ms, 2),
'baseline_mean': round(self.baseline_latency['mean'], 2),
'threshold': round(self.baseline_latency['mean'] + self.threshold_sigma * self.baseline_latency['std'], 2)
})
# Détection token burst (possible injection ou loop)
if tokens > self.baseline_tokens['mean'] + 2 * self.baseline_tokens['std']:
anomalies.append({
'type': 'TOKEN_BURST',
'severity': 'CRITICAL',
'current_tokens': tokens,
'baseline_mean': round(self.baseline_tokens['mean'], 2),
'ratio': round(tokens / (self.baseline_tokens['mean'] + 1e-6), 2)
})
# Détection coût excessif
cost_ratio = cost / (np.mean(self.cost_buffer) + 1e-6) if len(self.cost_buffer) > 1 else 1
if cost_ratio > 2.0:
anomalies.append({
'type': 'COST_SPIKE',
'severity': 'WARNING',
'current_cost': round(cost, 6),
'cost_ratio': round(cost_ratio, 2)
})
if anomalies:
self.anomaly_count += len(anomalies)
self._trigger_alerts(anomalies)
def _trigger_alerts(self, anomalies: List[Dict]):
"""Déclenche les callbacks d'alerte"""
for callback in self.alert_callbacks:
try:
callback(anomalies)
except Exception as e:
print(f"Erreur dans callback d'alerte: {e}")
def register_alert_callback(self, callback: Callable[[List[Dict]], None]):
"""Enregistre une fonction de callback pour les alertes"""
self.alert_callbacks.append(callback)
def get_stats(self) -> Dict:
"""Retourne les statistiques courantes"""
with self._lock:
return {
'total_checks': self.total_checks,
'anomaly_count': self.anomaly_count,
'anomaly_rate': round(self.anomaly_count / max(self.total_checks, 1) * 100, 2),
'baseline': {
'latency': self.baseline_latency,
'tokens': self.baseline_tokens
} if self.baseline_latency else None,
'current_buffers': {
'latency_last10': list(self.latency_buffer)[-10:],
'tokens_avg_recent': round(np.mean(list(self.token_buffer)[-10:]), 0) if len(self.token_buffer) >= 10 else None
}
}
class InteractiveDebugger:
"""
Debugger interactif avec pause, step-by-step et rollback
Optimisé pour HolySheep AI avec latence <50ms
"""
def __init__(self, api_key: str, model: str = "deepseek-chat"):
self.api_key = api_key
self.model = model
self.base_url = "https://api.holysheep.ai/v1"
self.checkpoints: List[Dict] = []
self.current_step = 0
self.is_paused = False
self.breakpoints: List[int] = []
self.anomaly_detector = AnomalyDetector()
self._setup_default_alerts()
def _setup_default_alerts(self):
"""Configure les alertes par défaut"""
def console_alert(anomalies):
print("⚠️ ALERTE D'ANOMALIE DÉTECTÉE:")
for anomaly in anomalies:
print(f" [{anomaly['severity']}] {anomaly['type']}")
print(f" Détails: {anomaly}")
self.anomaly_detector.register_alert_callback(console_alert)
def set_breakpoint(self, step: int):
"""Définit un point d'arrêt à un step donné"""
self.breakpoints.append(step)
print(f"🔴 Breakpoint défini au step {step}")
def create_checkpoint(self, state: Dict, messages: List[Dict]):
"""Crée un checkpoint de l'état actuel"""
checkpoint = {
'step': self.current_step,
'timestamp': time.time(),
'state': state.copy(),
'messages': [m.copy() for m in messages],
'checkpoint_id': len(self.checkpoints)
}
self.checkpoints.append(checkpoint)
print(f"💾 Checkpoint {checkpoint['checkpoint_id']} créé au step {self.current_step}")
return checkpoint['checkpoint_id']
def rollback_to_checkpoint(self, checkpoint_id: int) -> List[Dict]:
"""Restaure l'état à un checkpoint précédent"""
if checkpoint_id >= len(self.checkpoints):
raise ValueError(f"Checkpoint {checkpoint_id} inexistant")
checkpoint = self.checkpoints[checkpoint_id]
self.current_step = checkpoint['step']
self.checkpoints = self.checkpoints[:checkpoint_id]
print(f"↩️ Rollback vers checkpoint {checkpoint_id}")
return checkpoint['messages']
def execute_step(self, messages: List[Dict], tools: List[Dict] = None) -> Tuple[Dict, bool]:
"""
Exécute un step avec pause possible et monitoring
Retourne (résultat, doit_continuer)
"""
import requests
# Vérification breakpoint
if self.current_step in self.breakpoints:
self.is_paused = True
print(f"⏸️ Breakpoint atteint au step {self.current_step}")
# Pause si demandé
while self.is_paused:
command = input("🐛 Debugger (c=continuer, q=quitter, s=statut): ")
if command == 'c':
self.is_paused = False
elif command == 'q':
return {}, False
elif command == 's':
print(json.dumps(self.anomaly_detector.get_stats(), indent=2))
# Exécution avec timing précis
start = time.perf_counter()
payload = {
"model": self.model,
"messages": messages,
"temperature": 0.7,
"max_tokens": 4096
}
if tools:
payload["tools"] = tools
response = requests.post(
f"{self.base_url}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json=payload,
timeout=30
)
end = time.perf_counter()
latency_ms = (end - start) * 1000
result = response.json()
output_tokens = result.get("usage", {}).get("completion_tokens", 0)
# Monitoring d'anomalies
cost = (output_tokens / 1_000_000) * 0.00042 # DeepSeek V3.2
has_error = response.status_code != 200
self.anomaly_detector.add_metric(latency_ms, output_tokens, cost, has_error)
# Création automatique de checkpoint
if self.current_step % 5 == 0:
self.create_checkpoint({'latency': latency_ms, 'tokens': output_tokens}, messages)
self.current_step += 1
return result, True
=== INTÉGRATION HOLYSHEEP ===
import json
debugger = InteractiveDebugger(
api_key="YOUR_HOLYSHEEP_API_KEY",
model="deepseek-chat" # $0.42/MTok - optimal pour debugging intensif
)
debugger.set_breakpoint(3)
debugger.anomaly_detector.register_alert_callback(
lambda anomalies: print(f"🚨 {len(anomalies)} anomalie(s) détectée(s)")
)
print("Interactive Debugger initialisé avec HolySheep AI")
print(f"Statut: {json.dumps(debugger.anomaly_detector.get_stats(), indent=2)}")
3. Système de replay et post-mortem
#!/usr/bin/env python3
"""
Système de replay pour reproduire et analyser les bugs
Permet le post-mortem complet avec HolySheep AI
"""
import json
import time
from pathlib import Path
from typing import List, Dict, Optional, Tuple
from dataclasses import dataclass
from datetime import datetime
import hashlib
@dataclass
class ReplayEvent:
"""Événement unique dans une session de replay"""
timestamp: str
event_type: str # 'request', 'response', 'error', 'tool_call'
request_data: Optional[Dict] = None
response_data: Optional[Dict] = None
error_data: Optional[Dict] = None
duration_ms: float = 0.0
event_hash: str = ""
class SessionReplay:
"""
Enregistre et rejoule les sessions d'agent pour debugging
Fonctionne avec tout modèle supporté par HolySheep AI
"""
def __init__(self, session_id: str, output_dir: str = "./replays"):
self.session_id = session_id
self.output_dir = Path(output_dir)
self.output_dir.mkdir(exist_ok=True)
self.events: List[ReplayEvent] = []
self.start_time = datetime.utcnow().isoformat()
# Cache pour optimisation replay
self.response_cache: Dict[str, Dict] = {}
def record_request(self, model: str, messages: List[Dict],
tools: List[Dict] = None) -> str:
"""Enregistre une requête avant envoi"""
request_data = {
'model': model,
'messages': messages,
'tools': tools,
'timestamp': datetime.utcnow().isoformat()
}
event = ReplayEvent(
timestamp=datetime.utcnow().isoformat(),
event_type='request',
request_data=request_data
)
event.event_hash = self._compute_hash(request_data)
self.events.append(event)
return event.event_hash
def record_response(self, request_hash: str, response: Dict, duration_ms: float):
"""Enregistre une réponse pour le replay futur"""
# Recherche de l'événement request correspondant
for event in self.events:
if event.event_hash == request_hash and event.event_type == 'request':
event.response_data = response
event.duration_ms = duration_ms
# Cache la réponse pour replay rapide
self.response_cache[request_hash] = {
'response': response,
'duration_ms': duration_ms
}
# Crée un nouvel événement response
response_event = ReplayEvent(
timestamp=datetime.utcnow().isoformat(),
event_type='response',
response_data=response,
duration_ms=duration_ms
)
self.events.append(response_event)
break
def record_error(self, request_hash: str, error: Exception):
"""Enregistre une erreur pour analyse post-mortem"""
for event in self.events:
if event.event_hash == request_hash:
event.error_data = {
'type': type(error).__name__,
'message': str(error),
'timestamp': datetime.utcnow().isoformat()
}
break
def _compute_hash(self, data: Dict) -> str:
"""Calcule un hash unique pour identifier les requêtes"""
normalized = json.dumps(data, sort_keys=True, default=str)
return hashlib.sha256(normalized.encode()).hexdigest()[:16]
def replay_event(self, event_index: int, use_cache: bool = True) -> Tuple[bool, Optional[Dict]]:
"""
Rejoue un événement spécifique
Si use_cache=True, utilise la réponse cachée (replay instantané)
Sinon, rejoue via HolySheep AI (nouveau coût)
"""
if event_index >= len(self.events):
return False, None
event = self.events[event_index]
if use_cache and event.event_hash in self.response_cache:
cached = self.response_cache[event.event_hash]
print(f"🎭 Replay (cached) - Event {event_index}: {event.event_type}")
print(f" Durée: {cached['duration_ms']:.2f}ms")
return True, cached['response']
elif event.request_data:
import requests
print(f"🎭 Replay (live) - Event {event_index}: {event.event_type}")
print(f" Modèle: {event.request_data['model']}")
start = time.perf_counter()
response = requests.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={
"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
"Content-Type": "application/json"
},
json={
"model": event.request_data['model'],
"messages": event.request_data['messages'],
"temperature": 0.7
},
timeout=30
)
duration = (time.perf_counter() - start) * 1000
result = response.json()
print(f" Durée: {duration:.2f}ms")
return True, result
return False, None
def find_errors(self) -> List[Tuple[int, Dict]]:
"""Retourne tous les événements avec erreurs"""
errors = []
for i, event in enumerate(self.events):
if event.error_data:
errors.append((i, event.error_data))
return errors
def analyze_token_pattern(self) -> Dict:
"""Analyse les patterns de consommation de tokens"""
token_patterns = []
for event in self.events:
if event.response_data and 'usage' in event.response_data:
usage = event.response_data['usage']
token_patterns.append({
'step': len(token_patterns),
'prompt_tokens': usage.get('prompt_tokens', 0),
'completion_tokens': usage.get('completion_tokens', 0),
'total_tokens': usage.get('total_tokens', 0)
})
if not token_patterns:
return {}
import numpy as np
completion_tokens = [p['completion_tokens'] for p in token_patterns]
return {
'total_requests': len(token_patterns),
'avg_completion_tokens': round(np.mean(completion_tokens), 2),
'max_completion_tokens': max(completion_tokens),
'min_completion_tokens': min(completion_tokens),
'std_completion_tokens': round(np.std(completion_tokens), 2),
'pattern': token_patterns
}
def generate_report(self) -> str:
"""Génère un rapport complet de debugging"""
errors = self.find_errors()
token_analysis = self.analyze_token_pattern()
report = f"""
╔══════════════════════════════════════════════════════════════╗
║ RAPPORT DE DEBUGGING - SESSION {self.session_id} ║
╠══════════════════════════════════════════════════════════════╣
║ Date: {self.start_time}
║ Total événements: {len(self.events)}
║ Erreurs détectées: {len(errors)}
╚══════════════════════════════════════════════════════════════╝
{'='*60}
ANALYSE DES ERREURS
{'='*60}
"""
if errors:
for idx, error in errors:
report += f"""
❌ Erreur #{idx + 1}:
Type: {error['type']}
Message: {error['message']}
Timestamp: {error['timestamp']}
"""
else:
report += "\n✅ Aucune erreur détectée dans cette session.\n"
if token_analysis:
report += f"""
{'='*60}
ANALYSE DES TOKENS
{'='*60}
📊 Requêtes: {token_analysis['total_requests']}
📈 Moyenne tokens output: {token_analysis['avg_completion_tokens']}
📉 Min/Max: {token_analysis['min_completion_tokens']} / {token_analysis['max_completion_tokens']}
📐 Écart-type: {token_analysis['std_completion_tokens']}
"""
return report
def save(self, filename: Optional[str] = None):
"""Sauvegarde la session complète"""
if filename is None:
filename = f"replay_{self.session_id}_{int(time.time())}.json"
filepath = self.output_dir / filename
data = {
'session_id': self.session_id,
'start_time': self.start_time,
'save_time': datetime.utcnow().isoformat(),
'events': [
{
'timestamp': e.timestamp,
'event_type': e.event_type,
'request_data': e.request_data,
'response_data': e.response_data,
'error_data': e.error_data,
'duration_ms': e.duration_ms,
'event_hash': e.event_hash
}
for e in self.events
]
}
with open(filepath, 'w', encoding='utf-8') as f:
json.dump(data, f, indent=2, ensure_ascii=False)
print(f"💾 Session sauvegardée: {filepath}")
return str(filepath)
@classmethod
def load(cls, filepath: str) -> 'SessionReplay':
"""Charge une session sauvegardée"""
with open(filepath, 'r', encoding='utf-8') as f:
data = json.load(f)
replay = cls(session_id=data['session_id'])
replay.start_time = data['start_time']
for event_data in data['events']:
event = ReplayEvent(
timestamp=event_data['timestamp'],
event_type=event_data['event_type'],
request_data=event_data.get('request_data'),
response_data=event_data.get('response_data'),
error_data=event_data.get('error_data'),
duration_ms=event_data.get('duration_ms', 0.0),
event_hash=event_data.get('event_hash', '')
)
replay.events.append(event)
# Reconstruire le cache
if event.response_data:
replay.response_cache[event.event_hash] = {
'response': event.response_data,
'duration_ms': event.duration_ms
}
print(f"📂 Session chargée: {len(replay.events)} événements")
return replay
=== UTILISATION ===
if __name__ == "__main__":
# Démarrer une nouvelle session de replay
replay = SessionReplay("debug-001", output_dir="./debug_replays")
# Enregistrer une série de requêtes
request_hash = replay.record_request(
model="deepseek-chat",
messages=[
{"role": "system", "content": "Tu es un assistant technique."},
{"role": "user", "content": "Explique le concept de token dans les LLMs."}
]
)
# Simuler une réponse (en réel, ceci viendrait de l'API)
mock_response = {
"choices": [{
"message": {
"content": "Les tokens sont les unités de base du texte..."
}
}],
"usage": {
"prompt_tokens": 50,
"completion_tokens": 150,
"total_tokens": 200
}
}
replay.record_response(request_hash, mock_response, duration_ms=45.2)
# Générer et afficher le rapport
report = replay.generate_report()
print(report)
# Sauvegarder pour analyse future
filepath = replay.save()
# Charger et analyser une session passée
loaded_replay = SessionReplay.load(filepath)