Dom/rpa_vision_v3
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(agent_chat): Ajout des composants conversationnels

Browse Source 
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>
This commit is contained in:
Dom
2026-01-15 15:20:05 +01:00
parent c6a857b96b
commit bc096a3891
6 changed files with 2271 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
agent_chat/__init__.py
View File

@@ -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",
]

227
agent_chat/app.py
View File

@@ -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,6 +76,7 @@ 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:
@@ -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/<action>', methods=['POST'])
def api_gpu_action(action):
"""Contrôler le GPU Resource Manager."""

408
agent_chat/confirmation.py Normal file
View File

@@ -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()

489
agent_chat/conversation_manager.py Normal file
View File

@@ -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()

448
agent_chat/intent_parser.py Normal file
View File

@@ -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()

633
agent_chat/response_generator.py Normal file
View File

@@ -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()