diff --git a/core/cognition/__init__.py b/core/cognition/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/core/cognition/working_memory.py b/core/cognition/working_memory.py new file mode 100644 index 000000000..4a135e931 --- /dev/null +++ b/core/cognition/working_memory.py @@ -0,0 +1,203 @@ +""" +Mémoire de travail de Léa — contexte cognitif pendant l'exécution. + +Donne à Léa la conscience de "où elle en est" : +- Quel objectif elle poursuit +- Quel écran elle voit +- Ce qu'elle vient de faire +- Ce qu'elle doit faire ensuite +- Ce qu'elle a appris en cours de route + +Sans ça, chaque étape est indépendante — Léa est amnésique entre +deux actions. Avec ça, elle raisonne en contexte. +""" + +import logging +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any, Dict, List, Optional + +logger = logging.getLogger(__name__) + + +@dataclass +class Observation: + """Ce que Léa observe sur l'écran à un instant donné.""" + timestamp: datetime + window_title: str = "" + application: str = "" + ocr_text: str = "" + ui_pattern: Optional[str] = None + screen_description: str = "" + confidence: float = 0.0 + + +@dataclass +class ActionRecord: + """Une action que Léa a effectuée.""" + timestamp: datetime + action_type: str + target: str = "" + result: str = "" + success: bool = True + duration_ms: float = 0 + + +@dataclass +class CognitiveContext: + """Contexte cognitif complet — la "pensée" de Léa à un instant donné. + + C'est le bloc-notes interne qui est réinjecté à chaque décision. + Le VLM reçoit ce contexte pour raisonner en connaissance de cause. + """ + + # Objectif global (ce que Léa essaie d'accomplir) + objective: str = "" + + # Étape courante dans le plan + current_step: int = 0 + total_steps: int = 0 + current_step_description: str = "" + + # Ce que Léa voit maintenant + current_observation: Optional[Observation] = None + + # Historique des N dernières actions (mémoire court terme) + action_history: List[ActionRecord] = field(default_factory=list) + max_history: int = 10 + + # Ce que Léa a appris pendant cette session + learned_facts: List[str] = field(default_factory=list) + + # Plan : les étapes restantes + remaining_steps: List[str] = field(default_factory=list) + + # État émotionnel / confiance + confidence: float = 1.0 + needs_help: bool = False + help_reason: str = "" + + # Métadonnées + session_id: str = "" + machine_id: str = "" + started_at: Optional[datetime] = None + + def record_action(self, action_type: str, target: str = "", + result: str = "", success: bool = True, + duration_ms: float = 0): + """Enregistre une action dans l'historique.""" + self.action_history.append(ActionRecord( + timestamp=datetime.now(), + action_type=action_type, + target=target, + result=result, + success=success, + duration_ms=duration_ms, + )) + if len(self.action_history) > self.max_history: + self.action_history = self.action_history[-self.max_history:] + + if not success: + self.confidence = max(0, self.confidence - 0.2) + else: + self.confidence = min(1.0, self.confidence + 0.05) + + def observe(self, window_title: str = "", application: str = "", + ocr_text: str = "", ui_pattern: Optional[str] = None, + screen_description: str = ""): + """Met à jour l'observation courante.""" + self.current_observation = Observation( + timestamp=datetime.now(), + window_title=window_title, + application=application, + ocr_text=ocr_text, + ui_pattern=ui_pattern, + screen_description=screen_description, + ) + + def advance_step(self): + """Passe à l'étape suivante du plan.""" + self.current_step += 1 + if self.remaining_steps: + self.current_step_description = self.remaining_steps.pop(0) + + def learn(self, fact: str): + """Enregistre un fait appris pendant l'exécution.""" + if fact not in self.learned_facts: + self.learned_facts.append(fact) + logger.info(f"Fait appris: {fact}") + + def ask_for_help(self, reason: str): + """Signale que Léa a besoin d'aide.""" + self.needs_help = True + self.help_reason = reason + self.confidence = max(0, self.confidence - 0.3) + logger.warning(f"Léa demande de l'aide: {reason}") + + def to_prompt_context(self) -> str: + """Génère le contexte à injecter dans le prompt VLM. + + C'est ce texte qui donne au VLM la conscience de la situation. + """ + lines = [] + + if self.objective: + lines.append(f"OBJECTIF : {self.objective}") + + if self.current_step > 0: + lines.append(f"PROGRESSION : étape {self.current_step}/{self.total_steps}") + if self.current_step_description: + lines.append(f"ÉTAPE EN COURS : {self.current_step_description}") + + if self.current_observation: + obs = self.current_observation + if obs.window_title: + lines.append(f"FENÊTRE ACTIVE : {obs.window_title}") + if obs.application: + lines.append(f"APPLICATION : {obs.application}") + if obs.ui_pattern: + lines.append(f"DIALOGUE DÉTECTÉ : {obs.ui_pattern}") + + if self.action_history: + last_actions = self.action_history[-3:] + lines.append("DERNIÈRES ACTIONS :") + for a in last_actions: + status = "OK" if a.success else "ÉCHEC" + lines.append(f" - {a.action_type} '{a.target}' → {status}") + + if self.learned_facts: + lines.append("FAITS APPRIS :") + for fact in self.learned_facts[-5:]: + lines.append(f" - {fact}") + + if self.remaining_steps: + lines.append("PROCHAINES ÉTAPES :") + for step in self.remaining_steps[:3]: + lines.append(f" - {step}") + + lines.append(f"CONFIANCE : {self.confidence:.0%}") + + if self.needs_help: + lines.append(f"BESOIN D'AIDE : {self.help_reason}") + + return "\n".join(lines) + + def to_dict(self) -> Dict[str, Any]: + """Sérialise le contexte pour le stockage/transport.""" + return { + "objective": self.objective, + "current_step": self.current_step, + "total_steps": self.total_steps, + "current_step_description": self.current_step_description, + "confidence": self.confidence, + "needs_help": self.needs_help, + "help_reason": self.help_reason, + "action_count": len(self.action_history), + "learned_facts": self.learned_facts, + "remaining_steps": self.remaining_steps, + "last_observation": { + "window_title": self.current_observation.window_title, + "application": self.current_observation.application, + "ui_pattern": self.current_observation.ui_pattern, + } if self.current_observation else None, + }