#!/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 ne sais pas encore faire '{query}'. Montre-moi comment faire et je l'apprendrai !", "'{query}' m'est inconnu pour l'instant. Tu peux me montrer en enregistrant un workflow.", "Je ne connais pas '{query}'. Montre-moi et je m'en souviendrai !" ], "gesture": [ "{gesture_name} ({gesture_keys}) envoyé !", "Raccourci {gesture_name} ({gesture_keys}) exécuté.", ], "copilot": [ "Mode pas-à-pas activé pour '{workflow}'. Validez chaque étape.", ] }, 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.GREETING: { "default": [ "Bonjour ! Je suis votre assistant RPA. Comment puis-je vous aider ?", "Salut ! Que puis-je faire pour vous ?", "Bonjour ! Tapez une commande ou 'aide' pour voir ce que je peux faire.", ] }, 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.DATA_IMPORT: { "preview": [ "J'ai trouvé le fichier **{filename}** — {total_rows} lignes, colonnes : {columns}. Je l'importe dans la table '{table_name}' ?", "Fichier **{filename}** prêt : {total_rows} lignes avec les colonnes {columns}. On crée la table '{table_name}' ?", ], "imported": [ "Table **'{table_name}'** créée avec {row_count} lignes et {col_count} colonnes ({columns}). Vous pouvez maintenant utiliser 'Pour chaque ligne' dans un workflow !", "Import réussi ! Table **'{table_name}'** : {row_count} lignes, {col_count} colonnes ({columns}).", ], "list_tables": [ "Voici les tables disponibles :\n{tables_list}", "Tables dans la base :\n{tables_list}", ], "no_tables": [ "Aucune table n'a été importée pour l'instant. Envoyez-moi un fichier Excel pour commencer !", "La base est vide. Importez un fichier Excel pour créer votre première table.", ], "table_info": [ "La table **'{table_name}'** contient {row_count} lignes et {col_count} colonnes :\n{columns_detail}", ], "folder_list": [ "J'ai trouvé {count} fichiers Excel dans le dossier :\n{files_list}\n\nDites-moi lequel importer !", ], "folder_empty": [ "Aucun fichier Excel trouvé dans le dossier '{folder}'. Vérifiez le chemin.", ], "file_not_found": [ "Je n'ai pas trouvé le fichier '{file_path}'. Vérifiez le chemin ou envoyez-le via le bouton 📎.", "Fichier introuvable : '{file_path}'. Vous pouvez aussi glisser un fichier dans le chat.", ], "error": [ "Erreur lors de l'import : {error}", "L'import a échoué : {error}", ], "uploaded": [ "Fichier **{filename}** reçu ! Je l'analyse...", ], }, 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" ], "after_import": [ "montre les tables", "importer un autre fichier", "aide" ], "after_table_list": [ "importer un fichier Excel", "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("gesture"): # Geste primitif (raccourci clavier) template = random.choice(templates["gesture"]) message = template.format( gesture_name=result.get("gesture_name", "?"), gesture_keys=result.get("gesture_keys", "?"), ) suggestions = self.CONTEXTUAL_SUGGESTIONS["after_execute"] elif result.get("mode") == "copilot": template = random.choice(templates["copilot"]) message = template.format(workflow=result.get("workflow", "?")) suggestions = ["approuver", "passer", "annuler"] elif 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"]) query = result.get("query", intent.raw_query) message = template.format(query=query) suggestions = ["lister les workflows", "aide", "enregistrer un workflow"] 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_greeting( self, intent: ParsedIntent, context: Dict[str, Any], result: Dict[str, Any] ) -> GeneratedResponse: """Handler pour les salutations.""" templates = self.RESPONSE_TEMPLATES[IntentType.GREETING] message = random.choice(templates["default"]) 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_data_import( self, intent: ParsedIntent, context: Dict[str, Any], result: Dict[str, Any] ) -> GeneratedResponse: """Handler pour les imports de données (Excel/CSV).""" templates = self.RESPONSE_TEMPLATES[IntentType.DATA_IMPORT] if result.get("file_not_found"): template = random.choice(templates["file_not_found"]) message = template.format(file_path=result.get("file_path", "?")) suggestions = ["aide"] elif result.get("preview"): # Aperçu avant import template = random.choice(templates["preview"]) preview = result["preview"] cols_str = ", ".join(preview.get("columns", [])[:8]) if len(preview.get("columns", [])) > 8: cols_str += f"... (+{len(preview['columns']) - 8})" message = template.format( filename=result.get("filename", "?"), total_rows=preview.get("total_rows", 0), columns=cols_str, table_name=result.get("table_name", "?"), ) suggestions = ["oui", "non"] elif result.get("imported"): # Import réussi template = random.choice(templates["imported"]) imp = result["imported"] cols_str = ", ".join(list(imp.get("columns", {}).keys())[:6]) if len(imp.get("columns", {})) > 6: cols_str += "..." message = template.format( table_name=imp.get("table_name", "?"), row_count=imp.get("row_count", 0), col_count=imp.get("column_count", 0), columns=cols_str, ) suggestions = self.CONTEXTUAL_SUGGESTIONS["after_import"] elif result.get("tables_list") is not None: tables = result["tables_list"] if tables: lines = [] for t in tables: lines.append(f" **{t['name']}** ({t['row_count']} lignes)") template = random.choice(templates["list_tables"]) message = template.format(tables_list="\n".join(lines)) suggestions = self.CONTEXTUAL_SUGGESTIONS["after_table_list"] else: message = random.choice(templates["no_tables"]) suggestions = ["importer un fichier Excel"] elif result.get("table_info"): info = result["table_info"] cols_detail = "\n".join( f" {c['name']} ({c['type']})" for c in info.get("columns", []) if c["name"] not in ("_rowid", "_imported_at") ) template = random.choice(templates["table_info"]) message = template.format( table_name=info.get("table_name", "?"), row_count=info.get("row_count", 0), col_count=len([c for c in info.get("columns", []) if c["name"] not in ("_rowid", "_imported_at")]), columns_detail=cols_detail, ) suggestions = self.CONTEXTUAL_SUGGESTIONS["after_table_list"] elif result.get("folder_files") is not None: files = result["folder_files"] if files: files_list = "\n".join(f" {f}" for f in files) template = random.choice(templates["folder_list"]) message = template.format(count=len(files), files_list=files_list) else: template = random.choice(templates["folder_empty"]) message = template.format(folder=result.get("folder", "?")) suggestions = ["aide"] elif result.get("uploaded"): template = random.choice(templates["uploaded"]) message = template.format(filename=result.get("filename", "?")) suggestions = [] elif result.get("error"): template = random.choice(templates["error"]) message = template.format(error=result["error"]) suggestions = self.CONTEXTUAL_SUGGESTIONS["after_error"] else: message = "Je n'ai pas compris la demande d'import. Précisez le fichier ou dites 'montre les tables'." suggestions = ["montre les tables", "aide"] return GeneratedResponse( message=message, suggestions=suggestions, action_required=result.get("needs_confirmation", False), action_type="data_import_confirm" if result.get("needs_confirmation") else None, metadata=result, ) 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()