feat(cognition): mémoire de travail — Léa sait où elle en est
Some checks failed
security-audit / Bandit (scan statique) (push) Successful in 13s
security-audit / pip-audit (CVE dépendances) (push) Successful in 10s
security-audit / Scan secrets (grep) (push) Successful in 9s
tests / Lint (ruff + black) (push) Successful in 15s
tests / Tests unitaires (sans GPU) (push) Failing after 15s
tests / Tests sécurité (critique) (push) Has been skipped
Some checks failed
security-audit / Bandit (scan statique) (push) Successful in 13s
security-audit / pip-audit (CVE dépendances) (push) Successful in 10s
security-audit / Scan secrets (grep) (push) Successful in 9s
tests / Lint (ruff + black) (push) Successful in 15s
tests / Tests unitaires (sans GPU) (push) Failing after 15s
tests / Tests sécurité (critique) (push) Has been skipped
CognitiveContext : bloc-notes interne réinjecté à chaque décision. - objective : ce que Léa essaie de faire - current_step : progression dans le plan - action_history : les N dernières actions (succès/échec) - learned_facts : faits appris pendant l'exécution - confidence : auto-évaluation (baisse sur échec) - needs_help : demande d'aide à l'humain - to_prompt_context() : génère le texte pour le VLM Module standalone, pas encore câblé dans l'executor. Testé sur scénario de facturation OSIRIS. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
0
core/cognition/__init__.py
Normal file
0
core/cognition/__init__.py
Normal file
203
core/cognition/working_memory.py
Normal file
203
core/cognition/working_memory.py
Normal file
@@ -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,
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user