rpa_vision_v3/docs/specs/design.md

Document de Design - Workflow Graph Implementation

Vue d'Ensemble

Ce document décrit le design détaillé de l'implémentation de l'architecture Workflow Graph pour RPA Vision V2. Le système transforme des captures d'écran brutes en workflows sémantiques appris à travers 5 couches d'abstraction progressive.

Philosophie : "Observer → Comprendre → Apprendre → Agir"

Le système ne travaille PAS avec des coordonnées de clics, mais avec une compréhension sémantique des interfaces : types d'éléments, rôles, contexte visuel et textuel.

Architecture en 5 Couches :

RawSession (Couche 0) → ScreenState (Couche 1) → UIElement Detection (Couche 2) 
→ State Embedding (Couche 3) → Workflow Graph (Couche 4)

Référence : Ce design s'appuie sur docs/reference/ARCHITECTURE_VISION_COMPLETE.md et docs/reference/ARCHITECTURE_ENRICHISSEMENTS.md.

Architecture

Architecture Globale

┌─────────────────────────────────────────────────────────────┐
│                    Couche 4 : Workflow Graph                │
│  WorkflowNode + WorkflowEdge + Learning States              │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────────┐  │
│  │ Graph Builder│  │ Node Matcher │  │ Action Executor  │  │
│  └──────────────┘  └──────────────┘  └──────────────────┘  │
└─────────────────────────────────────────────────────────────┘
                            ↕
┌─────────────────────────────────────────────────────────────┐
│                 Couche 3 : State Embedding                  │
│  Fusion Multi-Modale (Image + Text + UI + Context)         │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────────┐  │
│  │ Fusion Engine│  │ FAISS Index  │  │ Similarity Comp  │  │
│  └──────────────┘  └──────────────┘  └──────────────────┘  │
└─────────────────────────────────────────────────────────────┘
                            ↕
┌─────────────────────────────────────────────────────────────┐
│              Couche 2 : UIElement Detection                 │
│  Détection Sémantique (Type + Rôle + Embeddings)           │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────────┐  │
│  │ VLM Detector │  │ Classifier   │  │ Embedding Gen    │  │
│  └──────────────┘  └──────────────┘  └──────────────────┘  │
└─────────────────────────────────────────────────────────────┘
                            ↕
┌─────────────────────────────────────────────────────────────┐
│                 Couche 1 : ScreenState                      │
│  Analyse Multi-Modale (4 Niveaux)                          │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────────┐  │
│  │ Raw Capture  │  │ Perception   │  │ Semantic + Ctx   │  │
│  └──────────────┘  └──────────────┘  └──────────────────┘  │
└─────────────────────────────────────────────────────────────┘
                            ↕
┌─────────────────────────────────────────────────────────────┐
│                  Couche 0 : RawSession                      │
│  Capture Brute (Events + Screenshots)                      │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────────┐  │
│  │ Event Logger │  │ Screenshot   │  │ Session Manager  │  │
│  └──────────────┘  └──────────────┘  └──────────────────┘  │
└─────────────────────────────────────────────────────────────┘

Structure des Répertoires

geniusia2/
├── core/
│   ├── models/
│   │   ├── raw_session.py          # Couche 0 : RawSession
│   │   ├── screen_state.py         # Couche 1 : ScreenState
│   │   ├── ui_element.py           # Couche 2 : UIElement
│   │   ├── state_embedding.py      # Couche 3 : State Embedding
│   │   └── workflow_graph.py       # Couche 4 : Workflow Graph
│   ├── capture/
│   │   ├── event_capture.py        # Capture événements
│   │   └── screenshot_capture.py   # Capture screenshots
│   ├── detection/
│   │   ├── ui_detector.py          # Détection UI avec VLM
│   │   └── text_detector.py        # Détection texte
│   ├── embedding/
│   │   ├── fusion_engine.py        # Fusion multi-modale
│   │   ├── faiss_manager.py        # Gestion index FAISS
│   │   └── similarity.py           # Calculs de similarité
│   ├── graph/
│   │   ├── graph_builder.py        # Construction graphes
│   │   ├── node_matcher.py         # Matching nodes
│   │   ├── action_executor.py      # Exécution actions
│   │   └── learning_manager.py     # Gestion états apprentissage
│   └── persistence/
│       ├── json_serializer.py      # Sérialisation JSON
│       └── storage_manager.py      # Gestion stockage
└── data/
    ├── sessions/                   # RawSessions
    ├── screen_states/              # ScreenStates
    ├── embeddings/                 # Embeddings (.npy)
    ├── faiss_index/                # Index FAISS
    └── workflows/                  # Workflow Graphs

Composants et Interfaces

Couche 0 : RawSession

Responsabilité : Capture fidèle de tous les événements utilisateur avec screenshots.

Classe Principale : RawSession

@dataclass
class RawSession:
    schema_version: str = "rawsession_v1"
    session_id: str
    agent_version: str
    environment: Dict[str, Any]
    user: Dict[str, str]
    context: Dict[str, str]
    started_at: datetime
    ended_at: Optional[datetime]
    events: List[Event]
    screenshots: List[Screenshot]
    
    def add_event(self, event: Event) -> None:
        """Ajouter un événement à la session"""
        
    def add_screenshot(self, screenshot: Screenshot) -> None:
        """Ajouter un screenshot à la session"""
        
    def to_json(self) -> Dict[str, Any]:
        """Sérialiser en JSON"""
        
    @classmethod
    def from_json(cls, data: Dict[str, Any]) -> 'RawSession':
        """Désérialiser depuis JSON"""

Classe Event :

@dataclass
class Event:
    t: float  # Timestamp relatif en secondes
    type: str  # mouse_click, key_press, etc.
    window: WindowContext
    screenshot_id: Optional[str]
    # Champs spécifiques selon type
    data: Dict[str, Any]

Couche 1 : ScreenState

Responsabilité : Représentation structurée d'un écran à 4 niveaux d'abstraction.

Classe Principale : ScreenState

@dataclass
class ScreenState:
    screen_state_id: str
    timestamp: datetime
    session_id: str
    window: WindowContext
    
    # Niveau 1 : Raw
    raw: RawLevel
    
    # Niveau 2 : Perception
    perception: PerceptionLevel
    
    # Niveau 3 : Sémantique UI
    ui_elements: List[UIElement]
    
    # Niveau 4 : Contexte Métier
    context: ContextLevel
    
    metadata: Dict[str, Any]
    
    def to_json(self) -> Dict[str, Any]:
        """Sérialiser en JSON"""
        
    @classmethod
    def from_json(cls, data: Dict[str, Any]) -> 'ScreenState':
        """Désérialiser depuis JSON"""

Niveaux :

@dataclass
class RawLevel:
    screenshot_path: str
    capture_method: str
    file_size_bytes: int

@dataclass
class PerceptionLevel:
    embedding: EmbeddingRef
    detected_text: List[str]
    text_detection_method: str
    confidence_avg: float

@dataclass
class ContextLevel:
    current_workflow_candidate: Optional[str]
    workflow_step: Optional[int]
    user_id: str
    tags: List[str]
    business_variables: Dict[str, Any]

Couche 2 : UIElement Detection

Responsabilité : Détection sémantique des éléments UI avec types, rôles et embeddings.

Classe Principale : UIElement

@dataclass
class UIElement:
    element_id: str
    type: str  # button, text_input, checkbox, etc.
    role: str  # primary_action, cancel, form_input, etc.
    bbox: Tuple[int, int, int, int]
    center: Tuple[int, int]
    
    label: str
    label_confidence: float
    
    embeddings: UIElementEmbeddings
    visual_features: VisualFeatures
    
    tags: List[str]
    confidence: float
    
    metadata: Dict[str, Any]
    
    def to_json(self) -> Dict[str, Any]:
        """Sérialiser en JSON"""

Embeddings Duaux :

@dataclass
class UIElementEmbeddings:
    image: EmbeddingRef  # Embedding de l'image croppée
    text: EmbeddingRef   # Embedding du texte détecté

Détecteur UI :

class UIDetector:
    def __init__(self, vlm_model: str, clip_model: str):
        self.vlm = VLMClient(vlm_model)
        self.clip = CLIPEmbedder(clip_model)
    
    def detect_elements(self, screenshot: np.ndarray, 
                       window_context: WindowContext) -> List[UIElement]:
        """Détecter tous les éléments UI dans un screenshot"""
        # 1. Proposer régions d'intérêt avec VLM
        # 2. Caractériser chaque élément (crop, OCR, embeddings)
        # 3. Classifier type et rôle
        # 4. Retourner liste d'UIElements

Couche 3 : State Embedding

Responsabilité : Fusion multi-modale en vecteur unique (fingerprint d'écran).

Classe Principale : StateEmbedding

@dataclass
class StateEmbedding:
    embedding_id: str
    vector_id: str  # Chemin vers .npy
    dimensions: int
    fusion_method: str  # "weighted" ou "concat_projection"
    
    components: Dict[str, EmbeddingComponent]
    metadata: Dict[str, Any]
    
    def get_vector(self) -> np.ndarray:
        """Charger le vecteur depuis le fichier"""
        
    def compute_similarity(self, other: 'StateEmbedding') -> float:
        """Calculer similarité cosinus avec autre embedding"""

Moteur de Fusion :

class FusionEngine:
    def __init__(self, method: str = "weighted", 
                 weights: Optional[Dict[str, float]] = None):
        self.method = method
        self.weights = weights or {
            "image": 0.5,
            "text": 0.3,
            "title": 0.1,
            "ui": 0.1
        }
    
    def fuse(self, 
             img_emb: np.ndarray,
             text_emb: np.ndarray,
             title_emb: np.ndarray,
             ui_emb: np.ndarray) -> np.ndarray:
        """Fusionner tous les embeddings en un seul vecteur"""
        if self.method == "weighted":
            return self._weighted_fusion(img_emb, text_emb, title_emb, ui_emb)
        elif self.method == "concat_projection":
            return self._concat_projection(img_emb, text_emb, title_emb, ui_emb)
    
    def _weighted_fusion(self, img_emb, text_emb, title_emb, ui_emb) -> np.ndarray:
        """Fusion pondérée simple"""
        fused = (
            self.weights["image"] * normalize(img_emb) +
            self.weights["text"] * normalize(text_emb) +
            self.weights["title"] * normalize(title_emb) +
            self.weights["ui"] * normalize(ui_emb)
        )
        return normalize(fused)

Gestionnaire FAISS :

class FAISSManager:
    def __init__(self, index_path: str, dimension: int = 512):
        self.index_path = index_path
        self.dimension = dimension
        self.index = self._load_or_create_index()
        self.metadata_store: Dict[int, Dict[str, Any]] = {}
    
    def add_embedding(self, embedding: np.ndarray, 
                     metadata: Dict[str, Any]) -> int:
        """Ajouter un embedding à l'index"""
        idx = self.index.ntotal
        self.index.add(embedding.reshape(1, -1))
        self.metadata_store[idx] = metadata
        return idx
    
    def search_similar(self, query: np.ndarray, 
                      k: int = 5) -> List[SearchResult]:
        """Chercher les k plus proches voisins"""
        distances, indices = self.index.search(query.reshape(1, -1), k)
        results = []
        for dist, idx in zip(distances[0], indices[0]):
            results.append(SearchResult(
                id=int(idx),
                distance=float(dist),
                similarity=1.0 - float(dist),  # Cosine similarity
                metadata=self.metadata_store.get(int(idx), {})
            ))
        return results

Couche 4 : Workflow Graph

Responsabilité : Modélisation des workflows en graphes avec apprentissage progressif.

Classe WorkflowNode :

@dataclass
class WorkflowNode:
    node_id: str
    label: str
    description: str
    screen_template: ScreenTemplate
    metadata: Dict[str, Any]
    
    def matches(self, screen_state: ScreenState, 
                state_embedding: StateEmbedding) -> Tuple[bool, float]:
        """Vérifier si un ScreenState correspond à ce node"""
        # 1. Vérifier contraintes fenêtre
        # 2. Vérifier texte requis
        # 3. Vérifier éléments UI requis
        # 4. Vérifier similarité embedding
        # Retourner (match, confidence)

@dataclass
class ScreenTemplate:
    window: WindowConstraints
    required_text_any: List[str]
    required_ui_elements: List[UIElementConstraint]
    embedding_prototype: EmbeddingPrototype
    optional_elements: List[UIElementConstraint]

@dataclass
class EmbeddingPrototype:
    provider: str
    vector_id: str
    min_cosine_similarity: float
    sample_count: int

Classe WorkflowEdge :

@dataclass
class WorkflowEdge:
    edge_id: str
    from_node: str
    to_node: str
    action: Action
    constraints: EdgeConstraints
    post_conditions: PostConditions
    stats: EdgeStats
    metadata: Dict[str, Any]
    
    def can_execute(self, current_state: ScreenState) -> Tuple[bool, str]:
        """Vérifier si l'edge peut être exécuté"""
        # Vérifier pre-conditions
        
    def execute(self, executor: ActionExecutor) -> ExecutionResult:
        """Exécuter l'action de cet edge"""

@dataclass
class Action:
    type: str  # mouse_click, key_press, text_input, compound
    target: TargetSpec
    parameters: Dict[str, Any]

@dataclass
class TargetSpec:
    role: str  # Rôle sémantique de l'élément cible
    selection_policy: str  # first, last, by_similarity
    fallback_strategy: str  # visual_similarity, position

Classe Workflow :

@dataclass
class Workflow:
    workflow_id: str
    name: str
    description: str
    version: int
    
    learning_state: str  # OBSERVATION, COACHING, AUTO_CANDIDATE, AUTO_CONFIRMÉ
    
    created_at: datetime
    updated_at: datetime
    
    entry_nodes: List[str]
    end_nodes: List[str]
    
    nodes: List[WorkflowNode]
    edges: List[WorkflowEdge]
    
    safety_rules: SafetyRules
    stats: WorkflowStats
    learning: LearningConfig
    metadata: Dict[str, Any]
    
    def get_node(self, node_id: str) -> Optional[WorkflowNode]:
        """Récupérer un node par ID"""
        
    def get_outgoing_edges(self, node_id: str) -> List[WorkflowEdge]:
        """Récupérer tous les edges sortants d'un node"""
        
    def to_json(self) -> Dict[str, Any]:
        """Sérialiser en JSON"""

Graph Builder

Responsabilité : Construire automatiquement des Workflow Graphs depuis des RawSessions.

class GraphBuilder:
    def __init__(self, 
                 faiss_manager: FAISSManager,
                 fusion_engine: FusionEngine,
                 ui_detector: UIDetector):
        self.faiss = faiss_manager
        self.fusion = fusion_engine
        self.ui_detector = ui_detector
    
    def build_from_session(self, session: RawSession) -> Optional[Workflow]:
        """Construire un workflow depuis une session"""
        # 1. Créer ScreenStates pour tous les screenshots
        screen_states = self._create_screen_states(session)
        
        # 2. Calculer State Embeddings
        embeddings = self._compute_embeddings(screen_states)
        
        # 3. Détecter patterns répétés
        patterns = self._detect_patterns(screen_states, embeddings)
        
        # 4. Construire nodes et edges
        if patterns:
            return self._build_workflow(patterns)
        return None
    
    def _detect_patterns(self, 
                        screen_states: List[ScreenState],
                        embeddings: List[StateEmbedding]) -> List[Pattern]:
        """Détecter séquences répétées"""
        # Utiliser clustering sur embeddings
        # Identifier transitions récurrentes
        # Retourner patterns détectés

Node Matcher

Responsabilité : Matcher un ScreenState actuel contre les WorkflowNodes existants.

class NodeMatcher:
    def __init__(self, 
                 faiss_manager: FAISSManager,
                 fusion_engine: FusionEngine):
        self.faiss = faiss_manager
        self.fusion = fusion_engine
    
    def match(self, 
             screen_state: ScreenState,
             workflow: Workflow) -> Optional[NodeMatch]:
        """Trouver le node correspondant au ScreenState actuel"""
        # 1. Calculer State Embedding
        state_emb = self._compute_state_embedding(screen_state)
        
        # 2. Chercher dans FAISS les prototypes similaires
        candidates = self.faiss.search_similar(state_emb.get_vector(), k=5)
        
        # 3. Filtrer par workflow_id
        candidates = [c for c in candidates 
                     if c.metadata.get('workflow_id') == workflow.workflow_id]
        
        # 4. Valider contraintes pour chaque candidat
        for candidate in candidates:
            node = workflow.get_node(candidate.metadata['node_id'])
            if node:
                matches, confidence = node.matches(screen_state, state_emb)
                if matches:
                    return NodeMatch(
                        node=node,
                        confidence=confidence,
                        embedding_similarity=candidate.similarity
                    )
        
        return None

@dataclass
class NodeMatch:
    node: WorkflowNode
    confidence: float
    embedding_similarity: float

Action Executor

Responsabilité : Exécuter les actions définies dans WorkflowEdges.

class ActionExecutor:
    def __init__(self, input_controller):
        self.input = input_controller
    
    def execute_edge(self, 
                    edge: WorkflowEdge,
                    current_state: ScreenState) -> ExecutionResult:
        """Exécuter l'action d'un edge"""
        # 1. Vérifier pre-conditions
        can_execute, reason = edge.can_execute(current_state)
        if not can_execute:
            return ExecutionResult(success=False, reason=reason)
        
        # 2. Trouver élément cible par rôle
        target_element = self._find_target_element(
            edge.action.target, 
            current_state
        )
        if not target_element:
            return ExecutionResult(success=False, 
                                 reason="Target element not found")
        
        # 3. Exécuter action
        if edge.action.type == "mouse_click":
            self._execute_click(target_element, edge.action.parameters)
        elif edge.action.type == "text_input":
            self._execute_text_input(target_element, edge.action.parameters)
        elif edge.action.type == "compound":
            self._execute_compound(edge.action, current_state)
        
        # 4. Attendre post-conditions
        success = self._wait_for_postconditions(edge.post_conditions)
        
        return ExecutionResult(success=success)
    
    def _find_target_element(self, 
                            target: TargetSpec,
                            state: ScreenState) -> Optional[UIElement]:
        """Trouver élément UI par rôle sémantique"""
        candidates = [el for el in state.ui_elements 
                     if el.role == target.role]
        
        if not candidates:
            return None
        
        if target.selection_policy == "first":
            return candidates[0]
        elif target.selection_policy == "last":
            return candidates[-1]
        elif target.selection_policy == "by_similarity":
            # Utiliser embedding similarity
            return self._select_by_similarity(candidates, target)
        
        return candidates[0]

@dataclass
class ExecutionResult:
    success: bool
    reason: Optional[str] = None
    execution_time_ms: Optional[float] = None

Learning Manager

Responsabilité : Gérer les états d'apprentissage et transitions.

class LearningManager:
    def __init__(self):
        self.workflows: Dict[str, Workflow] = {}
    
    def update_workflow_stats(self, 
                             workflow_id: str,
                             execution_result: ExecutionResult) -> None:
        """Mettre à jour statistiques après exécution"""
        workflow = self.workflows[workflow_id]
        workflow.stats.total_executions += 1
        
        if execution_result.success:
            workflow.stats.success_count += 1
        else:
            workflow.stats.failure_count += 1
        
        # Vérifier si transition d'état nécessaire
        self._check_state_transition(workflow)
    
    def _check_state_transition(self, workflow: Workflow) -> None:
        """Vérifier et effectuer transitions d'état si nécessaire"""
        current_state = workflow.learning_state
        
        if current_state == "OBSERVATION":
            if self._can_transition_to_coaching(workflow):
                self._transition_to(workflow, "COACHING")
        
        elif current_state == "COACHING":
            if self._can_transition_to_auto_candidate(workflow):
                self._transition_to(workflow, "AUTO_CANDIDATE")
        
        elif current_state == "AUTO_CANDIDATE":
            if self._can_transition_to_auto_confirmed(workflow):
                # Nécessite validation utilisateur
                self._request_user_approval(workflow)
        
        elif current_state == "AUTO_CONFIRMÉ":
            if self._should_rollback(workflow):
                self._transition_to(workflow, "COACHING")
    
    def _can_transition_to_coaching(self, workflow: Workflow) -> bool:
        """Vérifier critères pour OBSERVATION → COACHING"""
        return (workflow.stats.observed_runs >= 5 and
                workflow.stats.avg_similarity >= 0.90)
    
    def _can_transition_to_auto_candidate(self, workflow: Workflow) -> bool:
        """Vérifier critères pour COACHING → AUTO_CANDIDATE"""
        return (workflow.stats.assist_runs >= 10 and
                workflow.stats.success_rate >= 0.90)
    
    def _can_transition_to_auto_confirmed(self, workflow: Workflow) -> bool:
        """Vérifier critères pour AUTO_CANDIDATE → AUTO_CONFIRMÉ"""
        return (workflow.stats.auto_candidate_runs >= 20 and
                workflow.stats.success_rate >= 0.95)
    
    def _should_rollback(self, workflow: Workflow) -> bool:
        """Vérifier si rollback nécessaire"""
        return workflow.stats.recent_confidence < 0.90

Modèles de Données

Format JSON Complet

RawSession :

{
  "schema_version": "rawsession_v1",
  "session_id": "sess_2025-11-22T10-15-00_user1",
  "agent_version": "0.2.0",
  "environment": {
    "platform": "linux",
    "hostname": "dev-machine",
    "screen": {"primary_resolution": [1920, 1080]}
  },
  "user": {"id": "user1", "label": "Developer User"},
  "context": {"customer": "Demo", "training_label": "workflow_test"},
  "started_at": "2025-11-22T10:15:00Z",
  "ended_at": "2025-11-22T10:30:00Z",
  "events": [
    {
      "t": 0.523,
      "type": "mouse_click",
      "button": "left",
      "pos": [800, 400],
      "window": {"title": "App", "app_name": "app.exe"},
      "screenshot_id": "shot_0001"
    }
  ],
  "screenshots": [
    {
      "screenshot_id": "shot_0001",
      "relative_path": "shots/shot_0001.png",
      "captured_at": "2025-11-22T10:15:00.523Z"
    }
  ]
}

ScreenState :

{
  "screen_state_id": "screen_2025-11-22T10-15-32.123Z",
  "timestamp": "2025-11-22T10:15:32.123Z",
  "session_id": "sess_2025-11-22T10-15-00_user1",
  "window": {
    "app_name": "app",
    "window_title": "Main Window",
    "screen_resolution": [1920, 1080]
  },
  "raw": {
    "screenshot_path": "data/screens/2025-11-22/10-15-32.png",
    "capture_method": "mss",
    "file_size_bytes": 245678
  },
  "perception": {
    "embedding": {
      "provider": "openclip_ViT-B-32",
      "vector_id": "data/embeddings/screens/screen_2025-11-22T10-15-32.123Z.npy",
      "dimensions": 512
    },
    "detected_text": ["Button", "Input", "Submit"],
    "text_detection_method": "qwen_vl",
    "confidence_avg": 0.92
  },
  "ui_elements": [
    {
      "element_id": "el_btn_001",
      "type": "button",
      "role": "primary_action",
      "bbox": [100, 200, 200, 240],
      "center": [150, 220],
      "label": "Submit",
      "label_confidence": 0.96,
      "embeddings": {
        "image": {
          "provider": "openclip_ViT-B-32",
          "vector_id": "data/embeddings/elements/el_btn_001_img.npy",
          "dimensions": 512
        },
        "text": {
          "provider": "openclip_ViT-B-32",
          "vector_id": "data/embeddings/elements/el_btn_001_txt.npy",
          "dimensions": 512
        }
      },
      "visual_features": {
        "dominant_color": "#4CAF50",
        "has_icon": false,
        "shape": "rectangle",
        "size_category": "medium"
      },
      "tags": ["action", "primary"],
      "confidence": 0.94
    }
  ],
  "context": {
    "current_workflow_candidate": null,
    "workflow_step": null,
    "user_id": "user1",
    "tags": ["demo"],
    "business_variables": {}
  },
  "metadata": {
    "processing_time_ms": 245,
    "ui_elements_count": 5
  }
}

Workflow Graph :

{
  "workflow_id": "WF_demo_workflow",
  "name": "Demo Workflow",
  "description": "Simple demo workflow for testing",
  "version": 1,
  "learning_state": "OBSERVATION",
  "created_at": "2025-11-22T10:45:00Z",
  "updated_at": "2025-11-22T10:45:00Z",
  "entry_nodes": ["N1_start"],
  "end_nodes": ["N3_end"],
  "nodes": [
    {
      "node_id": "N1_start",
      "label": "Start Screen",
      "description": "Initial screen with form",
      "screen_template": {
        "window": {
          "app_name_any_of": ["app"],
          "title_contains_any_of": ["Main"]
        },
        "required_text_any": ["Submit", "Input"],
        "required_ui_elements": [
          {
            "role": "primary_action",
            "type_any_of": ["button"],
            "min_count": 1
          }
        ],
        "embedding_prototype": {
          "provider": "openclip_ViT-B-32",
          "vector_id": "data/embeddings/workflows/WF_demo/N1_prototype.npy",
          "min_cosine_similarity": 0.85,
          "sample_count": 5
        }
      },
      "metadata": {
        "created_at": "2025-11-22T10:45:00Z",
        "observation_count": 5
      }
    }
  ],
  "edges": [
    {
      "edge_id": "E1_submit",
      "from_node": "N1_start",
      "to_node": "N2_processing",
      "action": {
        "type": "mouse_click",
        "target": {
          "role": "primary_action",
          "selection_policy": "first",
          "fallback_strategy": "visual_similarity"
        },
        "parameters": {
          "click_offset": [0, 0],
          "wait_after_ms": 500
        }
      },
      "constraints": {
        "max_delay_seconds": 5,
        "pre_conditions": ["element:primary_action_visible"],
        "post_conditions": ["window_title_changed"]
      },
      "post_conditions": {
        "expected_node": "N2_processing",
        "min_similarity": 0.85,
        "timeout_seconds": 5
      },
      "stats": {
        "manual_executions": 5,
        "assist_executions": 0,
        "auto_executions": 0,
        "success_count": 5,
        "failure_count": 0,
        "avg_execution_time_ms": 1200
      }
    }
  ],
  "safety_rules": {
    "forbidden_text_clicks": ["Delete", "Remove"],
    "forbidden_roles": ["delete_action"],
    "require_confirmation_for": ["irreversible_action"]
  },
  "stats": {
    "observed_runs": 5,
    "assist_runs": 0,
    "auto_candidate_runs": 0,
    "auto_confirmed_runs": 0,
    "success_rate_overall": 1.0,
    "avg_duration_seconds": 30.5
  },
  "learning": {
    "state": "OBSERVATION",
    "thresholds": {
      "min_observed_runs_for_coaching": 5,
      "min_assist_runs_for_auto_candidate": 10,
      "min_auto_candidate_runs_for_auto_confirmed": 20
    }
  }
}

Gestion des Erreurs

Échecs de Matching

Scénario : Aucun node ne correspond au ScreenState actuel.

Gestion :

1. Logger le ScreenState non matché avec screenshot
2. Calculer similarité avec tous les nodes existants
3. Si similarité proche (0.75-0.84), suggérer mise à jour du node
4. Si similarité faible (<0.75), suggérer création d'un nouveau node
5. Notifier l'utilisateur et mettre en pause l'exécution

Échecs de Détection UI

Scénario : Élément cible non trouvé par rôle.

Gestion :

1. Logger l'échec avec contexte (node, edge, rôle recherché)
2. Essayer stratégie de fallback (visual similarity)
3. Si fallback échoue, essayer position approximative
4. Si tout échoue, notifier utilisateur et demander correction
5. Mettre à jour le template du node avec feedback

Violations de Post-Conditions

Scénario : Post-conditions non satisfaites après exécution.

Gestion :

1. Logger la violation avec détails (attendu vs réel)
2. Attendre timeout configuré
3. Si toujours pas satisfait, marquer exécution comme échec
4. Incrémenter compteur d'échecs pour cet edge
5. Si échecs répétés (>3), marquer edge comme problématique

Changements d'UI Détectés

Scénario : Similarité d'embedding chute significativement.

Gestion :

1. Détecter changement (similarité < 0.70 vs prototype)
2. Capturer nouveau screenshot pour analyse
3. Mettre en pause l'exécution automatique
4. Notifier utilisateur du changement détecté
5. Proposer ré-apprentissage du node affecté

Stratégie de Test

Tests Unitaires

Composants à Tester :

1. RawSession :

   • Sérialisation/désérialisation JSON
   • Ajout d'événements et screenshots
   • Validation de schéma

2. ScreenState :

   • Création des 4 niveaux
   • Sérialisation/désérialisation JSON
   • Validation de structure

3. UIElement :

   • Détection de types et rôles
   • Génération d'embeddings duaux
   • Calcul de features visuelles

4. StateEmbedding :

   • Fusion pondérée
   • Fusion par concaténation
   • Calcul de similarité cosinus

5. FAISSManager :

   • Ajout d'embeddings
   • Recherche de similarité
   • Sauvegarde/chargement d'index

6. WorkflowNode :

   • Matching avec ScreenState
   • Validation de contraintes
   • Calcul de confiance

7. WorkflowEdge :

   • Validation de pre-conditions
   • Exécution d'actions
   • Vérification de post-conditions

8. LearningManager :

   • Transitions d'états
   • Calcul de métriques
   • Détection de rollback

Framework : pytest avec fixtures

Tests d'Intégration

Scénarios :

1. Pipeline Complet RawSession → Workflow :

   • Capturer session
   • Créer ScreenStates
   • Détecter UI elements
   • Calculer embeddings
   • Construire workflow graph
   • Vérifier structure du graphe

2. Matching et Exécution :

   • Charger workflow existant
   • Matcher ScreenState actuel
   • Trouver edge sortant
   • Exécuter action
   • Vérifier transition

3. Apprentissage Progressif :

   • Simuler 5 observations → COACHING
   • Simuler 10 assistances → AUTO_CANDIDATE
   • Simuler 20 exécutions → AUTO_CONFIRMÉ
   • Vérifier transitions

4. Gestion d'Erreurs :

   • Simuler échec de matching
   • Simuler élément non trouvé
   • Simuler violation post-conditions
   • Vérifier récupération

Tests de Performance

Métriques Cibles :

Opération	Temps Cible	Méthode de Test
Compute State Embedding	< 100ms	pytest-benchmark
FAISS Search	< 50ms	pytest-benchmark
UI Detection	< 200ms	pytest-benchmark
Action Execution	< 50ms	pytest-benchmark
End-to-End Processing	< 400ms	pytest-benchmark

Outils : pytest-benchmark, cProfile

Tests de Bout en Bout

Workflow de Test :

1. Apprentissage Complet :

   • Capturer 5 sessions d'un workflow simple
   • Vérifier construction automatique du graphe
   • Vérifier transition OBSERVATION → COACHING
   • Exécuter en mode assisté 10 fois
   • Vérifier transition COACHING → AUTO_CANDIDATE

2. Robustesse UI :

   • Apprendre workflow sur UI version 1
   • Modifier légèrement UI (couleurs, positions)
   • Vérifier que matching fonctionne toujours
   • Modifier significativement UI
   • Vérifier détection de changement

3. Multi-Workflows :

   • Apprendre 3 workflows différents
   • Vérifier isolation des workflows
   • Vérifier matching correct pour chaque workflow
   • Vérifier pas de confusion entre workflows

Correctness Properties

A property is a characteristic or behavior that should hold true across all valid executions of a system-essentially, a formal statement about what the system should do. Properties serve as the bridge between human-readable specifications and machine-verifiable correctness guarantees.

Property 1: RawSession Serialization Round Trip

For any valid RawSession object, serializing to JSON then deserializing should produce an equivalent RawSession with all events and screenshots preserved.

Validates: Requirements 1.4, 1.5

Property 2: ScreenState Multi-Level Consistency

For any ScreenState, all 4 levels (Raw, Perception, Sémantique UI, Contexte) should reference the same screenshot and timestamp.

Validates: Requirements 2.1, 2.2, 2.3, 2.4, 2.5

Property 3: UIElement Detection Confidence Bounds

For any detected UIElement, the confidence score should be between 0.0 and 1.0, and elements with confidence below threshold should not be included in results.

Validates: Requirements 3.6

Property 4: State Embedding Normalization

For any State Embedding vector, the L2 norm should be 1.0 (normalized vector).

Validates: Requirements 4.6

Property 5: State Embedding Similarity Symmetry

For any two State Embeddings A and B, similarity(A, B) should equal similarity(B, A).

Validates: Requirements 4.7

Property 6: State Embedding Similarity Bounds

For any two State Embeddings, the cosine similarity should be between -1.0 and 1.0.

Validates: Requirements 4.7

Property 7: WorkflowNode Matching Consistency

For any WorkflowNode and ScreenState, if the node matches with confidence C1, then matching again immediately should return the same confidence C1 (deterministic).

Validates: Requirements 9.1, 9.2, 9.3, 9.4, 9.5, 9.6

Property 8: WorkflowEdge Pre-Condition Validation

For any WorkflowEdge, if pre-conditions are not satisfied, execution should not proceed and should return failure.

Validates: Requirements 10.5

Property 9: Learning State Monotonic Progression

For any Workflow in learning state S, transitioning to state S' should only happen if S' is the next state in the progression (OBSERVATION → COACHING → AUTO_CANDIDATE → AUTO_CONFIRMÉ), except for rollback to COACHING.

Validates: Requirements 8.1, 8.2, 8.3, 8.4

Property 10: Learning State Rollback Condition

For any Workflow in AUTO_CONFIRMÉ state, if confidence drops below 0.90, the system should rollback to COACHING state.

Validates: Requirements 8.6

Property 11: FAISS Index Consistency

For any embedding added to FAISS index with metadata M, searching for that exact embedding should return it as the top result with metadata M.

Validates: Requirements 4.8, 12.3, 12.6

Property 12: Workflow Graph Structural Validity

For any Workflow Graph, all edges should reference existing nodes (no dangling references).

Validates: Requirements 7.2

Property 13: UIElement Role Uniqueness Per Type

For any ScreenState, if multiple UIElements have the same role, they should have different element_ids.

Validates: Requirements 3.3

Property 14: Embedding Prototype Sample Count

For any WorkflowNode with embedding prototype, the sample_count should be at least 1 and the prototype vector should exist.

Validates: Requirements 5.4

Property 15: Action Execution Timeout

For any WorkflowEdge execution, if post-conditions are not satisfied within timeout_seconds, the execution should be marked as failed.

Validates: Requirements 10.6

Property 16: Pattern Detection Minimum Repetitions

For any detected workflow pattern, it should have been observed at least 3 times before being proposed as a Workflow Graph.

Validates: Requirements 11.7

Property 17: State Embedding Component Weights Sum

For any weighted fusion configuration, the sum of all component weights (image + text + title + ui) should equal 1.0.

Validates: Requirements 4.5

Property 18: Workflow JSON Serialization Round Trip

For any valid Workflow Graph, serializing to JSON then deserializing should produce an equivalent Workflow with all nodes and edges preserved.

Validates: Requirements 7.6, 12.4, 12.5

Property 19: Performance Constraint - State Embedding

For any ScreenState, computing the State Embedding should complete in less than 100ms.

Validates: Requirements 15.1

Property 20: Performance Constraint - End-to-End

For any ScreenState processing (detection + embedding + matching), the total time should be less than 400ms.

Validates: Requirements 15.5

Considérations de Sécurité

Validation des Données

• Tous les JSON chargés doivent être validés contre leur schéma
• Les embeddings chargés doivent avoir les dimensions attendues
• Les workflow_ids et node_ids doivent être validés (format, unicité)

Isolation des Workflows

• Chaque workflow doit avoir son propre espace dans FAISS
• Les embeddings de différents workflows ne doivent pas interférer
• Les métadonnées doivent inclure workflow_id pour filtrage

Safety Rules

• Les actions interdites (forbidden_text_clicks, forbidden_roles) doivent être bloquées
• Les actions irréversibles doivent demander confirmation
• Les rollbacks doivent être possibles pour les 3 dernières actions

Logging et Audit

• Toutes les transitions d'état doivent être loggées
• Tous les échecs d'exécution doivent être loggées avec contexte
• Les changements d'UI détectés doivent être loggées avec screenshots

Optimisation des Performances

Embeddings

• Utiliser batch processing pour calculer plusieurs embeddings
• Mettre en cache les embeddings de prototypes
• Utiliser quantification FP16 pour modèles CLIP

FAISS

• Utiliser index IVF pour grands ensembles (>10k embeddings)
• Optimiser périodiquement l'index (compactage)
• Utiliser GPU si disponible pour recherche

UI Detection

• Limiter la résolution des screenshots (max 1920x1080)
• Utiliser ROI detection pour réduire zone de traitement
• Mettre en cache les résultats de détection pour frames similaires

Workflow Matching

• Pré-filtrer les candidats par window context
• Utiliser early stopping si confiance très élevée (>0.95)
• Mettre en cache le dernier node matché

Considérations de Déploiement

Dépendances

• Python 3.9+
• PyTorch 2.0+
• OpenCLIP
• FAISS (CPU ou GPU)
• Transformers (Hugging Face)
• NumPy, Pillow
• pytest, pytest-benchmark

Structure de Données

data/
├── sessions/
│   └── YYYY-MM-DD/
│       └── sess_*.json
├── screen_states/
│   └── YYYY-MM-DD/
│       └── screen_*.json
├── embeddings/
│   ├── screens/
│   │   └── *.npy
│   ├── elements/
│   │   └── *.npy
│   └── states/
│       └── *.npy
├── faiss_index/
│   ├── index.faiss
│   └── metadata.json
└── workflows/
    └── WF_*/
        ├── workflow.json
        └── prototypes/
            └── *.npy

Configuration

CONFIG = {
    "models": {
        "clip": "ViT-B-32",
        "vlm": "qwen2.5-vl:3b"
    },
    "embedding": {
        "dimension": 512,
        "fusion_method": "weighted",
        "weights": {
            "image": 0.5,
            "text": 0.3,
            "title": 0.1,
            "ui": 0.1
        }
    },
    "matching": {
        "min_similarity": 0.85,
        "faiss_k": 5
    },
    "learning": {
        "observation_threshold": 5,
        "coaching_threshold": 10,
        "auto_candidate_threshold": 20,
        "min_success_rate": 0.90,
        "rollback_confidence": 0.90
    },
    "performance": {
        "max_embedding_time_ms": 100,
        "max_detection_time_ms": 200,
        "max_total_time_ms": 400
    }
}

Plan d'Implémentation

Phase 1 : Fondations (Semaines 1-2)

Objectif : Structures de données et sérialisation

• Implémenter classes de base (RawSession, ScreenState, UIElement, etc.)
• Implémenter sérialisation/désérialisation JSON
• Tests unitaires sur structures
• Validation de schémas

Livrables :

• geniusia2/core/models/*.py
• Tests unitaires complets

Phase 2 : Embeddings et FAISS (Semaines 3-4)

Objectif : Système d'embeddings fonctionnel

• Implémenter FusionEngine
• Implémenter FAISSManager
• Implémenter calculs de similarité
• Tests de performance

Livrables :

• geniusia2/core/embedding/*.py
• Benchmarks de performance

Phase 3 : UI Detection (Semaines 5-6)

Objectif : Détection sémantique d'éléments UI

• Intégrer VLM pour détection
• Implémenter classification type/rôle
• Générer embeddings duaux
• Tests avec screenshots réels

Livrables :

• geniusia2/core/detection/*.py
• Dataset de test avec screenshots

Phase 4 : Workflow Graph (Semaines 7-9)

Objectif : Construction et matching de graphes

• Implémenter WorkflowNode et WorkflowEdge
• Implémenter GraphBuilder
• Implémenter NodeMatcher
• Tests d'intégration

Livrables :

• geniusia2/core/graph/*.py
• Tests d'intégration complets

Phase 5 : Exécution et Apprentissage (Semaines 10-12)

Objectif : Exécution d'actions et états d'apprentissage

• Implémenter ActionExecutor
• Implémenter LearningManager
• Implémenter transitions d'états
• Tests end-to-end

Livrables :

• geniusia2/core/graph/action_executor.py
• geniusia2/core/graph/learning_manager.py
• Tests end-to-end

Phase 6 : Optimisation et Production (Semaines 13-14)

Objectif : Optimisation et déploiement

• Optimiser performances (caching, batching)
• Ajouter monitoring et logging
• Documentation complète
• Tests de charge

Livrables :

• Système optimisé
• Documentation utilisateur
• Guide de déploiement