#!/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 GREETING = "greeting" # Salutation 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 DATA_IMPORT = "data_import" # Importer des données (Excel, CSV) SMALL_TALK = "small_talk" # Conversation informelle (merci, café, ça va...) 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.DATA_IMPORT: [ # Import de fichiers Excel/CSV r"(?:importe|charge|lis|lire)\s+(?:le\s+fichier\s+|les\s+(?:données|feuilles)\s+(?:de|du|excel\s+du)\s+)?(.+\.xlsx?)\b", r"(?:importe|charge|lis|lire)\s+(?:le\s+fichier\s+|les\s+(?:données|feuilles)\s+(?:de|du|excel\s+du)\s+)?(.+\.csv)\b", r"(?:importe|charge|lis|lire)\s+(?:le\s+fichier\s+)?excel\s+(.+)", r"(?:importe|charge|lis|lire)\s+(?:les\s+)?(?:feuilles?\s+)?excel\s+(?:du\s+dossier\s+|de\s+)(.+)", r"(?:crée?|créer?)\s+une?\s+table\s+(?:à\s+partir\s+d[eu]'?\s*)(.+\.xlsx?)\b", # Lister les tables r"(?:montre|liste|affiche|voir)\s*(?:-moi\s+)?(?:les\s+)?tables?\b", r"(?:quelles?\s+)?tables?\s+(?:sont\s+)?(?:disponibles?|dans\s+la\s+base)", r"liste\s+(?:des?\s+)?tables?\s+(?:de\s+)?(?:la\s+)?(?:base)?", # Infos sur une table r"(?:combien\s+de\s+lignes?\s+(?:dans|pour)\s+(?:la\s+)?table\s+)(\w+)", r"(?:info|détails?|describe?|colonnes?|structure)\s+(?:de\s+)?(?:la\s+)?table\s+(\w+)", ], IntentType.EXECUTE: [ # Verbes d'action explicites r"(?:lance[rz]?|exécute[rz]?|démarre[rz]?|fai[st]|run|start|execute)\s+(.+)", r"(?:je veux|je voudrais|peux-tu|pouvez-vous)\s+(.+)", r"(?:facturer?|créer?|générer?|exporter?)\s+(.+)", r"^(.+)\s+(?:maintenant|tout de suite|svp|stp)$", # Langage humain — demande de replay r"(?:refai[st](?:es)?|refaire|recommence[rz]?|rejoue[rz]?)\s+(?:la\s+)?(?:tâche\s+)?(.+)", # Gestes courants (UI actions) — doivent rester EXECUTE r"(?:ferme[rz]?|ouvr[eir]+[sz]?|clique[rz]?|sélectionne[rz]?|coche[rz]?|décoche[rz]?)\s+(.+)", r"(?:copie[rz]?|colle[rz]?|coupe[rz]?|supprime[rz]?|efface[rz]?)\s+(.+)", r"(?:tape[rz]?|écri[rstv]+[sz]?|saisi[rstv]*[sz]?|rempli[rstv]*[sz]?|entre[rz]?)\s+(.+)", r"(?:scroll(?:e[rz]?)?|défile[rz]?|fait(?:es)?\s+défiler)\s*(.+)?", r"(?:glisse[rz]?|drag(?:ue)?[rz]?|déplace[rz]?|bouge[rz]?)\s+(.+)", r"(?:double[- ]?clique[rz]?|clic\s+droit)\s+(.+)?", r"(?:enregistre[rz]?|sauvegarde[rz]?|save)\s+(.+)?", r"(?:imprime[rz]?|print)\s+(.+)?", r"(?:envoie[rz]?|send|mail(?:e[rz]?)?|transmet[sz]?)\s+(.+)", r"(?:télécharge[rz]?|download|upload)\s+(.+)?", r"(?:actualise[rz]?|rafraîchi[rstv]*[sz]?|refresh|recharge[rz]?)\s*(.+)?", r"(?:valide[rz]?|confirme[rz]?|soumets?|submit)\s+(.+)", r"(?:connecte[rz]?|login|log\s*in|sign\s*in)\s*(.+)?", r"(?:déconnecte[rz]?|logout|log\s*out|sign\s*out)\s*(.+)?", # Raccourcis clavier r"(?:ctrl|alt|shift|maj)\s*\+\s*\w+", # Langage humain — demande d'apprentissage (déclenche l'enregistrement) r"(?:apprends|apprenez)[- ]moi\s+(.+)", ], IntentType.LIST: [ r"(?:liste|montre|affiche|quels?\s+sont)\s+(?:les\s+|des\s+)?(?:workflows?|tâches?|processus|automatisations?)", r"(?:quels?|quelles?)\s+(?:workflows?|tâches?|processus|automatisations?)", r"liste\s+des\s+(?:workflows?|tâches?)", r"(?:workflows?|tâches?|processus)\s+disponibles?", r"(?:voir|afficher)\s+(?:les\s+|tous\s+les\s+|mes\s+)?(?:workflows?|tâches?)", # Langage humain — demande de liste r"(?:qu'est-ce que\s+(?:tu|vous)\s+sai[st]\s+faire)", r"(?:que\s+sai[st]-(?:tu|vous)\s+faire)", r"mes\s+tâches?", ], # SMALL_TALK doit être AVANT QUERY pour que "qui es-tu" ne soit pas # capturé par le pattern générique "qui + ..." de QUERY IntentType.SMALL_TALK: [ # Remerciements r"^(?:merci|thanks?|thx|super|génial|parfait|cool|nickel|impec|impeccable|excellent|formidable)(?:\s.*)?$", # Adieux r"^(?:au revoir|à plus|bye|bonne nuit|à bientôt|à demain|ciao|tchao|tchuss|adieu)(?:\s.*)?$", # Compliments r"^(?:bien joué|bravo|top|chapeau|impressionnant|pas mal|bien fait|beau travail|good job|nice|trop bien|magnifique)(?:\s.*)?$", # Mécontentement r"^(?:c'est nul|nul|pas bien|pas top|pas ouf|bof|mauvais|moche|horrible|catastrophe|c'est pas bon|ça craint|erreur|bug|naze|pourri)(?:\s.*)?$", # Humour / boissons / nourriture / détente r"(?:une? (?:café|coca|thé|chocolat|verre|jus|bière|apéro|croissant|gâteau|bonbon|pause|pizza|glace)|café|coca|thé|chocolat|fais-moi rire|blague|raconte.+blague|drôle|rigol[eo]|mdr|lol|haha|ptdr|xd|😂|🤣|j'ai faim|j'ai soif|pause|il fait (?:chaud|froid|beau)|je suis (?:fatigué|crevé|motivé|content)|la flemme|trop bien|trop cool|vive .+|c'est la vie|oh là là|waouh|wow)", # Identité — qui es-tu ? r"(?:qui es[- ]tu|t'es qui|comment tu t'appelles|c'est quoi ton (?:nom|prénom)|t'es quoi|vous êtes qui|tu es quoi|tu t'appelles comment)", # Sentiments — ça va ? r"(?:ça va|comment (?:ça |tu |vous )?va[st]?|comment allez[- ]vous|tu vas bien|la forme|en forme|et toi|et vous)", ], IntentType.QUERY: [ # Questions directes avec mots interrogatifs r"(?:comment|pourquoi|quand|où|qui)\s+(.+)\??", r"(?:explique|décris|détaille)\s+(.+)", r"(?:qu'est-ce que|c'est quoi)\s+(.+)", # Questions avec "quel/quelle/quels/quelles" (exclure workflows → LIST) r"(?:quels?|quelles?)\s+(?!workflows?|processus|automatisations?)(.+)\??", # "quoi" comme question (pas une commande, pas "quoi faire" = HELP) r"^(?:c'est\s+)?quoi\s+(?!faire)(.+)\??$", r"^quoi\s*\?+$", # Questions indirectes r"(?:dis[- ]moi|raconte|informe[- ]moi)\s+(.+)", r"(?:je\s+(?:me\s+)?demande|je\s+(?:ne\s+)?comprends?\s+pas)\s+(.+)", ], IntentType.HELP: [ r"^(?:aide|help|assistance|sos)$", r"comment ça (?:marche|fonctionne)\s*\??", r"comment (?:utiliser|ça s'utilise|on fait)\s*\??", r"\?{2,}", # "que peux-tu faire", "quoi faire" = demande d'aide r"(?:qu'est-ce que|que)\s+(?:je peux|tu peux|vous pouvez)\s+faire", r"^quoi\s+faire\s*\??$", r"(?:que\s+)?(?:puis-je|peux-tu|pouvez-vous|peut-on)\s+faire\s*\??", r"(?:besoin\s+d'aide|j'ai\s+besoin\s+d'aide)", ], IntentType.GREETING: [ r"^(?:bonjour|bonsoir|salut|hello|hi|hey|coucou|yo|wesh)(?:\s.*)?$", r"^(?:bonne?\s+(?:journée|soirée|nuit|matinée))$", ], IntentType.STATUS: [ r"(?:statut|status|état|où en est)", r"(?:ça avance|progression|progress)", r"(?:terminé|fini|done)\s*\?", ], IntentType.CANCEL: [ r"(?:annule[rz]?|stop|arrête[rz]?|cancel|abort)", r"(?:laisse[rz]?\s+tomber|oublie[rz]?)", # Langage humain — stop courant r"^(?:arrêtez|stoppe[rz]?)$", ], 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)$", ], } # Verbes d'action reconnus pour le fallback EXECUTE # Si aucun pattern ne matche, on vérifie la présence d'un de ces verbes # avant de classifier en EXECUTE ACTION_VERBS = { # Actions de workflow/exécution "lance", "lancer", "exécute", "exécuter", "démarre", "démarrer", "fait", "fais", "run", "start", "execute", # Actions métier "facture", "facturer", "crée", "créer", "génère", "générer", "exporte", "exporter", "importe", "importer", # Actions UI / gestes "ferme", "fermer", "ouvre", "ouvrir", "clique", "cliquer", "sélectionne", "sélectionner", "coche", "cocher", "décoche", "décocher", "copie", "copier", "colle", "coller", "coupe", "couper", "supprime", "supprimer", "efface", "effacer", "tape", "taper", "écris", "écrire", "saisis", "saisir", "remplis", "remplir", "entre", "entrer", "scroll", "scroller", "défile", "défiler", "glisse", "glisser", "déplace", "déplacer", "drag", "enregistre", "enregistrer", "sauvegarde", "sauvegarder", "save", "imprime", "imprimer", "print", "envoie", "envoyer", "send", "transmet", "transmettre", "télécharge", "télécharger", "download", "upload", "actualise", "actualiser", "rafraîchis", "rafraîchir", "refresh", "valide", "valider", "confirme", "confirmer", "soumets", "soumettre", "connecte", "connecter", "déconnecte", "déconnecter", "login", "logout", } # 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+)", ], "expression": [ # Expressions mathématiques : 5+2, 100*3, 12/4, 7-3, 2.5+3.1 r"(\d+(?:[.,]\d+)?\s*[+\-*/x×÷]\s*\d+(?:[.,]\d+)?)", ], "file_path": [ # Chemins Windows : C:\data\fichier.xlsx r"([A-Za-z]:\\[^\s,]+\.(?:xlsx?|csv))", # Chemins Unix : /data/fichier.xlsx r"(/[^\s,]+\.(?:xlsx?|csv))", # Noms de fichier simples : patients.xlsx r"(?:^|\s)([\w\-\.]+\.(?:xlsx?|csv))(?:\s|$)", ], "folder_path": [ # Dossiers Windows : C:\data\imports r"(?:dossier|répertoire|dir|directory)\s+([A-Za-z]:\\[^\s,]+)", r"([A-Za-z]:\\[^\s,]+)(?:\s|$)", # Dossiers Unix : /data/imports r"(?:dossier|répertoire|dir|directory)\s+(/[^\s,]+)", ], "table_name": [ # Noms de table (exclure les mots courants comme "à", "de", "la") r"(?:table|la\s+table)\s+['\"]?(\w{2,})['\"]?", ], } 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) # 4b. Enrichir les paramètres DATA_IMPORT avec l'action et le chemin if intent_type == IntentType.DATA_IMPORT: parameters = self._enrich_data_import_params(normalized, query, 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 _enrich_data_import_params( self, normalized: str, raw_query: str, parameters: Dict[str, Any], entities: List[Dict[str, Any]], ) -> Dict[str, Any]: """Enrichir les paramètres pour une intention DATA_IMPORT. Détermine l'action (import_file, import_folder, list_tables, table_info) et extrait le chemin de fichier / nom de table. """ # Déterminer l'action list_patterns = [ r"(?:montre|liste|affiche|voir)\s*(?:-moi\s+)?(?:les\s+)?tables?", r"(?:quelles?\s+)?tables?\s+(?:sont\s+)?(?:disponibles?|dans)", r"liste\s+(?:des?\s+)?tables?", ] info_patterns = [ r"combien\s+de\s+lignes?", r"(?:info|détails?|describe?|colonnes?|structure)\s+(?:de\s+)?(?:la\s+)?table", ] folder_patterns = [ r"(?:feuilles?\s+excel|fichiers?\s+excel)\s+(?:du|de)\s+(?:dossier|répertoire)", r"(?:importe|charge|lis)\s+(?:les\s+)?(?:feuilles?\s+)?excel\s+(?:du\s+dossier|de\s+)", ] action = "import_file" # Par défaut for pat in list_patterns: if re.search(pat, normalized, re.IGNORECASE): action = "list_tables" break if action == "import_file": for pat in info_patterns: if re.search(pat, normalized, re.IGNORECASE): action = "table_info" break if action == "import_file": for pat in folder_patterns: if re.search(pat, normalized, re.IGNORECASE): action = "import_folder" break parameters["action"] = action # Extraire le chemin de fichier depuis les entités for entity in entities: if entity["type"] == "file_path" and "file_path" not in parameters: parameters["file_path"] = entity["value"] elif entity["type"] == "folder_path" and "folder_path" not in parameters: parameters["folder_path"] = entity["value"] elif entity["type"] == "table_name" and "table_name" not in parameters: parameters["table_name"] = entity["value"] # Fallback : extraire un chemin de fichier depuis la requête brute if "file_path" not in parameters and action == "import_file": # Chercher un .xlsx/.xls/.csv dans la requête brute (supporte les chemins Windows) fp_match = re.search( r'([A-Za-z]:\\[^\s,]+\.(?:xlsx?|csv)|/[^\s,]+\.(?:xlsx?|csv)|[\w\-\.]+\.(?:xlsx?|csv))', raw_query, re.IGNORECASE, ) if fp_match: parameters["file_path"] = fp_match.group(1) # Extraire table_name pour table_info depuis la requête if action == "table_info" and "table_name" not in parameters: tm = re.search(r"table\s+['\"]?(\w+)['\"]?", normalized, re.IGNORECASE) if tm: parameters["table_name"] = tm.group(1) return parameters def _normalize_query(self, query: str) -> str: """Normaliser une requête pour le matching.""" # Convertir en minuscules normalized = query.lower() # Supprimer la ponctuation finale 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 # Fallback durci : ne classifier en EXECUTE que si un verbe d'action est présent if best_intent == IntentType.UNKNOWN and len(query.split()) >= 2: words = query.lower().split() # Vérifier si au moins un mot est un verbe d'action connu has_action_verb = any(word in self.ACTION_VERBS for word in words) if has_action_verb: best_intent = IntentType.EXECUTE best_confidence = 0.40 else: # Pas de verbe d'action reconnu → demander clarification best_intent = IntentType.CLARIFY best_confidence = 0.30 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 tâche, demander clarification if not workflow_hint: return True, "Quelle tâche souhaitez-vous lancer ?" # 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 Léa, une assistante chaleureuse. 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/refaire une tâche ou une action UI (geste). Inclut "apprends-moi", "refais la tâche", "lance" - list: l'utilisateur veut voir les tâches disponibles (mots-clés: liste, quels, tâches, qu'est-ce que tu sais faire, mes tâches) - query: l'utilisateur pose une question (comment, pourquoi, c'est quoi, quel) - status: l'utilisateur demande le statut d'exécution - cancel: l'utilisateur veut arrêter/annuler (arrête, stop, annule) - history: l'utilisateur veut voir l'historique - help: l'utilisateur demande de l'aide ou ce qu'il peut faire - greeting: l'utilisateur dit bonjour/salut/hello - confirm: l'utilisateur confirme (oui, ok, go) - deny: l'utilisateur refuse (non, annule) - small_talk: conversation informelle (merci, café, ça va, qui es-tu, bravo, c'est nul) - 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 = [ # EXECUTE — actions explicites "facturer le client Acme", "lance le workflow de facturation", "exporter le rapport en PDF pour Client ABC", "créer une facture de 1500€ pour Société XYZ", "facturer les clients de A à Z", # EXECUTE — gestes UI "ferme la fenêtre", "ouvre un nouvel onglet", "copier le texte", "lance la facturation", # LIST "quels workflows sont disponibles ?", "liste des workflows", # QUERY — questions "comment ça marche ?", "c'est quoi ce workflow", "pourquoi ce processus est lent ?", # HELP "aide", "quoi faire ?", "que peux-tu faire ?", # GREETING "bonjour", "salut", # Confirmations / annulations "oui", "annule", "statut", # SMALL_TALK — conversation informelle "merci", "un café", "ça va ?", "qui es-tu ?", "c'est nul", "bravo", "au revoir", "t'es qui", # Fallback — ne doit PAS être EXECUTE "blah blah test", ] 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()