diff --git a/agent_chat/__init__.py b/agent_chat/__init__.py index 7c09eb3a1..33ed4717f 100644 --- a/agent_chat/__init__.py +++ b/agent_chat/__init__.py @@ -1,5 +1,69 @@ """ -RPA Vision V3 - Command Interface +RPA Vision V3 - Agent Chat -Interface web légère pour communiquer avec le système RPA. +Interface conversationnelle pour communiquer avec le système RPA. +Style "Spotlight/Alfred" - minimaliste et efficace. + +Composants: +- IntentParser: Compréhension des intentions utilisateur +- ConfirmationLoop: Validation avant actions critiques +- ResponseGenerator: Réponses en langage naturel +- ConversationManager: Contexte multi-tour + +Auteur: Dom - Janvier 2026 """ + +from .intent_parser import ( + IntentParser, + IntentType, + ParsedIntent, + get_intent_parser +) + +from .confirmation import ( + ConfirmationLoop, + PendingConfirmation, + ConfirmationStatus, + RiskLevel, + get_confirmation_loop +) + +from .response_generator import ( + ResponseGenerator, + ResponseTone, + GeneratedResponse, + get_response_generator +) + +from .conversation_manager import ( + ConversationManager, + ConversationSession, + ConversationTurn, + ConversationContext, + get_conversation_manager +) + +__all__ = [ + # Intent Parser + "IntentParser", + "IntentType", + "ParsedIntent", + "get_intent_parser", + # Confirmation + "ConfirmationLoop", + "PendingConfirmation", + "ConfirmationStatus", + "RiskLevel", + "get_confirmation_loop", + # Response Generator + "ResponseGenerator", + "ResponseTone", + "GeneratedResponse", + "get_response_generator", + # Conversation Manager + "ConversationManager", + "ConversationSession", + "ConversationTurn", + "ConversationContext", + "get_conversation_manager", +] diff --git a/agent_chat/app.py b/agent_chat/app.py index 6af229638..db4e41968 100644 --- a/agent_chat/app.py +++ b/agent_chat/app.py @@ -1,14 +1,22 @@ #!/usr/bin/env python3 """ -RPA Vision V3 - Interface de Commande Web +RPA Vision V3 - Agent Chat -Interface web légère pour communiquer avec le système RPA. +Interface conversationnelle pour communiquer avec le système RPA. Style "Spotlight/Alfred" - minimaliste et efficace. +Composants intégrés: +- IntentParser: Compréhension des intentions utilisateur +- ConfirmationLoop: Validation avant actions critiques +- ResponseGenerator: Réponses en langage naturel +- ConversationManager: Contexte multi-tour + Usage: - python command_interface/app.py - + python agent_chat/app.py + Puis ouvrir: http://localhost:5002 + +Auteur: Dom - Janvier 2026 """ import asyncio @@ -27,6 +35,12 @@ sys.path.insert(0, str(Path(__file__).parent.parent)) from core.workflow import SemanticMatcher, VariableManager +# Import des composants conversationnels +from .intent_parser import IntentParser, IntentType, get_intent_parser +from .confirmation import ConfirmationLoop, ConfirmationStatus, RiskLevel, get_confirmation_loop +from .response_generator import ResponseGenerator, get_response_generator +from .conversation_manager import ConversationManager, get_conversation_manager + # GPU Resource Manager (optional) try: from core.gpu import get_gpu_resource_manager, ExecutionMode @@ -44,6 +58,11 @@ socketio = SocketIO(app, cors_allowed_origins="*") # Global state matcher: Optional[SemanticMatcher] = None gpu_manager = None +intent_parser: Optional[IntentParser] = None +confirmation_loop: Optional[ConfirmationLoop] = None +response_generator: Optional[ResponseGenerator] = None +conversation_manager: Optional[ConversationManager] = None + execution_status = { "running": False, "workflow": None, @@ -57,7 +76,8 @@ command_history: List[Dict[str, Any]] = [] def init_system(): """Initialiser tous les composants du système.""" global matcher, gpu_manager - + global intent_parser, confirmation_loop, response_generator, conversation_manager + # 1. SemanticMatcher try: matcher = SemanticMatcher("data/workflows") @@ -65,7 +85,7 @@ def init_system(): except Exception as e: logger.error(f"✗ SemanticMatcher: {e}") matcher = None - + # 2. GPU Resource Manager if GPU_AVAILABLE: try: @@ -75,6 +95,21 @@ def init_system(): logger.warning(f"⚠ GPU Resource Manager: {e}") gpu_manager = None + # 3. Composants conversationnels + try: + intent_parser = get_intent_parser(use_llm=False) # LLM optionnel + confirmation_loop = get_confirmation_loop() + response_generator = get_response_generator() + conversation_manager = get_conversation_manager() + logger.info("✓ Composants conversationnels initialisés") + except Exception as e: + logger.error(f"✗ Composants conversationnels: {e}") + # Fallback aux composants de base + intent_parser = IntentParser(use_llm=False) + confirmation_loop = ConfirmationLoop() + response_generator = ResponseGenerator() + conversation_manager = ConversationManager() + # ============================================================================= # Routes Web @@ -241,6 +276,192 @@ def api_history(): return jsonify({"history": command_history[-20:]}) +@app.route('/api/chat', methods=['POST']) +def api_chat(): + """ + Endpoint conversationnel principal. + + Utilise le flux complet: + 1. IntentParser: Analyse l'intention + 2. ConversationManager: Gère le contexte multi-tour + 3. ConfirmationLoop: Valide les actions sensibles + 4. ResponseGenerator: Génère la réponse + """ + data = request.json + message = data.get('message', '').strip() + session_id = data.get('session_id') + + if not message: + return jsonify({"error": "Message vide"}), 400 + + # 1. Obtenir ou créer la session + session = conversation_manager.get_or_create_session(session_id=session_id) + + # 2. Parser l'intention + intent = intent_parser.parse(message) + + # 3. Résoudre les références anaphoriques (ex: "le même", "celui-ci") + intent = conversation_manager.resolve_references(session, intent) + + # 4. Construire le contexte + context = conversation_manager.get_context_summary(session) + context["execution_status"] = execution_status + + # 5. Traiter selon le type d'intention + result = {} + action_taken = None + + if intent.intent_type == IntentType.CONFIRM: + # Confirmer une action en attente + pending = conversation_manager.get_pending_confirmation(session) + if pending: + confirmation_loop.confirm(pending.id) + conversation_manager.clear_pending_confirmation(session) + result = {"confirmed": True, "workflow": pending.workflow_name} + action_taken = "confirmed" + + # Lancer l'exécution + socketio.start_background_task( + execute_workflow_from_confirmation, pending, session.session_id + ) + else: + result = {"confirmed": False} + + elif intent.intent_type == IntentType.DENY: + # Refuser une action en attente + pending = conversation_manager.get_pending_confirmation(session) + if pending: + confirmation_loop.deny(pending.id) + conversation_manager.clear_pending_confirmation(session) + result = {"denied": True} + action_taken = "denied" + + elif intent.intent_type == IntentType.EXECUTE: + # Exécuter un workflow + if matcher and intent.workflow_hint: + match = matcher.find_workflow(intent.workflow_hint, min_confidence=0.2) + + if match: + # Évaluer le risque + risk = confirmation_loop.evaluate_risk( + match.workflow_name, + {**match.extracted_params, **intent.parameters} + ) + + if confirmation_loop.requires_confirmation(risk): + # Créer une demande de confirmation + conf = confirmation_loop.create_confirmation_request( + workflow_name=match.workflow_name, + parameters={**match.extracted_params, **intent.parameters}, + action_type="execute", + risk_level=risk + ) + conversation_manager.set_pending_confirmation(session, conf) + + # Générer la réponse de confirmation + response = response_generator.generate_confirmation_request(conf) + result = {"needs_confirmation": True, "confirmation": conf.to_dict()} + action_taken = "confirmation_requested" + + else: + # Exécuter directement + all_params = {**match.extracted_params, **intent.parameters} + result = { + "success": True, + "workflow": match.workflow_name, + "params": all_params, + "confidence": match.confidence + } + action_taken = "executed" + + socketio.start_background_task(execute_workflow, match, all_params) + else: + result = {"not_found": True, "query": intent.workflow_hint} + else: + result = {"error": "Pas de workflow spécifié"} + + elif intent.intent_type == IntentType.LIST: + # Lister les workflows + if matcher: + workflows = [ + {"name": wf.name, "description": wf.description} + for wf in matcher.get_all_workflows() + ] + result = {"workflows": workflows} + else: + result = {"workflows": []} + action_taken = "listed" + + elif intent.intent_type == IntentType.STATUS: + result = {"execution": execution_status} + action_taken = "status_checked" + + elif intent.intent_type == IntentType.CANCEL: + if execution_status.get("running"): + execution_status["running"] = False + execution_status["message"] = "Annulé" + result = {"cancelled": True} + else: + result = {"cancelled": False} + action_taken = "cancelled" + + elif intent.intent_type == IntentType.HISTORY: + result = {"history": command_history[-10:]} + action_taken = "history_shown" + + elif intent.intent_type == IntentType.HELP: + result = {} + action_taken = "help_shown" + + elif intent.clarification_needed: + result = {"clarification_needed": True} + action_taken = "clarification_requested" + + # 6. Générer la réponse (si pas déjà fait pour confirmation) + if action_taken != "confirmation_requested": + response = response_generator.generate(intent, context, result) + + # 7. Enregistrer le tour dans la conversation + conversation_manager.add_turn( + session=session, + user_message=message, + intent=intent, + response=response.message, + action_taken=action_taken, + result=result + ) + + # 8. Retourner la réponse + return jsonify({ + "session_id": session.session_id, + "intent": intent.to_dict(), + "response": response.to_dict(), + "result": result, + "context": { + "current_workflow": session.context.current_workflow, + "has_pending_confirmation": session.context.pending_confirmation is not None + } + }) + + +def execute_workflow_from_confirmation(confirmation, session_id): + """Exécuter un workflow après confirmation.""" + global execution_status + + if not matcher: + return + + # Trouver le workflow + match = matcher.find_workflow(confirmation.workflow_name, min_confidence=0.1) + if not match: + return + + # Utiliser les paramètres confirmés (ou modifiés) + params = confirmation.modified_parameters or confirmation.parameters + + execute_workflow(match, params) + + @app.route('/api/gpu/', methods=['POST']) def api_gpu_action(action): """Contrôler le GPU Resource Manager.""" diff --git a/agent_chat/confirmation.py b/agent_chat/confirmation.py new file mode 100644 index 000000000..b6a0168ec --- /dev/null +++ b/agent_chat/confirmation.py @@ -0,0 +1,408 @@ +#!/usr/bin/env python3 +""" +RPA Vision V3 - ConfirmationLoop +Système de confirmation avant les actions critiques. + +Ce module gère la validation utilisateur pour les actions sensibles : +- Exécution de workflows +- Actions destructives ou irréversibles +- Actions sur des données sensibles + +Auteur: Dom - Janvier 2026 +""" + +import logging +from dataclasses import dataclass, field +from enum import Enum +from typing import Dict, Any, List, Optional, Callable +from datetime import datetime, timedelta + +logger = logging.getLogger(__name__) + + +class RiskLevel(Enum): + """Niveaux de risque des actions.""" + LOW = "low" # Pas de confirmation requise + MEDIUM = "medium" # Confirmation simple + HIGH = "high" # Confirmation détaillée + CRITICAL = "critical" # Double confirmation + + +class ConfirmationStatus(Enum): + """Statuts de confirmation.""" + PENDING = "pending" + CONFIRMED = "confirmed" + DENIED = "denied" + EXPIRED = "expired" + MODIFIED = "modified" # Confirmé avec modifications + + +@dataclass +class PendingConfirmation: + """Une confirmation en attente.""" + id: str + action_type: str + workflow_name: str + parameters: Dict[str, Any] + risk_level: RiskLevel + created_at: datetime + expires_at: datetime + status: ConfirmationStatus = ConfirmationStatus.PENDING + confirmation_message: str = "" + details: List[str] = field(default_factory=list) + modified_parameters: Optional[Dict[str, Any]] = None + + def is_expired(self) -> bool: + return datetime.now() > self.expires_at + + def to_dict(self) -> Dict[str, Any]: + return { + "id": self.id, + "action_type": self.action_type, + "workflow_name": self.workflow_name, + "parameters": self.parameters, + "risk_level": self.risk_level.value, + "status": self.status.value, + "confirmation_message": self.confirmation_message, + "details": self.details, + "expires_in_seconds": max(0, (self.expires_at - datetime.now()).total_seconds()), + "modified_parameters": self.modified_parameters + } + + +class ConfirmationLoop: + """ + Gestionnaire de confirmations pour l'agent conversationnel. + + Gère le flux de confirmation pour les actions sensibles : + 1. Analyse du risque de l'action + 2. Génération du message de confirmation + 3. Attente de la réponse utilisateur + 4. Validation et exécution + """ + + # Configuration par niveau de risque + RISK_CONFIG = { + RiskLevel.LOW: { + "requires_confirmation": False, + "timeout_seconds": 0, + "message_template": "Exécution de {workflow_name}..." + }, + RiskLevel.MEDIUM: { + "requires_confirmation": True, + "timeout_seconds": 60, + "message_template": "Voulez-vous exécuter '{workflow_name}' ?" + }, + RiskLevel.HIGH: { + "requires_confirmation": True, + "timeout_seconds": 120, + "message_template": "⚠️ Action importante : '{workflow_name}'\n\nParamètres :\n{params_list}\n\nConfirmez-vous cette action ?" + }, + RiskLevel.CRITICAL: { + "requires_confirmation": True, + "timeout_seconds": 180, + "message_template": "🚨 ACTION CRITIQUE : '{workflow_name}'\n\n{details}\n\nCette action est irréversible. Tapez 'confirmer' pour continuer." + } + } + + # Mots-clés associés aux niveaux de risque + RISK_KEYWORDS = { + RiskLevel.CRITICAL: [ + "supprimer", "delete", "effacer", "purger", "reset", "formater", + "production", "live", "client", "facturation", "paiement" + ], + RiskLevel.HIGH: [ + "modifier", "update", "éditer", "changer", "migrer", "transfert", + "export", "envoyer", "publier", "synchroniser" + ], + RiskLevel.MEDIUM: [ + "créer", "create", "générer", "ajouter", "nouveau", "copier" + ] + } + + def __init__(self, default_timeout: int = 60): + """ + Initialiser le gestionnaire de confirmations. + + Args: + default_timeout: Timeout par défaut en secondes + """ + self.default_timeout = default_timeout + self.pending_confirmations: Dict[str, PendingConfirmation] = {} + self._confirmation_counter = 0 + + def evaluate_risk( + self, + workflow_name: str, + parameters: Dict[str, Any], + action_type: str = "execute" + ) -> RiskLevel: + """ + Évaluer le niveau de risque d'une action. + + Args: + workflow_name: Nom du workflow + parameters: Paramètres de l'action + action_type: Type d'action + + Returns: + RiskLevel correspondant + """ + text_to_analyze = f"{workflow_name} {action_type} {str(parameters)}".lower() + + # Vérifier les mots-clés par ordre de criticité + for risk_level in [RiskLevel.CRITICAL, RiskLevel.HIGH, RiskLevel.MEDIUM]: + for keyword in self.RISK_KEYWORDS.get(risk_level, []): + if keyword in text_to_analyze: + logger.debug(f"Risk {risk_level.value} detected by keyword: {keyword}") + return risk_level + + return RiskLevel.LOW + + def create_confirmation_request( + self, + workflow_name: str, + parameters: Dict[str, Any], + action_type: str = "execute", + risk_level: Optional[RiskLevel] = None, + custom_message: Optional[str] = None + ) -> PendingConfirmation: + """ + Créer une demande de confirmation. + + Args: + workflow_name: Nom du workflow + parameters: Paramètres de l'action + action_type: Type d'action + risk_level: Niveau de risque (auto-détecté si non fourni) + custom_message: Message personnalisé + + Returns: + PendingConfirmation créée + """ + # Auto-détecter le risque si non fourni + if risk_level is None: + risk_level = self.evaluate_risk(workflow_name, parameters, action_type) + + # Générer un ID unique + self._confirmation_counter += 1 + confirmation_id = f"conf_{self._confirmation_counter}_{datetime.now().strftime('%H%M%S')}" + + # Calculer le timeout + config = self.RISK_CONFIG[risk_level] + timeout = config["timeout_seconds"] or self.default_timeout + expires_at = datetime.now() + timedelta(seconds=timeout) + + # Générer le message de confirmation + if custom_message: + message = custom_message + else: + message = self._generate_confirmation_message( + workflow_name, parameters, risk_level + ) + + # Générer les détails + details = self._generate_details(workflow_name, parameters, risk_level) + + # Créer la confirmation + confirmation = PendingConfirmation( + id=confirmation_id, + action_type=action_type, + workflow_name=workflow_name, + parameters=parameters, + risk_level=risk_level, + created_at=datetime.now(), + expires_at=expires_at, + confirmation_message=message, + details=details + ) + + # Stocker si confirmation requise + if config["requires_confirmation"]: + self.pending_confirmations[confirmation_id] = confirmation + logger.info(f"Confirmation request created: {confirmation_id} ({risk_level.value})") + + return confirmation + + def _generate_confirmation_message( + self, + workflow_name: str, + parameters: Dict[str, Any], + risk_level: RiskLevel + ) -> str: + """Générer le message de confirmation.""" + config = self.RISK_CONFIG[risk_level] + template = config["message_template"] + + # Formater la liste des paramètres + params_list = "\n".join([f" • {k}: {v}" for k, v in parameters.items()]) + if not params_list: + params_list = " (aucun paramètre)" + + # Générer les détails pour les actions critiques + details = "" + if risk_level == RiskLevel.CRITICAL: + details = "Cette action peut affecter des données importantes." + + return template.format( + workflow_name=workflow_name, + params_list=params_list, + details=details + ) + + def _generate_details( + self, + workflow_name: str, + parameters: Dict[str, Any], + risk_level: RiskLevel + ) -> List[str]: + """Générer les détails de l'action.""" + details = [] + + # Détails basiques + details.append(f"Workflow: {workflow_name}") + + if parameters: + for key, value in parameters.items(): + details.append(f" {key}: {value}") + + # Avertissements selon le risque + if risk_level == RiskLevel.CRITICAL: + details.append("⚠️ Cette action est irréversible") + elif risk_level == RiskLevel.HIGH: + details.append("⚠️ Cette action modifiera des données") + + return details + + def confirm( + self, + confirmation_id: str, + modified_parameters: Optional[Dict[str, Any]] = None + ) -> Optional[PendingConfirmation]: + """ + Confirmer une action en attente. + + Args: + confirmation_id: ID de la confirmation + modified_parameters: Paramètres modifiés (optionnel) + + Returns: + La confirmation mise à jour, ou None si non trouvée + """ + confirmation = self.pending_confirmations.get(confirmation_id) + + if not confirmation: + logger.warning(f"Confirmation not found: {confirmation_id}") + return None + + if confirmation.is_expired(): + confirmation.status = ConfirmationStatus.EXPIRED + logger.info(f"Confirmation expired: {confirmation_id}") + return confirmation + + if modified_parameters: + confirmation.status = ConfirmationStatus.MODIFIED + confirmation.modified_parameters = modified_parameters + else: + confirmation.status = ConfirmationStatus.CONFIRMED + + logger.info(f"Confirmation confirmed: {confirmation_id}") + return confirmation + + def deny(self, confirmation_id: str) -> Optional[PendingConfirmation]: + """ + Refuser une action en attente. + + Args: + confirmation_id: ID de la confirmation + + Returns: + La confirmation mise à jour, ou None si non trouvée + """ + confirmation = self.pending_confirmations.get(confirmation_id) + + if not confirmation: + logger.warning(f"Confirmation not found: {confirmation_id}") + return None + + confirmation.status = ConfirmationStatus.DENIED + logger.info(f"Confirmation denied: {confirmation_id}") + return confirmation + + def get_pending(self, confirmation_id: Optional[str] = None) -> Optional[PendingConfirmation]: + """ + Obtenir une ou la dernière confirmation en attente. + + Args: + confirmation_id: ID spécifique (optionnel) + + Returns: + La confirmation ou None + """ + if confirmation_id: + return self.pending_confirmations.get(confirmation_id) + + # Retourner la dernière confirmation en attente non expirée + for conf in reversed(list(self.pending_confirmations.values())): + if conf.status == ConfirmationStatus.PENDING and not conf.is_expired(): + return conf + + return None + + def cleanup_expired(self) -> int: + """ + Nettoyer les confirmations expirées. + + Returns: + Nombre de confirmations nettoyées + """ + expired_ids = [ + conf_id for conf_id, conf in self.pending_confirmations.items() + if conf.is_expired() or conf.status != ConfirmationStatus.PENDING + ] + + for conf_id in expired_ids: + del self.pending_confirmations[conf_id] + + if expired_ids: + logger.info(f"Cleaned up {len(expired_ids)} confirmations") + + return len(expired_ids) + + def requires_confirmation(self, risk_level: RiskLevel) -> bool: + """Vérifier si un niveau de risque requiert une confirmation.""" + return self.RISK_CONFIG[risk_level]["requires_confirmation"] + + +# Singleton pour utilisation globale +_confirmation_loop: Optional[ConfirmationLoop] = None + +def get_confirmation_loop() -> ConfirmationLoop: + """Obtenir l'instance globale du gestionnaire de confirmations.""" + global _confirmation_loop + if _confirmation_loop is None: + _confirmation_loop = ConfirmationLoop() + return _confirmation_loop + + +if __name__ == "__main__": + # Tests rapides + loop = ConfirmationLoop() + + test_cases = [ + ("facturation_client", {"client": "Acme"}, "execute"), + ("supprimer_donnees", {"table": "clients"}, "execute"), + ("export_rapport", {"format": "pdf"}, "execute"), + ("mise_a_jour_production", {"env": "live"}, "update"), + ] + + print("=== Tests ConfirmationLoop ===\n") + for workflow, params, action in test_cases: + risk = loop.evaluate_risk(workflow, params, action) + conf = loop.create_confirmation_request(workflow, params, action) + + print(f"Workflow: {workflow}") + print(f" Risk: {risk.value}") + print(f" Requires confirmation: {loop.requires_confirmation(risk)}") + print(f" Message: {conf.confirmation_message[:100]}...") + print() diff --git a/agent_chat/conversation_manager.py b/agent_chat/conversation_manager.py new file mode 100644 index 000000000..77fd3194b --- /dev/null +++ b/agent_chat/conversation_manager.py @@ -0,0 +1,489 @@ +#!/usr/bin/env python3 +""" +RPA Vision V3 - ConversationManager +Gestionnaire de contexte multi-tour pour l'agent conversationnel. + +Ce module gère : +- L'historique des échanges +- Le contexte courant (workflow en cours, paramètres accumulés) +- La résolution des références anaphoriques ("le", "ça", "celui-ci") +- Les sessions utilisateur + +Auteur: Dom - Janvier 2026 +""" + +import json +import logging +from dataclasses import dataclass, field +from datetime import datetime, timedelta +from typing import Dict, Any, List, Optional, Tuple +from pathlib import Path +import hashlib + +from .intent_parser import ParsedIntent, IntentType +from .confirmation import PendingConfirmation + +logger = logging.getLogger(__name__) + + +@dataclass +class ConversationTurn: + """Un tour de conversation.""" + turn_id: int + timestamp: datetime + user_message: str + intent: ParsedIntent + response: str + action_taken: Optional[str] = None + result: Optional[Dict[str, Any]] = None + + def to_dict(self) -> Dict[str, Any]: + return { + "turn_id": self.turn_id, + "timestamp": self.timestamp.isoformat(), + "user_message": self.user_message, + "intent": self.intent.to_dict(), + "response": self.response, + "action_taken": self.action_taken, + "result": self.result + } + + +@dataclass +class ConversationContext: + """Contexte courant de la conversation.""" + # Workflow en cours de discussion + current_workflow: Optional[str] = None + current_workflow_params: Dict[str, Any] = field(default_factory=dict) + + # Confirmation en attente + pending_confirmation: Optional[PendingConfirmation] = None + + # Entités mentionnées (pour références anaphoriques) + mentioned_entities: Dict[str, Any] = field(default_factory=dict) + + # Dernier résultat d'exécution + last_execution_result: Optional[Dict[str, Any]] = None + + # État de la conversation + awaiting_clarification: bool = False + clarification_for: Optional[str] = None + + def to_dict(self) -> Dict[str, Any]: + return { + "current_workflow": self.current_workflow, + "current_workflow_params": self.current_workflow_params, + "pending_confirmation": self.pending_confirmation.to_dict() if self.pending_confirmation else None, + "mentioned_entities": self.mentioned_entities, + "last_execution_result": self.last_execution_result, + "awaiting_clarification": self.awaiting_clarification, + "clarification_for": self.clarification_for + } + + +@dataclass +class ConversationSession: + """Session de conversation utilisateur.""" + session_id: str + user_id: Optional[str] + created_at: datetime + last_activity: datetime + turns: List[ConversationTurn] = field(default_factory=list) + context: ConversationContext = field(default_factory=ConversationContext) + + def to_dict(self) -> Dict[str, Any]: + return { + "session_id": self.session_id, + "user_id": self.user_id, + "created_at": self.created_at.isoformat(), + "last_activity": self.last_activity.isoformat(), + "turns_count": len(self.turns), + "context": self.context.to_dict() + } + + +class ConversationManager: + """ + Gestionnaire de conversations multi-tour. + + Maintient le contexte à travers plusieurs échanges : + - Historique des tours + - Entités mentionnées + - Workflow et paramètres en cours + - Résolution des références + """ + + # Mots de référence anaphorique + ANAPHORIC_REFERENCES = { + "le": "workflow", + "la": "workflow", + "celui-ci": "workflow", + "celle-ci": "workflow", + "ça": "last_action", + "cela": "last_action", + "lui": "entity", + "elle": "entity", + "ce client": "client", + "cette facture": "invoice", + "le même": "last_value", + "pareil": "last_value", + "idem": "last_value" + } + + def __init__( + self, + session_timeout_minutes: int = 30, + max_history_turns: int = 50, + persistence_path: Optional[Path] = None + ): + """ + Initialiser le gestionnaire de conversations. + + Args: + session_timeout_minutes: Timeout d'inactivité + max_history_turns: Nombre max de tours à garder + persistence_path: Chemin pour persister les sessions + """ + self.session_timeout = timedelta(minutes=session_timeout_minutes) + self.max_history_turns = max_history_turns + self.persistence_path = persistence_path + self.sessions: Dict[str, ConversationSession] = {} + + if persistence_path: + self._load_sessions() + + def get_or_create_session( + self, + session_id: Optional[str] = None, + user_id: Optional[str] = None + ) -> ConversationSession: + """ + Obtenir ou créer une session de conversation. + + Args: + session_id: ID de session (généré si non fourni) + user_id: ID utilisateur (optionnel) + + Returns: + ConversationSession active + """ + # Générer un ID si non fourni + if not session_id: + session_id = self._generate_session_id(user_id) + + # Vérifier si la session existe et n'est pas expirée + if session_id in self.sessions: + session = self.sessions[session_id] + if datetime.now() - session.last_activity < self.session_timeout: + session.last_activity = datetime.now() + return session + else: + # Session expirée, créer une nouvelle + logger.info(f"Session expired: {session_id}") + del self.sessions[session_id] + + # Créer une nouvelle session + session = ConversationSession( + session_id=session_id, + user_id=user_id, + created_at=datetime.now(), + last_activity=datetime.now() + ) + self.sessions[session_id] = session + logger.info(f"New session created: {session_id}") + + return session + + def add_turn( + self, + session: ConversationSession, + user_message: str, + intent: ParsedIntent, + response: str, + action_taken: Optional[str] = None, + result: Optional[Dict[str, Any]] = None + ) -> ConversationTurn: + """ + Ajouter un tour à la conversation. + + Args: + session: Session de conversation + user_message: Message utilisateur + intent: Intention parsée + response: Réponse générée + action_taken: Action effectuée + result: Résultat de l'action + + Returns: + ConversationTurn créé + """ + turn = ConversationTurn( + turn_id=len(session.turns) + 1, + timestamp=datetime.now(), + user_message=user_message, + intent=intent, + response=response, + action_taken=action_taken, + result=result + ) + + session.turns.append(turn) + session.last_activity = datetime.now() + + # Mettre à jour le contexte + self._update_context(session, intent, result) + + # Limiter l'historique + if len(session.turns) > self.max_history_turns: + session.turns = session.turns[-self.max_history_turns:] + + # Persister si configuré + if self.persistence_path: + self._save_session(session) + + return turn + + def resolve_references( + self, + session: ConversationSession, + intent: ParsedIntent + ) -> ParsedIntent: + """ + Résoudre les références anaphoriques dans l'intention. + + Args: + session: Session de conversation + intent: Intention à résoudre + + Returns: + ParsedIntent avec références résolues + """ + resolved_intent = intent + context = session.context + + # Vérifier si le message contient des références + message_lower = intent.raw_query.lower() + + for reference, ref_type in self.ANAPHORIC_REFERENCES.items(): + if reference in message_lower: + if ref_type == "workflow" and context.current_workflow: + # Référence au workflow courant + if not resolved_intent.workflow_hint: + resolved_intent.workflow_hint = context.current_workflow + logger.debug(f"Resolved '{reference}' to workflow: {context.current_workflow}") + + elif ref_type == "last_action" and context.last_execution_result: + # Référence à la dernière action + resolved_intent.parameters.update( + context.last_execution_result.get("params", {}) + ) + + elif ref_type == "entity" and context.mentioned_entities: + # Référence à une entité mentionnée + for entity_type, entity_value in context.mentioned_entities.items(): + if entity_type not in resolved_intent.parameters: + resolved_intent.parameters[entity_type] = entity_value + + elif ref_type in context.mentioned_entities: + # Référence à un type d'entité spécifique + if ref_type not in resolved_intent.parameters: + resolved_intent.parameters[ref_type] = context.mentioned_entities[ref_type] + + elif ref_type == "last_value" and context.current_workflow_params: + # Répéter les derniers paramètres + for key, value in context.current_workflow_params.items(): + if key not in resolved_intent.parameters: + resolved_intent.parameters[key] = value + + return resolved_intent + + def _update_context( + self, + session: ConversationSession, + intent: ParsedIntent, + result: Optional[Dict[str, Any]] + ): + """Mettre à jour le contexte après un tour.""" + context = session.context + + # Mettre à jour le workflow courant + if intent.intent_type == IntentType.EXECUTE: + if intent.workflow_hint: + context.current_workflow = intent.workflow_hint + if intent.parameters: + context.current_workflow_params.update(intent.parameters) + + # Mettre à jour les entités mentionnées + for entity in intent.entities: + entity_type = entity["type"] + entity_value = entity["value"] + context.mentioned_entities[entity_type] = entity_value + + # Mettre à jour le dernier résultat + if result and result.get("success"): + context.last_execution_result = result + + # Gérer les clarifications + if intent.clarification_needed: + context.awaiting_clarification = True + context.clarification_for = intent.workflow_hint + elif intent.intent_type in [IntentType.CONFIRM, IntentType.DENY]: + context.awaiting_clarification = False + context.clarification_for = None + + def get_context_summary(self, session: ConversationSession) -> Dict[str, Any]: + """ + Obtenir un résumé du contexte pour les prompts LLM. + + Args: + session: Session de conversation + + Returns: + Dict résumant le contexte + """ + context = session.context + + # Résumé des derniers tours + recent_turns = [] + for turn in session.turns[-5:]: + recent_turns.append({ + "user": turn.user_message, + "intent": turn.intent.intent_type.value, + "action": turn.action_taken + }) + + return { + "current_workflow": context.current_workflow, + "current_params": context.current_workflow_params, + "mentioned_entities": context.mentioned_entities, + "awaiting_clarification": context.awaiting_clarification, + "recent_turns": recent_turns + } + + def set_pending_confirmation( + self, + session: ConversationSession, + confirmation: PendingConfirmation + ): + """Définir une confirmation en attente.""" + session.context.pending_confirmation = confirmation + + def get_pending_confirmation( + self, + session: ConversationSession + ) -> Optional[PendingConfirmation]: + """Obtenir la confirmation en attente.""" + return session.context.pending_confirmation + + def clear_pending_confirmation(self, session: ConversationSession): + """Effacer la confirmation en attente.""" + session.context.pending_confirmation = None + + def cleanup_expired_sessions(self) -> int: + """ + Nettoyer les sessions expirées. + + Returns: + Nombre de sessions nettoyées + """ + now = datetime.now() + expired = [ + sid for sid, session in self.sessions.items() + if now - session.last_activity > self.session_timeout + ] + + for sid in expired: + del self.sessions[sid] + logger.info(f"Session cleaned up: {sid}") + + return len(expired) + + def _generate_session_id(self, user_id: Optional[str]) -> str: + """Générer un ID de session unique.""" + base = f"{user_id or 'anon'}_{datetime.now().isoformat()}" + return hashlib.md5(base.encode()).hexdigest()[:16] + + def _load_sessions(self): + """Charger les sessions depuis le stockage.""" + if not self.persistence_path: + return + + sessions_file = self.persistence_path / "sessions.json" + if sessions_file.exists(): + try: + with open(sessions_file, 'r') as f: + data = json.load(f) + # TODO: Reconstruire les sessions depuis JSON + logger.info(f"Loaded {len(data)} sessions") + except Exception as e: + logger.warning(f"Failed to load sessions: {e}") + + def _save_session(self, session: ConversationSession): + """Sauvegarder une session.""" + if not self.persistence_path: + return + + try: + self.persistence_path.mkdir(parents=True, exist_ok=True) + session_file = self.persistence_path / f"session_{session.session_id}.json" + + with open(session_file, 'w') as f: + json.dump(session.to_dict(), f, indent=2, default=str) + except Exception as e: + logger.warning(f"Failed to save session: {e}") + + +# Singleton pour utilisation globale +_conversation_manager: Optional[ConversationManager] = None + +def get_conversation_manager( + persistence_path: Optional[Path] = None +) -> ConversationManager: + """Obtenir l'instance globale du gestionnaire de conversations.""" + global _conversation_manager + if _conversation_manager is None: + _conversation_manager = ConversationManager(persistence_path=persistence_path) + return _conversation_manager + + +if __name__ == "__main__": + # Tests rapides + from .intent_parser import IntentParser + + parser = IntentParser() + manager = ConversationManager() + + # Simuler une conversation + session = manager.get_or_create_session(user_id="test_user") + + test_conversation = [ + "facturer le client Acme", + "oui", + "et pour le client Beta ?", # Référence implicite + "le même format", # Référence anaphorique + "statut", + ] + + print("=== Tests ConversationManager ===\n") + for message in test_conversation: + intent = parser.parse(message) + + # Résoudre les références + resolved = manager.resolve_references(session, intent) + + print(f"User: {message}") + print(f"Intent: {resolved.intent_type.value}") + print(f"Workflow: {resolved.workflow_hint}") + print(f"Params: {resolved.parameters}") + + # Ajouter le tour + manager.add_turn( + session=session, + user_message=message, + intent=resolved, + response="[Réponse simulée]", + action_taken=resolved.intent_type.value + ) + + print(f"Context workflow: {session.context.current_workflow}") + print() diff --git a/agent_chat/intent_parser.py b/agent_chat/intent_parser.py new file mode 100644 index 000000000..fc1df1607 --- /dev/null +++ b/agent_chat/intent_parser.py @@ -0,0 +1,448 @@ +#!/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+)?(?:workflows?|processus|automatisations?)", + r"(?:qu'est-ce que|que)\s+(?:je peux|tu peux)\s+faire", + r"(?:workflows?|processus)\s+disponibles?", + ], + 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"): + """ + Initialiser le parseur d'intentions. + + Args: + use_llm: Utiliser un LLM pour l'analyse (optionnel) + llm_endpoint: URL de l'endpoint Ollama + """ + self.use_llm = use_llm + self.llm_endpoint = llm_endpoint + self.llm_available = False + + if use_llm: + self._check_llm_availability() + + 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 + + prompt = f"""Analyse cette requête utilisateur pour un système RPA. + +Requête: "{query}" + +Contexte: {json.dumps(context) if context else "Aucun"} + +Réponds en JSON avec: +- intent: execute|query|list|configure|help|status|cancel|history|confirm|deny|unknown +- confidence: 0.0 à 1.0 +- workflow_hint: le workflow demandé si applicable +- parameters: dict des paramètres extraits +- clarification_needed: true/false +- clarification_question: question à poser si besoin + +JSON:""" + + response = requests.post( + f"{self.llm_endpoint}/api/generate", + json={ + "model": "qwen2.5:7b", + "prompt": prompt, + "stream": False, + "options": {"temperature": 0.1} + }, + timeout=10 + ) + + if response.status_code == 200: + result = response.json().get("response", "") + # Extraire le JSON de la réponse + json_match = re.search(r'\{[^{}]+\}', result, re.DOTALL) + if json_match: + data = json.loads(json_match.group(0)) + + intent_str = data.get("intent", "unknown") + try: + intent_type = IntentType(intent_str) + except ValueError: + intent_type = IntentType.UNKNOWN + + return ParsedIntent( + intent_type=intent_type, + confidence=data.get("confidence", 0.5), + 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 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) -> IntentParser: + """Obtenir l'instance globale du parseur d'intentions.""" + global _intent_parser + if _intent_parser is None: + _intent_parser = IntentParser(use_llm=use_llm) + return _intent_parser + + +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() diff --git a/agent_chat/response_generator.py b/agent_chat/response_generator.py new file mode 100644 index 000000000..97a54fdea --- /dev/null +++ b/agent_chat/response_generator.py @@ -0,0 +1,633 @@ +#!/usr/bin/env python3 +""" +RPA Vision V3 - ResponseGenerator +Générateur de réponses en langage naturel pour l'agent conversationnel. + +Ce module génère des réponses adaptées au contexte : +- Confirmations et résumés d'exécution +- Réponses aux questions +- Messages d'erreur formatés +- Suggestions et aide + +Auteur: Dom - Janvier 2026 +""" + +import logging +import random +from dataclasses import dataclass +from enum import Enum +from typing import Dict, Any, List, Optional + +from .intent_parser import IntentType, ParsedIntent +from .confirmation import PendingConfirmation, ConfirmationStatus, RiskLevel + +logger = logging.getLogger(__name__) + + +class ResponseTone(Enum): + """Ton des réponses.""" + FORMAL = "formal" # Formel, professionnel + FRIENDLY = "friendly" # Amical, décontracté + CONCISE = "concise" # Bref, direct + DETAILED = "detailed" # Détaillé, explicatif + + +@dataclass +class GeneratedResponse: + """Une réponse générée.""" + message: str + suggestions: List[str] + action_required: bool = False + action_type: Optional[str] = None + metadata: Dict[str, Any] = None + + def to_dict(self) -> Dict[str, Any]: + return { + "message": self.message, + "suggestions": self.suggestions, + "action_required": self.action_required, + "action_type": self.action_type, + "metadata": self.metadata or {} + } + + +class ResponseGenerator: + """ + Générateur de réponses en langage naturel. + + Adapte les réponses selon le contexte, le ton souhaité + et le type d'interaction. + """ + + # Templates de réponses par type d'intention + RESPONSE_TEMPLATES = { + IntentType.EXECUTE: { + "success": [ + "J'ai lancé le workflow '{workflow}'. {details}", + "Le workflow '{workflow}' est en cours d'exécution. {details}", + "C'est parti pour '{workflow}' ! {details}" + ], + "error": [ + "Impossible d'exécuter '{workflow}': {error}", + "Erreur lors du lancement de '{workflow}': {error}", + "Le workflow '{workflow}' a échoué: {error}" + ], + "not_found": [ + "Je n'ai pas trouvé de workflow correspondant à '{query}'.", + "Aucun workflow ne correspond à '{query}'. Voulez-vous voir la liste ?", + "'{query}' ne correspond à aucun workflow connu." + ] + }, + IntentType.LIST: { + "success": [ + "Voici les workflows disponibles :\n{list}", + "J'ai trouvé {count} workflows :\n{list}", + ], + "empty": [ + "Aucun workflow n'est configuré pour le moment.", + "La liste des workflows est vide." + ] + }, + IntentType.QUERY: { + "found": [ + "Voici ce que j'ai trouvé sur '{topic}' :\n{answer}", + "À propos de '{topic}' :\n{answer}" + ], + "not_found": [ + "Je n'ai pas d'information sur '{topic}'.", + "Je ne peux pas répondre à cette question sur '{topic}'." + ] + }, + IntentType.HELP: { + "general": [ + "Je suis votre assistant RPA. Voici ce que je peux faire :\n\n" + "• Exécuter des workflows : \"lance facturation client Acme\"\n" + "• Lister les workflows : \"quels workflows sont disponibles ?\"\n" + "• Voir le statut : \"où en est l'exécution ?\"\n" + "• Annuler : \"annule\"\n\n" + "Tapez votre commande en langage naturel !", + ] + }, + IntentType.STATUS: { + "running": [ + "Exécution en cours : '{workflow}'\nProgression : {progress}%\n{message}", + "Le workflow '{workflow}' s'exécute ({progress}%): {message}" + ], + "idle": [ + "Aucune exécution en cours. Système prêt.", + "Tout est calme. Que puis-je faire pour vous ?" + ], + "completed": [ + "Dernière exécution : '{workflow}' - {status}", + "'{workflow}' est terminé : {status}" + ] + }, + IntentType.CANCEL: { + "success": [ + "Exécution annulée.", + "J'ai arrêté le workflow en cours.", + "Annulation effectuée." + ], + "nothing": [ + "Rien à annuler, aucune exécution en cours.", + "Il n'y a pas d'exécution active." + ] + }, + IntentType.HISTORY: { + "success": [ + "Voici vos dernières commandes :\n{history}", + "Historique récent :\n{history}" + ], + "empty": [ + "Pas encore d'historique.", + "Vous n'avez pas encore exécuté de commandes." + ] + }, + IntentType.CONFIRM: { + "accepted": [ + "Très bien, j'exécute '{workflow}'.", + "C'est parti pour '{workflow}' !", + "Confirmé. Lancement de '{workflow}'." + ], + "no_pending": [ + "Il n'y a rien à confirmer.", + "Aucune action en attente de confirmation." + ] + }, + IntentType.DENY: { + "cancelled": [ + "Action annulée.", + "D'accord, j'annule.", + "Compris, on oublie." + ] + }, + IntentType.CLARIFY: { + "question": [ + "{question}", + ] + }, + IntentType.UNKNOWN: { + "default": [ + "Je n'ai pas compris. Pouvez-vous reformuler ?", + "Désolé, je ne comprends pas '{query}'. Tapez 'aide' pour voir les commandes.", + "'{query}' ? Je ne suis pas sûr de comprendre." + ] + } + } + + # Suggestions par contexte + CONTEXTUAL_SUGGESTIONS = { + "after_execute": [ + "voir le statut", + "annuler", + "liste des workflows" + ], + "after_error": [ + "aide", + "liste des workflows", + "réessayer" + ], + "after_list": [ + "exécuter un workflow", + "aide" + ], + "idle": [ + "facturer client X", + "liste des workflows", + "aide" + ] + } + + def __init__(self, tone: ResponseTone = ResponseTone.FRIENDLY): + """ + Initialiser le générateur de réponses. + + Args: + tone: Ton des réponses + """ + self.tone = tone + + def generate( + self, + intent: ParsedIntent, + context: Dict[str, Any], + result: Optional[Dict[str, Any]] = None + ) -> GeneratedResponse: + """ + Générer une réponse adaptée. + + Args: + intent: L'intention parsée + context: Contexte de la conversation + result: Résultat de l'action (optionnel) + + Returns: + GeneratedResponse formatée + """ + result = result or {} + + # Sélectionner le handler approprié + handler = getattr(self, f"_handle_{intent.intent_type.value}", self._handle_unknown) + return handler(intent, context, result) + + def generate_confirmation_request( + self, + confirmation: PendingConfirmation + ) -> GeneratedResponse: + """ + Générer une demande de confirmation. + + Args: + confirmation: La confirmation en attente + + Returns: + GeneratedResponse avec la demande + """ + message = confirmation.confirmation_message + + # Ajouter des émojis selon le risque + if confirmation.risk_level == RiskLevel.CRITICAL: + message = f"🚨 {message}" + elif confirmation.risk_level == RiskLevel.HIGH: + message = f"⚠️ {message}" + + suggestions = ["oui", "non", "modifier les paramètres"] + + return GeneratedResponse( + message=message, + suggestions=suggestions, + action_required=True, + action_type="confirmation", + metadata={"confirmation_id": confirmation.id} + ) + + def generate_execution_progress( + self, + workflow_name: str, + progress: int, + step: str, + current: int, + total: int + ) -> GeneratedResponse: + """ + Générer un message de progression. + + Args: + workflow_name: Nom du workflow + progress: Pourcentage de progression + step: Étape actuelle + current: Numéro de l'étape + total: Nombre total d'étapes + + Returns: + GeneratedResponse avec la progression + """ + # Barre de progression visuelle + bar_length = 20 + filled = int(bar_length * progress / 100) + bar = "█" * filled + "░" * (bar_length - filled) + + message = f"**{workflow_name}** [{bar}] {progress}%\n\nÉtape {current}/{total}: {step}" + + return GeneratedResponse( + message=message, + suggestions=["annuler"] if progress < 100 else [], + action_required=False, + metadata={ + "workflow": workflow_name, + "progress": progress, + "step": step + } + ) + + def generate_execution_result( + self, + workflow_name: str, + success: bool, + message: str, + duration: Optional[float] = None + ) -> GeneratedResponse: + """ + Générer un message de résultat d'exécution. + + Args: + workflow_name: Nom du workflow + success: Succès ou échec + message: Message détaillé + duration: Durée d'exécution en secondes + + Returns: + GeneratedResponse avec le résultat + """ + if success: + emoji = "✅" + status = "terminé avec succès" + suggestions = self.CONTEXTUAL_SUGGESTIONS["idle"] + else: + emoji = "❌" + status = "échoué" + suggestions = self.CONTEXTUAL_SUGGESTIONS["after_error"] + + response_message = f"{emoji} **{workflow_name}** {status}\n\n{message}" + + if duration: + response_message += f"\n\nDurée: {duration:.1f}s" + + return GeneratedResponse( + message=response_message, + suggestions=suggestions, + action_required=False, + metadata={ + "workflow": workflow_name, + "success": success, + "duration": duration + } + ) + + # --- Handlers par type d'intention --- + + def _handle_execute( + self, + intent: ParsedIntent, + context: Dict[str, Any], + result: Dict[str, Any] + ) -> GeneratedResponse: + """Handler pour les intentions d'exécution.""" + templates = self.RESPONSE_TEMPLATES[IntentType.EXECUTE] + + if result.get("success"): + template = random.choice(templates["success"]) + workflow = result.get("workflow", intent.workflow_hint or "inconnu") + details = "" + + if result.get("params"): + params_str = ", ".join([f"{k}={v}" for k, v in result["params"].items()]) + details = f"Paramètres: {params_str}" + + message = template.format(workflow=workflow, details=details) + suggestions = self.CONTEXTUAL_SUGGESTIONS["after_execute"] + + elif result.get("not_found"): + template = random.choice(templates["not_found"]) + message = template.format(query=intent.raw_query) + suggestions = self.CONTEXTUAL_SUGGESTIONS["after_error"] + + else: + template = random.choice(templates["error"]) + workflow = intent.workflow_hint or intent.raw_query + error = result.get("error", "Erreur inconnue") + message = template.format(workflow=workflow, error=error) + suggestions = self.CONTEXTUAL_SUGGESTIONS["after_error"] + + return GeneratedResponse( + message=message, + suggestions=suggestions, + action_required=False + ) + + def _handle_list( + self, + intent: ParsedIntent, + context: Dict[str, Any], + result: Dict[str, Any] + ) -> GeneratedResponse: + """Handler pour les listes de workflows.""" + templates = self.RESPONSE_TEMPLATES[IntentType.LIST] + workflows = result.get("workflows", []) + + if workflows: + template = random.choice(templates["success"]) + workflow_list = "\n".join([f"• **{wf['name']}**: {wf.get('description', '')}" for wf in workflows[:10]]) + message = template.format(list=workflow_list, count=len(workflows)) + suggestions = self.CONTEXTUAL_SUGGESTIONS["after_list"] + else: + message = random.choice(templates["empty"]) + suggestions = ["aide"] + + return GeneratedResponse( + message=message, + suggestions=suggestions, + action_required=False + ) + + def _handle_help( + self, + intent: ParsedIntent, + context: Dict[str, Any], + result: Dict[str, Any] + ) -> GeneratedResponse: + """Handler pour l'aide.""" + templates = self.RESPONSE_TEMPLATES[IntentType.HELP] + message = random.choice(templates["general"]) + + return GeneratedResponse( + message=message, + suggestions=self.CONTEXTUAL_SUGGESTIONS["idle"], + action_required=False + ) + + def _handle_status( + self, + intent: ParsedIntent, + context: Dict[str, Any], + result: Dict[str, Any] + ) -> GeneratedResponse: + """Handler pour le statut.""" + templates = self.RESPONSE_TEMPLATES[IntentType.STATUS] + execution = result.get("execution", {}) + + if execution.get("running"): + template = random.choice(templates["running"]) + message = template.format( + workflow=execution.get("workflow", "Inconnu"), + progress=execution.get("progress", 0), + message=execution.get("message", "") + ) + suggestions = ["annuler"] + else: + message = random.choice(templates["idle"]) + suggestions = self.CONTEXTUAL_SUGGESTIONS["idle"] + + return GeneratedResponse( + message=message, + suggestions=suggestions, + action_required=False + ) + + def _handle_cancel( + self, + intent: ParsedIntent, + context: Dict[str, Any], + result: Dict[str, Any] + ) -> GeneratedResponse: + """Handler pour l'annulation.""" + templates = self.RESPONSE_TEMPLATES[IntentType.CANCEL] + + if result.get("cancelled"): + message = random.choice(templates["success"]) + else: + message = random.choice(templates["nothing"]) + + return GeneratedResponse( + message=message, + suggestions=self.CONTEXTUAL_SUGGESTIONS["idle"], + action_required=False + ) + + def _handle_history( + self, + intent: ParsedIntent, + context: Dict[str, Any], + result: Dict[str, Any] + ) -> GeneratedResponse: + """Handler pour l'historique.""" + templates = self.RESPONSE_TEMPLATES[IntentType.HISTORY] + history = result.get("history", []) + + if history: + template = random.choice(templates["success"]) + history_list = "\n".join([ + f"• {h['timestamp'][:19]} - {h['command']} → {h['status']}" + for h in history[-5:] + ]) + message = template.format(history=history_list) + else: + message = random.choice(templates["empty"]) + + return GeneratedResponse( + message=message, + suggestions=self.CONTEXTUAL_SUGGESTIONS["idle"], + action_required=False + ) + + def _handle_confirm( + self, + intent: ParsedIntent, + context: Dict[str, Any], + result: Dict[str, Any] + ) -> GeneratedResponse: + """Handler pour la confirmation.""" + templates = self.RESPONSE_TEMPLATES[IntentType.CONFIRM] + + if result.get("confirmed"): + template = random.choice(templates["accepted"]) + message = template.format(workflow=result.get("workflow", "")) + suggestions = self.CONTEXTUAL_SUGGESTIONS["after_execute"] + else: + message = random.choice(templates["no_pending"]) + suggestions = self.CONTEXTUAL_SUGGESTIONS["idle"] + + return GeneratedResponse( + message=message, + suggestions=suggestions, + action_required=False + ) + + def _handle_deny( + self, + intent: ParsedIntent, + context: Dict[str, Any], + result: Dict[str, Any] + ) -> GeneratedResponse: + """Handler pour le refus.""" + templates = self.RESPONSE_TEMPLATES[IntentType.DENY] + message = random.choice(templates["cancelled"]) + + return GeneratedResponse( + message=message, + suggestions=self.CONTEXTUAL_SUGGESTIONS["idle"], + action_required=False + ) + + def _handle_clarify( + self, + intent: ParsedIntent, + context: Dict[str, Any], + result: Dict[str, Any] + ) -> GeneratedResponse: + """Handler pour les demandes de clarification.""" + question = intent.clarification_question or "Pouvez-vous préciser ?" + + return GeneratedResponse( + message=question, + suggestions=[], + action_required=True, + action_type="clarification" + ) + + def _handle_query( + self, + intent: ParsedIntent, + context: Dict[str, Any], + result: Dict[str, Any] + ) -> GeneratedResponse: + """Handler pour les questions.""" + templates = self.RESPONSE_TEMPLATES[IntentType.QUERY] + topic = intent.workflow_hint or intent.raw_query + + if result.get("answer"): + template = random.choice(templates["found"]) + message = template.format(topic=topic, answer=result["answer"]) + else: + template = random.choice(templates["not_found"]) + message = template.format(topic=topic) + + return GeneratedResponse( + message=message, + suggestions=self.CONTEXTUAL_SUGGESTIONS["idle"], + action_required=False + ) + + def _handle_unknown( + self, + intent: ParsedIntent, + context: Dict[str, Any], + result: Dict[str, Any] + ) -> GeneratedResponse: + """Handler pour les intentions non reconnues.""" + templates = self.RESPONSE_TEMPLATES[IntentType.UNKNOWN] + template = random.choice(templates["default"]) + message = template.format(query=intent.raw_query) + + return GeneratedResponse( + message=message, + suggestions=["aide", "liste des workflows"], + action_required=False + ) + + +# Singleton pour utilisation globale +_response_generator: Optional[ResponseGenerator] = None + +def get_response_generator(tone: ResponseTone = ResponseTone.FRIENDLY) -> ResponseGenerator: + """Obtenir l'instance globale du générateur de réponses.""" + global _response_generator + if _response_generator is None: + _response_generator = ResponseGenerator(tone=tone) + return _response_generator + + +if __name__ == "__main__": + # Tests rapides + from .intent_parser import IntentParser + + parser = IntentParser() + generator = ResponseGenerator() + + test_cases = [ + ("facturer client Acme", {"success": True, "workflow": "facturation", "params": {"client": "Acme"}}), + ("liste des workflows", {"workflows": [{"name": "facturation", "description": "Facturer un client"}]}), + ("aide", {}), + ("statut", {"execution": {"running": True, "workflow": "test", "progress": 50, "message": "En cours"}}), + ("blablabla inconnu", {}), + ] + + print("=== Tests ResponseGenerator ===\n") + for query, result in test_cases: + intent = parser.parse(query) + response = generator.generate(intent, {}, result) + + print(f"Query: {query}") + print(f"Response: {response.message[:100]}...") + print(f"Suggestions: {response.suggestions}") + print()