bc096a3891
Nouveaux composants pour l'agent conversationnel : - IntentParser: Analyse des intentions utilisateur (règles + LLM optionnel) - ConfirmationLoop: Validation avant actions critiques (niveaux de risque) - ResponseGenerator: Génération de réponses en langage naturel - ConversationManager: Gestion du contexte multi-tour Endpoint /api/chat ajouté pour le flux conversationnel complet. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
634 lines
20 KiB
Python
634 lines
20 KiB
Python
#!/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()
|