Feat: Action analyser_avec_ia (Ollama qwen2.5-vl)
Nouvelle action d'intelligence artificielle: - Analyse de contenu visuel via Ollama - 8 types d'analyse prédéfinis: general, formulaire, erreur, boutons, tableau, menu, validation, extraction - Prompts personnalisables - Support température et max_tokens - Variable de sortie configurable Modèle par défaut: qwen2.5-vl:7b Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Dom
@@ -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'
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user