Feat: Intégration système d'apprentissage VWB
- Création service learning_integration.py (pont VWB <-> LearningManager) - Enregistrement automatique des workflows à la création - Enregistrement des résultats d'exécution (succès/échec + confiance) - Endpoints API: /workflows/<id>/feedback et /workflows/<id>/learning - Boutons feedback (pouce vert/rouge) dans VWBExecutorExtension - Fix: VariableAutocomplete inputRef pour setSelectionRange - Amélioration: Chips cliquables pour insérer les variables Le système apprend maintenant des exécutions et feedbacks utilisateur. États: OBSERVATION -> COACHING -> AUTO_CANDIDATE -> AUTO_CONFIRMED Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Dom
227
visual_workflow_builder/backend/services/learning_integration.py
Normal file
227
visual_workflow_builder/backend/services/learning_integration.py
Normal file
@@ -0,0 +1,227 @@
|
||||
"""
|
||||
Learning Integration Service - Connecte VWB au système d'apprentissage
|
||||
Auteur: Dom, Claude - 14 janvier 2026
|
||||
|
||||
Ce service fait le pont entre le Visual Workflow Builder et le système
|
||||
d'apprentissage du core RPA Vision.
|
||||
"""
|
||||
|
||||
import logging
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Optional, Dict, Any
|
||||
|
||||
# Ajouter le chemin du core au path
|
||||
CORE_PATH = Path(__file__).parent.parent.parent.parent
|
||||
if str(CORE_PATH) not in sys.path:
|
||||
sys.path.insert(0, str(CORE_PATH))
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Singleton pour le LearningManager
|
||||
_learning_manager = None
|
||||
_initialized = False
|
||||
|
||||
|
||||
def _get_learning_manager():
|
||||
"""Obtient l'instance singleton du LearningManager."""
|
||||
global _learning_manager, _initialized
|
||||
|
||||
if _initialized:
|
||||
return _learning_manager
|
||||
|
||||
_initialized = True
|
||||
|
||||
try:
|
||||
from core.learning.learning_manager import LearningManager
|
||||
from core.models.workflow_graph import LearningState
|
||||
_learning_manager = LearningManager()
|
||||
logger.info("✓ LearningManager initialisé avec succès")
|
||||
except ImportError as e:
|
||||
logger.warning(f"⚠ Impossible d'importer LearningManager: {e}")
|
||||
_learning_manager = None
|
||||
except Exception as e:
|
||||
logger.error(f"✗ Erreur initialisation LearningManager: {e}")
|
||||
_learning_manager = None
|
||||
|
||||
return _learning_manager
|
||||
|
||||
|
||||
def _create_minimal_workflow(visual_workflow) -> Optional[Any]:
|
||||
"""
|
||||
Crée un objet Workflow minimal pour le LearningManager
|
||||
à partir d'un VisualWorkflow ou SimpleWorkflow.
|
||||
"""
|
||||
try:
|
||||
from core.models.workflow_graph import (
|
||||
Workflow,
|
||||
LearningState,
|
||||
WorkflowStats as CoreWorkflowStats,
|
||||
SafetyRules,
|
||||
LearningConfig
|
||||
)
|
||||
from datetime import datetime
|
||||
|
||||
# Récupérer les attributs (compatible VisualWorkflow et SimpleWorkflow)
|
||||
wf_id = getattr(visual_workflow, 'id', str(visual_workflow))
|
||||
wf_name = getattr(visual_workflow, 'name', 'Unknown')
|
||||
wf_desc = getattr(visual_workflow, 'description', '') or ''
|
||||
wf_created_by = getattr(visual_workflow, 'created_by', 'unknown')
|
||||
|
||||
# Créer un workflow minimal avec tous les paramètres requis
|
||||
workflow = Workflow(
|
||||
workflow_id=wf_id,
|
||||
name=wf_name,
|
||||
description=wf_desc,
|
||||
version=1,
|
||||
learning_state=LearningState.OBSERVATION,
|
||||
created_at=datetime.now(),
|
||||
updated_at=datetime.now(),
|
||||
entry_nodes=[],
|
||||
end_nodes=[],
|
||||
nodes=[],
|
||||
edges=[],
|
||||
safety_rules=SafetyRules(),
|
||||
stats=CoreWorkflowStats(),
|
||||
learning=LearningConfig(),
|
||||
metadata={'created_by': wf_created_by, 'source': 'VWB'}
|
||||
)
|
||||
|
||||
return workflow
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Erreur création workflow minimal: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return None
|
||||
|
||||
|
||||
def register_workflow_for_learning(visual_workflow) -> bool:
|
||||
"""
|
||||
Enregistre un workflow VWB dans le système d'apprentissage.
|
||||
|
||||
Args:
|
||||
visual_workflow: Instance de VisualWorkflow
|
||||
|
||||
Returns:
|
||||
True si enregistré avec succès, False sinon
|
||||
"""
|
||||
manager = _get_learning_manager()
|
||||
|
||||
if manager is None:
|
||||
logger.debug("LearningManager non disponible, enregistrement ignoré")
|
||||
return False
|
||||
|
||||
try:
|
||||
workflow = _create_minimal_workflow(visual_workflow)
|
||||
if workflow is None:
|
||||
return False
|
||||
|
||||
manager.register_workflow(workflow)
|
||||
logger.info(f"✓ Workflow '{visual_workflow.name}' (ID: {visual_workflow.id}) enregistré pour apprentissage")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"✗ Erreur enregistrement workflow: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def record_workflow_execution(workflow_id: str, success: bool, confidence: float = 0.8) -> bool:
|
||||
"""
|
||||
Enregistre une exécution de workflow pour l'apprentissage.
|
||||
|
||||
Args:
|
||||
workflow_id: ID du workflow
|
||||
success: True si l'exécution a réussi
|
||||
confidence: Score de confiance (0.0 à 1.0)
|
||||
|
||||
Returns:
|
||||
True si enregistré avec succès, False sinon
|
||||
"""
|
||||
manager = _get_learning_manager()
|
||||
|
||||
if manager is None:
|
||||
logger.debug("LearningManager non disponible, exécution non enregistrée")
|
||||
return False
|
||||
|
||||
try:
|
||||
manager.record_execution(workflow_id, success, confidence)
|
||||
logger.info(f"✓ Exécution enregistrée: workflow={workflow_id}, success={success}, confidence={confidence:.2f}")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"✗ Erreur enregistrement exécution: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def get_workflow_learning_state(workflow_id: str) -> Optional[str]:
|
||||
"""
|
||||
Obtient l'état d'apprentissage d'un workflow.
|
||||
|
||||
Returns:
|
||||
Nom de l'état (OBSERVATION, COACHING, AUTO_CANDIDATE, AUTO_CONFIRMED)
|
||||
ou None si non trouvé
|
||||
"""
|
||||
manager = _get_learning_manager()
|
||||
|
||||
if manager is None:
|
||||
return None
|
||||
|
||||
try:
|
||||
state = manager.get_workflow_state(workflow_id)
|
||||
if state:
|
||||
return state.value
|
||||
return None
|
||||
except Exception as e:
|
||||
logger.error(f"Erreur récupération état: {e}")
|
||||
return None
|
||||
|
||||
|
||||
def get_workflow_stats(workflow_id: str) -> Optional[Dict[str, Any]]:
|
||||
"""
|
||||
Obtient les statistiques d'apprentissage d'un workflow.
|
||||
|
||||
Returns:
|
||||
Dict avec les stats ou None si non trouvé
|
||||
"""
|
||||
manager = _get_learning_manager()
|
||||
|
||||
if manager is None:
|
||||
return None
|
||||
|
||||
try:
|
||||
stats = manager.get_workflow_stats(workflow_id)
|
||||
if stats:
|
||||
return {
|
||||
"workflow_id": stats.workflow_id,
|
||||
"learning_state": stats.learning_state.value,
|
||||
"observation_count": stats.observation_count,
|
||||
"execution_count": stats.execution_count,
|
||||
"success_count": stats.success_count,
|
||||
"failure_count": stats.failure_count,
|
||||
"success_rate": stats.success_rate,
|
||||
"avg_confidence": stats.avg_confidence,
|
||||
"last_execution": stats.last_execution.isoformat() if stats.last_execution else None
|
||||
}
|
||||
return None
|
||||
except Exception as e:
|
||||
logger.error(f"Erreur récupération stats: {e}")
|
||||
return None
|
||||
|
||||
|
||||
def should_auto_execute(workflow_id: str) -> bool:
|
||||
"""
|
||||
Vérifie si un workflow peut s'exécuter automatiquement.
|
||||
|
||||
Returns:
|
||||
True si le workflow est en AUTO_CANDIDATE ou AUTO_CONFIRMED
|
||||
"""
|
||||
manager = _get_learning_manager()
|
||||
|
||||
if manager is None:
|
||||
return False
|
||||
|
||||
try:
|
||||
return manager.should_execute_automatically(workflow_id)
|
||||
except Exception:
|
||||
return False
|
||||
Reference in New Issue
Block a user