Spaces:
Configuration error
Configuration error
| # 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] | |
| } | |