# core/hermes_loop.py """ Boucle Hermès — méthode Fable en 7 étapes : Ancrer → Raisonner → Agir → Observer → Réévaluer → Vérifier → Narrer """ import time import logging from typing import Dict, Any log = logging.getLogger("vortex.hermes") class HermesLoop: def __init__(self, llm_engine, memory, error_tree, semantic_router): self.llm = llm_engine self.memory = memory self.error_tree = error_tree self.router = semantic_router async def run(self, task: str) -> Dict[str, Any]: log.info(f"[Hermès] Démarrage : {task[:80]}...") start_time = time.perf_counter() steps = [] deliverable = "Non produit" narration = "Non généré" # Leçons passées pertinentes (Harbor) lessons_ctx = "" try: similar = self.error_tree.search_similar(task) if similar: lessons_ctx = f"\n[Leçons passées] {similar[0].get('lesson') or similar[0].get('fix') or ''}" except Exception: pass step_defs = [ ("anchor", "Cadrer le problème", "contraintes listées, objectif clair", 0.9), ("reason", "Explorer les approches", "≥3 options explorées, ≥2 écartées", 0.85), ("act", "Produire la solution", "artefact complet, syntaxe valide", 0.85), ("observe", "Capturer le résultat", "métriques chiffrées, logs", 0.7), ("reeval", "Comparer à l'attendu", "écart quantifié < 5%", 0.7), ("verify", "Valider formellement", "tests automatiques passés", 0.8), ("narrate", "Documenter et livrer", "rapport structuré, recommandations", 0.9), ] context = f"Tâche : {task}{lessons_ctx}\n" for step_name, action, eval_criteria, default_score in step_defs: prompt = ( f"{context}\nÉtape : {step_name}\nAction : {action}\n" f"Consigne : {eval_criteria}\nRésultat :" ) score = default_score try: resp = await self.llm.call( agent="hermes", system="Tu es un assistant méthodique. Suis les consignes étape par étape.", user=prompt, max_tokens=300, temperature=0.3, use_cache=False ) result = resp.content if hasattr(resp, "content") else str(resp) except Exception as e: log.error(f"[Hermès] Erreur étape {step_name}: {e}") result = f"[Erreur] {e}" score = 0.0 try: self.error_tree.add_error(str(e), f"Hermès step={step_name} task={task[:100]}") except Exception: pass if step_name == "act": deliverable = result if step_name == "narrate": narration = result if step_name == "observe": score = min(1.0, len(result) / 500) * 0.7 + 0.3 steps.append({ "step": step_name, "action": action, "result": result[:1000], "score": round(score, 3), "eval_criteria": eval_criteria }) context += f"{step_name}: {result[:200]}\n" overall_score = sum(s["score"] for s in steps) / len(steps) if steps else 0.0 elapsed = time.perf_counter() - start_time log.info(f"[Hermès] Terminé en {elapsed:.2f}s, score {overall_score:.3f}") # Mémorisation du cycle try: from core.memory import MemoryTier self.memory.ingest( f"Hermès: {task[:150]} → score={overall_score:.2f}", MemoryTier.EPISODIC, "hermes", importance=overall_score, confidence=overall_score ) except Exception as e: log.warning(f"[Hermès] Mémorisation échouée: {e}") return { "task": task, "method": "Hermès/Fable", "elapsed_s": round(elapsed, 2), "overall_score": round(overall_score, 3), "steps": steps, "deliverable": deliverable[:1500], "narration": narration[:1500] }