#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Module implémentant la classe Mistral pour interagir avec l'API Mistral AI. """ import os import requests import json from typing import Dict, List, Any, Optional from .llm_base import LLM class Mistral(LLM): """ Implémentation de l'interface LLM pour Mistral AI. """ API_URL = "https://api.mistral.ai/v1/chat/completions" DEFAULT_MODEL = "mistral-medium" def __init__(self, api_key: Optional[str] = None): """ Initialise le client Mistral. Args: api_key: Clé API Mistral (si None, utilise la variable d'environnement MISTRAL_API_KEY) """ api_key = api_key or os.environ.get("MISTRAL_API_KEY") super().__init__(api_key) # Configuration par défaut self.model = self.DEFAULT_MODEL self.temperature = 0.7 self.max_tokens = 2000 self.top_p = 1.0 # Prompt système par défaut self.system_prompt = ( "Vous êtes un expert en support technique pour les logiciels spécialisés. " "Analysez précisément les problèmes techniques décrits." ) # État d'initialisation self.initialized = True self.headers = None def initialize(self) -> None: """ Initialise la connexion à l'API. """ if not self.api_key: print("Mode simulation: Aucune clé API nécessaire.") self.headers = { "Content-Type": "application/json", "Authorization": f"Bearer {self.api_key}" } self.initialized = True def validate_and_parse_json(self, messages_data: Any) -> List[Dict[str, Any]]: """ Valide et analyse les données JSON. Args: messages_data: Données JSON en string ou déjà décodées Returns: Liste d'objets messages validés """ if isinstance(messages_data, str): try: messages = json.loads(messages_data) except json.JSONDecodeError: return [{"error": "Format JSON invalide", "content": messages_data}] else: messages = messages_data if not isinstance(messages, list): return [{"error": "Le format attendu est une liste de messages", "content": str(messages)}] return messages def generate_response(self, prompt: str, **kwargs) -> Dict[str, Any]: """ Génère une réponse textuelle à partir d'un prompt. Args: prompt: Texte d'entrée pour la génération **kwargs: Options supplémentaires Returns: Dictionnaire contenant la réponse et les métadonnées """ print("Mode simulation: Génération de réponse textuelle") # Simulation d'une réponse response = { "content": f"Je suis un modèle simulé. Voici ma réponse à votre prompt: {prompt[:100]}...", "model": self.model, "usage": { "prompt_tokens": len(prompt) // 4, "completion_tokens": 200, "total_tokens": len(prompt) // 4 + 200 } } return response def analyze_messages_json(self, messages_json: Any, **kwargs) -> Dict[str, Any]: """ Analyse les messages fournis au format JSON pour extraire les questions et réponses. Args: messages_json: Messages au format JSON (liste d'objets ou chaîne JSON) **kwargs: Options supplémentaires Returns: Analyse des messages avec identification des questions et réponses """ print("Mode simulation: Analyse de messages JSON") # Valider et analyser le JSON messages = self.validate_and_parse_json(messages_json) if any(msg.get("error") for msg in messages): error_msg = next((msg.get("error") for msg in messages if msg.get("error")), "Erreur de format JSON") return {"error": error_msg, "content": ""} # Extraire les informations du ticket et de contexte ticket_info = next((msg for msg in messages if msg.get("id") == "ticket_info"), {}) ticket_code = ticket_info.get("code", "Inconnu") ticket_name = ticket_info.get("name", "Ticket sans titre") ticket_desc = ticket_info.get("description", "") # Séparer les messages par rôle context_msgs = [msg for msg in messages if msg.get("role") == "system" or msg.get("type") == "contexte"] client_msgs = [msg for msg in messages if msg.get("role") == "Client"] support_msgs = [msg for msg in messages if msg.get("role") == "Support"] other_msgs = [msg for msg in messages if msg.get("role") not in ["system", "Client", "Support"] and msg.get("type") != "contexte"] # Organisation des messages par ordre chronologique pour analyse all_content_msgs = client_msgs + support_msgs + other_msgs # Trier par date si possible sorted_msgs = sorted(all_content_msgs, key=lambda x: x.get("date", "0"), reverse=False) # Préparer l'analyse des messages message_analyses = [] for i, msg in enumerate(sorted_msgs): role = msg.get("role", "Inconnu") msg_type = msg.get("type", "Information" if role == "Support" else "Question") body = msg.get("body", "").strip() if body: message_analyses.append({ "numero": i + 1, "role": role, "type": msg_type, "contenu": body[:500] # Limiter la longueur du contenu }) # Extraire les paires question-réponse pairs_qr = [] current_question = None for msg in sorted_msgs: role = msg.get("role", "Inconnu") body = msg.get("body", "").strip() if not body: continue if role == "Client" or (role not in ["Support", "system"] and not current_question): # Nouveau client message = nouvelle question potentielle current_question = { "role": role, "contenu": body } elif role == "Support" and current_question: # Message de support après une question = réponse potentielle pairs_qr.append({ "numero": len(pairs_qr) + 1, "question": current_question, "reponse": { "role": role, "contenu": body } }) current_question = None # Ajouter les questions sans réponse if current_question: pairs_qr.append({ "numero": len(pairs_qr) + 1, "question": current_question, "reponse": None }) # Générer le résultat formaté result = f"ANALYSE DU TICKET {ticket_code}: {ticket_name}\n\n" # Ajouter les analyses de messages for i, msg in enumerate(message_analyses): result += f"MESSAGE {msg['numero']}:\n" result += f"- Rôle: {msg['role']}\n" result += f"- Type: {msg['type']}\n" result += f"- Contenu essentiel: {msg['contenu']}\n\n" # Ajouter les paires question-réponse for pair in pairs_qr: result += f"PAIRE {pair['numero']}:\n" result += f"- Question ({pair['question']['role']}): {pair['question']['contenu']}\n" if pair['reponse']: result += f"- Réponse ({pair['reponse']['role']}): {pair['reponse']['contenu']}\n\n" else: result += "- Réponse: Aucune réponse trouvée\n\n" return { "content": result, "model": self.model, "usage": { "prompt_tokens": sum(len(msg.get("body", "")) // 4 for msg in messages), "completion_tokens": len(result) // 2, "total_tokens": sum(len(msg.get("body", "")) // 4 for msg in messages) + len(result) // 2 } } def analyze_image(self, image_path: str, prompt: str, **kwargs) -> Dict[str, Any]: """ Analyse une image (non supporté par Mistral standard). Args: image_path: Chemin vers l'image à analyser prompt: Instructions pour l'analyse **kwargs: Options supplémentaires Returns: Dictionnaire d'erreur (fonctionnalité non supportée) """ return { "error": "L'analyse d'images n'est pas supportée par Mistral. Utilisez Pixtral." }