diff --git a/visual_workflow_builder/backend/actions/intelligence/__init__.py b/visual_workflow_builder/backend/actions/intelligence/__init__.py new file mode 100644 index 000000000..527e332b9 --- /dev/null +++ b/visual_workflow_builder/backend/actions/intelligence/__init__.py @@ -0,0 +1,21 @@ +""" +Actions Intelligence VWB - Module d'initialisation +Auteur : Dom, Claude - 14 janvier 2026 + +Ce module contient les actions d'intelligence artificielle +pour le Visual Workflow Builder. + +Actions disponibles : +- VWBAnalyserAvecIAAction : Analyse de contenu visuel avec IA (Ollama) +""" + +from .analyser_avec_ia import VWBAnalyserAvecIAAction, VWBAIAnalyzeTextAction + +__all__ = [ + 'VWBAnalyserAvecIAAction', + 'VWBAIAnalyzeTextAction', # Alias anglais +] + +__version__ = '1.0.0' +__author__ = 'Dom, Claude' +__date__ = '14 janvier 2026' diff --git a/visual_workflow_builder/backend/actions/intelligence/analyser_avec_ia.py b/visual_workflow_builder/backend/actions/intelligence/analyser_avec_ia.py new file mode 100644 index 000000000..5e42974e6 --- /dev/null +++ b/visual_workflow_builder/backend/actions/intelligence/analyser_avec_ia.py @@ -0,0 +1,377 @@ +""" +Action Analyser avec IA - Analyse de contenu visuel via Ollama +Auteur : Dom, Claude - 14 janvier 2026 + +Cette action permet d'analyser du contenu visuel (texte, images, interfaces) +en utilisant un modèle de vision IA (Ollama avec qwen2.5-vl ou similaire). + +Cas d'usage : +- Comprendre le contenu d'une zone d'écran +- Extraire des informations structurées +- Valider visuellement des états d'interface +- Analyser des documents ou formulaires +""" + +from typing import Dict, Any, List, Optional, Tuple +from datetime import datetime +import time +import base64 +import io +import requests + +from ..base_action import BaseVWBAction, VWBActionResult, VWBActionStatus +from ...contracts.error import VWBErrorType, create_vwb_error +from ...contracts.visual_anchor import VWBVisualAnchor + + +# Configuration Ollama par défaut +OLLAMA_DEFAULT_URL = "http://localhost:11434" +OLLAMA_DEFAULT_MODEL = "qwen2.5-vl:7b" + + +class VWBAnalyserAvecIAAction(BaseVWBAction): + """ + Action d'analyse de contenu visuel avec IA. + + Utilise Ollama avec un modèle de vision (qwen2.5-vl) pour : + - Analyser le contenu d'une zone d'écran + - Répondre à des questions sur l'interface + - Extraire des informations structurées + - Valider des états visuels + """ + + def __init__( + self, + action_id: str, + parameters: Dict[str, Any], + screen_capturer=None + ): + """ + Initialise l'action d'analyse IA. + + Args: + action_id: Identifiant unique de l'action + parameters: Paramètres de l'analyse + screen_capturer: Instance du ScreenCapturer (optionnel) + """ + super().__init__( + action_id=action_id, + name="Analyser avec IA", + description="Analyse du contenu visuel avec intelligence artificielle", + parameters=parameters, + screen_capturer=screen_capturer + ) + + # Zone à analyser (ancre visuelle ou région) + self.ancre_visuelle: Optional[VWBVisualAnchor] = ( + parameters.get('visual_anchor') or + parameters.get('ancre_visuelle') + ) + self.region = parameters.get('region') # {x, y, width, height} + + # Prompt d'analyse + self.prompt = parameters.get('prompt', parameters.get('question', '')) + self.prompt_systeme = parameters.get('prompt_systeme', parameters.get('system_prompt', '')) + + # Type d'analyse prédéfini + self.type_analyse = parameters.get('type_analyse', parameters.get('analysis_type', 'general')) + + # Configuration Ollama + self.ollama_url = parameters.get('ollama_url', OLLAMA_DEFAULT_URL) + self.ollama_model = parameters.get('ollama_model', parameters.get('model', OLLAMA_DEFAULT_MODEL)) + + # Options + self.timeout_ms = parameters.get('timeout_ms', 30000) + self.temperature = parameters.get('temperature', 0.3) + self.max_tokens = parameters.get('max_tokens', 1000) + + # Variable de sortie + self.variable_sortie = parameters.get('variable_sortie', parameters.get('output_variable', 'analyse_ia')) + + # Prompts prédéfinis par type d'analyse + self.prompts_predefinifs = { + 'general': "Décris ce que tu vois dans cette image de manière concise.", + 'formulaire': "Analyse ce formulaire. Liste les champs visibles, leur état (rempli/vide) et les valeurs si lisibles.", + 'erreur': "Y a-t-il un message d'erreur visible ? Si oui, quel est son contenu exact ?", + 'boutons': "Liste tous les boutons visibles avec leur texte et leur état apparent (actif/inactif/grisé).", + 'tableau': "Analyse ce tableau. Décris sa structure (colonnes, lignes) et résume son contenu.", + 'menu': "Décris les options de menu visibles et leur hiérarchie.", + 'validation': "Cette interface semble-t-elle dans un état valide ? Décris ce que tu observes.", + 'extraction': "Extrait toutes les informations textuelles visibles de manière structurée.", + } + + def validate_parameters(self) -> List[str]: + """Valide les paramètres de l'action.""" + erreurs = [] + + # Vérifier qu'on a une source d'image + if not self.ancre_visuelle and not self.region and not self.screen_capturer: + erreurs.append("Ancre visuelle, région ou screen_capturer requis") + + # Vérifier le prompt + if not self.prompt and self.type_analyse not in self.prompts_predefinifs: + erreurs.append("Prompt ou type d'analyse valide requis") + + # Vérifier le timeout + if self.timeout_ms < 5000: + erreurs.append("Timeout minimum: 5000ms (5 secondes)") + + if self.timeout_ms > 120000: + erreurs.append("Timeout maximum: 120000ms (2 minutes)") + + return erreurs + + def execute_core(self, step_id: str) -> VWBActionResult: + """ + Exécute l'analyse IA. + + Args: + step_id: Identifiant de l'étape + + Returns: + Résultat d'exécution avec l'analyse + """ + start_time = datetime.now() + + try: + # Étape 1: Capturer l'image à analyser + image_base64 = self._capturer_image() + + if not image_base64: + return self._create_error_result( + step_id=step_id, + start_time=start_time, + error_type=VWBErrorType.SCREEN_CAPTURE_FAILED, + message="Impossible de capturer l'image à analyser" + ) + + # Étape 2: Construire le prompt + prompt_final = self._construire_prompt() + + # Étape 3: Appeler Ollama + print(f"🤖 Analyse IA en cours ({self.ollama_model})...") + resultat_analyse = self._analyser_avec_ollama(image_base64, prompt_final) + + if resultat_analyse is None: + return self._create_error_result( + step_id=step_id, + start_time=start_time, + error_type=VWBErrorType.SYSTEM_ERROR, + message="Échec de l'analyse IA" + ) + + end_time = datetime.now() + execution_time = (end_time - start_time).total_seconds() * 1000 + + print(f"✅ Analyse IA terminée en {execution_time:.0f}ms") + print(f"📝 Résultat: {resultat_analyse[:200]}..." if len(resultat_analyse) > 200 else f"📝 Résultat: {resultat_analyse}") + + return VWBActionResult( + action_id=self.action_id, + step_id=step_id, + status=VWBActionStatus.SUCCESS, + start_time=start_time, + end_time=end_time, + execution_time_ms=execution_time, + output_data={ + 'analyse': resultat_analyse, + 'variable_sortie': self.variable_sortie, + 'type_analyse': self.type_analyse, + 'model': self.ollama_model, + 'prompt_utilise': prompt_final[:100] + '...' if len(prompt_final) > 100 else prompt_final + }, + evidence_list=self.evidence_list.copy() + ) + + except Exception as e: + return self._create_error_result( + step_id=step_id, + start_time=start_time, + error_type=VWBErrorType.SYSTEM_ERROR, + message=f"Erreur: {str(e)}", + technical_details={'exception': str(e)} + ) + + def _capturer_image(self) -> Optional[str]: + """ + Capture l'image à analyser. + + Returns: + Image en base64 ou None + """ + try: + # Option 1: Image depuis l'ancre visuelle + if self.ancre_visuelle: + if isinstance(self.ancre_visuelle, dict): + img = self.ancre_visuelle.get('screenshot') or self.ancre_visuelle.get('image_base64') + if img: + return img + elif isinstance(self.ancre_visuelle, VWBVisualAnchor): + if self.ancre_visuelle.screenshot_base64: + return self.ancre_visuelle.screenshot_base64 + + # Option 2: Capture d'une région spécifique + if self.region and self.screen_capturer: + return self._capturer_region(self.region) + + # Option 3: Capture de l'écran entier + if self.screen_capturer: + return self._capturer_ecran_complet() + + return None + + except Exception as e: + print(f"⚠️ Erreur capture: {e}") + return None + + def _capturer_region(self, region: Dict[str, int]) -> Optional[str]: + """Capture une région spécifique de l'écran.""" + try: + from PIL import Image + + # Capturer l'écran entier + img_array = self.screen_capturer.capture() + if img_array is None: + return None + + # Convertir en PIL et découper + pil_image = Image.fromarray(img_array) + + x = region.get('x', 0) + y = region.get('y', 0) + width = region.get('width', 100) + height = region.get('height', 100) + + cropped = pil_image.crop((x, y, x + width, y + height)) + + # Convertir en base64 + buffer = io.BytesIO() + cropped.save(buffer, format='PNG') + return base64.b64encode(buffer.getvalue()).decode('utf-8') + + except Exception as e: + print(f"⚠️ Erreur capture région: {e}") + return None + + def _capturer_ecran_complet(self) -> Optional[str]: + """Capture l'écran entier.""" + try: + from PIL import Image + + img_array = self.screen_capturer.capture() + if img_array is None: + return None + + pil_image = Image.fromarray(img_array) + + buffer = io.BytesIO() + pil_image.save(buffer, format='PNG', optimize=True) + return base64.b64encode(buffer.getvalue()).decode('utf-8') + + except Exception as e: + print(f"⚠️ Erreur capture écran: {e}") + return None + + def _construire_prompt(self) -> str: + """ + Construit le prompt final pour l'analyse. + + Returns: + Prompt complet + """ + # Utiliser le prompt personnalisé si fourni + if self.prompt: + prompt_base = self.prompt + # Sinon utiliser le prompt prédéfini selon le type + elif self.type_analyse in self.prompts_predefinifs: + prompt_base = self.prompts_predefinifs[self.type_analyse] + else: + prompt_base = self.prompts_predefinifs['general'] + + # Ajouter le prompt système si fourni + if self.prompt_systeme: + return f"{self.prompt_systeme}\n\n{prompt_base}" + + return prompt_base + + def _analyser_avec_ollama(self, image_base64: str, prompt: str) -> Optional[str]: + """ + Envoie l'image à Ollama pour analyse. + + Args: + image_base64: Image en base64 + prompt: Prompt d'analyse + + Returns: + Texte de l'analyse ou None + """ + try: + # Préparer la requête + payload = { + "model": self.ollama_model, + "prompt": prompt, + "images": [image_base64], + "stream": False, + "options": { + "temperature": self.temperature, + "num_predict": self.max_tokens, + } + } + + # Appeler l'API Ollama + response = requests.post( + f"{self.ollama_url}/api/generate", + json=payload, + timeout=self.timeout_ms / 1000 + ) + + if response.status_code == 200: + result = response.json() + return result.get('response', '').strip() + else: + print(f"⚠️ Erreur Ollama: {response.status_code} - {response.text[:200]}") + return None + + except requests.exceptions.Timeout: + print(f"⚠️ Timeout Ollama après {self.timeout_ms}ms") + return None + + except requests.exceptions.ConnectionError: + print(f"⚠️ Ollama non accessible à {self.ollama_url}") + return self._fallback_analyse(prompt) + + except Exception as e: + print(f"⚠️ Erreur Ollama: {e}") + return None + + def _fallback_analyse(self, prompt: str) -> Optional[str]: + """ + Fallback si Ollama n'est pas disponible. + + Args: + prompt: Prompt original + + Returns: + Message d'erreur informatif + """ + return f"[Analyse IA non disponible - Ollama non accessible]\nPrompt demandé: {prompt[:100]}..." + + def get_action_info(self) -> Dict[str, Any]: + """Retourne les informations de l'action.""" + return { + 'action_id': self.action_id, + 'name': self.name, + 'description': self.description, + 'type': 'analyser_avec_ia', + 'parameters': { + 'type_analyse': self.type_analyse, + 'prompt': self.prompt[:50] + '...' if len(self.prompt) > 50 else self.prompt, + 'model': self.ollama_model, + 'variable_sortie': self.variable_sortie + }, + 'status': self.current_status.value + } + + +# Alias pour compatibilité avec le catalogue anglais +VWBAIAnalyzeTextAction = VWBAnalyserAvecIAAction