feat: acteur gemma4 — décide PASSER/EXECUTER/STOPPER quand target_not_found

Quand le magnétoscope ne trouve pas la cible, au lieu de la pause
supervisée, gemma4 (Docker port 11435, think=True) reçoit le contexte
(action prévue + fenêtre active) et décide :
- PASSER : le résultat est déjà atteint (onglet actif, dialog ouvert)
- STOPPER : état incohérent (mauvaise app)
- EXECUTER : fallback vers la pause supervisée

Testé : gemma4 décide PASSER quand l'onglet est déjà actif (5s).

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Dom
2026-04-05 21:05:37 +02:00
parent 3bcf59e16f
commit 8a1dfc6e8b

View File

@@ -177,6 +177,70 @@ class ActionExecutorV1:
except Exception as e:
logger.error(f"Echec de l'ordre {action} : {e}")
# =========================================================================
# Acteur intelligent — décision gemma4 quand le magnétoscope bloque
# =========================================================================
def _actor_decide(self, action: dict, target_spec: dict) -> str:
"""Demander à gemma4 de décider quand le magnétoscope ne trouve pas la cible.
gemma4 reçoit le contexte (action prévue, état de l'écran) et décide :
- PASSER : l'état est déjà atteint (ex: onglet déjà actif)
- EXECUTER : l'action est nécessaire mais pas trouvable automatiquement
- STOPPER : l'état est incohérent, impossible de continuer
Appelle gemma4 en mode texte avec thinking (Docker port 11435).
Fallback : EXECUTER (pause supervisée) si gemma4 indisponible.
"""
import requests as _requests
gemma4_port = os.environ.get("GEMMA4_PORT", "11435")
by_text = target_spec.get("by_text", "")
window_title = target_spec.get("window_title", "")
# Récupérer le titre de la fenêtre ACTUELLE
try:
from ..window_info_crossplatform import get_active_window_info
current_info = get_active_window_info()
current_title = current_info.get("title", "")
except Exception:
current_title = ""
prompt = (
f"Tu es un robot RPA. L'action suivante est : cliquer sur '{by_text or 'un élément'}' "
f"dans '{window_title}'.\n"
f"La fenêtre active est \"{current_title}\".\n"
f"Dois-je faire cette action ?\n"
f"- EXECUTER : l'action est nécessaire\n"
f"- PASSER : le résultat est déjà atteint\n"
f"- STOPPER : état incohérent\n"
f"Réponds UN SEUL MOT."
)
try:
resp = _requests.post(
f"http://localhost:{gemma4_port}/api/chat",
json={
"model": "gemma4:e4b",
"messages": [{"role": "user", "content": prompt}],
"stream": False,
"think": True,
"options": {"temperature": 0.1, "num_predict": 500},
},
timeout=30,
)
content = resp.json().get("message", {}).get("content", "").strip().upper()
# Extraire le mot clé
for keyword in ("PASSER", "EXECUTER", "STOPPER"):
if keyword in content:
logger.info(f"Acteur gemma4 décide : {keyword}")
return keyword
logger.warning(f"Acteur gemma4 réponse inattendue : {content[:50]}")
return "EXECUTER"
except Exception as e:
logger.warning(f"Acteur gemma4 indisponible : {e}")
return "EXECUTER"
# =========================================================================
# Execution replay (polling serveur)
# =========================================================================
@@ -342,20 +406,28 @@ class ActionExecutorV1:
self.notifier.replay_target_not_found(target_desc)
return result
else:
# Cible invisible, pas de popup — PAUSE supervisée
# Cible invisible — demander à l'acteur (gemma4) de décider
target_desc = self._describe_target(target_spec)
decision = self._actor_decide(action, target_spec)
if decision == "PASSER":
print(f" [ACTEUR] Décision: PASSER — l'état est déjà atteint")
logger.info(f"Action {action_id} : acteur décide PASSER pour '{target_desc}'")
result["success"] = True
result["warning"] = "actor_skip"
elif decision == "STOPPER":
print(f" [ACTEUR] Décision: STOPPER — état incohérent")
logger.error(f"Action {action_id} : acteur décide STOPPER pour '{target_desc}'")
result["success"] = False
result["error"] = f"actor_stop:{target_desc}"
self.notifier.replay_target_not_found(target_desc)
else:
# EXECUTER ou décision inconnue → pause supervisée (fallback)
print(f" [ACTEUR] Décision: {decision} — pause supervisée")
logger.warning(f"Action {action_id} : acteur décide {decision}, pause")
result["success"] = False
result["error"] = "target_not_found"
result["target_description"] = target_desc
result["target_spec"] = target_spec
result["screenshot"] = self._capture_screenshot_b64()
result["warning"] = "visual_resolve_failed"
print(f" [ERREUR] Visual resolve échoué, pas de popup — PAUSE")
logger.error(
f"Action {action_id} : cible '{target_desc}' non trouvée, "
f"replay en pause supervisée"
)
# Notifier l'utilisateur via toast
self.notifier.replay_target_not_found(target_desc)
return result