Dom/rpa_vision_v3
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
7ea5d6b992b50b74631932ece9b3e89478bba610
rpa_vision_v3/agent_chat/intent_parser.py
Dom 848184db5c feat(agent_chat): Activer intégration LLM Ollama pour parsing intelligent 
- Activer use_llm=True par défaut dans app.py
- Améliorer le prompt LLM avec contexte des workflows disponibles
- Ajouter endpoints /api/llm/status et /api/llm/model pour configuration
- Permettre injection dynamique des workflows dans IntentParser
- Supporter changement de modèle à chaud

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-16 18:04:07 +01:00

525 lines
19 KiB
Python
Raw Blame History

#!/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()
Reference in New Issue View Git Blame Copy Permalink