rpa_vision_v3/core/graph

Graph Module - Construction de Workflow Graphs

Ce module implémente la construction automatique de graphes de workflows depuis des sessions enregistrées.

Architecture

graph/
├── __init__.py
├── graph_builder.py    # Construction de workflows depuis sessions
├── node_matcher.py     # Matching de ScreenStates contre nodes
└── README.md          # Ce fichier

GraphBuilder

Responsabilités

Le GraphBuilder analyse une RawSession pour construire automatiquement un Workflow complet :

1. Création de ScreenStates - Convertit les screenshots en états structurés
2. Calcul d'Embeddings - Génère des embeddings multi-modaux pour chaque état
3. Détection de Patterns - Utilise DBSCAN pour identifier les patterns répétés
4. Construction de Nodes - Crée des WorkflowNodes depuis les clusters
5. Construction d'Edges - Détecte les transitions entre états (TODO)

Algorithme de Détection de Patterns

Utilise DBSCAN (Density-Based Spatial Clustering of Applications with Noise) :

• Métrique : Similarité cosinus entre embeddings
• Paramètres :
  • eps : Distance maximum entre points (défaut: 0.15)
  • min_samples : Échantillons minimum par cluster (défaut: 2)
  • min_pattern_repetitions : Répétitions minimum pour un pattern (défaut: 3)

Avantages de DBSCAN :

• Détecte automatiquement le nombre de clusters
• Identifie le bruit (états uniques)
• Fonctionne bien avec des clusters de formes arbitraires

Exemple d'Utilisation

from core.graph.graph_builder import GraphBuilder
from core.models.raw_session import RawSession

# Créer le builder
builder = GraphBuilder(
    min_pattern_repetitions=3,
    clustering_eps=0.15
)

# Construire workflow depuis session
workflow = builder.build_from_session(
    session=raw_session,
    workflow_name="Login Workflow"
)

print(f"Workflow: {len(workflow.nodes)} nodes, {len(workflow.edges)} edges")

Configuration

GraphBuilder(
    embedding_builder=None,           # StateEmbeddingBuilder personnalisé
    faiss_manager=None,               # FAISSManager pour indexation
    min_pattern_repetitions=3,        # Répétitions min pour un pattern
    clustering_eps=0.15,              # Distance max DBSCAN
    clustering_min_samples=2          # Échantillons min par cluster
)

Méthodes Publiques

build_from_session(session, workflow_name=None) -> Workflow

Construit un workflow complet depuis une RawSession.

Args:

• session : RawSession à analyser
• workflow_name : Nom du workflow (optionnel)

Returns:

• Workflow avec nodes et edges

Raises:

• ValueError si la session est vide

Méthodes Privées

_create_screen_states(session) -> List[ScreenState]

Crée des ScreenStates depuis les screenshots.

Note: Pour l'instant, crée des états basiques. TODO: Enrichir avec détection UI.

_compute_embeddings(screen_states) -> List[np.ndarray]

Calcule les embeddings pour tous les états.

Utilise StateEmbeddingBuilder pour générer des embeddings multi-modaux (image + texte + UI).

_detect_patterns(embeddings, screen_states) -> Dict[int, List[int]]

Détecte les patterns répétés via clustering DBSCAN.

Returns: Dictionnaire {cluster_id: [indices des états]}

_build_nodes(clusters, screen_states, embeddings) -> List[WorkflowNode]

Construit des WorkflowNodes depuis les clusters.

Pour chaque cluster :

1. Calcule l'embedding prototype (moyenne normalisée)
2. Extrait les contraintes
3. Crée un ScreenTemplate
4. Crée un WorkflowNode

_create_screen_template(states, prototype_embedding) -> ScreenTemplate

Crée un ScreenTemplate depuis un cluster d'états.

TODO: Extraire intelligemment :

• window_title_pattern (regex depuis titres communs)
• required_text_patterns (texte présent dans tous les états)
• required_ui_elements (éléments UI communs)

_build_edges(nodes, screen_states, session) -> List[WorkflowEdge]

Construit des WorkflowEdges depuis les transitions.

TODO: Implémenter détection de transitions :

1. Identifier séquences d'états (state_i → state_j)
2. Extraire actions depuis événements RawSession
3. Mapper états vers nodes
4. Créer edges avec TargetSpec et conditions

NodeMatcher

Responsabilités

Le NodeMatcher trouve le WorkflowNode qui correspond le mieux à un ScreenState actuel.

Stratégies de Matching

1. Recherche FAISS (si disponible) : Recherche rapide dans l'index
2. Recherche Linéaire (fallback) : Compare avec tous les candidats
3. Validation de Contraintes : Vérifie titre fenêtre, texte requis, UI requis

Exemple d'Utilisation

from core.graph.node_matcher import NodeMatcher

# Créer le matcher
matcher = NodeMatcher(similarity_threshold=0.85)

# Matcher un état contre des nodes candidats
result = matcher.match(current_state, candidate_nodes)

if result:
    node, confidence = result
    print(f"Matched {node.node_id} with confidence {confidence:.2f}")
else:
    print("No match found")

Configuration

NodeMatcher(
    embedding_builder=None,      # StateEmbeddingBuilder personnalisé
    faiss_manager=None,          # FAISSManager pour recherche rapide
    similarity_threshold=0.85    # Seuil de similarité minimum
)

Méthodes Publiques

match(current_state, candidate_nodes) -> Optional[Tuple[WorkflowNode, float]]

Trouve le node qui matche le mieux l'état actuel.

Returns: (node, confidence) si match trouvé, None sinon

validate_constraints(state, node) -> bool

Valide les contraintes du node contre l'état.

Returns: True si toutes les contraintes sont satisfaites

Tests

Tests Unitaires

# Lancer les tests
python -m pytest tests/unit/test_graph_builder.py -v
python -m pytest tests/unit/test_node_matcher.py -v

Test d'Intégration

# Test rapide
python test_phase_a_b.py

Qualité du Code

• ✅ Type Hints : Toutes les fonctions sont typées
• ✅ Docstrings : Documentation complète (Google style)
• ✅ Logging : Logs informatifs à tous les niveaux
• ✅ Error Handling : Validation des entrées
• ✅ No Diagnostics : Aucune erreur de linting/typing

Prochaines Étapes

Priorité Haute

1. Implémenter _build_edges()

   • Détecter transitions entre états
   • Extraire actions depuis événements
   • Créer TargetSpec avec rôles sémantiques

2. Enrichir _create_screen_template()

   • Extraire window_title_pattern
   • Extraire required_text_patterns
   • Extraire required_ui_elements

3. Tests Property-Based

   • Property 14: Embedding Prototype Sample Count
   • Property 16: Pattern Detection Minimum Repetitions

Priorité Moyenne

4. Optimisations

   • Batch processing pour embeddings
   • Cache pour prototypes
   • Parallélisation du clustering

5. Robustesse

   • Gestion des sessions très longues
   • Gestion des états sans patterns
   • Métriques de qualité des clusters

Références

• DBSCAN : Scikit-learn Documentation
• Workflow Graphs : Voir core/models/workflow_graph.py
• State Embeddings : Voir core/embedding/state_embedding_builder.py