MORSATIPIK / hermes_loop.py
RDS777's picture
Upload hermes_loop.py
55b81b3 verified
Raw
History Blame Contribute Delete
4.49 kB
# 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]
}