8f31ba95d3
- IntentParser: ajout pattern "expression" pour capturer 5+2, 100*3, etc.
- demo_calculator.json: text "${expression}=" avec default "2+2"
→ l'utilisateur peut dire "calcule 5+2" et le paramètre est injecté
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
529 lines
19 KiB
Python
529 lines
19 KiB
Python
#!/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+)",
|
||
],
|
||
"expression": [
|
||
# Expressions mathématiques : 5+2, 100*3, 12/4, 7-3, 2.5+3.1
|
||
r"(\d+(?:[.,]\d+)?\s*[+\-*/x×÷]\s*\d+(?:[.,]\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()
|