Geniusia_v2/docs/implementation/WORKFLOW_DETECTION_IMPLEMENTATION.md

🔄 Implémentation de la Détection de Workflows

Date : 19 Novembre 2025
Objectif : Implémenter la détection de workflows complets au lieu de micro-patterns

✅ Ce qui a été fait

1. SessionManager (geniusia2/core/session_manager.py)

Rôle : Segmenter les actions en sessions basées sur le temps et le contexte.

Fonctionnalités :

• ✅ Détection automatique de début/fin de session
• ✅ Timeout configurable (5 minutes par défaut)
• ✅ Détection de changement de fenêtre
• ✅ Historique des sessions
• ✅ Callback on_session_completed
• ✅ Statistiques (durée moyenne, actions par session)

Classe Session :

@dataclass
class Session:
    session_id: str
    start_time: datetime
    end_time: Optional[datetime]
    actions: List[Dict[str, Any]]
    window: Optional[str]

Méthodes principales :

• add_action(action) : Ajoute une action à la session courante
• should_start_new_session(action) : Détermine si nouvelle session nécessaire
• finalize_current_session() : Finalise la session courante
• get_recent_sessions(n) : Retourne les N sessions récentes
• get_stats() : Statistiques des sessions

2. WorkflowDetector (geniusia2/core/workflow_detector.py)

Rôle : Détecter les workflows répétés en analysant les sessions.

Fonctionnalités :

• ✅ Analyse de sessions pour détecter des similarités
• ✅ Seuil de répétition configurable (3 par défaut)
• ✅ Seuil de similarité configurable (75% par défaut)
• ✅ Création de workflows à partir de sessions similaires
• ✅ Callback on_workflow_detected
• ✅ Mise à jour des workflows existants

Classe Workflow :

@dataclass
class Workflow:
    workflow_id: str
    name: str
    steps: List[WorkflowStep]
    repetitions: int
    confidence: float
    last_seen: datetime

Classe WorkflowStep :

@dataclass
class WorkflowStep:
    step_id: int
    action_type: str
    target_description: str
    position: tuple
    window: str
    embedding: Optional[np.ndarray]
    screenshot: Optional[np.ndarray]

Méthodes principales :

• analyze_sessions(sessions) : Analyse les sessions pour détecter des workflows
• _calculate_session_similarity(sessions) : Calcule la similarité entre sessions
• _compare_two_sessions(s1, s2) : Compare deux sessions
• _create_workflow_from_sessions(sessions) : Crée un workflow
• get_workflows() : Retourne tous les workflows détectés
• get_stats() : Statistiques des workflows

3. Intégration dans EventCapture

Modifications dans geniusia2/core/event_capture.py :

# Initialisation
self.session_manager = SessionManager(logger, self.config)
self.workflow_detector = WorkflowDetector(logger, self.config)

# Connexion des callbacks
self.session_manager.on_session_completed = self._on_session_completed
self.workflow_detector.on_workflow_detected = self._on_workflow_detected

Nouveaux callbacks :

• _on_session_completed(session) : Appelé quand une session est terminée
• _on_workflow_detected(workflow) : Appelé quand un workflow est détecté

Nouvelles méthodes publiques :

• capture_event(action) : Capture un événement manuellement (pour tests)
• get_workflows() : Retourne les workflows détectés
• get_sessions(count) : Retourne les sessions récentes
• get_workflow_stats() : Statistiques complètes
• force_finalize_session() : Force la finalisation de la session

🔄 Flux de Détection

1. Utilisateur effectue des actions
   ↓
2. EventCapture capture les événements
   ↓
3. SessionManager segmente en sessions
   ├─ Timeout (5 min) → Nouvelle session
   ├─ Changement fenêtre → Nouvelle session
   └─ Session terminée → Callback
   ↓
4. WorkflowDetector analyse les sessions
   ├─ Compare les 3+ dernières sessions
   ├─ Calcule la similarité (75% minimum)
   └─ Si similaires → Workflow détecté
   ↓
5. Workflow créé/mis à jour
   ├─ Génération d'un nom descriptif
   ├─ Création des étapes (WorkflowStep)
   └─ Callback → Notification

📊 Algorithme de Détection

Similarité entre Sessions

def _compare_two_sessions(session1, session2):
    # 1. Vérifier la longueur (tolérance de 2 actions)
    if abs(len(actions1) - len(actions2)) > 2:
        return 0.0
    
    # 2. Comparer action par action
    for i in range(min_len):
        # Type d'action identique → +1 point
        if action1.type == action2.type:
            matches += 1
            
            # Même fenêtre → +0.5 point bonus
            if action1.window == action2.window:
                matches += 0.5
    
    # 3. Score normalisé (0-1)
    return matches / (min_len * 1.5)

Détection de Workflow

def analyze_sessions(sessions):
    # 1. Prendre N sessions consécutives (N = min_repetitions)
    for i in range(len(sessions) - N + 1):
        session_group = sessions[i:i + N]
        
        # 2. Calculer similarité moyenne
        similarity = calculate_session_similarity(session_group)
        
        # 3. Si similarité >= 75% → Workflow !
        if similarity >= 0.75:
            workflow = create_workflow_from_sessions(session_group)
            
            # 4. Vérifier si workflow existe déjà
            existing = find_existing_workflow(workflow)
            
            if existing:
                # Mettre à jour
                existing.repetitions += 1
                existing.confidence += 0.05
            else:
                # Nouveau workflow
                workflows.append(workflow)
                on_workflow_detected(workflow)

🎯 Configuration

Dans config.py, ajouter :

"workflow": {
    # Timeout de session (secondes)
    "session_timeout": 300,  # 5 minutes
    
    # Nombre minimum de répétitions pour détecter un workflow
    "min_repetitions": 3,
    
    # Seuil de similarité pour considérer deux sessions comme identiques
    "similarity_threshold": 0.75  # 75%
}

🧪 Tests

Test Basique

from core.event_capture import EventCapture
from core.config import get_config

# Initialiser
config = get_config()
event_capture = EventCapture(
    logger=logger,
    pattern_threshold=3,
    config=config
)

# Simuler des actions répétées
for i in range(3):
    # Session 1
    event_capture.capture_event({
        "action_type": "click",
        "position": (100, 50),
        "window": "Desktop",
        "timestamp": datetime.now()
    })
    
    event_capture.capture_event({
        "action_type": "type",
        "text": "office",
        "window": "Desktop",
        "timestamp": datetime.now()
    })
    
    # Finaliser la session
    event_capture.force_finalize_session()

# Vérifier les workflows
workflows = event_capture.get_workflows()
print(f"Workflows détectés: {len(workflows)}")

Test Complet

Voir test_workflow_suggestion.py pour un test complet avec :

• Création de 3 sessions similaires
• Détection automatique de workflow
• Suggestion basée sur le début d'un workflow

📈 Avantages

Avant (Micro-Patterns)

• ❌ Détecte seulement les clics répétés
• ❌ Inutilisable en pratique
• ❌ Ne correspond pas à l'usage réel

Après (Macro-Workflows)

• ✅ Détecte les workflows complets
• ✅ Utilisable en pratique
• ✅ Correspond à l'usage réel
• ✅ Vrai RPA Vision !

🚀 Prochaines Étapes

Court Terme

1. ✅ Créer SessionManager Fait !
2. ✅ Créer WorkflowDetector Fait !
3. ✅ Intégrer dans EventCapture Fait !
4. 🔄 Tester avec workflow réel (OnlyOffice)
5. 🔄 Ajuster les seuils si nécessaire

Moyen Terme

6. Intégrer dans SuggestionManager
7. Afficher suggestions de workflow dans la GUI
8. Permettre l'exécution de workflows complets

Long Terme

9. Apprentissage des variations de workflows
10. Workflows conditionnels (if/else)
11. Workflows avec boucles

📝 Exemple Concret

Scénario : Ouvrir OnlyOffice

Jour 1 - 10h00 :

Session 1:
  1. Clic sur lanceur (100, 50)
  2. Type "office"
  3. Clic sur OnlyOffice (150, 100)
  4. Clic "Nouveau document" (200, 150)

Jour 2 - 14h30 :

Session 2:
  1. Clic sur lanceur (100, 50)
  2. Type "office"
  3. Clic sur OnlyOffice (150, 100)
  4. Clic "Nouveau document" (200, 150)

Jour 3 - 16h00 :

Session 3:
  1. Clic sur lanceur (100, 50)
  2. Type "office"
  3. Clic sur OnlyOffice (150, 100)
  4. Clic "Nouveau document" (200, 150)

Résultat :

🎉 Workflow détecté !
   Nom: "Click → Type → Click (+ 1 étapes)"
   Répétitions: 3x
   Confiance: 80%
   Étapes: 4

Jour 4 - Suggestion :

Utilisateur fait:
  1. Clic sur lanceur
  2. Type "office"

Système détecte:
  "Tu es au début du workflow 'Ouvrir OnlyOffice'"
  
Suggestion:
  "Je peux continuer ce workflow pour toi ?"
  [Oui] [Non] [Corriger]

✅ Checklist d'Implémentation

• Créer SessionManager
• Créer WorkflowDetector
• Ajouter imports dans EventCapture
• Connecter les callbacks
• Ajouter méthodes publiques
• Corriger les appels de méthodes
• Ajouter capture_event() pour tests
• Tester avec workflow réel
• Ajuster les seuils
• Intégrer dans SuggestionManager
• Ajouter GUI pour workflows

🎊 Conclusion

Le système de détection de workflows est maintenant implémenté !

Capacités :

• ✅ Segmentation en sessions
• ✅ Détection de workflows répétés
• ✅ Callbacks pour notifications
• ✅ Statistiques complètes

Prêt pour les tests ! 🚀

---

Implémenté le 19 Novembre 2025