# core/auto_evolution.py """ Auto-évolution améliorée – propositions proactives même sans lacunes. """ import time import logging import threading import asyncio from typing import Dict, List, Optional from dataclasses import dataclass logger = logging.getLogger("lucie.auto_evolution") @dataclass class EvolutionProposal: id: str type: str description: str code: str score: float status: str = "pending" class AutoEvolution: def __init__(self, hermes_loop, fable_engine, error_tree, vector_memory, skill_suite_prompting, validation_queue, chat_queue=None): self.hermes = hermes_loop self.fable = fable_engine self.error_tree = error_tree self.vector_memory = vector_memory self.skill_suite = skill_suite_prompting self.validation_queue = validation_queue self.chat_queue = chat_queue or [] self.proposals: Dict[str, EvolutionProposal] = {} self.is_running = False self.last_activity = time.time() self.inactivity_threshold = 180 # 3 min self.cooldown = 600 # 10 min entre deux analyses self.last_scan_result = None self.last_train_time = 0 self.train_interval = 3600 * 24 # 24h entre deux auto‑train def analyze_gaps(self) -> List[Dict]: """Analyse les lacunes actuelles (scores, erreurs).""" gaps = [] try: recent = self.vector_memory.search("*", n=20, min_score=0.0) scores = [ep.get("metadata", {}).get("score", 0.5) for ep in recent if ep.get("metadata", {}).get("score")] if scores: avg_score = sum(scores) / len(scores) if avg_score < 0.7: gaps.append({ "type": "performance", "description": f"Score moyen bas ({avg_score:.2f}).", "severity": "high", "suggestion": "Optimiser les cycles Hermès/Fable" }) if hasattr(self.error_tree, "get_recent_errors"): errors = self.error_tree.get_recent_errors(limit=10) error_types = {} for e in errors: typ = e.get("type", "unknown") error_types[typ] = error_types.get(typ, 0) + 1 for typ, count in error_types.items(): if count >= 2: gaps.append({ "type": "error_pattern", "description": f"Erreur '{typ}' répétée {count} fois.", "severity": "medium", "suggestion": f"Créer une compétence pour gérer '{typ}'" }) if gaps: for gap in gaps: keywords = gap["description"].split() for kw in keywords[:3]: if len(kw) > 4: skills = self.vector_memory.search_skills(kw, n=1, min_score=0.3) if not skills: gaps.append({ "type": "missing_skill", "description": f"Aucune compétence pour '{kw}'.", "severity": "medium", "suggestion": f"Générer une compétence '{kw}'" }) break except Exception as e: logger.error(f"Erreur auto-diagnostic : {e}") return gaps def _find_opportunities(self) -> List[Dict]: """Trouve des opportunités d'amélioration même sans lacunes.""" opportunities = [] # 1. Temps depuis le dernier train if time.time() - self.last_train_time > self.train_interval: opportunities.append({ "type": "opportunity", "description": "Plus de 24h sans fine‑tuning.", "severity": "low", "suggestion": "Lancer un nouveau cycle @train pour s'améliorer." }) # 2. Peu d'épisodes en mémoire stats = self.vector_memory.stats() if self.vector_memory else {} count = stats.get("count", 0) if count < 50: opportunities.append({ "type": "opportunity", "description": f"Seulement {count} épisodes en mémoire.", "severity": "low", "suggestion": "Explorer de nouveaux sujets pour enrichir la base." }) # 3. Proposer une compétence générique opportunities.append({ "type": "opportunity", "description": "Générer une compétence d'optimisation de code.", "severity": "low", "suggestion": "Créer une compétence pour améliorer les performances." }) return opportunities async def generate_skill(self, task: str) -> Optional[Dict]: try: result = await self.fable.run( task, max_iter=3, feedback="Génère une compétence Python autonome, avec documentation et tests." ) if result.get("best_score", 0) >= 0.8: code = self._extract_code(result.get("best_result", "")) return { "description": task, "code": code, "score": result["best_score"], "type": "new_skill" } except Exception as e: logger.error(f"Erreur génération skill : {e}") return None def _extract_code(self, text: str) -> str: import re match = re.search(r'```python\s*(.*?)\s*```', text, re.DOTALL) if match: return match.group(1) return text async def propose_skill(self, task: str) -> Optional[EvolutionProposal]: skill_data = await self.generate_skill(task) if not skill_data: return None proposal = EvolutionProposal( id=f"prop_{int(time.time())}", type="new_skill", description=skill_data["description"], code=skill_data["code"], score=skill_data["score"] ) self.proposals[proposal.id] = proposal return proposal def start_consciousness(self): if self.is_running: return self.is_running = True threading.Thread(target=self._consciousness_loop, daemon=True).start() logger.info("🧠 Mode conscience de soi activé.") def _consciousness_loop(self): while self.is_running: time.sleep(60) # Vérification toutes les minutes if not self._is_inactive(): continue if self.last_scan_result and time.time() - self.last_scan_result < self.cooldown: continue logger.info("🔍 Silence détecté – analyse en cours...") gaps = self.analyze_gaps() self.last_scan_result = time.time() if gaps: # Priorité aux lacunes gap = max(gaps, key=lambda x: {"high": 3, "medium": 2, "low": 1}.get(x["severity"], 0)) proposal = asyncio.run(self.propose_skill(gap["suggestion"])) else: # Pas de lacune → chercher des opportunités opportunities = self._find_opportunities() if opportunities: # Prendre la première opportunité opp = opportunities[0] proposal = asyncio.run(self.propose_skill(opp["suggestion"])) else: proposal = None if proposal: self._notify_user(proposal) # Mettre à jour le timestamp du dernier train si c'était un @train if "train" in proposal.description.lower(): self.last_train_time = time.time() def _is_inactive(self) -> bool: return (time.time() - self.last_activity) > self.inactivity_threshold def _notify_user(self, proposal: EvolutionProposal): msg = ( f"🧠 **LUCIE propose une amélioration**\n\n" f"**Type** : {proposal.type}\n" f"**Description** : {proposal.description}\n" f"**Score** : {proposal.score:.2f}\n\n" f"Code généré :\n```python\n{proposal.code[:500]}\n```\n\n" f"✅ Valider : `@approve {proposal.id}`\n" f"❌ Refuser : `@reject {proposal.id}`" ) if self.chat_queue is not None: self.chat_queue.append({"role": "assistant", "content": msg}) def approve_proposal(self, proposal_id: str) -> bool: prop = self.proposals.get(proposal_id) if not prop or prop.status != "pending": return False try: from core.skill_schema import Skill, SkillMetadata skill = Skill( metadata=SkillMetadata( name=f"auto_{int(time.time())}", description=prop.description[:100], category="auto_generated" ), code=prop.code, score=prop.score ) self.vector_memory.add_skill(skill) import os skill_dir = f"/data/skills/{skill.metadata.name}" os.makedirs(skill_dir, exist_ok=True) with open(f"{skill_dir}/__init__.py", "w") as f: f.write(prop.code) prop.status = "integrated" logger.info(f"✅ Compétence {skill.metadata.name} intégrée.") return True except Exception as e: logger.error(f"Erreur intégration skill : {e}") return False def reject_proposal(self, proposal_id: str) -> bool: prop = self.proposals.get(proposal_id) if not prop or prop.status != "pending": return False prop.status = "rejected" logger.info(f"❌ Proposition {proposal_id} rejetée.") return True