28 KiB
Document de Design
Vue d'Ensemble
RPA Vision V2 est un système d'automatisation à apprentissage progressif construit sur une architecture cognitive qui suit le paradigme Observer → Réfléchir → Agir. Le système utilise des modèles de vision par ordinateur de pointe pour la détection d'éléments UI, des modèles de langage pour le raisonnement visuel, et des mécanismes de mémoire adaptative pour apprendre les flux de travail utilisateur de manière incrémentale à travers trois modes opérationnels : Shadow, Assisté et Autopilot.
L'architecture est modulaire, avec une séparation claire entre la logique centrale (orchestration, apprentissage, embeddings), les composants IA (détection vision, raisonnement LLM), l'interface utilisateur (GUI PyQt5), et la persistance des données (logs chiffrés, index FAISS, profils utilisateur).
Architecture
Architecture de Haut Niveau
┌─────────────────────────────────────────────────────────────────┐
│ Couche Interface Utilisateur │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────────┐ │
│ │ GUI Minimale │ │ Dialogue de │ │ Tableau de Bord │ │
│ │ (PyQt5) │ │ Correction │ │ Résumé │ │
│ └──────────────┘ └──────────────┘ └──────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
↕
┌─────────────────────────────────────────────────────────────────┐
│ Couche Orchestration Centrale │
│ ┌──────────────────────────────────────────────────────────┐ │
│ │ Orchestrateur (Boucle Cognitive) │ │
│ │ Observer → Réfléchir → Agir → Apprendre → Répéter │ │
│ └──────────────────────────────────────────────────────────┘ │
│ ↕ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────────┐ │
│ │ Gestionnaire │ │ Gestionnaire │ │ Rejeu Async │ │
│ │Apprentissage │ │ Embeddings │ │ │ │
│ └──────────────┘ └──────────────┘ └──────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
↕
┌─────────────────────────────────────────────────────────────────┐
│ Couche IA/Vision │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────────┐ │
│ │ Utils Vision │ │ Gestionnaire │ │ Utils Image │ │
│ │ OWL-v2/DINO │ │ LLM │ │ Capture/ROI │ │
│ └──────────────┘ └──────────────┘ └──────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
↕
┌─────────────────────────────────────────────────────────────────┐
│ Couche Persistance des Données │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────────┐ │
│ │ Logs │ │ Profils │ │ Index FAISS │ │
│ │ Chiffrés │ │ Utilisateur │ │ (Embeddings) │ │
│ └──────────────┘ └──────────────┘ └──────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
Structure des Répertoires
/AI/amina/geniusia2/
├── core/
│ ├── orchestrator.py # Boucle cognitive principale
│ ├── replay_async.py # Moteur de rejeu d'actions
│ ├── embeddings_manager.py # Gestion OpenCLIP + FAISS
│ ├── learning_manager.py # État d'apprentissage & transitions de mode
│ ├── logger.py # Journalisation chiffrée AES-256
│ ├── config.py # Configuration globale
│ └── utils/
│ ├── image_utils.py # Capture d'écran, extraction ROI
│ ├── vision_utils.py # Interfaces modèles vision
│ └── input_utils.py # PyAutoGUI/PyDirectInput (AZERTY)
├── gui/
│ ├── minimal_gui.py # Interface PyQt5 principale
│ └── dialogs/
│ ├── correction_dialog.py
│ └── summary_dashboard.py
├── data/
│ ├── user_profiles/ # Profils d'apprentissage tâches (JSON)
│ ├── logs/ # Logs d'actions chiffrés
│ └── faiss_index/ # Index de mémoire visuelle
└── models/
├── openclip/ # Encodeurs visuels
├── owl_v2/ # Détecteur open-vocabulary
└── qwen2.5_vl/ # Modèle vision-langage (Ollama)
Composants et Interfaces
1. Orchestrateur (orchestrator.py)
Responsabilité : Boucle cognitive principale implémentant le paradigme Observer → Réfléchir → Agir.
Méthodes Clés :
run(): Boucle d'événements principalecapture_context(): Capture d'écran et détection de fenêtredetect_elements(frame, intent): Détection d'éléments UI basée sur la visionreason_about_action(detections, context): Prise de décision basée sur LLMexecute_or_suggest(decision): Exécution d'action dépendante du modelearn_from_feedback(feedback): Mise à jour de l'état d'apprentissage
Interfaces :
class Orchestrator:
def __init__(self, config, learning_manager, vision_utils, llm_manager, gui):
self.config = config
self.learning_manager = learning_manager
self.vision = vision_utils
self.llm = llm_manager
self.gui = gui
self.running = False
def run(self):
"""Boucle cognitive principale"""
while self.running:
# Observer
frame = self.capture_context()
intent = self.learning_manager.get_current_intent()
# Réfléchir
detections = self.vision.detect(frame, intent)
decision = self.llm.reason_about_action(detections, context)
# Agir (dépendant du mode)
mode = self.learning_manager.get_mode()
if mode == "shadow":
self.learning_manager.observe(decision)
elif mode == "assist":
feedback = self.gui.show_suggestion(decision)
self.learning_manager.confirm_action(feedback)
elif mode == "auto":
self.execute_action(decision)
self.learning_manager.record_execution(decision)
2. Gestionnaire d'Apprentissage (learning_manager.py)
Responsabilité : Suit la progression d'apprentissage, gère les transitions de mode, calcule les scores de confiance.
Attributs Clés :
mode: Mode opérationnel actuel (shadow/assist/auto)tasks: Dictionnaire des Séquence_Actions apprisescurrent_context: Contexte de tâche actifembeddings_index: Référence à l'index FAISSlogger: Référence au logger chiffré
Méthodes Clés :
observe(action): Enregistrer l'observation en mode Shadowsuggest_action(context): Générer une suggestion en mode Assistéconfirm_action(feedback): Traiter la validation/correction utilisateurevaluate_task(task_id): Calculer confiance et concordanceshould_transition_to_auto(task_id): Vérifier l'éligibilité autopilotrollback_if_low_confidence(task_id): Rétrograder au mode Assistécalculate_confidence(vision_conf, llm_score, history): Calculer le score de confiance
Interfaces :
class LearningManager:
def __init__(self, embeddings_manager, logger, config):
self.mode = "shadow" # Mode initial
self.tasks = {} # task_id -> TaskProfile
self.embeddings_manager = embeddings_manager
self.logger = logger
self.config = config
def calculate_confidence(self, vision_conf, llm_score, task_id):
"""Calculer le score de confiance pondéré"""
history_score = self.tasks[task_id].get_historical_performance()
return 0.6 * vision_conf + 0.3 * llm_score + 0.1 * history_score
def should_transition_to_auto(self, task_id):
"""Vérifier si la tâche remplit les critères autopilot"""
task = self.tasks[task_id]
return (task.observation_count >= 20 and
task.concordance_rate >= 0.95)
def rollback_if_low_confidence(self, task_id):
"""Rétrograder au mode Assisté si confiance faible"""
task = self.tasks[task_id]
if task.mode == "auto" and task.confidence_score < 0.90:
task.mode = "assist"
self.logger.log_mode_transition(task_id, "auto", "assist", "low_confidence")
3. Gestionnaire d'Embeddings (embeddings_manager.py)
Responsabilité : Gère les embeddings visuels en utilisant OpenCLIP et l'indexation FAISS pour la recherche de similarité.
Méthodes Clés :
encode_image(image): Générer un embedding 512-d avec OpenCLIPadd_to_index(embedding, metadata): Stocker l'embedding dans FAISSsearch_similar(embedding, k=5): Trouver les k embeddings les plus similairesrebuild_index(): Reconstruire l'index FAISS après mises à jourget_embedding_similarity(emb1, emb2): Calculer la similarité cosinus
Interfaces :
class EmbeddingsManager:
def __init__(self, model_path, index_path):
self.clip_model = self.load_openclip(model_path)
self.faiss_index = self.load_or_create_index(index_path)
self.metadata_store = {} # id -> mapping métadonnées
def encode_image(self, image):
"""Générer un embedding 512-d"""
with torch.no_grad():
embedding = self.clip_model.encode_image(image)
return embedding.cpu().numpy()
def add_to_index(self, embedding, metadata):
"""Stocker l'embedding avec métadonnées"""
idx = self.faiss_index.ntotal
self.faiss_index.add(embedding.reshape(1, -1))
self.metadata_store[idx] = metadata
return idx
def search_similar(self, query_embedding, k=5):
"""Trouver les k plus proches voisins"""
distances, indices = self.faiss_index.search(
query_embedding.reshape(1, -1), k
)
results = [
{
"id": int(idx),
"distance": float(dist),
"metadata": self.metadata_store[int(idx)]
}
for dist, idx in zip(distances[0], indices[0])
]
return results
4. Utils Vision (vision_utils.py)
Responsabilité : Interface vers les modèles de vision (OWL-v2, Grounding DINO, YOLO-World) pour la détection d'éléments UI.
Méthodes Clés :
detect_with_owlv2(prompt, frame): Détection open-vocabulary avec OWL-v2detect_with_dino(prompt, frame): Détection Grounding DINOdetect_with_yolo(prompt, frame): Détection YOLO-Worldselect_best_detection(detections, context): Choisir la détection la plus confianteextract_roi(frame, bbox): Extraire la région d'intérêt
Format de Sortie de Détection :
{
"label": "valider_button",
"confidence": 0.93,
"bbox": (x, y, w, h), # Coordonnées en pixels
"embedding": np.array([...]), # Embedding visuel 512-d
"model": "owl-v2"
}
5. Gestionnaire LLM (llm_manager.py)
Responsabilité : Interface vers Ollama pour le raisonnement visuel utilisant Qwen 2.5-VL ou CogVLM.
Méthodes Clés :
reason_about_detections(detections, context, intent): Sélectionner la meilleure actiongenerate_with_vision(prompt, images): Génération multi-modalescore_action_relevance(action, intent): Calculer la confiance LLM
Interfaces :
class LLMManager:
def __init__(self, model_name="qwen2.5-vl:3b", ollama_host="localhost:11434"):
self.model_name = model_name
self.ollama_client = ollama.Client(host=ollama_host)
def reason_about_detections(self, detections, context, intent):
"""Utiliser VLM pour sélectionner la meilleure action"""
prompt = f"""Étant donné ces éléments UI et l'intention utilisateur '{intent}',
avec quel élément devrait-on interagir ?
Éléments : {[d['label'] for d in detections]}
Contexte : {context}
Répondre avec le label de l'élément et la confiance (0-1)."""
images = [d['roi_image'] for d in detections]
response = self.ollama_client.generate(
model=self.model_name,
prompt=prompt,
images=images
)
return self.parse_llm_response(response)
6. Logger (logger.py)
Responsabilité : Journalisation chiffrée AES-256 avec format JSON structuré.
Méthodes Clés :
log_action(action_data): Enregistrer l'action avec chiffrementlog_correction(correction_data): Enregistrer la correction utilisateurlog_mode_transition(task_id, from_mode, to_mode, reason): Enregistrer les changements de modeget_logs(task_id=None, start_time=None, end_time=None): Interroger les logsencrypt_entry(data): Chiffrement AES-256decrypt_entry(encrypted_data): Déchiffrement AES-256
Format d'Entrée de Log :
{
"timestamp": "2025-11-13T10:32:04.123Z",
"window": "Dolibarr - Facturation",
"action": "click",
"element": "valider_button",
"bbox": [450, 320, 120, 40],
"confidence": 0.97,
"mode": "auto",
"result": "success",
"task_id": "ouvrir_facture_001"
}
7. GUI Minimale (minimal_gui.py)
Responsabilité : Interface PyQt5 pour l'affichage des modes, suggestions et retours utilisateur.
Composants Clés :
- Indicateur de mode (👀 Shadow | 🤝 Assisté | 🤖 Auto)
- Contrôles Start/Stop/Pause
- Superposition de suggestion avec éléments UI surlignés
- Raccourcis clavier (Entrée/Échap/Alt+C/Ctrl+Pause)
- Système de notifications pour alertes et confirmations
Interfaces :
class MinimalGUI(QMainWindow):
def __init__(self, orchestrator):
super().__init__()
self.orchestrator = orchestrator
self.init_ui()
self.setup_shortcuts()
def show_suggestion(self, decision):
"""Afficher la superposition de suggestion et attendre l'entrée utilisateur"""
overlay = SuggestionOverlay(decision)
overlay.show()
return overlay.wait_for_feedback() # Retourne : "accept", "reject", ou "correct"
def show_notification(self, message, notification_type="info"):
"""Afficher une notification non intrusive"""
notification = QNotification(message, notification_type)
notification.show()
def update_mode_indicator(self, mode):
"""Mettre à jour l'icône de mode dans l'UI"""
icons = {"shadow": "👀", "assist": "🤝", "auto": "🤖"}
self.mode_label.setText(f"{icons[mode]} {mode.capitalize()}")
8. Tableau de Bord Résumé (summary_dashboard.py)
Responsabilité : Afficher les statistiques de tâches, niveaux de confiance et historique d'exécution.
Fonctionnalités Clés :
- Tableau de métriques en temps réel
- Filtrage et recherche de tâches
- Visualisation des tendances de confiance
- Historique des corrections
- Fonctionnalité d'export
Métriques du Tableau de Bord :
{
"task_id": "ouvrir_facture",
"task_name": "Ouvrir Facture",
"mode": "auto",
"confidence": 97.3,
"observation_count": 45,
"concordance_rate": 98.2,
"correction_count": 1,
"last_execution": "2025-11-12T15:32:04",
"avg_latency_ms": 320,
"success_rate": 98.9
}
Modèles de Données
TaskProfile
@dataclass
class TaskProfile:
task_id: str
task_name: str
mode: str # "shadow", "assist", "auto"
observation_count: int
concordance_rate: float
confidence_score: float
correction_count: int
last_execution: datetime
window_whitelist: List[str]
action_sequence: List[Action]
embeddings: List[np.ndarray]
metadata: Dict[str, Any]
def to_json(self):
"""Sérialiser en JSON pour la persistance"""
return {
"task_id": self.task_id,
"task_name": self.task_name,
"mode": self.mode,
"observation_count": self.observation_count,
"concordance_rate": self.concordance_rate,
"confidence_score": self.confidence_score,
"correction_count": self.correction_count,
"last_execution": self.last_execution.isoformat(),
"window_whitelist": self.window_whitelist,
"action_sequence": [a.to_dict() for a in self.action_sequence],
"metadata": self.metadata
}
Action
@dataclass
class Action:
action_type: str # "click", "type", "scroll", "wait"
target_element: str
bbox: Tuple[int, int, int, int]
confidence: float
embedding: np.ndarray
timestamp: datetime
window_title: str
def to_dict(self):
return {
"action_type": self.action_type,
"target_element": self.target_element,
"bbox": list(self.bbox),
"confidence": self.confidence,
"timestamp": self.timestamp.isoformat(),
"window_title": self.window_title
}
Detection
@dataclass
class Detection:
label: str
confidence: float
bbox: Tuple[int, int, int, int]
embedding: np.ndarray
model_source: str # "owl-v2", "dino", "yolo"
roi_image: np.ndarray
Gestion des Erreurs
Échecs des Modèles de Vision
Scénario : Le modèle de vision ne parvient pas à détecter d'éléments UI.
Gestion :
- Logger l'erreur avec contexte (fenêtre, capture d'écran)
- Réessayer avec un modèle de vision alternatif (OWL-v2 → DINO → YOLO)
- Si tous les modèles échouent, notifier l'utilisateur et mettre en pause l'automatisation
- Demander une correction manuelle ou une ré-observation
Échecs du Raisonnement LLM
Scénario : Service Ollama indisponible ou LLM retourne une réponse à faible confiance.
Gestion :
- Revenir à la détection pure basée sur la vision (bbox avec confiance la plus élevée)
- Réduire temporairement le poids du LLM dans le calcul de confiance
- Logger l'opération en mode dégradé
- Notifier l'utilisateur de la capacité de raisonnement réduite
Violations de Seuil de Confiance
Scénario : Le score de confiance tombe en dessous du seuil pendant le mode Autopilot.
Gestion :
- Geler immédiatement l'exécution de l'action
- Faire passer la tâche en mode Assisté
- Afficher une notification expliquant la baisse de confiance
- Demander la validation utilisateur pour la prochaine action
- Logger la transition de mode avec raison
Violations de Liste Blanche
Scénario : Le système tente une action dans une fenêtre non présente dans la liste blanche.
Gestion :
- Bloquer l'action immédiatement
- Logger la violation de sécurité avec détails de la fenêtre
- Afficher une alerte à l'utilisateur
- Inviter l'utilisateur à ajouter la fenêtre à la liste blanche ou abandonner la tâche
Échecs de Rollback
Scénario : L'opération de rollback ne peut pas restaurer l'état précédent.
Gestion :
- Logger l'échec du rollback avec les actions tentées
- Notifier l'utilisateur du rollback partiel
- Fournir des instructions d'annulation manuelle
- Mettre en pause l'automatisation jusqu'à confirmation de l'état par l'utilisateur
Corruption de l'Index FAISS
Scénario : L'index FAISS devient corrompu ou inaccessible.
Gestion :
- Tenter la reconstruction de l'index à partir des embeddings stockés
- Si la reconstruction échoue, créer un nouvel index et marquer les tâches pour ré-observation
- Notifier l'utilisateur de la perte de données et des tâches affectées
- Continuer l'opération avec un contexte historique réduit
Stratégie de Test
Tests Unitaires
Composants Centraux :
LearningManager: Tester le calcul de confiance, transitions de mode, évaluation de tâcheEmbeddingsManager: Tester l'encodage, l'indexation, la recherche de similaritéLogger: Tester le chiffrement/déchiffrement, l'interrogation des logsVision Utils: Tester le parsing de détection, l'extraction ROILLM Manager: Tester la génération de prompts, le parsing de réponses
Framework de Test : pytest avec fixtures pour données mock
Tests d'Intégration
Scénarios :
- Transition Shadow → Assisté : Vérifier que l'accumulation d'observations déclenche le changement de mode
- Transition Assisté → Autopilot : Vérifier les exigences de seuil de confiance et de compteur d'observations
- Rollback Autopilot → Assisté : Vérifier que la faible confiance déclenche la rétrogradation
- Boucle de Retour de Correction : Vérifier que les corrections mettent à jour les embeddings et améliorent les prédictions futures
- Application de Liste Blanche : Vérifier que les actions sont bloquées dans les fenêtres non présentes dans la liste blanche
- Opération de Rollback : Vérifier que les 3 dernières actions peuvent être inversées
Tests de Bout en Bout
Flux de Travail de Test :
- Cycle d'Apprentissage Complet : Observation Shadow → Validation Assistée → Exécution Autopilot
- Adaptation au Changement d'UI : Introduire des changements d'UI et vérifier que le système détecte et s'adapte
- Gestion Multi-Tâches : Apprendre et exécuter plusieurs tâches simultanément
- Conformité Sécurité : Vérifier que tous les logs sont chiffrés, liste blanche appliquée, rollback fonctionnel
Environnement de Test : Application UI contrôlée (ex : application web de test avec éléments connus)
Tests de Performance
Métriques :
- Latence observation-vers-suggestion (cible : <400ms)
- Temps d'inférence du modèle de vision
- Temps de raisonnement LLM
- Temps de recherche de similarité FAISS
- Surcharge de chiffrement des logs
Outils : pytest-benchmark, cProfile pour le profilage
Tests d'Acceptation Utilisateur
Scénarios :
- Utilisateur non technique apprend une tâche simple (ex : remplissage de formulaire)
- Utilisateur avancé automatise un flux de travail complexe (ex : saisie de données multi-étapes)
- Administrateur examine les logs et le tableau de bord
- Utilisateur corrige les mauvaises détections et vérifie l'amélioration de l'apprentissage
Critères de Succès :
- Taux de concordance ≥95% après 20 observations
- Taux de correction <3% en mode Autopilot
- Satisfaction utilisateur avec la réactivité et la clarté de l'UI
Considérations de Sécurité
Chiffrement
- Tous les logs chiffrés avec AES-256-CBC
- Clés de chiffrement stockées dans un trousseau sécurisé (keychain système)
- Politique de rotation des clés : tous les 90 jours
Contrôle d'Accès
- Liste blanche appliquée au niveau de l'orchestrateur
- Aucune action permise en dehors des fenêtres de la liste blanche
- Modifications de la liste blanche nécessitent une confirmation administrateur
Piste d'Audit
- Toutes les actions loggées avec contexte complet
- Logs immuables (ajout uniquement)
- Intégrité des logs vérifiée avec signatures HMAC
Confidentialité
- Captures d'écran stockées temporairement en mémoire uniquement
- Embeddings anonymisés (pas de PII dans les métadonnées)
- Profils utilisateur stockés localement, jamais transmis
Mécanismes de Sécurité
- Arrêt d'urgence Ctrl+Pause (niveau interruption matérielle)
- Capacité de rollback pour les 3 dernières actions
- Rétrogradation automatique de mode en cas de perte de confiance
- Override manuel toujours disponible
Optimisation des Performances
Optimisation des Modèles de Vision
- Quantification des modèles (FP16 ou INT8) pour inférence plus rapide
- Traitement par lots de multiples détections
- Détection basée sur ROI pour réduire la zone de traitement
- Mise en cache des modèles pour éviter les chargements répétés
Optimisation FAISS
- Utiliser l'index IVF (Inverted File) pour les grands ensembles d'embeddings
- Optimisation et compactage périodiques de l'index
- Accélération GPU pour la recherche de similarité (si disponible)
Optimisation LLM
- Utiliser des modèles quantifiés plus petits (Qwen 3B paramètres)
- Mise en cache des prompts pour requêtes répétées
- Regroupement de multiples requêtes de raisonnement
- Déploiement Ollama local pour minimiser la latence
Optimisation GUI
- Mises à jour UI non bloquantes utilisant signaux/slots Qt
- Chargement paresseux des données du tableau de bord
- Rendu efficace de superposition avec capture d'écran minimale
- Gestion d'entrée utilisateur avec anti-rebond
Considérations de Déploiement
Exigences Système
- OS : Linux (Ubuntu 20.04+), Windows 10+, macOS 11+
- RAM : 16GB minimum (32GB recommandé pour grands modèles)
- GPU : GPU NVIDIA avec 8GB+ VRAM (optionnel mais recommandé)
- Stockage : 20GB pour modèles et données
- Python : 3.9+
Dépendances
- PyTorch 2.0+
- Transformers (Hugging Face)
- OpenCLIP
- FAISS (variante CPU ou GPU)
- Ollama (pour LLM local)
- PyQt5
- PyAutoGUI / PyDirectInput
- cryptography (pour chiffrement AES)
Étapes d'Installation
- Installer Ollama et télécharger le modèle Qwen 2.5-VL
- Télécharger les modèles de vision (OWL-v2, Grounding DINO)
- Installer les dépendances Python via requirements.txt
- Initialiser l'index FAISS et les clés de chiffrement
- Configurer la liste blanche et les paramètres initiaux
- Lancer la GUI et commencer l'observation en mode Shadow
Configuration
Configuration globale dans core/config.py :
CONFIG = {
"models": {
"vision": "owl-v2", # ou "dino", "yolo"
"llm": "qwen2.5-vl:3b",
"clip": "ViT-B-32"
},
"thresholds": {
"autopilot_observations": 20,
"autopilot_concordance": 0.95,
"confidence_min": 0.90,
"rollback_confidence": 0.85
},
"performance": {
"max_latency_ms": 400,
"max_correction_rate": 0.03
},
"security": {
"encryption_algorithm": "AES-256-CBC",
"log_retention_days": 90,
"key_rotation_days": 90
}
}