#!/usr/bin/env python3 """ RPA Vision V3 - IntentParser Parseur d'intentions basé sur LLM pour l'agent conversationnel. Ce module analyse les requêtes utilisateur en langage naturel et extrait : - L'intention principale (execute, query, configure, help, etc.) - Les paramètres associés - Le niveau de confiance Auteur: Dom - Janvier 2026 """ import json import logging import re from dataclasses import dataclass, field from enum import Enum from typing import Dict, Any, List, Optional, Tuple from pathlib import Path logger = logging.getLogger(__name__) class IntentType(Enum): """Types d'intentions supportées.""" EXECUTE = "execute" # Exécuter un workflow QUERY = "query" # Poser une question sur les workflows LIST = "list" # Lister les workflows disponibles CONFIGURE = "configure" # Configurer un paramètre HELP = "help" # Demander de l'aide STATUS = "status" # Vérifier le statut CANCEL = "cancel" # Annuler l'exécution en cours HISTORY = "history" # Voir l'historique CONFIRM = "confirm" # Confirmer une action DENY = "deny" # Refuser une action CLARIFY = "clarify" # Demander une clarification UNKNOWN = "unknown" # Intention non reconnue @dataclass class ParsedIntent: """Résultat du parsing d'une intention.""" intent_type: IntentType confidence: float raw_query: str workflow_hint: Optional[str] = None parameters: Dict[str, Any] = field(default_factory=dict) entities: List[Dict[str, Any]] = field(default_factory=list) clarification_needed: bool = False clarification_question: Optional[str] = None def to_dict(self) -> Dict[str, Any]: return { "intent_type": self.intent_type.value, "confidence": self.confidence, "raw_query": self.raw_query, "workflow_hint": self.workflow_hint, "parameters": self.parameters, "entities": self.entities, "clarification_needed": self.clarification_needed, "clarification_question": self.clarification_question } class IntentParser: """ Parseur d'intentions pour l'agent conversationnel. Utilise une combinaison de règles et optionnellement un LLM pour comprendre les requêtes utilisateur. """ # Patterns pour la détection d'intentions par règles INTENT_PATTERNS = { IntentType.EXECUTE: [ r"(?:lance|exécute|démarre|fait|run|start|execute)\s+(.+)", r"(?:je veux|je voudrais|peux-tu)\s+(.+)", r"(?:facturer?|créer?|générer?|exporter?)\s+(.+)", r"^(.+)\s+(?:maintenant|tout de suite|svp|stp)$", ], IntentType.LIST: [ r"(?:liste|montre|affiche|quels sont)\s+(?:les\s+|des\s+)?(?:workflows?|processus|automatisations?)", r"liste\s+des\s+workflows?", r"(?:qu'est-ce que|que)\s+(?:je peux|tu peux)\s+faire", r"(?:workflows?|processus)\s+disponibles?", r"(?:voir|afficher)\s+(?:les\s+|tous\s+les\s+)?workflows?", ], IntentType.QUERY: [ r"(?:comment|pourquoi|quand|où|qui)\s+(.+)\?", r"(?:explique|décris|détaille)\s+(.+)", r"(?:qu'est-ce que|c'est quoi)\s+(.+)", ], IntentType.HELP: [ r"(?:aide|help|assistance|sos)", r"(?:comment ça marche|comment utiliser)", r"\?{2,}", ], IntentType.STATUS: [ r"(?:statut|status|état|où en est)", r"(?:ça avance|progression|progress)", r"(?:terminé|fini|done)\s*\?", ], IntentType.CANCEL: [ r"(?:annule|stop|arrête|cancel|abort)", r"(?:laisse tomber|oublie)", ], IntentType.HISTORY: [ r"(?:historique|history|dernières?\s+commandes?)", r"(?:qu'est-ce que j'ai fait|actions? précédentes?)", ], IntentType.CONFIRM: [ r"^(?:oui|yes|ok|d'accord|go|lance|confirme|valide|c'est bon)$", r"^(?:vas-y|fais-le|proceed|continue)$", ], IntentType.DENY: [ r"^(?:non|no|annule|stop|pas ça|mauvais)$", r"^(?:arrête|ne fais pas|cancel)$", ], } # Patterns pour l'extraction d'entités ENTITY_PATTERNS = { "client": [ r"(?:client|société|entreprise)\s+([A-Z][a-zA-Z0-9\s]+)", r"pour\s+([A-Z][a-zA-Z0-9\s]+?)(?:\s|$|,)", ], "date": [ r"(\d{1,2}[/-]\d{1,2}[/-]\d{2,4})", r"(aujourd'hui|demain|hier|lundi|mardi|mercredi|jeudi|vendredi)", ], "format": [ r"(?:en|format|type)\s+(pdf|excel|csv|json|xml)", ], "amount": [ r"(\d+(?:[.,]\d+)?)\s*(?:€|euros?|dollars?|\$)", ], "range": [ r"de\s+([A-Za-z])\s+à\s+([A-Za-z])", r"(\d+)\s*(?:-|à|to)\s*(\d+)", ], } def __init__( self, use_llm: bool = False, llm_endpoint: str = "http://localhost:11434", llm_model: str = "qwen2.5:7b" ): """ Initialiser le parseur d'intentions. Args: use_llm: Utiliser un LLM pour l'analyse (optionnel) llm_endpoint: URL de l'endpoint Ollama llm_model: Modèle Ollama à utiliser """ self.use_llm = use_llm self.llm_endpoint = llm_endpoint self.llm_model = llm_model self.llm_available = False self._workflows_cache: List[Dict[str, Any]] = [] if use_llm: self._check_llm_availability() def set_workflows(self, workflows: List[Dict[str, Any]]): """ Injecter la liste des workflows disponibles pour enrichir le contexte LLM. Args: workflows: Liste de dict avec 'name', 'description', 'tags' """ self._workflows_cache = workflows logger.info(f"IntentParser: {len(workflows)} workflows injectés") def _check_llm_availability(self) -> bool: """Vérifier si le LLM est disponible.""" try: import requests response = requests.get(f"{self.llm_endpoint}/api/tags", timeout=2) self.llm_available = response.status_code == 200 if self.llm_available: logger.info("✓ LLM disponible pour IntentParser") return self.llm_available except Exception as e: logger.warning(f"LLM non disponible: {e}") self.llm_available = False return False def parse(self, query: str, context: Optional[Dict[str, Any]] = None) -> ParsedIntent: """ Parser une requête utilisateur. Args: query: La requête en langage naturel context: Contexte de la conversation (optionnel) Returns: ParsedIntent avec l'intention détectée """ query = query.strip() if not query: return ParsedIntent( intent_type=IntentType.UNKNOWN, confidence=0.0, raw_query=query ) # Normaliser la requête normalized = self._normalize_query(query) # 1. Détecter l'intention par règles intent_type, rule_confidence = self._detect_intent_by_rules(normalized) # 2. Extraire les entités entities = self._extract_entities(query) # 3. Extraire le hint de workflow workflow_hint = self._extract_workflow_hint(normalized, intent_type) # 4. Construire les paramètres depuis les entités parameters = self._entities_to_parameters(entities) # 5. Si le LLM est disponible et la confiance est basse, utiliser le LLM if self.use_llm and self.llm_available and rule_confidence < 0.7: llm_result = self._parse_with_llm(query, context) if llm_result and llm_result.confidence > rule_confidence: return llm_result # 6. Vérifier si une clarification est nécessaire clarification_needed, clarification_question = self._check_clarification_needed( intent_type, parameters, workflow_hint ) return ParsedIntent( intent_type=intent_type, confidence=rule_confidence, raw_query=query, workflow_hint=workflow_hint, parameters=parameters, entities=entities, clarification_needed=clarification_needed, clarification_question=clarification_question ) def _normalize_query(self, query: str) -> str: """Normaliser une requête pour le matching.""" # Convertir en minuscules normalized = query.lower() # Supprimer la ponctuation excessive normalized = re.sub(r'[!.]+$', '', normalized) # Normaliser les espaces normalized = re.sub(r'\s+', ' ', normalized).strip() return normalized def _detect_intent_by_rules(self, query: str) -> Tuple[IntentType, float]: """Détecter l'intention par matching de patterns.""" best_intent = IntentType.UNKNOWN best_confidence = 0.0 for intent_type, patterns in self.INTENT_PATTERNS.items(): for pattern in patterns: match = re.search(pattern, query, re.IGNORECASE) if match: # Calculer la confiance basée sur la qualité du match match_length = len(match.group(0)) query_length = len(query) confidence = min(0.9, 0.5 + (match_length / query_length) * 0.4) if confidence > best_confidence: best_confidence = confidence best_intent = intent_type # Si aucune intention trouvée mais la requête ressemble à une commande if best_intent == IntentType.UNKNOWN and len(query.split()) >= 2: # Supposer que c'est une demande d'exécution best_intent = IntentType.EXECUTE best_confidence = 0.4 return best_intent, best_confidence def _extract_entities(self, query: str) -> List[Dict[str, Any]]: """Extraire les entités nommées de la requête.""" entities = [] for entity_type, patterns in self.ENTITY_PATTERNS.items(): for pattern in patterns: matches = re.finditer(pattern, query, re.IGNORECASE) for match in matches: groups = match.groups() value = groups[0] if len(groups) == 1 else groups entities.append({ "type": entity_type, "value": value, "start": match.start(), "end": match.end(), "text": match.group(0) }) return entities def _extract_workflow_hint(self, query: str, intent_type: IntentType) -> Optional[str]: """Extraire un indice sur le workflow demandé.""" if intent_type not in [IntentType.EXECUTE, IntentType.QUERY]: return None # Supprimer les mots-clés d'intention hint = query for pattern in self.INTENT_PATTERNS.get(intent_type, []): match = re.search(pattern, query, re.IGNORECASE) if match and match.groups(): hint = match.group(1) break # Nettoyer le hint hint = re.sub(r'^\s*(le|la|les|un|une|des)\s+', '', hint) hint = hint.strip() return hint if hint else None def _entities_to_parameters(self, entities: List[Dict[str, Any]]) -> Dict[str, Any]: """Convertir les entités en paramètres.""" parameters = {} for entity in entities: entity_type = entity["type"] value = entity["value"] # Gérer les plages if entity_type == "range" and isinstance(value, tuple): parameters["start"] = value[0] parameters["end"] = value[1] else: # Si le paramètre existe déjà, le convertir en liste if entity_type in parameters: if isinstance(parameters[entity_type], list): parameters[entity_type].append(value) else: parameters[entity_type] = [parameters[entity_type], value] else: parameters[entity_type] = value return parameters def _check_clarification_needed( self, intent_type: IntentType, parameters: Dict[str, Any], workflow_hint: Optional[str] ) -> Tuple[bool, Optional[str]]: """Vérifier si une clarification est nécessaire.""" if intent_type == IntentType.EXECUTE: # Si pas de hint de workflow, demander clarification if not workflow_hint: return True, "Quel workflow souhaitez-vous exécuter ?" # Si le hint est trop vague if len(workflow_hint.split()) <= 1: return True, f"Pouvez-vous préciser ce que vous entendez par '{workflow_hint}' ?" return False, None def _parse_with_llm(self, query: str, context: Optional[Dict[str, Any]]) -> Optional[ParsedIntent]: """Utiliser le LLM pour parser la requête.""" try: import requests # Construire le contexte des workflows disponibles workflows_context = "" if self._workflows_cache: workflow_names = [w.get("name", "") for w in self._workflows_cache[:15]] workflows_context = f"\nWorkflows disponibles: {', '.join(workflow_names)}" prompt = f"""Tu es un assistant RPA. Analyse cette requête utilisateur. REQUÊTE: "{query}" {workflows_context} {f"Contexte conversation: {json.dumps(context, ensure_ascii=False)}" if context else ""} INTENTIONS POSSIBLES: - execute: l'utilisateur veut lancer/exécuter un workflow - list: l'utilisateur veut voir les workflows disponibles (mots-clés: liste, quels, workflows, disponibles, montrer) - query: l'utilisateur pose une question sur un workflow - status: l'utilisateur demande le statut d'exécution - cancel: l'utilisateur veut annuler - history: l'utilisateur veut voir l'historique - help: l'utilisateur demande de l'aide - confirm: l'utilisateur confirme (oui, ok, go) - deny: l'utilisateur refuse (non, annule) - unknown: impossible à déterminer Réponds UNIQUEMENT en JSON valide (pas de texte avant/après): {{"intent": "...", "confidence": 0.0-1.0, "workflow_hint": "...", "parameters": {{}}, "clarification_needed": false, "clarification_question": null}}""" response = requests.post( f"{self.llm_endpoint}/api/generate", json={ "model": self.llm_model, "prompt": prompt, "stream": False, "options": { "temperature": 0.1, "num_predict": 200 } }, timeout=15 ) if response.status_code == 200: result = response.json().get("response", "").strip() logger.debug(f"LLM response: {result[:200]}") # Extraire le JSON de la réponse (supporte JSON imbriqué) json_match = re.search(r'\{.*\}', result, re.DOTALL) if json_match: try: data = json.loads(json_match.group(0)) except json.JSONDecodeError: # Fallback: essayer de parser un JSON simple simple_match = re.search(r'\{[^{}]+\}', result) if simple_match: data = json.loads(simple_match.group(0)) else: return None intent_str = data.get("intent", "unknown") try: intent_type = IntentType(intent_str) except ValueError: intent_type = IntentType.UNKNOWN confidence = float(data.get("confidence", 0.5)) # Boost confidence for LLM results confidence = min(0.95, confidence + 0.1) return ParsedIntent( intent_type=intent_type, confidence=confidence, raw_query=query, workflow_hint=data.get("workflow_hint"), parameters=data.get("parameters", {}), entities=[], clarification_needed=data.get("clarification_needed", False), clarification_question=data.get("clarification_question") ) except requests.exceptions.Timeout: logger.warning("LLM timeout - falling back to rules") except Exception as e: logger.warning(f"LLM parsing failed: {e}") return None # Singleton pour utilisation globale _intent_parser: Optional[IntentParser] = None def get_intent_parser( use_llm: bool = False, llm_model: str = "qwen2.5:7b", llm_endpoint: str = "http://localhost:11434" ) -> IntentParser: """ Obtenir l'instance globale du parseur d'intentions. Args: use_llm: Activer le LLM (Ollama) llm_model: Modèle à utiliser (qwen2.5:7b par défaut) llm_endpoint: URL de l'endpoint Ollama """ global _intent_parser if _intent_parser is None: _intent_parser = IntentParser( use_llm=use_llm, llm_endpoint=llm_endpoint, llm_model=llm_model ) elif use_llm and not _intent_parser.use_llm: # Réactiver le LLM si demandé _intent_parser.use_llm = True _intent_parser.llm_model = llm_model _intent_parser._check_llm_availability() return _intent_parser def reset_intent_parser(): """Réinitialiser le singleton (pour tests).""" global _intent_parser _intent_parser = None if __name__ == "__main__": # Tests rapides parser = IntentParser(use_llm=False) test_queries = [ "facturer le client Acme", "lance le workflow de facturation", "quels workflows sont disponibles ?", "aide", "oui", "annule", "statut", "exporter le rapport en PDF pour Client ABC", "créer une facture de 1500€ pour Société XYZ", "facturer les clients de A à Z", ] print("=== Tests IntentParser ===\n") for query in test_queries: result = parser.parse(query) print(f"Query: {query}") print(f" Intent: {result.intent_type.value} ({result.confidence:.2f})") print(f" Workflow: {result.workflow_hint}") print(f" Params: {result.parameters}") if result.clarification_needed: print(f" Clarification: {result.clarification_question}") print()