488 lines
17 KiB
Python
488 lines
17 KiB
Python
# agent_v0/server_v1/replay_memory.py
|
|
"""
|
|
replay_memory — Greffe de TargetMemoryStore (Fiche #18) sur le pipeline V4.
|
|
|
|
Phase 1 du plan apprentissage Léa (docs/PLAN_APPRENTISSAGE_LEA.md).
|
|
|
|
Le runtime V4 appelle :
|
|
- `memory_lookup()` AVANT la cascade coûteuse (OCR/template/VLM)
|
|
- `memory_record_success()` APRÈS validation post-condition (`title_match` strict)
|
|
- `memory_record_failure()` sur les échecs
|
|
|
|
Fingerprint léger V4 : les coordonnées clic (x_pct, y_pct) sont stockées dans
|
|
les deux premières valeurs de `TargetFingerprint.bbox`, et la méthode de
|
|
résolution ayant réussi dans le champ `etype`.
|
|
|
|
Signature d'écran V4 : `sha256(normalize(window_title))[:16]`. Simple et
|
|
robuste aux données variables car les titres de fenêtre restent stables.
|
|
Les faux positifs (même titre, écrans différents) sont rattrapés par la
|
|
post-condition qui décrémentera la fiabilité via `record_failure()`.
|
|
|
|
Critère de fiabilité : 2 succès minimum et < 30% d'échecs pour déclencher
|
|
un hit (paramètres de `TargetMemoryStore.lookup`). C'est exactement la
|
|
cristallisation par répétition que l'on veut — Léa est un stagiaire qui
|
|
apprend de l'observation.
|
|
|
|
Auteur : Dom, Alice — avril 2026
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import logging
|
|
import os
|
|
import unicodedata
|
|
from typing import Any, Dict, Optional
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# =========================================================================
|
|
# Singleton du store persistant
|
|
# =========================================================================
|
|
|
|
_MEMORY_SINGLETON: Optional[Any] = None
|
|
_MEMORY_DISABLED = False
|
|
|
|
_GENERIC_BUTTON_TEXTS = {
|
|
"annuler",
|
|
"cancel",
|
|
"enregistrer",
|
|
"non",
|
|
"no",
|
|
"ok",
|
|
"oui",
|
|
"ouvrir",
|
|
"open",
|
|
"remplacer",
|
|
"replace",
|
|
"save",
|
|
"yes",
|
|
}
|
|
|
|
|
|
def get_memory_store():
|
|
"""Retourne le `TargetMemoryStore` partagé, ou None si indisponible.
|
|
|
|
Lazy-init : le store n'est créé qu'au premier appel, ce qui évite
|
|
d'importer `core.learning.target_memory_store` à l'import du module
|
|
(et donc d'éviter les effets de bord sur le démarrage du serveur).
|
|
"""
|
|
global _MEMORY_SINGLETON, _MEMORY_DISABLED
|
|
|
|
if _MEMORY_DISABLED:
|
|
return None
|
|
if _MEMORY_SINGLETON is not None:
|
|
return _MEMORY_SINGLETON
|
|
|
|
try:
|
|
from core.learning.target_memory_store import TargetMemoryStore
|
|
|
|
base_path = os.environ.get("RPA_LEARNING_DIR", "data/learning")
|
|
_MEMORY_SINGLETON = TargetMemoryStore(base_path=base_path)
|
|
logger.info(
|
|
"replay_memory: TargetMemoryStore initialisé (base=%s)", base_path,
|
|
)
|
|
return _MEMORY_SINGLETON
|
|
except Exception as exc:
|
|
logger.warning(
|
|
"replay_memory: TargetMemoryStore indisponible (%s) — "
|
|
"l'apprentissage persistant est désactivé", exc,
|
|
)
|
|
_MEMORY_DISABLED = True
|
|
return None
|
|
|
|
|
|
# =========================================================================
|
|
# Normalisation de texte et hash
|
|
# =========================================================================
|
|
|
|
|
|
def _norm_text(s: str) -> str:
|
|
"""Normalise un texte pour un hash stable (accents, casse, NBSP, espaces)."""
|
|
if not s:
|
|
return ""
|
|
s = s.replace("\u00A0", " ").strip().lower()
|
|
s = unicodedata.normalize("NFKD", s)
|
|
s = "".join(ch for ch in s if not unicodedata.combining(ch))
|
|
return " ".join(s.split())
|
|
|
|
|
|
def _memory_lookup_skip_reason(target_spec: Dict[str, Any]) -> str:
|
|
"""Retourne la raison pour laquelle la mémoire ne doit pas court-circuiter.
|
|
|
|
Les clics qui changent de fenêtre doivent être résolus visuellement à
|
|
l'instant T : une coordonnée apprise peut être une bonne piste, mais pas
|
|
une décision finale. Pour les boutons très génériques, on exige au moins
|
|
un contexte de fenêtre/interaction dans la clé mémoire afin d'éviter les
|
|
collisions entre « Enregistrer », « OK », « Oui », etc.
|
|
"""
|
|
if not isinstance(target_spec, dict):
|
|
return ""
|
|
|
|
hints = target_spec.get("context_hints") or {}
|
|
if bool(hints.get("requires_window_transition")):
|
|
return "window_transition_requires_visual_confirmation"
|
|
|
|
button_text = _norm_text(str(target_spec.get("by_text") or ""))
|
|
if button_text not in _GENERIC_BUTTON_TEXTS:
|
|
return ""
|
|
|
|
before = (
|
|
hints.get("expected_window_before")
|
|
or hints.get("button_expected_before_window")
|
|
or hints.get("window_title")
|
|
or target_spec.get("window_title")
|
|
)
|
|
after = (
|
|
hints.get("expected_window_after")
|
|
or hints.get("button_expected_after_window")
|
|
or hints.get("expected_after_window")
|
|
)
|
|
interaction = hints.get("interaction") or hints.get("foreground_dialog_id")
|
|
role = target_spec.get("by_role")
|
|
if not (before and role and (after or interaction)):
|
|
return "generic_button_missing_context"
|
|
return ""
|
|
|
|
|
|
def compute_screen_sig(window_title: str) -> str:
|
|
"""Calcule la signature d'écran V4 à partir du titre de fenêtre.
|
|
|
|
Le `window_title` est strict depuis la phase "controle des étapes"
|
|
(post-condition `title_match` obligatoire). C'est notre clé naturelle.
|
|
"""
|
|
norm = _norm_text(window_title)
|
|
if not norm:
|
|
return ""
|
|
return hashlib.sha256(norm.encode("utf-8")).hexdigest()[:16]
|
|
|
|
|
|
def _round_float_list(values: Any, precision: int = 4) -> Optional[tuple[float, ...]]:
|
|
"""Normaliser une liste de coordonnées flottantes pour le hash mémoire."""
|
|
if not isinstance(values, (list, tuple)):
|
|
return None
|
|
out = []
|
|
for value in values:
|
|
try:
|
|
out.append(round(float(value), precision))
|
|
except (TypeError, ValueError):
|
|
return None
|
|
return tuple(out)
|
|
|
|
|
|
def _int_pair(values: Any) -> Optional[tuple[int, int]]:
|
|
"""Extraire une paire entière stable pour les hints spatiaux."""
|
|
if not isinstance(values, (list, tuple)) or len(values) < 2:
|
|
return None
|
|
try:
|
|
return int(values[0]), int(values[1])
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def _should_reuse_recorded_window_relative_coords(fp: Any) -> bool:
|
|
"""Décider si on doit remplacer la mémoire apprise par la position source.
|
|
|
|
Cette réécriture n'est légitime que pour les entrées faibles de type
|
|
`position_fallback`/`v4_unknown`, où la mémoire ne contient pas une vraie
|
|
localisation visuelle robuste mais seulement un clic écran dépendant de la
|
|
résolution. Pour les méthodes visuelles apprises (template, SoM, OCR...),
|
|
réinjecter un vieux `click_relative` source crée des collisions et des
|
|
dérives sur des boutons homonymes (`Enregistrer`, `OK`, etc.).
|
|
"""
|
|
method = str(getattr(fp, "etype", "") or "").strip().lower()
|
|
return method in {"position_fallback", "v4_unknown"}
|
|
|
|
|
|
class _TargetSpecLike:
|
|
"""Adaptateur dict → objet pour `TargetMemoryStore._hash_target_spec()`.
|
|
|
|
Le hash interne de TargetMemoryStore utilise `getattr(spec, "by_role", ...)`
|
|
qui ne fonctionne pas avec un dict brut. On expose les attributs nécessaires.
|
|
|
|
On intègre aussi `resolve_order`, `vlm_description` et des indices
|
|
spatiaux (SoM, click_relative) dans `context_hints` pour qu'ils entrent
|
|
dans le hash. Sinon, deux actions `Enregistrer` dans la même fenêtre
|
|
mais à des emplacements différents collisionnent.
|
|
"""
|
|
|
|
__slots__ = ("by_role", "by_text", "by_position", "context_hints")
|
|
|
|
def __init__(self, d: Dict[str, Any]):
|
|
self.by_role = d.get("by_role", "") or ""
|
|
self.by_text = d.get("by_text", "") or ""
|
|
self.by_position = d.get("by_position")
|
|
|
|
hints = dict(d.get("context_hints") or {})
|
|
resolve_order = d.get("resolve_order")
|
|
if resolve_order:
|
|
hints["_resolve_order"] = "|".join(resolve_order) if isinstance(
|
|
resolve_order, list
|
|
) else str(resolve_order)
|
|
if d.get("vlm_description"):
|
|
hints["_vlm_desc"] = str(d["vlm_description"])
|
|
if d.get("anchor_hint"):
|
|
hints["_anchor_hint"] = str(d["anchor_hint"])
|
|
|
|
som_element = d.get("som_element") or {}
|
|
som_bbox = _round_float_list(som_element.get("bbox_norm"))
|
|
if som_bbox:
|
|
hints["_som_bbox"] = som_bbox
|
|
som_center = _round_float_list(som_element.get("center_norm"), precision=5)
|
|
if som_center:
|
|
hints["_som_center"] = som_center
|
|
|
|
window_capture = d.get("window_capture") or {}
|
|
click_relative = _int_pair(window_capture.get("click_relative"))
|
|
window_size = _int_pair(window_capture.get("window_size"))
|
|
if click_relative and window_size:
|
|
hints["_window_rel"] = f"{click_relative[0]},{click_relative[1]}@{window_size[0]}x{window_size[1]}"
|
|
|
|
self.context_hints = hints
|
|
|
|
|
|
# =========================================================================
|
|
# Lookup — consulté AVANT la cascade coûteuse
|
|
# =========================================================================
|
|
|
|
|
|
def memory_lookup(
|
|
window_title: str,
|
|
target_spec: Dict[str, Any],
|
|
) -> Optional[Dict[str, Any]]:
|
|
"""Cherche une résolution apprise pour cette cible sur cet écran.
|
|
|
|
Returns:
|
|
Dict compatible avec le format de sortie de `_resolve_target_sync`
|
|
(resolved, method, x_pct, y_pct, score, ...) si une entrée fiable
|
|
est trouvée. None sinon.
|
|
"""
|
|
skip_reason = _memory_lookup_skip_reason(target_spec)
|
|
if skip_reason:
|
|
logger.info("memory_lookup SKIP : %s", skip_reason)
|
|
return None
|
|
|
|
store = get_memory_store()
|
|
if store is None:
|
|
return None
|
|
|
|
screen_sig = compute_screen_sig(window_title)
|
|
if not screen_sig:
|
|
return None
|
|
|
|
try:
|
|
spec_shim = _TargetSpecLike(target_spec)
|
|
fp = store.lookup(screen_sig, spec_shim)
|
|
except Exception as exc:
|
|
logger.debug("memory_lookup: erreur lookup (%s)", exc)
|
|
return None
|
|
|
|
if fp is None:
|
|
return None
|
|
|
|
# Fingerprint léger : bbox = (x_pct, y_pct, 0, 0)
|
|
try:
|
|
x_pct = float(fp.bbox[0])
|
|
y_pct = float(fp.bbox[1])
|
|
except (TypeError, IndexError, ValueError):
|
|
logger.debug("memory_lookup: fingerprint bbox invalide")
|
|
return None
|
|
|
|
# Quand l'entrée mémoire provient d'un simple `position_fallback`, les
|
|
# coordonnées stockées reflètent surtout la géométrie écran source. Dans
|
|
# ce cas précis, réutiliser la position relative enregistrée dans la
|
|
# fenêtre source reste préférable si elle existe.
|
|
#
|
|
# En revanche, pour une méthode visuelle réellement apprise
|
|
# (`anchor_template`, `som_*`, `hybrid_text_direct`, ...), remplacer les
|
|
# coords mémorisées par un vieux `click_relative` crée des dérives sur
|
|
# des cibles textuelles homonymes. On garde donc les coords apprises.
|
|
window_capture = target_spec.get("window_capture") or {}
|
|
click_relative = window_capture.get("click_relative")
|
|
window_size = window_capture.get("window_size")
|
|
if (
|
|
_should_reuse_recorded_window_relative_coords(fp)
|
|
and (
|
|
isinstance(click_relative, (list, tuple))
|
|
and len(click_relative) >= 2
|
|
and isinstance(window_size, (list, tuple))
|
|
and len(window_size) >= 2
|
|
)
|
|
):
|
|
try:
|
|
rel_x = float(click_relative[0])
|
|
rel_y = float(click_relative[1])
|
|
win_w = float(window_size[0])
|
|
win_h = float(window_size[1])
|
|
if win_w > 1 and win_h > 1:
|
|
x_pct = rel_x / win_w
|
|
y_pct = rel_y / win_h
|
|
logger.info(
|
|
"memory_lookup: coords fenêtre source réutilisées "
|
|
"(click_relative=%s, window_size=%s) -> (%.4f, %.4f)",
|
|
click_relative,
|
|
window_size,
|
|
x_pct,
|
|
y_pct,
|
|
)
|
|
except (TypeError, ValueError, ZeroDivisionError):
|
|
logger.debug("memory_lookup: window_capture invalide, fallback bbox")
|
|
|
|
# Sanity check : les pourcentages doivent être dans [0, 1]
|
|
if not (0.0 <= x_pct <= 1.0 and 0.0 <= y_pct <= 1.0):
|
|
logger.warning(
|
|
"memory_lookup: coords invalides (%.3f, %.3f) pour sig=%s — "
|
|
"entrée ignorée",
|
|
x_pct, y_pct, screen_sig,
|
|
)
|
|
return None
|
|
|
|
method = fp.etype or "memory"
|
|
confidence = float(getattr(fp, "confidence", 0.9) or 0.9)
|
|
|
|
logger.info(
|
|
"memory_lookup HIT : sig=%s method=%s coords=(%.4f, %.4f) conf=%.2f "
|
|
"target='%s'",
|
|
screen_sig, method, x_pct, y_pct, confidence,
|
|
(target_spec.get("by_text") or "")[:60],
|
|
)
|
|
|
|
return {
|
|
"resolved": True,
|
|
"method": f"memory_{method}",
|
|
"x_pct": x_pct,
|
|
"y_pct": y_pct,
|
|
"score": confidence,
|
|
"from_memory": True,
|
|
"screen_sig": screen_sig,
|
|
}
|
|
|
|
|
|
# =========================================================================
|
|
# Record — appelé APRÈS validation post-condition
|
|
# =========================================================================
|
|
|
|
|
|
def memory_record_success(
|
|
window_title: str,
|
|
target_spec: Dict[str, Any],
|
|
x_pct: float,
|
|
y_pct: float,
|
|
method: str,
|
|
confidence: float = 0.9,
|
|
) -> bool:
|
|
"""Enregistre une résolution réussie dans la mémoire persistante.
|
|
|
|
À appeler APRÈS validation de la post-condition (`title_match` strict).
|
|
"""
|
|
store = get_memory_store()
|
|
if store is None:
|
|
return False
|
|
|
|
screen_sig = compute_screen_sig(window_title)
|
|
if not screen_sig:
|
|
return False
|
|
|
|
# Sanity check : coordonnées dans [0, 1]
|
|
try:
|
|
x_pct = float(x_pct)
|
|
y_pct = float(y_pct)
|
|
except (TypeError, ValueError):
|
|
logger.debug("memory_record_success: coords non numériques, skip")
|
|
return False
|
|
if not (0.0 <= x_pct <= 1.0 and 0.0 <= y_pct <= 1.0):
|
|
logger.warning(
|
|
"memory_record_success: coords hors [0,1] (%.3f, %.3f), skip — "
|
|
"probable input parasite (target='%s' method=%s)",
|
|
x_pct, y_pct,
|
|
(target_spec.get("by_text") or "")[:60], method,
|
|
)
|
|
return False
|
|
# Rejeter (0.0, 0.0) exact : coin haut-gauche = signature de bruit
|
|
# (curseur NoMachine, événement OS parasite, listener pynput sans clic
|
|
# humain réel). Cf. bug observé replay_sess_63a1313b 2026-05-24 18:31-18:32.
|
|
if x_pct == 0.0 and y_pct == 0.0:
|
|
logger.warning(
|
|
"memory_record_success: coords (0.0, 0.0) rejetées — "
|
|
"signature de bruit (target='%s' method=%s)",
|
|
(target_spec.get("by_text") or "")[:60], method,
|
|
)
|
|
return False
|
|
|
|
try:
|
|
from core.learning.target_memory_store import TargetFingerprint
|
|
|
|
# Stripper les préfixes "memory_" empilés pour ne garder que
|
|
# la méthode de résolution originale (ex: template_matching).
|
|
# Sans ça, le cycle lookup → record → lookup empile "memory_"
|
|
# indéfiniment : memory_memory_memory_template_matching.
|
|
method_clean = method or "v4_unknown"
|
|
while method_clean.startswith("memory_"):
|
|
method_clean = method_clean[len("memory_"):]
|
|
method_clean = method_clean or "v4_unknown"
|
|
fingerprint = TargetFingerprint(
|
|
element_id=f"v4_{method_clean}",
|
|
bbox=(x_pct, y_pct, 0.0, 0.0),
|
|
role=target_spec.get("by_role", "") or None,
|
|
etype=method_clean,
|
|
label=(target_spec.get("by_text") or "")[:200] or None,
|
|
confidence=float(confidence),
|
|
)
|
|
|
|
spec_shim = _TargetSpecLike(target_spec)
|
|
store.record_success(
|
|
screen_signature=screen_sig,
|
|
target_spec=spec_shim,
|
|
fingerprint=fingerprint,
|
|
strategy_used=method_clean,
|
|
confidence=float(confidence),
|
|
)
|
|
|
|
logger.info(
|
|
"memory_record_success: sig=%s method=%s coords=(%.4f, %.4f) "
|
|
"target='%s'",
|
|
screen_sig, method_clean, x_pct, y_pct,
|
|
(target_spec.get("by_text") or "")[:60],
|
|
)
|
|
return True
|
|
|
|
except Exception as exc:
|
|
logger.warning("memory_record_success: échec (%s)", exc)
|
|
return False
|
|
|
|
|
|
def memory_record_failure(
|
|
window_title: str,
|
|
target_spec: Dict[str, Any],
|
|
error_message: str,
|
|
) -> bool:
|
|
"""Incrémente le `fail_count` pour cette (signature, target).
|
|
|
|
Appelé quand l'action échoue OU quand la post-condition n'est pas
|
|
satisfaite. Le `TargetMemoryStore.lookup()` ignorera cette entrée
|
|
si le ratio d'échecs dépasse 30%.
|
|
"""
|
|
store = get_memory_store()
|
|
if store is None:
|
|
return False
|
|
|
|
screen_sig = compute_screen_sig(window_title)
|
|
if not screen_sig:
|
|
return False
|
|
|
|
try:
|
|
spec_shim = _TargetSpecLike(target_spec)
|
|
store.record_failure(
|
|
screen_signature=screen_sig,
|
|
target_spec=spec_shim,
|
|
error_message=(error_message or "unknown")[:200],
|
|
)
|
|
logger.debug(
|
|
"memory_record_failure: sig=%s error='%s'",
|
|
screen_sig, (error_message or "")[:80],
|
|
)
|
|
return True
|
|
except Exception as exc:
|
|
logger.debug("memory_record_failure: échec (%s)", exc)
|
|
return False
|