David Blanc Brioir
05fc6a3994
- Reorder dataclass fields: required fields before default fields - Replace Unicode arrow (→) with ASCII (->) for cp1252 compatibility - Fixes Python dataclass initialization errors Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
246 lines
7.3 KiB
Python
246 lines
7.3 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
OccasionLogger - Logging des occasions d'expérience.
|
|
|
|
Chaque occasion est loggée avec:
|
|
- Trigger (type, contenu)
|
|
- Préhension (pensées/docs récupérés)
|
|
- Concrescence (réponse, outils utilisés)
|
|
- Satisfaction (nouvel état, paramètres)
|
|
- Profils avant/après
|
|
|
|
Les logs sont stockés en JSON pour analyse et debugging.
|
|
"""
|
|
|
|
import json
|
|
from dataclasses import dataclass, asdict, field
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import List, Optional, Dict, Any
|
|
|
|
|
|
@dataclass
|
|
class OccasionLog:
|
|
"""Structure de log pour une occasion."""
|
|
|
|
# Identifiants
|
|
occasion_id: int
|
|
timestamp: str
|
|
|
|
# Trigger
|
|
trigger_type: str # "user", "timer", "event"
|
|
trigger_content: str
|
|
|
|
# Préhension
|
|
previous_state_id: int
|
|
prehended_thoughts_count: int
|
|
prehended_docs_count: int
|
|
|
|
# Concrescence
|
|
response_summary: str
|
|
|
|
# Satisfaction
|
|
new_state_id: int
|
|
alpha_used: float
|
|
beta_used: float
|
|
|
|
# Champs avec valeurs par défaut (doivent être après les champs obligatoires)
|
|
prehended_thoughts: List[str] = field(default_factory=list) # Résumés des pensées
|
|
new_thoughts: List[str] = field(default_factory=list)
|
|
tools_used: List[str] = field(default_factory=list)
|
|
profile_before: Dict[str, Dict[str, float]] = field(default_factory=dict)
|
|
profile_after: Dict[str, Dict[str, float]] = field(default_factory=dict)
|
|
processing_time_ms: int = 0
|
|
token_count: Optional[int] = None
|
|
|
|
|
|
class OccasionLogger:
|
|
"""Gère le logging des occasions en fichiers JSON."""
|
|
|
|
def __init__(self, log_dir: str = "logs/occasions"):
|
|
"""
|
|
Args:
|
|
log_dir: Répertoire de stockage des logs
|
|
"""
|
|
self.log_dir = Path(log_dir)
|
|
self.log_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
def log(self, occasion: OccasionLog) -> Path:
|
|
"""
|
|
Enregistre une occasion.
|
|
|
|
Args:
|
|
occasion: OccasionLog à enregistrer
|
|
|
|
Returns:
|
|
Chemin du fichier créé
|
|
"""
|
|
filename = f"occasion_{occasion.occasion_id:06d}.json"
|
|
filepath = self.log_dir / filename
|
|
|
|
with open(filepath, 'w', encoding='utf-8') as f:
|
|
json.dump(asdict(occasion), f, indent=2, ensure_ascii=False)
|
|
|
|
print(f"[OccasionLogger] Occasion {occasion.occasion_id} -> {filepath}")
|
|
return filepath
|
|
|
|
def get_occasion(self, occasion_id: int) -> Optional[OccasionLog]:
|
|
"""
|
|
Récupère une occasion par son ID.
|
|
|
|
Args:
|
|
occasion_id: ID de l'occasion
|
|
|
|
Returns:
|
|
OccasionLog ou None si non trouvé
|
|
"""
|
|
filename = f"occasion_{occasion_id:06d}.json"
|
|
filepath = self.log_dir / filename
|
|
|
|
if not filepath.exists():
|
|
return None
|
|
|
|
with open(filepath, 'r', encoding='utf-8') as f:
|
|
data = json.load(f)
|
|
|
|
return OccasionLog(**data)
|
|
|
|
def get_recent_occasions(self, limit: int = 10) -> List[OccasionLog]:
|
|
"""
|
|
Récupère les N dernières occasions.
|
|
|
|
Args:
|
|
limit: Nombre max d'occasions à retourner
|
|
|
|
Returns:
|
|
Liste des occasions (plus récentes d'abord)
|
|
"""
|
|
files = sorted(self.log_dir.glob("occasion_*.json"), reverse=True)
|
|
|
|
occasions = []
|
|
for f in files[:limit]:
|
|
with open(f, 'r', encoding='utf-8') as fp:
|
|
data = json.load(fp)
|
|
occasions.append(OccasionLog(**data))
|
|
|
|
return occasions
|
|
|
|
def get_last_occasion_id(self) -> int:
|
|
"""Retourne l'ID de la dernière occasion (-1 si aucune)."""
|
|
files = sorted(self.log_dir.glob("occasion_*.json"), reverse=True)
|
|
if not files:
|
|
return -1
|
|
|
|
# Extraire l'ID du nom de fichier
|
|
filename = files[0].stem # occasion_000042
|
|
try:
|
|
return int(filename.split('_')[1])
|
|
except (IndexError, ValueError):
|
|
return -1
|
|
|
|
def get_profile_evolution(
|
|
self,
|
|
component: str,
|
|
last_n: int = 20
|
|
) -> List[tuple]:
|
|
"""
|
|
Retourne l'évolution d'une composante sur les N dernières occasions.
|
|
|
|
Args:
|
|
component: Nom de la composante (ex: "curiosity")
|
|
last_n: Nombre d'occasions à considérer
|
|
|
|
Returns:
|
|
Liste de tuples (occasion_id, valeur)
|
|
"""
|
|
occasions = self.get_recent_occasions(last_n)
|
|
|
|
evolution = []
|
|
for occ in reversed(occasions): # Ordre chronologique
|
|
# Chercher la composante dans le profil après
|
|
for category, comps in occ.profile_after.items():
|
|
if component in comps:
|
|
evolution.append((occ.occasion_id, comps[component]))
|
|
break
|
|
|
|
return evolution
|
|
|
|
def get_statistics(self, last_n: int = 100) -> Dict[str, Any]:
|
|
"""
|
|
Calcule des statistiques sur les occasions récentes.
|
|
|
|
Args:
|
|
last_n: Nombre d'occasions à analyser
|
|
|
|
Returns:
|
|
Dictionnaire de statistiques
|
|
"""
|
|
occasions = self.get_recent_occasions(last_n)
|
|
|
|
if not occasions:
|
|
return {"count": 0}
|
|
|
|
# Statistiques de base
|
|
processing_times = [o.processing_time_ms for o in occasions]
|
|
thoughts_created = [len(o.new_thoughts) for o in occasions]
|
|
tools_counts = [len(o.tools_used) for o in occasions]
|
|
|
|
# Répartition des triggers
|
|
trigger_types = {}
|
|
for o in occasions:
|
|
trigger_types[o.trigger_type] = trigger_types.get(o.trigger_type, 0) + 1
|
|
|
|
return {
|
|
"count": len(occasions),
|
|
"processing_time": {
|
|
"avg_ms": sum(processing_times) / len(processing_times),
|
|
"min_ms": min(processing_times),
|
|
"max_ms": max(processing_times),
|
|
},
|
|
"thoughts_created": {
|
|
"total": sum(thoughts_created),
|
|
"avg_per_occasion": sum(thoughts_created) / len(thoughts_created),
|
|
},
|
|
"tools": {
|
|
"avg_per_occasion": sum(tools_counts) / len(tools_counts),
|
|
},
|
|
"trigger_distribution": trigger_types,
|
|
}
|
|
|
|
|
|
# Test
|
|
if __name__ == "__main__":
|
|
logger = OccasionLogger("tests/temp_logs")
|
|
|
|
# Créer une occasion test
|
|
occasion = OccasionLog(
|
|
occasion_id=1,
|
|
timestamp=datetime.now().isoformat(),
|
|
trigger_type="user",
|
|
trigger_content="Test question sur Whitehead",
|
|
previous_state_id=0,
|
|
prehended_thoughts_count=5,
|
|
prehended_docs_count=2,
|
|
prehended_thoughts=["Pensée 1", "Pensée 2"],
|
|
response_summary="Réponse détaillée sur le processus...",
|
|
new_thoughts=["Nouvelle insight sur le devenir"],
|
|
tools_used=["search_thoughts", "search_library"],
|
|
new_state_id=1,
|
|
alpha_used=0.85,
|
|
beta_used=0.15,
|
|
profile_before={"epistemic": {"curiosity": 0.5, "certainty": 0.3}},
|
|
profile_after={"epistemic": {"curiosity": 0.55, "certainty": 0.32}},
|
|
processing_time_ms=1500
|
|
)
|
|
|
|
# Logger
|
|
filepath = logger.log(occasion)
|
|
print(f"Logged to: {filepath}")
|
|
|
|
# Relire
|
|
loaded = logger.get_occasion(1)
|
|
print(f"Loaded: trigger_type={loaded.trigger_type}, new_state_id={loaded.new_state_id}")
|
|
|
|
# Stats
|
|
print(f"Stats: {logger.get_statistics()}")
|